🐛 Source MSSQL: fix data type (smalldatetime, smallmoney) conversion from mssql source (airbytehq#5609) (airbytehq#7386)

* Fix data type (smalldatetime, smallmoney) conversion from mssql source (airbytehq#5609)

* Fixed code format

* Bumb new version

* Update documentation (mssql.md)

* formating

* fixed converter properties

* aligned converter utils with airbytehq#7339

Co-authored-by: Andrii Leonets <[email protected]>

Author: 2 people

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.6",
+  "dockerImageTag": "0.3.8",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/mssql",
   "icon": "mssql.svg"
 }

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

@@ -0,0 +1,65 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.debezium.internals;
+
+import io.debezium.spi.converter.CustomConverter;
+import io.debezium.spi.converter.RelationalColumn;
+import java.math.BigDecimal;
+import java.util.Objects;
+import java.util.Properties;
+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 String SMALLDATETIME_TYPE = "SMALLDATETIME";
+  private final String SMALLMONEY_TYPE = "SMALLMONEY";
+
+  @Override
+  public void configure(Properties props) {}
+
+  @Override
+  public void converterFor(final RelationalColumn field,
+                           final ConverterRegistration<SchemaBuilder> registration) {
+    if (SMALLDATETIME_TYPE.equalsIgnoreCase(field.typeName())) {
+      registerDate(field, registration);
+    } else if (SMALLMONEY_TYPE.equalsIgnoreCase(field.typeName())) {
+      registerMoney(field, registration);
+    }
+
+  }
+
+  private void registerDate(final RelationalColumn field,
+                            final ConverterRegistration<SchemaBuilder> registration) {
+    registration.register(SchemaBuilder.string(), input -> {
+      if (Objects.isNull(input)) {
+        return DebeziumConverterUtils.convertDefaultValue(field);
+      }
+
+      return DebeziumConverterUtils.convertDate(input);
+    });
+  }
+
+  private void registerMoney(final RelationalColumn field,
+                             final ConverterRegistration<SchemaBuilder> registration) {
+    registration.register(SchemaBuilder.float64(), input -> {
+      if (Objects.isNull(input)) {
+        return DebeziumConverterUtils.convertDefaultValue(field);
+      }
+
+      if (input instanceof BigDecimal) {
+        return ((BigDecimal) input).doubleValue();
+      }
+
+      LOGGER.warn("Uncovered money class type '{}'. Use default converter",
+          input.getClass().getName());
+      return input.toString();
+    });
+  }
+
+}

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

@@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
 
 RUN tar xf ${APPLICATION}.tar --strip-components=1
 
-LABEL io.airbyte.version=0.1.0
+LABEL io.airbyte.version=0.1.1
 LABEL io.airbyte.name=airbyte/source-mssql-strict-encrypt

airbyte-integrations/connectors/source-mssql/Dockerfile

@@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
 
 RUN tar xf ${APPLICATION}.tar --strip-components=1
 
-LABEL io.airbyte.version=0.3.6
+LABEL io.airbyte.version=0.3.8
 LABEL io.airbyte.name=airbyte/source-mssql

airbyte-integrations/connectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlCdcProperties.java

@@ -26,6 +26,9 @@ static Properties getDebeziumProperties() {
     // https://debezium.io/documentation/reference/1.4/connectors/sqlserver.html#sqlserver-property-provide-transaction-metadata
     props.setProperty("provide.transaction.metadata", "false");
 
+    props.setProperty("converters", "mssql_converter");
+    props.setProperty("mssql_converter.type", "io.airbyte.integrations.debezium.internals.MSSQLConverter");
+
     return props;
   }
 

docs/integrations/sources/mssql.md

@@ -294,6 +294,7 @@ If you do not see a type in this list, assume that it is coerced into a string.
 
 | Version | Date | Pull Request | Subject |  |
 | :--- | :--- | :--- | :--- | :--- |
+| 0.3.8 | 2021-10-26 | [7386](https://github.com/airbytehq/airbyte/pull/7386) | Fixed data type (smalldatetime, smallmoney) conversion from mssql source |  |
 | 0.3.7 | 2021-09-30 | [6585](https://github.com/airbytehq/airbyte/pull/6585) | Improved SSH Tunnel key generation steps |  |
 | 0.3.6 | 2021-09-17 | [6318](https://github.com/airbytehq/airbyte/pull/6318) | Added option to connect to DB via SSH |  |
 | 0.3.4 | 2021-08-13 | [4699](https://github.com/airbytehq/airbyte/pull/4699) | Added json config validator |  |