🐛 Snowflake destination: snowflake s3 destination COPY is writing records from different table in the same raw table fix

airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/424892c4-daac-4491-b35d-c6688ba547ba.json

@@ -2,6 +2,6 @@
   "destinationDefinitionId": "424892c4-daac-4491-b35d-c6688ba547ba",
   "name": "Snowflake",
   "dockerRepository": "airbyte/destination-snowflake",
-  "dockerImageTag": "0.3.13",
+  "dockerImageTag": "0.3.14",
   "documentationUrl": "https://docs.airbyte.io/integrations/destinations/snowflake"
 }

airbyte-config/init/src/main/resources/seed/destination_definitions.yaml

@@ -42,7 +42,7 @@
 - destinationDefinitionId: 424892c4-daac-4491-b35d-c6688ba547ba
   name: Snowflake
   dockerRepository: airbyte/destination-snowflake
-  dockerImageTag: 0.3.13
+  dockerImageTag: 0.3.14
   documentationUrl: https://docs.airbyte.io/integrations/destinations/snowflake
 - destinationDefinitionId: 4816b78f-1489-44c1-9060-4b19d5fa9362
   name: S3

airbyte-db/lib/src/main/java/io/airbyte/db/mongodb/MongoUtils.java

@@ -110,8 +110,8 @@ private static void readBson(final Document document, final ObjectNode o, final
   }
 
   /**
-   * Gets 10.000 documents from collection, gathers all unique fields and its type. In case when one field has different types in 2 and more
-   * documents, the type is set to String.
+   * Gets 10.000 documents from collection, gathers all unique fields and its type. In case when one
+   * field has different types in 2 and more documents, the type is set to String.
    *
    * @param collection mongo collection
    * @return map of unique fields and its type

airbyte-integrations/connectors/destination-jdbc/Dockerfile

@@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
 
 RUN tar xf ${APPLICATION}.tar --strip-components=1
 
-LABEL io.airbyte.version=0.3.3
+LABEL io.airbyte.version=0.3.4
 LABEL io.airbyte.name=airbyte/destination-jdbc

airbyte-integrations/connectors/destination-jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/copy/s3/S3StreamCopier.java

@@ -92,7 +92,7 @@ public S3StreamCopier(String stagingFolder,
     this.s3Client = client;
     this.s3Config = s3Config;
 
-    this.s3StagingFile = String.join("/", stagingFolder, schemaName, streamName);
+    this.s3StagingFile = prepareS3StagingFile(stagingFolder, streamName);
     LOGGER.info("S3 upload part size: {} MB", s3Config.getPartSize());
     // The stream transfer manager lets us greedily stream into S3. The native AWS SDK does not
     // have support for streaming multipart uploads;
@@ -118,6 +118,47 @@ public S3StreamCopier(String stagingFolder,
     }
   }
 
+  public S3StreamCopier(String stagingFolder,
+                        DestinationSyncMode destSyncMode,
+                        String schema,
+                        String streamName,
+                        String s3FileName,
+                        AmazonS3 client,
+                        JdbcDatabase db,
+                        S3Config s3Config,
+                        ExtendedNameTransformer nameTransformer,
+                        SqlOperations sqlOperations) {
+    this.destSyncMode = destSyncMode;
+    this.schemaName = schema;
+    this.streamName = streamName;
+    this.db = db;
+    this.nameTransformer = nameTransformer;
+    this.sqlOperations = sqlOperations;
+    this.tmpTableName = nameTransformer.getTmpTableName(streamName);
+    this.s3Client = client;
+    this.s3Config = s3Config;
+
+    this.s3StagingFile = prepareS3StagingFile(stagingFolder, s3FileName);
+    LOGGER.info("S3 upload part size: {} MB", s3Config.getPartSize());
+    this.multipartUploadManager =
+        new StreamTransferManager(s3Config.getBucketName(), s3StagingFile, client)
+            .numUploadThreads(DEFAULT_UPLOAD_THREADS)
+            .queueCapacity(DEFAULT_QUEUE_CAPACITY)
+            .partSize(s3Config.getPartSize());
+    this.outputStream = multipartUploadManager.getMultiPartOutputStreams().get(0);
+
+    var writer = new PrintWriter(outputStream, true, StandardCharsets.UTF_8);
+    try {
+      this.csvPrinter = new CSVPrinter(writer, CSVFormat.DEFAULT);
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private String prepareS3StagingFile(String stagingFolder, String s3FileName) {
+    return String.join("/", stagingFolder, schemaName, s3FileName);
+  }
+
   @Override
   public void write(UUID id, String jsonDataString, Timestamp emittedAt) throws Exception {
     csvPrinter.printRecord(id, jsonDataString, emittedAt);

airbyte-integrations/connectors/destination-snowflake/Dockerfile

@@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
 
 RUN tar xf ${APPLICATION}.tar --strip-components=1
 
-LABEL io.airbyte.version=0.3.13
+LABEL io.airbyte.version=0.3.14
 LABEL io.airbyte.name=airbyte/destination-snowflake

airbyte-integrations/connectors/destination-snowflake/src/main/java/io/airbyte/integrations/destination/snowflake/SnowflakeS3StreamCopier.java

@@ -25,16 +25,21 @@
 package io.airbyte.integrations.destination.snowflake;
 
 import com.amazonaws.services.s3.AmazonS3;
+import io.airbyte.commons.string.Strings;
 import io.airbyte.db.jdbc.JdbcDatabase;
 import io.airbyte.integrations.destination.ExtendedNameTransformer;
 import io.airbyte.integrations.destination.jdbc.SqlOperations;
 import io.airbyte.integrations.destination.jdbc.copy.s3.S3Config;
 import io.airbyte.integrations.destination.jdbc.copy.s3.S3StreamCopier;
 import io.airbyte.protocol.models.DestinationSyncMode;
 import java.sql.SQLException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class SnowflakeS3StreamCopier extends S3StreamCopier {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(SnowflakeS3StreamCopier.class);
+
   public SnowflakeS3StreamCopier(String stagingFolder,
                                  DestinationSyncMode destSyncMode,
                                  String schema,
@@ -44,7 +49,8 @@ public SnowflakeS3StreamCopier(String stagingFolder,
                                  S3Config s3Config,
                                  ExtendedNameTransformer nameTransformer,
                                  SqlOperations sqlOperations) {
-    super(stagingFolder, destSyncMode, schema, streamName, client, db, s3Config, nameTransformer, sqlOperations);
+    super(stagingFolder, destSyncMode, schema, streamName, Strings.addRandomSuffix("", "", 3) + "_" + streamName, client, db, s3Config,
+        nameTransformer, sqlOperations);
   }
 
   @Override

airbyte-integrations/connectors/source-mongodb-v2/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/MongoDbSourceAcceptanceTest.java

@@ -115,10 +115,10 @@ protected ConfiguredAirbyteCatalog getConfiguredCatalog() throws Exception {
             .withDestinationSyncMode(DestinationSyncMode.APPEND)
             .withCursorField(List.of("_id"))
             .withStream(CatalogHelpers.createAirbyteStream(
-                    "test.acceptance_test",
-                    Field.of("_id", JsonSchemaPrimitive.STRING),
-                    Field.of("id", JsonSchemaPrimitive.STRING),
-                    Field.of("name", JsonSchemaPrimitive.STRING))
+                "test.acceptance_test",
+                Field.of("_id", JsonSchemaPrimitive.STRING),
+                Field.of("id", JsonSchemaPrimitive.STRING),
+                Field.of("name", JsonSchemaPrimitive.STRING))
                 .withSupportedSyncModes(Lists.newArrayList(SyncMode.INCREMENTAL))
                 .withDefaultCursorField(List.of("_id")))));
   }

docs/integrations/destinations/snowflake.md

@@ -189,6 +189,7 @@ Finally, you need to add read/write permissions to your bucket with that email.
 
 | Version | Date      | Pull Request | Subject |
 | :------ | :-------- | :-----       | :------ |
+| 0.3.14  | 2021-09-08 | [#5924](https://github.com/airbytehq/airbyte/pull/5924) | Fixed AWS S3 Staging COPY is writing records from different table in the same raw table  |
 | 0.3.13  | 2021-09-01 | [#5784](https://github.com/airbytehq/airbyte/pull/5784) | Updated query timeout from 30 minutes to 3 hours |
 | 0.3.12  | 2021-07-30 | [#5125](https://github.com/airbytehq/airbyte/pull/5125) | Enable `additionalPropertities` in spec.json |
 | 0.3.11  | 2021-07-21 | [#3555](https://github.com/airbytehq/airbyte/pull/3555) | Partial Success in BufferedStreamConsumer |