Destination Snowflake : fixed duplicate rows on retries

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

@@ -2,7 +2,7 @@
   "destinationDefinitionId": "424892c4-daac-4491-b35d-c6688ba547ba",
   "name": "Snowflake",
   "dockerRepository": "airbyte/destination-snowflake",
-  "dockerImageTag": "0.4.1",
+  "dockerImageTag": "0.4.2",
   "documentationUrl": "https://docs.airbyte.io/integrations/destinations/snowflake",
   "icon": "snowflake.svg"
 }

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

@@ -179,7 +179,7 @@
 - name: Snowflake
   destinationDefinitionId: 424892c4-daac-4491-b35d-c6688ba547ba
   dockerRepository: airbyte/destination-snowflake
-  dockerImageTag: 0.4.1
+  dockerImageTag: 0.4.2
   documentationUrl: https://docs.airbyte.io/integrations/destinations/snowflake
   icon: snowflake.svg
 - name: MariaDB ColumnStore

airbyte-integrations/connectors/destination-snowflake/Dockerfile

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

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

@@ -25,6 +25,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -45,8 +46,9 @@ public class SnowflakeInternalStagingConsumerFactory {
   private static final Logger LOGGER = LoggerFactory.getLogger(SnowflakeInternalStagingConsumerFactory.class);
 
   private static final long MAX_BATCH_SIZE_BYTES = 1024 * 1024 * 1024 / 4; // 256mb
+  private final String CURRENT_SYNC_PATH = UUID.randomUUID().toString();
 
-  public static AirbyteMessageConsumer create(final Consumer<AirbyteMessage> outputRecordCollector,
+  public AirbyteMessageConsumer create(final Consumer<AirbyteMessage> outputRecordCollector,
                                               final JdbcDatabase database,
                                               final SnowflakeStagingSqlOperations sqlOperations,
                                               final SnowflakeSQLNameTransformer namingResolver,
@@ -130,7 +132,7 @@ private static AirbyteStreamNameNamespacePair toNameNamespacePair(final WriteCon
     return new AirbyteStreamNameNamespacePair(config.getStreamName(), config.getNamespace());
   }
 
-  private static RecordWriter recordWriterFunction(final JdbcDatabase database,
+  private RecordWriter recordWriterFunction(final JdbcDatabase database,
                                                    final SqlOperations snowflakeSqlOperations,
                                                    final List<WriteConfig> writeConfigs,
                                                    final ConfiguredAirbyteCatalog catalog,
@@ -149,13 +151,13 @@ private static RecordWriter recordWriterFunction(final JdbcDatabase database,
       final WriteConfig writeConfig = pairToWriteConfig.get(pair);
       final String schemaName = writeConfig.getOutputSchemaName();
       final String tableName = writeConfig.getOutputTableName();
-      final String stageName = namingResolver.getStageName(schemaName, tableName);
+      final String path = namingResolver.getStagingPath(schemaName, tableName, CURRENT_SYNC_PATH);
 
-      snowflakeSqlOperations.insertRecords(database, records, schemaName, stageName);
+      snowflakeSqlOperations.insertRecords(database, records, schemaName, path);
     };
   }
 
-  private static OnCloseFunction onCloseFunction(final JdbcDatabase database,
+  private OnCloseFunction onCloseFunction(final JdbcDatabase database,
                                                  final SnowflakeStagingSqlOperations sqlOperations,
                                                  final List<WriteConfig> writeConfigs,
                                                  final SnowflakeSQLNameTransformer namingResolver) {
@@ -170,11 +172,18 @@ private static OnCloseFunction onCloseFunction(final JdbcDatabase database,
           LOGGER.info("Finalizing stream {}. schema {}, tmp table {}, final table {}", writeConfig.getStreamName(), schemaName, srcTableName,
               dstTableName);
 
-          final String stageName = namingResolver.getStageName(schemaName, dstTableName);
-          sqlOperations.copyIntoTmpTableFromStage(database, stageName, srcTableName, schemaName);
-          LOGGER.info("Uploading data from stage:  stream {}. schema {}, tmp table {}, stage {}", writeConfig.getStreamName(), schemaName,
-              srcTableName,
-              stageName);
+          final String path = namingResolver.getStagingPath(schemaName, dstTableName, CURRENT_SYNC_PATH);
+          LOGGER.info("Uploading data from stage:  stream {}. schema {}, tmp table {}, stage path {}", writeConfig.getStreamName(), schemaName,
+                  srcTableName,
+                  path);
+          try {
+            sqlOperations.copyIntoTmpTableFromStage(database, path, srcTableName, schemaName);
+          } catch (Exception e){
+            sqlOperations.cleanUpStage(database, path);
+            LOGGER.info("Cleaning stage path {}", path);
+            throw new RuntimeException("Failed to upload data from stage "+ path, e);
+          }
+
           sqlOperations.createTableIfNotExists(database, schemaName, dstTableName);
           switch (writeConfig.getSyncMode()) {
             case OVERWRITE -> queryList.add(sqlOperations.truncateTableQuery(database, schemaName, dstTableName));

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

@@ -71,7 +71,7 @@ public JsonNode toJdbcConfig(final JsonNode config) {
   public AirbyteMessageConsumer getConsumer(final JsonNode config,
                                             final ConfiguredAirbyteCatalog catalog,
                                             final Consumer<AirbyteMessage> outputRecordCollector) {
-    return SnowflakeInternalStagingConsumerFactory.create(outputRecordCollector, getDatabase(config),
+    return new SnowflakeInternalStagingConsumerFactory().create(outputRecordCollector, getDatabase(config),
         new SnowflakeStagingSqlOperations(), new SnowflakeSQLNameTransformer(), config, catalog);
   }
 

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

@@ -17,4 +17,7 @@ public String getStageName(String schemaName, String outputTableName) {
     return schemaName.concat(outputTableName).replaceAll("-", "_").toUpperCase();
   }
 
+  public String getStagingPath(String schemaName, String tableName, String currentSyncPath) {
+    return (getStageName(schemaName,tableName)+"/staged/"+currentSyncPath).toUpperCase();
+  }
 }

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

@@ -77,9 +77,12 @@ public String createTableQuery(final JdbcDatabase database, final String schemaN
         schemaName, tableName, JavaBaseConstants.COLUMN_NAME_AB_ID, JavaBaseConstants.COLUMN_NAME_DATA, JavaBaseConstants.COLUMN_NAME_EMITTED_AT);
   }
 
+  public void cleanUpStage(JdbcDatabase database, String path) throws SQLException {
+    database.execute(String.format("REMOVE @%s;", path));
+  }
+
   @Override
   public boolean isSchemaExists(JdbcDatabase database, String outputSchema) throws Exception {
     return database.query(SHOW_SCHEMAS).map(schemas -> schemas.get(NAME).asText()).anyMatch(outputSchema::equalsIgnoreCase);
   }
-
 }

docs/integrations/destinations/snowflake.md

@@ -198,6 +198,7 @@ Finally, you need to add read/write permissions to your bucket with that email.
 
 | Version | Date       | Pull Request | Subject |
 |:--------|:-----------| :-----       | :------ |
+| 0.4.2   | 2022-01-10 | [#9141](https://github.com/airbytehq/airbyte/pull/9141) | Fixed duplicate rows on retries |
 | 0.4.1   | 2021-01-06 | [#9311](https://github.com/airbytehq/airbyte/pull/9311) | Update сreating schema during check |
 | 0.4.0   | 2021-12-27 | [#9063](https://github.com/airbytehq/airbyte/pull/9063) | Updated normalization to produce permanent tables |
 | 0.3.24  | 2021-12-23 | [#8869](https://github.com/airbytehq/airbyte/pull/8869) | Changed staging approach to Byte-Buffered |