Destination Snowflake: Revisiting merge instead of insert+delete (#43367

)
airbytehq · Aug 20, 2024 · 141daac · 141daac
1 parent a2ff415
commit 141daac
Show file tree

Hide file tree

Showing 9 changed files with 2,213 additions and 1,993 deletions.
diff --git a/airbyte-integrations/connectors/destination-snowflake/metadata.yaml b/airbyte-integrations/connectors/destination-snowflake/metadata.yaml
@@ -5,7 +5,7 @@ data:
   connectorSubtype: database
   connectorType: destination
   definitionId: 424892c4-daac-4491-b35d-c6688ba547ba
-  dockerImageTag: 3.11.8
+  dockerImageTag: 3.11.9
   dockerRepository: airbyte/destination-snowflake
   documentationUrl: https://docs.airbyte.com/integrations/destinations/snowflake
   githubIssueLabel: destination-snowflake

diff --git a/...ake/src/main/kotlin/io/airbyte/integrations/destination/snowflake/SnowflakeDestination.kt b/...ake/src/main/kotlin/io/airbyte/integrations/destination/snowflake/SnowflakeDestination.kt
@@ -201,7 +201,9 @@ constructor(
             getRetentionPeriodDays(
                 config[RETENTION_PERIOD_DAYS],
             )
-        val sqlGenerator = SnowflakeSqlGenerator(retentionPeriodDays)
+        val useMergeForUpsert =
+            config.has(USE_MERGE_FOR_UPSERT) && config[USE_MERGE_FOR_UPSERT].asBoolean(false)
+        val sqlGenerator = SnowflakeSqlGenerator(retentionPeriodDays, useMergeForUpsert)
         val database = getDatabase(getDataSource(config))
         val databaseName = config[JdbcUtils.DATABASE_KEY].asText()
         val rawTableSchemaName: String =
@@ -285,6 +287,7 @@ constructor(
         const val RAW_SCHEMA_OVERRIDE: String = "raw_data_schema"
         const val RETENTION_PERIOD_DAYS: String = "retention_period_days"
         const val DISABLE_TYPE_DEDUPE: String = "disable_type_dedupe"
+        const val USE_MERGE_FOR_UPSERT: String = "use_merge_for_upsert"
         @JvmField
         val SCHEDULED_EXECUTOR_SERVICE: ScheduledExecutorService =
             Executors.newScheduledThreadPool(1)

diff --git a/...in/io/airbyte/integrations/destination/snowflake/typing_deduping/SnowflakeSqlGenerator.kt b/...in/io/airbyte/integrations/destination/snowflake/typing_deduping/SnowflakeSqlGenerator.kt
diff --git a/airbyte-integrations/connectors/destination-snowflake/src/main/resources/spec.json b/airbyte-integrations/connectors/destination-snowflake/src/main/resources/spec.json
@@ -181,6 +181,13 @@
         "description": "The number of days of Snowflake Time Travel to enable on the tables. See <a href=\"https://docs.snowflake.com/en/user-guide/data-time-travel#data-retention-period\">Snowflake's documentation</a> for more information. Setting a nonzero value will incur increased storage costs in your Snowflake instance.",
         "title": "Data Retention Period (days)",
         "order": 13
+      },
+      "use_merge_for_upsert": {
+        "type": "boolean",
+        "default": false,
+        "description": "Use MERGE for de-duplication of final tables. This option no effect if Final tables are disabled or Sync mode is not DEDUPE",
+        "title": "Use MERGE for De-duplication of final tables",
+        "order": 14
       }
     }
   },