🎉 Updated timestamp transformation with microseconds

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

@@ -140,7 +140,7 @@
 - name: Cockroachdb
   sourceDefinitionId: 9fa5862c-da7c-11eb-8d19-0242ac130003
   dockerRepository: airbyte/source-cockroachdb
-  dockerImageTag: 0.1.7
+  dockerImageTag: 0.1.8
   documentationUrl: https://docs.airbyte.io/integrations/sources/cockroachdb
   icon: cockroachdb.svg
   sourceType: database
@@ -321,7 +321,7 @@
 - name: IBM Db2
   sourceDefinitionId: 447e0381-3780-4b46-bb62-00a4e3c8b8e2
   dockerRepository: airbyte/source-db2
-  dockerImageTag: 0.1.7
+  dockerImageTag: 0.1.8
   documentationUrl: https://docs.airbyte.io/integrations/sources/db2
   icon: db2.svg
   sourceType: database
@@ -432,7 +432,7 @@
 - name: Microsoft SQL Server (MSSQL)
   sourceDefinitionId: b5ea17b1-f170-46dc-bc31-cc744ca984c1
   dockerRepository: airbyte/source-mssql
-  dockerImageTag: 0.3.15
+  dockerImageTag: 0.3.16
   documentationUrl: https://docs.airbyte.io/integrations/sources/mssql
   icon: mssql.svg
   sourceType: database
@@ -474,7 +474,7 @@
 - name: MySQL
   sourceDefinitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
   dockerRepository: airbyte/source-mysql
-  dockerImageTag: 0.5.4
+  dockerImageTag: 0.5.5
   documentationUrl: https://docs.airbyte.io/integrations/sources/mysql
   icon: mysql.svg
   sourceType: database
@@ -508,7 +508,7 @@
 - name: Oracle DB
   sourceDefinitionId: b39a7370-74c3-45a6-ac3a-380d48520a83
   dockerRepository: airbyte/source-oracle
-  dockerImageTag: 0.3.12
+  dockerImageTag: 0.3.13
   documentationUrl: https://docs.airbyte.io/integrations/sources/oracle
   icon: oracle.svg
   sourceType: database
@@ -578,7 +578,7 @@
 - name: Postgres
   sourceDefinitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
   dockerRepository: airbyte/source-postgres
-  dockerImageTag: 0.4.6
+  dockerImageTag: 0.4.7
   documentationUrl: https://docs.airbyte.io/integrations/sources/postgres
   icon: postgresql.svg
   sourceType: database
@@ -704,7 +704,7 @@
 - name: Snowflake
   sourceDefinitionId: e2d65910-8c8b-40a1-ae7d-ee2416b2bfa2
   dockerRepository: airbyte/source-snowflake
-  dockerImageTag: 0.1.7
+  dockerImageTag: 0.1.8
   documentationUrl: https://docs.airbyte.io/integrations/sources/snowflake
   icon: snowflake.svg
   sourceType: database

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

@@ -1205,7 +1205,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-cockroachdb:0.1.7"
+- dockerImage: "airbyte/source-cockroachdb:0.1.8"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/cockroachdb"
     connectionSpecification:
@@ -3189,7 +3189,7 @@
         - - "client_secret"
         oauthFlowOutputParameters:
         - - "refresh_token"
-- dockerImage: "airbyte/source-db2:0.1.7"
+- dockerImage: "airbyte/source-db2:0.1.8"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/db2"
     connectionSpecification:
@@ -4283,7 +4283,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-mssql:0.3.15"
+- dockerImage: "airbyte/source-mssql:0.3.16"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/mssql"
     connectionSpecification:
@@ -4982,7 +4982,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-mysql:0.5.4"
+- dockerImage: "airbyte/source-mysql:0.5.5"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/mysql"
     connectionSpecification:
@@ -5371,7 +5371,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-oracle:0.3.12"
+- dockerImage: "airbyte/source-oracle:0.3.13"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/oracle"
     connectionSpecification:
@@ -5961,7 +5961,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-postgres:0.4.6"
+- dockerImage: "airbyte/source-postgres:0.4.7"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/sources/postgres"
     connectionSpecification:
@@ -7410,7 +7410,7 @@
         - - "client_secret"
         oauthFlowOutputParameters:
         - - "refresh_token"
