destination-snowflake: remove contention on state table (#43440)

airbytehq · Aug 9, 2024 · 2a15fbd · 2a15fbd
1 parent f8b559e
commit 2a15fbd
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 6 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.6
+  dockerImageTag: 3.11.7
   dockerRepository: airbyte/destination-snowflake
   documentationUrl: https://docs.airbyte.com/integrations/destinations/snowflake
   githubIssueLabel: destination-snowflake

diff --git a/...airbyte/integrations/destination/snowflake/typing_deduping/SnowflakeDestinationHandler.kt b/...airbyte/integrations/destination/snowflake/typing_deduping/SnowflakeDestinationHandler.kt
@@ -500,6 +500,16 @@ class SnowflakeDestinationHandler(
         return database.queryJsons(sql)
     }
 
+    override fun getDeleteStatesSql(destinationStates: Map<StreamId, SnowflakeState>): String {
+        if (Math.random() < 0.01) {
+            LOGGER.info("actually deleting states")
+            return super.getDeleteStatesSql(destinationStates)
+        } else {
+            LOGGER.info("skipping state deletion")
+            return "SELECT 1" // We still need to send a valid SQL query.
+        }
+    }
+
     companion object {
         private val LOGGER: Logger =
             LoggerFactory.getLogger(SnowflakeDestinationHandler::class.java)

diff --git a/...integrations/destination/snowflake/typing_deduping/AbstractSnowflakeTypingDedupingTest.kt b/...integrations/destination/snowflake/typing_deduping/AbstractSnowflakeTypingDedupingTest.kt
@@ -99,10 +99,10 @@ abstract class AbstractSnowflakeTypingDedupingTest : BaseTypingDedupingTest() {
         database!!.execute(
             String.format(
                 """
-            DROP TABLE IF EXISTS "%s"."%s";
-            DROP SCHEMA IF EXISTS "%s" CASCADE
-            
-            """.trimIndent(),
+                DROP TABLE IF EXISTS "%s"."%s";
+                DROP SCHEMA IF EXISTS "%s" CASCADE
+                
+                """.trimIndent(),
                 rawSchema, // Raw table is still lowercase.
                 StreamId.concatenateRawTableName(namespaceOrDefault, streamName),
                 namespaceOrDefault.uppercase(Locale.getDefault()),

diff --git a/docs/integrations/destinations/snowflake.md b/docs/integrations/destinations/snowflake.md
@@ -268,7 +268,8 @@ desired namespace.
 
 | Version         | Date       | Pull Request                                               | Subject                                                                                                                                                                          |
 |:----------------|:-----------|:-----------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| 3.11.6          | 2024-08-07 | [\#43332](https://github.com/airbytehq/airbyte/pull/43332) | bump Java CDK |
+| 3.11.7          | 2024-08-09 | [\#43440](https://github.com/airbytehq/airbyte/pull/43440) | remove contention on state table by deleting rows ony once every 100 updates |
+| 3.11.6          | 2024-08-09 | [\#43332](https://github.com/airbytehq/airbyte/pull/43332) | bump Java CDK |
 | 3.11.5          | 2024-08-07 | [\#43348](https://github.com/airbytehq/airbyte/pull/43348) | SnowflakeSqlGen cleanup to Kotlin string interpolation                                                                                                                           |
 | 3.11.4          | 2024-07-18 | [\#41940](https://github.com/airbytehq/airbyte/pull/41940) | Update host regex to allow connecting to LocalStack Snowflake                                                                                                                    |
 | 3.11.3          | 2024-07-15 | [\#41968](https://github.com/airbytehq/airbyte/pull/41968) | Don't hang forever on empty stream list; shorten error message on INCOMPLETE stream status                                                                                       |