🐛Source-mssql: added support for missed data types (airbytehq#9094)

* [7728] Source-mssql: added support for missed data types

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/b5ea17b1-f170-46dc-bc31-cc744ca984c1.json

@@ -2,7 +2,7 @@
   "sourceDefinitionId": "b5ea17b1-f170-46dc-bc31-cc744ca984c1",
   "name": "Microsoft SQL Server (MSSQL)",
   "dockerRepository": "airbyte/source-mssql",
-  "dockerImageTag": "0.3.12",
+  "dockerImageTag": "0.3.13",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/mssql",
   "icon": "mssql.svg"
 }

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

@@ -398,7 +398,7 @@
 - name: Microsoft SQL Server (MSSQL)
   sourceDefinitionId: b5ea17b1-f170-46dc-bc31-cc744ca984c1
   dockerRepository: airbyte/source-mssql
-  dockerImageTag: 0.3.12
+  dockerImageTag: 0.3.13
   documentationUrl: https://docs.airbyte.io/integrations/sources/mssql
   icon: mssql.svg
   sourceType: database

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

@@ -3846,7 +3846,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-mssql:0.3.12"
+- dockerImage: "airbyte/source-mssql:0.3.13"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/mssql"
     connectionSpecification:

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

@@ -52,6 +52,10 @@ public static String toISO8601String(final java.util.Date date) {
     return DATE_FORMAT.format(date);
   }
 
+  public static String toISOTimeString(final LocalDateTime dateTime) {
+    return DateTimeFormatter.ISO_TIME.format(dateTime.toLocalTime());
+  }
+
   public static String toISO8601String(final LocalDate date) {
     return toISO8601String(date.atStartOfDay());
   }

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

@@ -13,6 +13,7 @@
 import java.sql.DatabaseMetaData;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.util.List;
 import java.util.Spliterator;
@@ -166,6 +167,21 @@ public Stream<JsonNode> query(final String sql, final String... params) throws S
     }, sourceOperations::rowToJson);
   }
 
+  public ResultSetMetaData queryMetadata(final String sql, final String... params) throws SQLException {
+    try (final Stream<ResultSetMetaData> q = query(c -> {
+      PreparedStatement statement = c.prepareStatement(sql);
+      int i = 1;
+      for (String param : params) {
+        statement.setString(i, param);
+        ++i;
+      }
+      return statement;
+    },
+        ResultSet::getMetaData)) {
+      return q.findFirst().orElse(null);
+    }
+  }
+
   public abstract DatabaseMetaData getMetaData() throws SQLException;
 
 }

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

@@ -4,20 +4,24 @@
 
 package io.airbyte.integrations.debezium.internals;
 
+import io.airbyte.db.DataTypeUtils;
 import io.debezium.spi.converter.CustomConverter;
 import io.debezium.spi.converter.RelationalColumn;
 import java.math.BigDecimal;
+import java.sql.Timestamp;
 import java.util.Objects;
 import java.util.Properties;
+import java.util.Set;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class MSSQLConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {
 
-  private final Logger LOGGER = LoggerFactory.getLogger(MSSQLConverter.class);;
+  private final Logger LOGGER = LoggerFactory.getLogger(MSSQLConverter.class);
 
-  private final String SMALLDATETIME_TYPE = "SMALLDATETIME";
+  private final Set<String> DATE_TYPES = Set.of("DATE", "DATETIME", "DATETIME2", "DATETIMEOFFSET", "SMALLDATETIME");
+  private final String TIME_TYPE = "TIME";
   private final String SMALLMONEY_TYPE = "SMALLMONEY";
 
   @Override
@@ -26,12 +30,13 @@ public void configure(Properties props) {}
   @Override
   public void converterFor(final RelationalColumn field,
                            final ConverterRegistration<SchemaBuilder> registration) {
-    if (SMALLDATETIME_TYPE.equalsIgnoreCase(field.typeName())) {
+    if (DATE_TYPES.contains(field.typeName().toUpperCase())) {
       registerDate(field, registration);
     } else if (SMALLMONEY_TYPE.equalsIgnoreCase(field.typeName())) {
       registerMoney(field, registration);
+    } else if (TIME_TYPE.equalsIgnoreCase(field.typeName())) {
+      registerTime(field, registration);
     }
-
   }
 
   private void registerDate(final RelationalColumn field,
@@ -45,6 +50,23 @@ private void registerDate(final RelationalColumn field,
     });
   }
 
+  private void registerTime(final RelationalColumn field,
+                            final ConverterRegistration<SchemaBuilder> registration) {
+    registration.register(SchemaBuilder.string(), input -> {
+      if (Objects.isNull(input)) {
+        return DebeziumConverterUtils.convertDefaultValue(field);
+      }
+
+      if (input instanceof Timestamp) {
+        return DataTypeUtils.toISOTimeString(((Timestamp) input).toLocalDateTime());
+      }
+
+      LOGGER.warn("Uncovered time class type '{}'. Use default converter",
+          input.getClass().getName());
+      return input.toString();
+    });
+  }
+
   private void registerMoney(final RelationalColumn field,
                              final ConverterRegistration<SchemaBuilder> registration) {
     registration.register(SchemaBuilder.float64(), input -> {

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.5
+LABEL io.airbyte.version=0.1.6
 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.12
+LABEL io.airbyte.version=0.3.13
 LABEL io.airbyte.name=airbyte/source-mssql