🐛Destination-Snowflake: updated check method to handle more errors (#18970)

* [16833] Destination-Snowflake: updated check method to handle "No Active Warehouse" error or user has incorrect permissions

Author: etsybaev

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

@@ -306,7 +306,7 @@
 - name: Snowflake
  destinationDefinitionId: 424892c4-daac-4491-b35d-c6688ba547ba
  dockerRepository: airbyte/destination-snowflake
- dockerImageTag: 0.4.38
+ dockerImageTag: 0.4.39
  documentationUrl: https://docs.airbyte.com/integrations/destinations/snowflake
  icon: snowflake.svg
  normalizationRepository: airbyte/normalization-snowflake

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

@@ -5408,7 +5408,7 @@
  supported_destination_sync_modes:
  - "overwrite"
  - "append"
-- dockerImage: "airbyte/destination-snowflake:0.4.38"
+- dockerImage: "airbyte/destination-snowflake:0.4.39"
  spec:
  documentationUrl: "https://docs.airbyte.com/integrations/destinations/snowflake"
  connectionSpecification:

airbyte-integrations/connectors/destination-jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/AbstractJdbcDestination.java

@@ -8,6 +8,7 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.commons.exceptions.ConnectionErrorException;
+import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.map.MoreMaps;
 import io.airbyte.db.factory.DataSourceFactory;
 import io.airbyte.db.jdbc.DefaultJdbcDatabase;
@@ -21,8 +22,10 @@
 import io.airbyte.protocol.models.AirbyteConnectionStatus;
 import io.airbyte.protocol.models.AirbyteConnectionStatus.Status;
 import io.airbyte.protocol.models.AirbyteMessage;
+import io.airbyte.protocol.models.AirbyteRecordMessage;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
 import java.sql.SQLException;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
@@ -84,11 +87,38 @@ public AirbyteConnectionStatus check(final JsonNode config) {
  }
  }
 