-- dockerImage: "airbyte/source-snowflake:0.1.7"
+- dockerImage: "airbyte/source-snowflake:0.1.8"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/snowflake"
     connectionSpecification:

airbyte-db/lib/src/main/java/io/airbyte/db/DataTypeUtils.java

@@ -14,9 +14,13 @@
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.function.Function;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class DataTypeUtils {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(DataTypeUtils.class);
+
   public static final String DATE_FORMAT_PATTERN = "yyyy-MM-dd'T'HH:mm:ss'Z'";
   public static final DateFormat DATE_FORMAT = new SimpleDateFormat(DATE_FORMAT_PATTERN); // Quoted "Z" to indicate UTC, no timezone offset
 
@@ -40,6 +44,25 @@ public static <T> T returnNullIfInvalid(final DataTypeSupplier<T> valueProducer,
     }
   }
 
+  public static String toISO8601StringWithMicroseconds(Instant instant) {
+
+    String dateWithMilliseconds = DATE_FORMAT_WITH_MILLISECONDS.format(Date.from(instant));
+    return dateWithMilliseconds.substring(0, 23) + calculateMicrosecondsString(instant.getNano()) + dateWithMilliseconds.substring(23);
+  }
+
+  private static String calculateMicrosecondsString(int nano) {
+    var microSeconds = (nano / 1000) % 1000;
+    String result;
+    if (microSeconds < 10) {
+      result = "00" + microSeconds;
+    } else if (microSeconds < 100) {
+      result = "0" + microSeconds;
+    } else {
+      result = "" + microSeconds;
+    }
+    return result;
+  }
+
   public static String toISO8601StringWithMilliseconds(final long epochMillis) {
     return DATE_FORMAT_WITH_MILLISECONDS.format(Date.from(Instant.ofEpochMilli(epochMillis)));
   }

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

@@ -19,6 +19,7 @@
 import java.sql.SQLException;
 import java.sql.Timestamp;
 import java.text.ParseException;
+import java.time.Instant;
 import java.util.Collections;
 import java.util.List;
 import java.util.StringJoiner;
