Destination Snowflake: Use safe executeMetadataQuery call (#37857)

Author: gisripa

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.1
+  dockerImageTag: 3.7.2
   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

@@ -132,12 +132,18 @@ private InitialRawTableStatus getInitialRawTableState(final StreamId id, final D
     if (destinationSyncMode == DestinationSyncMode.OVERWRITE) {
       return new InitialRawTableStatus(false, false, Optional.empty());
     }
-    final ResultSet tables = database.getMetaData().getTables(
-        databaseName,
-        id.getRawNamespace(),
-        id.getRawName(),
-        null);
-    if (!tables.next()) {
+    final boolean tableExists = database.executeMetadataQuery(databaseMetaData -> {
+      LOGGER.info("Retrieving table from Db metadata: {} {}",
+          id.getRawNamespace(),
+          id.getRawName());
+      try (final ResultSet tables = databaseMetaData.getTables(databaseName, id.getRawNamespace(), id.getRawName(), null)) {
+        return tables.next();
+      } catch (SQLException e) {
+        LOGGER.error("Failed to retrieve table metadata", e);
+        throw new RuntimeException(e);
+      }
+    });
+    if (!tableExists) {
       return new InitialRawTableStatus(false, false, Optional.empty());
     }
     // Snowflake timestamps have nanosecond precision, so decrement by 1ns