Adding logs to track schematization (#884)

snowflakedb · Jul 18, 2024 · 9905c7f · 9905c7f
1 parent 7172481
commit 9905c7f
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 3 deletions.
diff --git a/src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionServiceV1.java b/src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionServiceV1.java
@@ -425,6 +425,7 @@ public void appendMetaColIfNotExist(final String tableName) {
    */
   @Override
   public boolean hasSchemaEvolutionPermission(String tableName, String role) {
+    LOGGER.info("Checking schema evolution permission for table {}", tableName);
     checkConnection();
     InternalUtils.assertNotEmpty("tableName", tableName);
     String query = "show grants on table identifier(?)";

diff --git a/.../java/com/snowflake/kafka/connector/internal/streaming/BufferedTopicPartitionChannel.java b/.../java/com/snowflake/kafka/connector/internal/streaming/BufferedTopicPartitionChannel.java
@@ -669,6 +669,10 @@ public InsertRowsResponse get() throws Throwable {
               // Simply added to the final response if it's not schema related errors
               finalResponse.addError(insertError);
             } else {
+              LOGGER.info(
+                  "Triggering schema evolution. NonNullableColumns={}, extraColumns={}",
+                  String.join(",", nonNullableColumns),
+                  extraColNames == null ? "null" : String.join(",", extraColNames));
               SchematizationUtils.evolveSchemaIfNeeded(
                   this.conn,
                   this.channel.getTableName(),

diff --git a/src/main/java/com/snowflake/kafka/connector/internal/streaming/SnowflakeSinkServiceV2.java b/src/main/java/com/snowflake/kafka/connector/internal/streaming/SnowflakeSinkServiceV2.java
@@ -707,12 +707,19 @@ private void createTableIfNotExists(final String tableName) {
   private void populateSchemaEvolutionPermissions(String tableName) {
     if (!tableName2SchemaEvolutionPermission.containsKey(tableName)) {
       if (enableSchematization) {
-        tableName2SchemaEvolutionPermission.put(
-            tableName,
+        boolean hasSchemaEvolutionPermission =
             conn != null
                 && conn.hasSchemaEvolutionPermission(
-                    tableName, connectorConfig.get(SNOWFLAKE_ROLE)));
+                    tableName, connectorConfig.get(SNOWFLAKE_ROLE));
+        LOGGER.info(
+            "[SCHEMA_EVOLUTION_CACHE] Setting {} for table {}",
+            hasSchemaEvolutionPermission,
+            tableName);
+        tableName2SchemaEvolutionPermission.put(tableName, hasSchemaEvolutionPermission);
       } else {
+        LOGGER.info(
+            "[SCHEMA_EVOLUTION_CACHE] Schematization disabled. Setting false for table {}",
+            tableName);
         tableName2SchemaEvolutionPermission.put(tableName, false);
       }
     }