revert contention-reducing change in destination-snowflake (#38052)

as part of the move of destination-snowflake to the kotlin CDK, we tried improve concurrency by only `DELETE`ing from `_airbyte_destination_state` if it has some data to delete (by issuing an `IF EXISTS` in the same transaction.
Looks like it might be causing some stuck syncs, so we're reverting that "improvement"

Author: stephane-airbyte

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/db/jdbc/JdbcDatabase.kt

@@ -43,7 +43,9 @@ abstract class JdbcDatabase(protected val sourceOperations: JdbcCompatibleSource
         execute { connection: Connection ->
             connection.autoCommit = false
             for (s in queries) {
+                LOGGER.info("executing query within transaction: $s")
                 connection.createStatement().execute(s)
+                LOGGER.info("done executing query within transaction: $s")
             }
             connection.commit()
             connection.autoCommit = true

airbyte-integrations/connectors/destination-snowflake/gradle.properties

@@ -1,2 +1,2 @@
-testExecutionConcurrency=-1
+testExecutionConcurrency=4
 JunitMethodExecutionTimeout=30 m

airbyte-integrations/connectors/destination-snowflake/metadata.yaml

@@ -5,7 +5,7 @@ data:
   connectorSubtype: database
   connectorType: destination
   definitionId: 424892c4-daac-4491-b35d-c6688ba547ba
-  dockerImageTag: 3.7.3
+  dockerImageTag: 3.7.4
   dockerRepository: airbyte/destination-snowflake
   documentationUrl: https://docs.airbyte.com/integrations/destinations/snowflake
   githubIssueLabel: destination-snowflake

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

@@ -42,7 +42,6 @@
 import java.util.UUID;
 import java.util.stream.Collectors;
 import net.snowflake.client.jdbc.SnowflakeSQLException;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.text.StringSubstitutor;
 import org.jooq.SQLDialect;
 import org.slf4j.Logger;
@@ -362,30 +361,4 @@ private String toJdbcTypeName(final AirbyteProtocolType airbyteProtocolType) {
     };
   }
 
-  protected String getDeleteStatesSql(Map<StreamId, ? extends SnowflakeState> destinationStates) {
-    // only doing the DELETE where there's rows to delete allows us to avoid taking a lock on the table
-    // when there's nothing to delete
-    // This is particularly relevant in the context of tests, where many instance of the snowflake
-    // destination could be run in parallel
-    String deleteStatesSql = super.getDeleteStatesSql(destinationStates);
-    StringBuilder sql = new StringBuilder();
-    // sql.append("BEGIN\n");
-    sql.append("  IF (EXISTS (").append(deleteStatesSql.replace("delete from", "SELECT 1 FROM ")).append(")) THEN\n");
-    sql.append("    ").append(deleteStatesSql).append(";\n");
-    sql.append("  END IF\n");
-    // sql.append("END;\n");
-    return sql.toString();
-  }
-
-  protected void executeWithinTransaction(List<String> statements) throws SQLException {
-    StringBuilder sb = new StringBuilder();
-    sb.append("BEGIN\n");
-    sb.append("  BEGIN TRANSACTION;\n    ");
-    sb.append(StringUtils.join(statements, ";\n    "));
-    sb.append(";\n  COMMIT;\n");
-    sb.append("END;");
-    LOGGER.info("executing SQL:" + sb);
-    getJdbcDatabase().execute(sb.toString());
-  }
-
 }

docs/integrations/destinations/snowflake.md

@@ -276,6 +276,7 @@ desired namespace.
 
 | Version         | Date       | Pull Request                                               | Subject                                                                                                                                                          |
 | :-------------- | :--------- | :--------------------------------------------------------- | :--------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| 3.7.4           | 2024-05-07 | [\#38052](https://github.com/airbytehq/airbyte/pull/38052) | Revert problematic optimization                                                                                                                                |
 | 3.7.3           | 2024-05-07 | [\#34612](https://github.com/airbytehq/airbyte/pull/34612) | Adopt CDK 0.33.2                                                                                                                                                 |
 | 3.7.2           | 2024-05-06 | [\#37857](https://github.com/airbytehq/airbyte/pull/37857) | Use safe executeMetadata call                                                                                                                                    |
 | 3.7.1           | 2024-04-30 | [\#36910](https://github.com/airbytehq/airbyte/pull/36910) | Bump CDK version                                                                                                                                                 |