🐛Destination-Snowflake: Added "Table/Stage already exists, but no permissions" handler on Sync action (#21912)

* [21843] Destination-Snowflake: Added "Table/Stage already exists, but no permissions" handler on Sync action

---------

Co-authored-by: Octavia Squidington III <[email protected]>

Author: etsybaev

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

@@ -348,7 +348,7 @@
 - name: Snowflake
   destinationDefinitionId: 424892c4-daac-4491-b35d-c6688ba547ba
   dockerRepository: airbyte/destination-snowflake
-  dockerImageTag: 0.4.46
+  dockerImageTag: 0.4.47
   documentationUrl: https://docs.airbyte.com/integrations/destinations/snowflake
   icon: snowflake.svg
   normalizationConfig:

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

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

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

@@ -67,7 +67,11 @@ protected Optional<ConfigErrorException> checkForKnownConfigExceptions(Exception
 
   @Override
   public void createTableIfNotExists(final JdbcDatabase database, final String schemaName, final String tableName) throws SQLException {
-    database.execute(createTableQuery(database, schemaName, tableName));
+    try {
+      database.execute(createTableQuery(database, schemaName, tableName));
+    } catch (SQLException e) {
+      throw checkForKnownConfigExceptions(e).orElseThrow(() -> e);
+    }
   }
 
   @Override

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.46
+LABEL io.airbyte.version=0.4.47
 LABEL io.airbyte.name=airbyte/destination-snowflake

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

@@ -130,7 +130,11 @@ protected String getListQuery(final String stageName, final String stagingPath,
   public void createStageIfNotExists(final JdbcDatabase database, final String stageName) throws Exception {
     final String query = getCreateStageQuery(stageName);
     LOGGER.debug("Executing query: {}", query);
-    database.execute(query);
+    try {
+      database.execute(query);
+    } catch (Exception e) {
+      throw checkForKnownConfigExceptions(e).orElseThrow(() -> e);
+    }
   }
 
   /**

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

@@ -83,7 +83,9 @@ protected String generateFilesList(final List<String> files) {
   @Override
   protected Optional<ConfigErrorException> checkForKnownConfigExceptions(Exception e) {
     if (e instanceof SnowflakeSQLException && e.getMessage().contains(NO_PRIVILEGES_ERROR_MESSAGE)) {
-      return Optional.of(new ConfigErrorException("Encountered Error with Snowflake Configuration: Current role does not have permissions on the target schema please verify your privileges", e));
+      return Optional.of(new ConfigErrorException(
+          "Encountered Error with Snowflake Configuration: Current role does not have permissions on the target schema please verify your privileges",
+          e));
     }
     return Optional.empty();
   }

airbyte-integrations/connectors/destination-snowflake/src/test/java/io/airbyte/integrations/destination/snowflake/SnowflakeInternalStagingSqlOperationsTest.java

@@ -5,10 +5,19 @@
 package io.airbyte.integrations.destination.snowflake;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import io.airbyte.commons.exceptions.ConfigErrorException;
+import io.airbyte.db.jdbc.JdbcDatabase;
+import java.sql.SQLException;
 import java.util.List;
+import net.snowflake.client.jdbc.SnowflakeSQLException;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.mockito.Mockito;
 
 class SnowflakeInternalStagingSqlOperationsTest {
 
@@ -66,4 +75,24 @@ void removeStage() {
     assertEquals(expectedQuery, actualRemoveQuery);
   }
 
+  @ParameterizedTest
+  @CsvSource({"TEST,false", "but current role has no privileges on it,true"})
+  public void testCreateStageIfNotExists(final String message, final boolean shouldCapture) {
+    final JdbcDatabase db = Mockito.mock(JdbcDatabase.class);
+    final String stageName = "foo";
+    try {
+      Mockito.doThrow(new SnowflakeSQLException(message)).when(db).execute(Mockito.anyString());
+    } catch (SQLException e) {
+      // This would not be expected, but the `execute` method above will flag as an unhandled exception
+      assert false;
+    }
+    final Exception exception = Assertions.assertThrows(Exception.class, () -> snowflakeStagingSqlOperations.createStageIfNotExists(db, stageName));
+    if (shouldCapture) {
+      assertInstanceOf(ConfigErrorException.class, exception);
+    } else {
+      assertInstanceOf(SnowflakeSQLException.class, exception);
+      assertEquals(exception.getMessage(), message);
+    }
+  }
+
 }

airbyte-integrations/connectors/destination-snowflake/src/test/java/io/airbyte/integrations/destination/snowflake/SnowflakeSqlOperationsTest.java

@@ -5,6 +5,8 @@
 package io.airbyte.integrations.destination.snowflake;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
@@ -20,7 +22,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import net.snowflake.client.jdbc.SnowflakeSQLException;
-import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
@@ -70,12 +71,34 @@ public void testCreateSchemaIfNotExists(final String message, final boolean shou
       // This would not be expected, but the `execute` method above will flag as an unhandled exception
       assert false;
     }
-    Exception exception = Assertions.assertThrows(Exception.class, () -> snowflakeSqlOperations.createSchemaIfNotExists(db, schemaName));
+    Exception exception = assertThrows(Exception.class, () -> snowflakeSqlOperations.createSchemaIfNotExists(db, schemaName));
     if (shouldCapture) {
-      Assertions.assertInstanceOf(ConfigErrorException.class, exception);
+      assertInstanceOf(ConfigErrorException.class, exception);
     } else {
-      Assertions.assertInstanceOf(SnowflakeSQLException.class, exception);
-      Assertions.assertEquals(exception.getMessage(), message);
+      assertInstanceOf(SnowflakeSQLException.class, exception);
+      assertEquals(exception.getMessage(), message);
+    }
+  }
+
+  @ParameterizedTest
+  @CsvSource({"TEST,false", "but current role has no privileges on it,true"})
+  public void testCreateTableIfNotExists(final String message, final boolean shouldCapture) {
+    final JdbcDatabase db = Mockito.mock(JdbcDatabase.class);
+    final String schemaName = "foo";
+    final String tableName = "bar";
+    try {
+      Mockito.doThrow(new SnowflakeSQLException(message)).when(db).execute(Mockito.anyString());
+    } catch (SQLException e) {
+      // This would not be expected, but the `execute` method above will flag as an unhandled exception
+      assert false;
+    }
+    final Exception exception =
+        assertThrows(Exception.class, () -> snowflakeSqlOperations.createTableIfNotExists(db, schemaName, tableName));
+    if (shouldCapture) {
+      assertInstanceOf(ConfigErrorException.class, exception);
+    } else {
+      assertInstanceOf(SnowflakeSQLException.class, exception);
+      assertEquals(exception.getMessage(), message);
     }
   }
 

docs/integrations/destinations/snowflake.md

@@ -287,6 +287,7 @@ Otherwise, make sure to grant the role the required permissions in the desired n
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                             |
 |:--------|:-----------|:-----------------------------------------------------------|:----------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.4.47  | 2023-01-30 | [\#21912](https://github.com/airbytehq/airbyte/pull/21912) | Catch "Create" Table and Stage Known Permissions and rethrow as ConfigExceptions                                                                      |
 | 0.4.46  | 2023-01-26 | [\#20631](https://github.com/airbytehq/airbyte/pull/20631) | Added support for destination checkpointing with staging                                                                                            |
 | 0.4.45  | 2023-01-25 | [#21087](https://github.com/airbytehq/airbyte/pull/21764)  | Catch Known Permissions and rethrow as ConfigExceptions                                                                                             |
 | 0.4.44  | 2023-01-20 | [#21087](https://github.com/airbytehq/airbyte/pull/21087)  | Wrap Authentication Errors as Config Exceptions                                                                                                     |