@@ -120,9 +121,8 @@ protected void putTime(final ObjectNode node, final String columnName, final Res
 
   protected void putTimestamp(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException {
     // https://www.cis.upenn.edu/~bcpierce/courses/629/jdkdocs/guide/jdbc/getstart/mapping.doc.html
-    final Timestamp t = resultSet.getTimestamp(index);
-    final java.util.Date d = new java.util.Date(t.getTime() + (t.getNanos() / 1000000));
-    node.put(columnName, DataTypeUtils.toISO8601String(d));
+    final Instant instant = resultSet.getTimestamp(index).toInstant();
+    node.put(columnName, DataTypeUtils.toISO8601StringWithMicroseconds(instant));
   }
 
   protected void putBinary(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException {
@@ -143,8 +143,14 @@ protected void setTimestamp(final PreparedStatement preparedStatement, final int
     // Parsing TIME as a TIMESTAMP might potentially break for ClickHouse cause it doesn't expect TIME
     // value in the following format
     try {
-      preparedStatement.setTimestamp(parameterIndex, Timestamp
-          .from(DataTypeUtils.DATE_FORMAT.parse(value).toInstant()));
+      var micro = value.substring(value.lastIndexOf('.') + 1, value.length() - 1);
+      var nanos = micro + "000";
+      var valueWithoutMicros = value.replace("." + micro, "");
+
+      var timestamp = Timestamp
+          .from(DataTypeUtils.DATE_FORMAT.parse(valueWithoutMicros).toInstant());
+      timestamp.setNanos(Integer.parseInt(nanos));
+      preparedStatement.setTimestamp(parameterIndex, timestamp);
     } catch (final ParseException e) {
       throw new RuntimeException(e);
     }

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

@@ -144,8 +144,8 @@ void testSetStatementField() throws SQLException {
       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:00Z");
-      sourceOperations.setStatementField(ps, 15, JDBCType.TIMESTAMP, "2001-09-29T03:00:00Z");
+      sourceOperations.setStatementField(ps, 14, JDBCType.TIME, "1970-01-01T05:00:00.000Z");
+      sourceOperations.setStatementField(ps, 15, JDBCType.TIMESTAMP, "2001-09-29T03:00:00.000Z");
       sourceOperations.setStatementField(ps, 16, JDBCType.BINARY, "61616161");
 
       ps.execute();

airbyte-integrations/bases/debezium/src/main/java/io/airbyte/integrations/debezium/internals/DebeziumConverterUtils.java

@@ -39,7 +39,7 @@ public static String convertDate(final Object input) {
     } else if (input instanceof Duration) {
       return DataTypeUtils.toISO8601String((Duration) input);
     } else if (input instanceof Timestamp) {
-      return DataTypeUtils.toISO8601String(((Timestamp) input).toLocalDateTime());
+      return DataTypeUtils.toISO8601StringWithMicroseconds((((Timestamp) input).toInstant()));
     } else if (input instanceof Number) {
       return DataTypeUtils.toISO8601String(
           new Timestamp(((Number) input).longValue()).toLocalDateTime());

airbyte-integrations/connectors/source-cockroachdb-strict-encrypt/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION source-cockroachdb-strict-encrypt
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.1.4
+LABEL io.airbyte.version=0.1.5
 LABEL io.airbyte.name=airbyte/source-cockroachdb-strict-encrypt

airbyte-integrations/connectors/source-cockroachdb/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION source-cockroachdb
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.1.7
+LABEL io.airbyte.version=0.1.8
 LABEL io.airbyte.name=airbyte/source-cockroachdb

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

@@ -327,8 +327,8 @@ protected void initTests() {
         TestDataHolder.builder()
             .sourceType("timestamp")
             .airbyteType(JsonSchemaType.STRING)
-            .addInsertValues("TIMESTAMP '2004-10-19 10:23:54'", "null")
-            .addExpectedValues("2004-10-19T10:23:54Z", null)
+            .addInsertValues("TIMESTAMP '2004-10-19 10:23:54'", "TIMESTAMP '2004-10-19 10:23:54.123456'", "null")
+            .addExpectedValues("2004-10-19T10:23:54.000000Z", "2004-10-19T10:23:54.123456Z", null)
             .build());
 
     addDataTypeTestData(

airbyte-integrations/connectors/source-db2-strict-encrypt/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION source-db2-strict-encrypt
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.1.4
+LABEL io.airbyte.version=0.1.5
 LABEL io.airbyte.name=airbyte/source-db2-strict-encrypt

airbyte-integrations/connectors/source-db2/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION source-db2
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.1.7
+LABEL io.airbyte.version=0.1.8
 LABEL io.airbyte.name=airbyte/source-db2

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

@@ -312,8 +312,9 @@ protected void initTests() {
             .createTablePatternSql(CREATE_TABLE_SQL)
             .sourceType("TIMESTAMP")
             .airbyteType(JsonSchemaType.STRING)
-            .addInsertValues("null", "'2018-03-22-12.00.00.123'", "'20180322125959'", "'20180101 12:00:59 PM'")
-            .addExpectedValues(null, "2018-03-22T12:00:00Z", "2018-03-22T12:59:59Z", "2018-01-01T12:00:59Z") // milliseconds values are erased
+            .addInsertValues("null", "'2018-03-22-12.00.00.123'", "'2018-03-22-12.00.00.123456'", "'20180322125959'", "'20180101 12:00:59 PM'")
+            .addExpectedValues(null, "2018-03-22T12:00:00.123000Z", "2018-03-22T12:00:00.123456Z", "2018-03-22T12:59:59.000000Z",
+                "2018-01-01T12:00:59.000000Z")
             .build());
   }
 

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

@@ -147,7 +147,7 @@ protected List<TableInfo<CommonField<Datatype>>> discoverInternal(final JdbcData
   }
 
   protected Predicate<JsonNode> excludeNotAccessibleTables(final Set<String> internalSchemas,
-                                                         final Set<JdbcPrivilegeDto> tablesWithSelectGrantPrivilege) {
+                                                           final Set<JdbcPrivilegeDto> tablesWithSelectGrantPrivilege) {
     return jsonNode -> {
       if (tablesWithSelectGrantPrivilege.isEmpty()) {
         return isNotInternalSchema(jsonNode, internalSchemas);
@@ -160,7 +160,8 @@ protected Predicate<JsonNode> excludeNotAccessibleTables(final Set<String> inter
     };
   }
 
-  // needs to override isNotInternalSchema for connectors that override getPrivilegesTableForCurrentUser()
+  // needs to override isNotInternalSchema for connectors that override
+  // getPrivilegesTableForCurrentUser()
   protected boolean isNotInternalSchema(JsonNode jsonNode, Set<String> internalSchemas) {
     return !internalSchemas.contains(jsonNode.get(INTERNAL_SCHEMA_NAME).asText());
   }

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

@@ -723,11 +723,11 @@ void testReadMultipleTablesIncrementally() throws Exception {
   }
 
   // when initial and final cursor fields are the same.
-  private void incrementalCursorCheck(
-                                      final String cursorField,
-                                      final String initialCursorValue,
-                                      final String endCursorValue,
-                                      final List<AirbyteMessage> expectedRecordMessages)
+  protected void incrementalCursorCheck(
+                                        final String cursorField,
+                                        final String initialCursorValue,
+                                        final String endCursorValue,
+                                        final List<AirbyteMessage> expectedRecordMessages)
       throws Exception {
     incrementalCursorCheck(cursorField, cursorField, initialCursorValue, endCursorValue,
         expectedRecordMessages);

airbyte-integrations/connectors/source-mssql-strict-encrypt/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION source-mssql-strict-encrypt
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.1.7
+LABEL io.airbyte.version=0.1.8
 LABEL io.airbyte.name=airbyte/source-mssql-strict-encrypt

airbyte-integrations/connectors/source-mssql/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION source-mssql
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.3.15
+LABEL io.airbyte.version=0.3.16
 LABEL io.airbyte.name=airbyte/source-mssql

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

@@ -340,34 +340,35 @@ protected void initTests() {
             .sourceType("smalldatetime")
             .airbyteType(JsonSchemaType.STRING)
             .addInsertValues("'1900-01-01'", "'2079-06-06'", "null")
-            .addExpectedValues("1900-01-01T00:00:00Z", "2079-06-06T00:00:00Z", null)
+            .addExpectedValues("1900-01-01T00:00:00.000000Z", "2079-06-06T00:00:00.000000Z", null)
             .createTablePatternSql(CREATE_TABLE_SQL)
             .build());
 
     addDataTypeTestData(
         TestDataHolder.builder()
             .sourceType("datetime")
             .airbyteType(JsonSchemaType.STRING)
-            .addInsertValues("'1753-01-01'", "'9999-12-31'", "null")
-            .addExpectedValues("1753-01-01T00:00:00Z", "9999-12-31T00:00:00Z", null)
+            .addInsertValues("'1753-01-01'", "'9999-12-31'", "'9999-12-31T13:00:04Z'",
+                "'9999-12-31T13:00:04.123Z'", "null")
+            .addExpectedValues("1753-01-01T00:00:00.000000Z", "9999-12-31T00:00:00.000000Z", "9999-12-31T13:00:04.000000Z",
+                "9999-12-31T13:00:04.123000Z", null)
             .createTablePatternSql(CREATE_TABLE_SQL)
             .build());
 
     addDataTypeTestData(
         TestDataHolder.builder()
             .sourceType("datetime2")
             .airbyteType(JsonSchemaType.STRING)
-            .addInsertValues("'0001-01-01'", "'9999-12-31'", "null")
-            .addExpectedValues("0001-01-01T00:00:00Z", "9999-12-31T00:00:00Z", null)
+            .addInsertValues("'0001-01-01'", "'9999-12-31'", "'9999-12-31T13:00:04.123456Z'", "null")
+            .addExpectedValues("0001-01-01T00:00:00.000000Z", "9999-12-31T00:00:00.000000Z", "9999-12-31T13:00:04.123456Z", null)
             .createTablePatternSql(CREATE_TABLE_SQL)
             .build());
 
     addDataTypeTestData(
         TestDataHolder.builder()
             .sourceType("time")
             .airbyteType(JsonSchemaType.STRING)
-            .addInsertValues("'00:00:00.0000000'", "'23:59:59.9999999'", "'00:00:00'", "'23:58'",
-                "null")
+            .addInsertValues("'00:00:00.0000000'", "'23:59:59.9999999'", "'00:00:00'", "'23:58'", "null")
             .addExpectedValues("00:00:00", "23:59:59.9999999", "00:00:00", "23:58:00", null)
             .createTablePatternSql(CREATE_TABLE_SQL)
             .build());