source-snowflake: TIMESTAMP_WITH_TIMEZONE column can't be used as cursor

[disi-pph]

Environment

• Airbyte version: 0.40.32
• OS Version / Instance: Debian GNU/Linux 10 (buster)
• Deployment: Docker Compose
• Source Connector and version: Snowflake 0.1.29
• Destination Connector and version: BigQuery 1.2.14
• Step where error happened: Sync job

Current Behavior

Similar to #13603 getting an error when attempting to run an incremental sync with deduplicated history on any table with a TIMESTAMP_WITH_TIMEZONE column as cursor:
2023-02-14 20:18:41 - Additional Failure Information: io.airbyte.commons.exceptions.ConfigErrorException: The following tables have invalid columns selected as cursor, please select a column with a well-defined ordering with no null values as a cursor. {tableName='redacted', cursorColumnName='HDB_LAST_SYNC', cursorSqlType=TIMESTAMP_WITH_TIMEZONE, cause=Unsupported cursor type},...

Expected Behavior

Sync works with TIMESTAMP_WITH_TIMEZONE as cursor field

Logs

8e523bfe_cdcb_47f8_8d58_2826f3a28ee8_logs_10888_txt.txt

Steps to Reproduce

1. Create a table with a TIMESTAMP_WITH_TIMEZONE column
2. Create an incremental sync using this column as a cursor
3. Run the sync

Are you willing to submit a PR?

Yes

[giacomo-calixa]

We are also affected by this bug, which also applies to cursor of type TIME_WITH_TIMEZONE.
We have a hacky solution that keeps SnowflakeSourceOperations.java as extending JdbcSourceOperations vs maybe the more apt PostgresSourceOperations

[giacomo-calixa]

A super simple work around that seems to be working for us.
We would like for the authors to review but I think this is not a comprehensive solution (it does not for instance cover the TIME_WITH_TIMEZONE type):

diff --git a/airbyte-integrations/connectors/source-snowflake/Dockerfile b/airbyte-integrations/connectors/source-snowflake/Dockerfile
index 6f9f02f09885..85cf8f6f81bb 100644
--- a/airbyte-integrations/connectors/source-snowflake/Dockerfile
+++ b/airbyte-integrations/connectors/source-snowflake/Dockerfile
@@ -16,5 +16,5 @@ ENV APPLICATION source-snowflake
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.1.28
+LABEL io.airbyte.version=0.1.30
 LABEL io.airbyte.name=airbyte/source-snowflake
diff --git a/airbyte-integrations/connectors/source-snowflake/src/main/java/io.airbyte.integrations.source.snowflake/SnowflakeSourceOperations.java b/airbyte-integrations/connectors/source-snowflake/src/main/java/io.airbyte.integrations.source.snowflake/SnowflakeSourceOperations.java
index 9240c2c6c486..0e885c65e6d7 100644
--- a/airbyte-integrations/connectors/source-snowflake/src/main/java/io.airbyte.integrations.source.snowflake/SnowflakeSourceOperations.java
+++ b/airbyte-integrations/connectors/source-snowflake/src/main/java/io.airbyte.integrations.source.snowflake/SnowflakeSourceOperations.java
@@ -24,12 +24,18 @@
 import java.sql.Timestamp;
 import java.time.LocalDate;
 import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.util.Set;
+import net.snowflake.client.jdbc.SnowflakeUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class SnowflakeSourceOperations extends JdbcSourceOperations {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(SnowflakeSourceOperations.class);
+  // HACK: See ____
+  // and https://github.com/airbytehq/airbyte/issues/23103
+  private static final Set<JDBCType> EXTRA_ALLOWED_CURSOR_TYPES = Set.of(JDBCType.TIMESTAMP_WITH_TIMEZONE);
 
   @Override
   protected void putDouble(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) {
@@ -101,9 +107,9 @@ public void copyToJsonField(final ResultSet resultSet, final int colIndex, final
     // TIMESTAMPLTZ data type detected as JDBCType.TIMESTAMP which is not correct
     if ("TIMESTAMPLTZ".equalsIgnoreCase(columnTypeName)) {
       putTimestampWithTimezone(json, columnName, resultSet, colIndex);
-      return;
+    } else {
+      super.copyToJsonField(resultSet, colIndex, json);
     }
-    super.copyToJsonField(resultSet, colIndex, json);
   }
 
   @Override
@@ -138,4 +144,26 @@ protected void putTime(final ObjectNode node, final String columnName, final Res
     node.put(columnName, DateTimeConverter.convertToTime(localTime));
   }
 
+  @Override
+  public boolean isCursorType(JDBCType type) {
+    return super.isCursorType(type) || EXTRA_ALLOWED_CURSOR_TYPES.contains(type);
+  }
+
+  @Override
+  public void setCursorField(PreparedStatement preparedStatement, int parameterIndex,
+      JDBCType cursorFieldType, String value) throws SQLException {
+    switch (cursorFieldType) {
+      case TIMESTAMP_WITH_TIMEZONE -> setTimestampWithTimezone(preparedStatement, parameterIndex, value);
+      default -> super.setCursorField(preparedStatement, parameterIndex, cursorFieldType, value);
+    }
+  }
+
+  private void setTimestampWithTimezone(final PreparedStatement preparedStatement,
+      final int parameterIndex, final String value) throws SQLException {
+
+    OffsetDateTime offsetDateTime = OffsetDateTime.parse(value);
+    var ts = Timestamp.from(offsetDateTime.toInstant());
+    // https://docs.snowflake.com/en/sql-reference/data-types-datetime#timestamp-ltz-timestamp-ntz-timestamp-tz
+    preparedStatement.setObject(parameterIndex, ts, SnowflakeUtil.EXTRA_TYPES_TIMESTAMP_LTZ);
+  }
 }```

[philippeboyd]

+1 on this, I was surprised to encounter this bug as I'm really wondering what was the logic behind such exception.

@disi-pph @giacomo-calixa has any of you work a PR for this bug?

[subodh1810]

This is fixed in version 0.1.34. Closing this

[kevin868]

Thanks @subodh1810 !
#24693
#24667