JDBC sources: consolidate date/time values mapping

airbyte-db/db-lib/src/main/java/io/airbyte/db/jdbc/AbstractJdbcCompatibleSourceOperations.java

@@ -24,9 +24,11 @@
 import java.text.ParseException;
 import java.time.Instant;
 import java.time.LocalDate;
+import java.time.LocalTime;
 import java.time.OffsetDateTime;
 import java.time.OffsetTime;
 import java.time.chrono.IsoEra;
+import java.time.format.DateTimeParseException;
 import java.util.Collections;
 import java.util.List;
 import java.util.StringJoiner;
@@ -124,11 +126,11 @@ protected void putString(final ObjectNode node, final String columnName, final R
   }
 
   protected void putDate(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException {
-    node.put(columnName, DataTypeUtils.toISO8601String(resultSet.getDate(index)));
+    node.put(columnName, DateTimeConverter.convertToDate(getObject(resultSet, index, LocalDate.class)));
   }
 
   protected void putTime(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException {
-    node.put(columnName, DataTypeUtils.toISO8601String(resultSet.getTime(index)));
+    node.put(columnName, DateTimeConverter.convertToTime(getObject(resultSet, index, LocalTime.class)));
   }
 
   protected void putTimestamp(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException {
@@ -146,7 +148,11 @@ protected void putDefault(final ObjectNode node, final String columnName, final
   }
 
   protected void setTime(final PreparedStatement preparedStatement, final int parameterIndex, final String value) throws SQLException {
-    setTimestamp(preparedStatement, parameterIndex, value);
+    try {
+      preparedStatement.setObject(parameterIndex, LocalTime.parse(value));
+    } catch (final DateTimeParseException e) {
+      setTimestamp(preparedStatement, parameterIndex, value);
+    }
   }
 
   protected void setTimestamp(final PreparedStatement preparedStatement, final int parameterIndex, final String value) throws SQLException {
@@ -177,6 +183,14 @@ protected void setTimestamp(final PreparedStatement preparedStatement, final int
   }
 
   protected void setDate(final PreparedStatement preparedStatement, final int parameterIndex, final String value) throws SQLException {
+    try {
+      preparedStatement.setObject(parameterIndex, LocalDate.parse(value));
+    } catch (final DateTimeParseException e) {
+      setDateAsTimestamp(preparedStatement, parameterIndex, value);
+    }
+  }
+
+  private void setDateAsTimestamp(PreparedStatement preparedStatement, int parameterIndex, String value) throws SQLException {
     try {
       final Timestamp from = Timestamp.from(DataTypeUtils.getDateFormat().parse(value).toInstant());
       preparedStatement.setDate(parameterIndex, new Date(from.getTime()));

airbyte-db/db-lib/src/main/java/io/airbyte/db/jdbc/DateTimeConverter.java

@@ -12,7 +12,10 @@
 import static io.airbyte.db.jdbc.AbstractJdbcCompatibleSourceOperations.resolveEra;
 import static java.time.ZoneOffset.UTC;
 
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import java.sql.Date;
+import java.sql.ResultSet;
+import java.sql.SQLException;
 import java.sql.Time;
 import java.sql.Timestamp;
 import java.time.Duration;
@@ -168,4 +171,15 @@ public static String convertToTime(final Object time) {
     }
   }
 
+  public static void putJavaSQLDate(ObjectNode node, String columnName, ResultSet resultSet, int index) throws SQLException {
+    final Date date = resultSet.getDate(index);
+    node.put(columnName, convertToDate(date));
+  }
+
+  public static void putJavaSQLTime(ObjectNode node, String columnName, ResultSet resultSet, int index) throws SQLException {
+    // resultSet.getTime() will lose nanoseconds precision
+    final LocalTime localTime = resultSet.getTimestamp(index).toLocalDateTime().toLocalTime();
+    node.put(columnName, convertToTime(localTime));
+  }
+
 }

airbyte-db/db-lib/src/test/java/io/airbyte/db/jdbc/TestJdbcUtils.java

@@ -163,8 +163,8 @@ void testSetStatementField() throws SQLException {
       sourceOperations.setStatementField(ps, 10, JDBCType.DECIMAL, "1");
       sourceOperations.setStatementField(ps, 11, JDBCType.CHAR, "a");
       sourceOperations.setStatementField(ps, 12, JDBCType.VARCHAR, "a");
-      sourceOperations.setStatementField(ps, 13, JDBCType.DATE, "2020-11-01T00:00:00Z");
-      sourceOperations.setStatementField(ps, 14, JDBCType.TIME, "1970-01-01T05:00:00.000Z");
+      sourceOperations.setStatementField(ps, 13, JDBCType.DATE, "2020-11-01");
+      sourceOperations.setStatementField(ps, 14, JDBCType.TIME, "05:00:00.000");
       sourceOperations.setStatementField(ps, 15, JDBCType.TIMESTAMP, "2001-09-29T03:00:00.000Z");
       sourceOperations.setStatementField(ps, 16, JDBCType.BINARY, "61616161");
 
@@ -390,10 +390,8 @@ private ObjectNode expectedValues() {
     expected.put("decimal", new BigDecimal(1));
     expected.put("char", "a");
     expected.put("varchar", "a");
-    // todo (cgardens) we should parse this to a date string
-    expected.put("date", "2020-11-01T00:00:00Z");
-    // todo (cgardens) we should parse this to a time string
-    expected.put("time", "1970-01-01T05:00:00Z");
+    expected.put("date", "2020-11-01");
+    expected.put("time", "05:00:00.000000");
     expected.put("timestamp", "2001-09-29T03:00:00.000000Z");
     expected.put("binary1", "aaaa".getBytes(Charsets.UTF_8));
     return expected;

airbyte-integrations/bases/debezium-v1-9-6/src/main/java/io/airbyte/integrations/debezium/internals/MSSQLConverter.java

@@ -8,6 +8,7 @@
 import com.microsoft.sqlserver.jdbc.Geometry;
 import com.microsoft.sqlserver.jdbc.SQLServerException;
 import io.airbyte.db.DataTypeUtils;
+import io.airbyte.db.jdbc.DateTimeConverter;
 import io.debezium.spi.converter.CustomConverter;
 import io.debezium.spi.converter.RelationalColumn;
 import java.math.BigDecimal;
@@ -107,7 +108,9 @@ private void registerDate(final RelationalColumn field,
       if (Objects.isNull(input)) {
         return DebeziumConverterUtils.convertDefaultValue(field);
       }
-
+      if (field.typeName().equalsIgnoreCase("DATE")) {
+        return DateTimeConverter.convertToDate(input);
+      }
       return DebeziumConverterUtils.convertDate(input);
     });
   }

airbyte-integrations/connectors/source-cockroachdb/src/main/java/io/airbyte/integrations/source/cockroachdb/CockroachJdbcSourceOperations.java

@@ -4,12 +4,15 @@
 
 package io.airbyte.integrations.source.cockroachdb;
 
+import static io.airbyte.db.DataTypeUtils.TIMETZ_FORMATTER;
+
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.db.jdbc.JdbcSourceOperations;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.time.OffsetTime;
 import java.util.Collections;
 
 /**
@@ -61,6 +64,9 @@ private void putCockroachSpecialDataType(final ResultSet resultSet, final int in
       if ("numeric".equalsIgnoreCase(columnType)) {
         final double value = resultSet.getDouble(index);
         node.put(columnName, value);
+      } else if ("timetz".equalsIgnoreCase(columnType)) {
+        final OffsetTime timetz = getObject(resultSet, index, OffsetTime.class);
+        node.put(columnName, timetz.format(TIMETZ_FORMATTER));
       } else {
         node.put(columnName, (Double) null);
       }

airbyte-integrations/connectors/source-cockroachdb/src/test-integration/java/io/airbyte/integrations/source/cockroachdb/CockroachDbSourceDatatypeTest.java

@@ -217,7 +217,7 @@ protected void initTests() {
             .sourceType("date")
             .airbyteType(JsonSchemaType.STRING)
             .addInsertValues("'1999-01-08'", "null")
-            .addExpectedValues("1999-01-08T00:00:00Z", null)
+            .addExpectedValues("1999-01-08", null)
             .build());
 
     addDataTypeTestData(
@@ -313,14 +313,12 @@ protected void initTests() {
             .addExpectedValues("a", "abc", "Миші йдуть;", "櫻花分店", "", null, "\\xF0\\x9F\\x9A\\x80")
             .build());
 
-    // JdbcUtils-> DATE_FORMAT is set as ""yyyy-MM-dd'T'HH:mm:ss'Z'"" for both Date and Time types.
-    // Time (04:05:06) would be represented like "1970-01-01T04:05:06Z"
     addDataTypeTestData(
         TestDataHolder.builder()
             .sourceType("time")
             .airbyteType(JsonSchemaType.STRING)
             .addInsertValues("'04:05:06'", null)
-            .addExpectedValues("1970-01-01T04:05:06Z")
+            .addExpectedValues("04:05:06.000000")
             .addNullExpectedValue()
             .build());
 
@@ -330,7 +328,7 @@ protected void initTests() {
             .sourceType("timetz")
             .airbyteType(JsonSchemaType.STRING)
             .addInsertValues("'04:05:06Z'", null)
-            .addExpectedValues("1970-01-01T04:05:06Z")
+            .addExpectedValues("04:05:06.000000Z")
             .addNullExpectedValue()
             .build());
 

airbyte-integrations/connectors/source-cockroachdb/src/test/java/io/airbyte/integrations/source/cockroachdb/CockroachDbJdbcSourceAcceptanceTest.java

@@ -170,22 +170,22 @@ protected List<AirbyteMessage> getTestMessages() {
                 .withData(Jsons.jsonNode(ImmutableMap
                     .of(COL_ID, ID_VALUE_1,
                         COL_NAME, "picard",
-                        COL_UPDATED_AT, "2004-10-19T00:00:00Z")))),
+                        COL_UPDATED_AT, "2004-10-19")))),
         new AirbyteMessage().withType(Type.RECORD)
             .withRecord(new AirbyteRecordMessage().withStream(streamName)
                 .withNamespace(getDefaultNamespace())
                 .withData(Jsons.jsonNode(ImmutableMap
                     .of(COL_ID, ID_VALUE_2,
                         COL_NAME, "crusher",
                         COL_UPDATED_AT,
-                        "2005-10-19T00:00:00Z")))),
+                        "2005-10-19")))),
         new AirbyteMessage().withType(Type.RECORD)
             .withRecord(new AirbyteRecordMessage().withStream(streamName)
                 .withNamespace(getDefaultNamespace())
                 .withData(Jsons.jsonNode(ImmutableMap
                     .of(COL_ID, ID_VALUE_3,
                         COL_NAME, "vash",
-                        COL_UPDATED_AT, "2006-10-19T00:00:00Z")))));
+                        COL_UPDATED_AT, "2006-10-19")))));
   }
 
   @Test
@@ -314,13 +314,13 @@ void testReadOneTableIncrementallyTwice() throws Exception {
             .withData(Jsons.jsonNode(ImmutableMap
                 .of(COL_ID, ID_VALUE_4,
                     COL_NAME, "riker",
-                    COL_UPDATED_AT, "2006-10-19T00:00:00Z")))));
+                    COL_UPDATED_AT, "2006-10-19")))));
     expectedMessages.add(new AirbyteMessage().withType(Type.RECORD)
         .withRecord(new AirbyteRecordMessage().withStream(streamName).withNamespace(namespace)
             .withData(Jsons.jsonNode(ImmutableMap
                 .of(COL_ID, ID_VALUE_5,
                     COL_NAME, "data",
-                    COL_UPDATED_AT, "2006-10-19T00:00:00Z")))));
+                    COL_UPDATED_AT, "2006-10-19")))));
     expectedMessages.add(new AirbyteMessage()
         .withType(Type.STATE)
         .withState(new AirbyteStateMessage()

airbyte-integrations/connectors/source-db2-strict-encrypt/src/test/resources/expected_spec.json

@@ -56,9 +56,7 @@
             "properties": {
               "encryption_method": {
                 "type": "string",
-                "const": "encrypted_verify_certificate",
-                "enum": ["encrypted_verify_certificate"],
-                "default": "encrypted_verify_certificate"
+                "const": "encrypted_verify_certificate"
               },
               "ssl_certificate": {
                 "title": "SSL PEM file",

airbyte-integrations/connectors/source-db2/src/main/java/io.airbyte.integrations.source.db2/Db2SourceOperations.java

@@ -4,12 +4,18 @@
 
 package io.airbyte.integrations.source.db2;
 
+import static io.airbyte.db.jdbc.DateTimeConverter.putJavaSQLDate;
+import static io.airbyte.db.jdbc.DateTimeConverter.putJavaSQLTime;
+
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.db.jdbc.JdbcSourceOperations;
+import java.sql.Date;
+import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.time.LocalDate;
 import java.util.Collections;
 import java.util.List;
 import org.slf4j.Logger;
@@ -69,4 +75,28 @@ private void putDecfloat(final ObjectNode node,
     }
   }
 
+  @Override
+  protected void putDate(final ObjectNode node,
+                         final String columnName,
+                         final ResultSet resultSet,
+                         final int index)
+      throws SQLException {
+    putJavaSQLDate(node, columnName, resultSet, index);
+  }
+
+  @Override
+  protected void putTime(final ObjectNode node,
+                         final String columnName,
+                         final ResultSet resultSet,
+                         final int index)
+      throws SQLException {
+    putJavaSQLTime(node, columnName, resultSet, index);
+  }
+
+  @Override
+  protected void setDate(final PreparedStatement preparedStatement, final int parameterIndex, final String value) throws SQLException {
+    final LocalDate date = LocalDate.parse(value);
+    preparedStatement.setDate(parameterIndex, Date.valueOf(date));
+  }
+
 }

airbyte-integrations/connectors/source-db2/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/Db2SourceDatatypeTest.java

@@ -303,15 +303,15 @@ protected void initTests() {
             .sourceType("DATE")
             .airbyteType(JsonSchemaType.STRING)
             .addInsertValues("null", "'0001-01-01'", "'9999-12-31'")
-            .addExpectedValues(null, "0001-01-01T00:00:00Z", "9999-12-31T00:00:00Z")
+            .addExpectedValues(null, "0001-01-01", "9999-12-31")
             .build());
     addDataTypeTestData(
         TestDataHolder.builder()
             .createTablePatternSql(CREATE_TABLE_SQL)
             .sourceType("TIME")
             .airbyteType(JsonSchemaType.STRING)
             .addInsertValues("null", "'00.00.00'", "'1:59 PM'", "'23.59.59'")
-            .addExpectedValues(null, "1970-01-01T00:00:00Z", "1970-01-01T13:59:00Z", "1970-01-01T23:59:59Z")
+            .addExpectedValues(null, "00:00:00.000000", "13:59:00.000000", "23:59:59.000000")
             .build());
     addDataTypeTestData(
         TestDataHolder.builder()

airbyte-integrations/connectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java

@@ -621,4 +621,5 @@ protected List<ConfiguredAirbyteStream> identifyStreamsToSnapshot(final Configur
         .map(Jsons::clone)
         .collect(Collectors.toList());
   }
+
 }

airbyte-integrations/connectors/source-jdbc/src/testFixtures/java/io/airbyte/integrations/source/jdbc/test/JdbcSourceAcceptanceTest.java

@@ -633,8 +633,8 @@ void testIncrementalDateCheckCursor() throws Exception {
   protected void incrementalDateCheck() throws Exception {
     incrementalCursorCheck(
         COL_UPDATED_AT,
-        "2005-10-18T00:00:00Z",
-        "2006-10-19T00:00:00Z",
+        "2005-10-18",
+        "2006-10-19",
         List.of(getTestMessages().get(1), getTestMessages().get(2)));
   }
 
@@ -702,13 +702,13 @@ protected List<AirbyteMessage> getExpectedAirbyteMessagesSecondSync(final String
             .withData(Jsons.jsonNode(Map
                 .of(COL_ID, ID_VALUE_4,
                     COL_NAME, "riker",
-                    COL_UPDATED_AT, "2006-10-19T00:00:00Z")))));
+                    COL_UPDATED_AT, "2006-10-19")))));
     expectedMessages.add(new AirbyteMessage().withType(Type.RECORD)
         .withRecord(new AirbyteRecordMessage().withStream(streamName).withNamespace(namespace)
             .withData(Jsons.jsonNode(Map
                 .of(COL_ID, ID_VALUE_5,
                     COL_NAME, "data",
-                    COL_UPDATED_AT, "2006-10-19T00:00:00Z")))));
+                    COL_UPDATED_AT, "2006-10-19")))));
     final DbStreamState state = new DbStreamState()
         .withStreamName(streamName)
         .withStreamNamespace(namespace)
@@ -1033,20 +1033,20 @@ protected List<AirbyteMessage> getTestMessages() {
                 .withData(Jsons.jsonNode(Map
                     .of(COL_ID, ID_VALUE_1,
                         COL_NAME, "picard",
-                        COL_UPDATED_AT, "2004-10-19T00:00:00Z")))),
+                        COL_UPDATED_AT, "2004-10-19")))),
         new AirbyteMessage().withType(Type.RECORD)
             .withRecord(new AirbyteRecordMessage().withStream(streamName).withNamespace(getDefaultNamespace())
                 .withData(Jsons.jsonNode(Map
                     .of(COL_ID, ID_VALUE_2,
                         COL_NAME, "crusher",
                         COL_UPDATED_AT,
-                        "2005-10-19T00:00:00Z")))),
+                        "2005-10-19")))),
         new AirbyteMessage().withType(Type.RECORD)
             .withRecord(new AirbyteRecordMessage().withStream(streamName).withNamespace(getDefaultNamespace())
                 .withData(Jsons.jsonNode(Map
                     .of(COL_ID, ID_VALUE_3,
                         COL_NAME, "vash",
-                        COL_UPDATED_AT, "2006-10-19T00:00:00Z")))));
+                        COL_UPDATED_AT, "2006-10-19")))));
   }
 
   protected List<AirbyteMessage> createExpectedTestMessages(final List<DbStreamState> states) {

airbyte-integrations/connectors/source-mssql/src/test-integration/java/io/airbyte/integrations/source/mssql/AbstractMssqlSourceDatatypeTest.java

@@ -144,7 +144,7 @@ protected void initTests() {
             .sourceType("date")
             .airbyteType(JsonSchemaType.STRING)
             .addInsertValues("'0001-01-01'", "'9999-12-31'", "'1999-01-08'", "null")
-            .addExpectedValues("0001-01-01T00:00:00Z", "9999-12-31T00:00:00Z", "1999-01-08T00:00:00Z", null)
+            .addExpectedValues("0001-01-01", "9999-12-31", "1999-01-08", null)
             .createTablePatternSql(CREATE_TABLE_SQL)
             .build());
 

airbyte-integrations/connectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/MySqlSourceOperations.java

@@ -212,16 +212,6 @@ public boolean isCursorType(MysqlType type) {
     return ALLOWED_CURSOR_TYPES.contains(type);
   }
 
-  @Override
-  protected void putDate(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException {
-    node.put(columnName, DateTimeConverter.convertToDate(getObject(resultSet, index, LocalDate.class)));
-  }
-
-  @Override
-  protected void putTime(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException {
-    node.put(columnName, DateTimeConverter.convertToTime(getObject(resultSet, index, LocalTime.class)));
-  }
-
   @Override
   protected void putTimestamp(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException {
     node.put(columnName, DateTimeConverter.convertToTimestamp(getObject(resultSet, index, LocalDateTime.class)));