+ /**
+ * This method is deprecated. It verifies table creation, but not insert right to a newly created
+ * table. Use attemptTableOperations with the attemptInsert argument instead.
+ */
+ @Deprecated
  public static void attemptSQLCreateAndDropTableOperations(final String outputSchema,
  final JdbcDatabase database,
  final NamingConventionTransformer namingResolver,
  final SqlOperations sqlOps)
  throws Exception {
+ attemptTableOperations(outputSchema, database, namingResolver, sqlOps, false);
+ }
+
+ /**
+ * Verifies if provided creds has enough permissions. Steps are: 1. Create schema if not exists. 2.
+ * Create test table. 3. Insert dummy record to newly created table if "attemptInsert" set to true.
+ * 4. Delete table created on step 2.
+ *
+ * @param outputSchema - schema to tests against.
+ * @param database - database to tests against.
+ * @param namingResolver - naming resolver.
+ * @param sqlOps - SqlOperations object
+ * @param attemptInsert - set true if need to make attempt to insert dummy records to newly created
+ * table. Set false to skip insert step.
+ * @throws Exception
+ */
+ public static void attemptTableOperations(final String outputSchema,
+ final JdbcDatabase database,
+ final NamingConventionTransformer namingResolver,
+ final SqlOperations sqlOps,
+ final boolean attemptInsert)
+ throws Exception {
  // verify we have write permissions on the target schema by creating a table with a random name,
  // then dropping that table
  try {
@@ -100,7 +130,14 @@ public static void attemptSQLCreateAndDropTableOperations(final String outputSch
  final String outputTableName = namingResolver.getIdentifier("_airbyte_connection_test_" + UUID.randomUUID().toString().replaceAll("-", ""));
  sqlOps.createSchemaIfNotExists(database, outputSchema);
  sqlOps.createTableIfNotExists(database, outputSchema, outputTableName);
- sqlOps.dropTableIfExists(database, outputSchema, outputTableName);
+ // verify if user has permission to make SQL INSERT queries
+ try {
+ if (attemptInsert) {
+ sqlOps.insertRecords(database, List.of(getDummyRecord()), outputSchema, outputTableName);
+ }
+ } finally {
+ sqlOps.dropTableIfExists(database, outputSchema, outputTableName);
+ }
  } catch (final SQLException e) {
  if (Objects.isNull(e.getCause()) || !(e.getCause() instanceof SQLException)) {
  throw new ConnectionErrorException(e.getSQLState(), e.getErrorCode(), e.getMessage(), e);
@@ -113,6 +150,19 @@ public static void attemptSQLCreateAndDropTableOperations(final String outputSch
  }
  }
 
+ /**
+ * Generates a dummy AirbyteRecordMessage with random values.
+ *
+ * @return AirbyteRecordMessage object with dummy values that may be used to test insert permission.
+ */
+ private static AirbyteRecordMessage getDummyRecord() {
+ final JsonNode dummyDataToInsert = Jsons.deserialize("{ \"field1\": true }");
+ return new AirbyteRecordMessage()
+ .withStream("stream1")
+ .withData(dummyDataToInsert)
+ .withEmittedAt(1602637589000L);
+ }
+
  protected DataSource getDataSource(final JsonNode config) {
  final JsonNode jdbcConfig = toJdbcConfig(config);
  return DataSourceFactory.create(

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

@@ -14,6 +14,7 @@
 import io.airbyte.integrations.base.AirbyteTraceMessageUtility;
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.destination.ExtendedNameTransformer;
+import io.airbyte.integrations.destination.NamingConventionTransformer;
 import io.airbyte.integrations.destination.jdbc.AbstractJdbcDestination;
 import io.airbyte.integrations.destination.jdbc.SqlOperations;
 import io.airbyte.protocol.models.AirbyteConnectionStatus;
@@ -68,7 +69,7 @@ public AirbyteConnectionStatus check(final JsonNode config) {
  final JdbcDatabase database = getDatabase(dataSource);
  final var nameTransformer = getNameTransformer();
  final var outputSchema = nameTransformer.convertStreamName(config.get(schemaFieldName).asText());
- AbstractJdbcDestination.attemptSQLCreateAndDropTableOperations(outputSchema, database, nameTransformer, getSqlOperations());
+ performCreateInsertTestOnDestination(outputSchema, database, nameTransformer);
 
  return new AirbyteConnectionStatus().withStatus(AirbyteConnectionStatus.Status.SUCCEEDED);
  } catch (final ConnectionErrorException ex) {
@@ -92,4 +93,11 @@ public AirbyteConnectionStatus check(final JsonNode config) {
  }
  }
 
+ protected void performCreateInsertTestOnDestination(final String outputSchema,
+ final JdbcDatabase database,
+ final NamingConventionTransformer nameTransformer)
+ throws Exception {
+ AbstractJdbcDestination.attemptSQLCreateAndDropTableOperations(outputSchema, database, nameTransformer, getSqlOperations());
+ }
+
 }

airbyte-integrations/connectors/destination-snowflake/Dockerfile

@@ -20,5 +20,5 @@ RUN tar xf ${APPLICATION}.tar --strip-components=1
 
 ENV ENABLE_SENTRY true
 
-LABEL io.airbyte.version=0.4.38
+LABEL io.airbyte.version=0.4.39
 LABEL io.airbyte.name=airbyte/destination-snowflake

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

@@ -47,7 +47,7 @@ dependencies {
 // this is a configuration to make mockito work with final classes
  testImplementation 'org.mockito:mockito-inline:2.13.0'
 
-
+ integrationTestJavaImplementation project(':airbyte-commons-worker')
  integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
  integrationTestJavaImplementation project(':airbyte-integrations:connectors:destination-snowflake')
  integrationTestJavaImplementation 'org.apache.commons:commons-lang3:3.11'

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

@@ -10,6 +10,8 @@
 import io.airbyte.db.jdbc.JdbcDatabase;
 import io.airbyte.integrations.base.AirbyteMessageConsumer;
 import io.airbyte.integrations.destination.ExtendedNameTransformer;
+import io.airbyte.integrations.destination.NamingConventionTransformer;
+import io.airbyte.integrations.destination.jdbc.AbstractJdbcDestination;
 import io.airbyte.integrations.destination.jdbc.SqlOperations;
 import io.airbyte.integrations.destination.jdbc.copy.CopyConsumerFactory;
 import io.airbyte.integrations.destination.jdbc.copy.CopyDestination;
@@ -63,6 +65,15 @@ public SqlOperations getSqlOperations() {
  return new SnowflakeSqlOperations();
  }
 
+ @Override
+ protected void performCreateInsertTestOnDestination(final String outputSchema,
+ final JdbcDatabase database,
+ final NamingConventionTransformer nameTransformer)
+ throws Exception {
+ AbstractJdbcDestination.attemptTableOperations(outputSchema, database, nameTransformer,
+ getSqlOperations(), true);
+ }
+
  private String getConfiguredSchema(final JsonNode config) {
  return config.get("schema").asText();
  }

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

@@ -60,7 +60,9 @@ public AirbyteConnectionStatus check(final JsonNode config) {
  try {
  final JdbcDatabase database = getDatabase(dataSource);
  final String outputSchema = super.getNamingResolver().getIdentifier(config.get("schema").asText());
- attemptSQLCreateAndDropTableOperations(outputSchema, database, nameTransformer, snowflakeGcsStagingSqlOperations);
+
+ attemptTableOperations(outputSchema, database, nameTransformer, snowflakeGcsStagingSqlOperations,
+ true);
  attemptWriteAndDeleteGcsObject(gcsConfig, outputSchema);
 
  return new AirbyteConnectionStatus().withStatus(AirbyteConnectionStatus.Status.SUCCEEDED);

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

@@ -46,8 +46,9 @@ public AirbyteConnectionStatus check(final JsonNode config) {
  try {
  final JdbcDatabase database = getDatabase(dataSource);
  final String outputSchema = nameTransformer.getIdentifier(config.get("schema").asText());
- attemptSQLCreateAndDropTableOperations(outputSchema, database, nameTransformer, snowflakeInternalStagingSqlOperations);
- attemptSQLCreateAndDropStages(outputSchema, database, nameTransformer, snowflakeInternalStagingSqlOperations);
+ attemptTableOperations(outputSchema, database, nameTransformer,
+ snowflakeInternalStagingSqlOperations, true);
+ attemptStageOperations(outputSchema, database, nameTransformer, snowflakeInternalStagingSqlOperations);
  return new AirbyteConnectionStatus().withStatus(AirbyteConnectionStatus.Status.SUCCEEDED);
  } catch (final Exception e) {
  LOGGER.error("Exception while checking connection: ", e);
@@ -63,10 +64,10 @@ public AirbyteConnectionStatus check(final JsonNode config) {
  }
  }
 
- private static void attemptSQLCreateAndDropStages(final String outputSchema,
-  final JdbcDatabase database,
-  final NamingConventionTransformer namingResolver,
-  final SnowflakeInternalStagingSqlOperations sqlOperations)
+ private static void attemptStageOperations(final String outputSchema,
+ final JdbcDatabase database,
+ final NamingConventionTransformer namingResolver,
+ final SnowflakeInternalStagingSqlOperations sqlOperations)
  throws Exception {
 
  // verify we have permissions to create/drop stage

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

@@ -60,8 +60,9 @@ public AirbyteConnectionStatus check(final JsonNode config) {
  try {
  final JdbcDatabase database = getDatabase(dataSource);
  final String outputSchema = super.getNamingResolver().getIdentifier(config.get("schema").asText());
- attemptSQLCreateAndDropTableOperations(outputSchema, database, nameTransformer, snowflakeS3StagingSqlOperations);
- attemptSQLCreateAndDropStages(outputSchema, database, nameTransformer, snowflakeS3StagingSqlOperations);
+ attemptTableOperations(outputSchema, database, nameTransformer, snowflakeS3StagingSqlOperations,
+ true);
+ attemptStageOperations(outputSchema, database, nameTransformer, snowflakeS3StagingSqlOperations);
  return new AirbyteConnectionStatus().withStatus(AirbyteConnectionStatus.Status.SUCCEEDED);
  } catch (final Exception e) {
  LOGGER.error("Exception while checking connection: ", e);
@@ -77,7 +78,7 @@ public AirbyteConnectionStatus check(final JsonNode config) {
  }
  }
 
- private static void attemptSQLCreateAndDropStages(final String outputSchema,
+ private static void attemptStageOperations(final String outputSchema,
  final JdbcDatabase database,
  final NamingConventionTransformer namingResolver,
  final SnowflakeS3StagingSqlOperations sqlOperations)