🐛 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]>
DonnyZhao · Nov 5, 2021 · 56db806 · 56db806
1 parent af1c62f
commit 56db806
Show file tree

Hide file tree

Showing 6 changed files with 72 additions and 3 deletions.
diff --git a/...ain/resources/config/STANDARD_SOURCE_DEFINITION/b5ea17b1-f170-46dc-bc31-cc744ca984c1.json b/...ain/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"
 }
diff --git a/...ses/debezium/src/main/java/io/airbyte/integrations/debezium/internals/MSSQLConverter.java b/...ses/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();
+    });
+  }
+
+}
diff --git a/airbyte-integrations/connectors/source-mssql-strict-encrypt/Dockerfile b/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
diff --git a/airbyte-integrations/connectors/source-mssql/Dockerfile b/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
diff --git a/...s/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlCdcProperties.java b/...s/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;
   }
 

diff --git a/docs/integrations/sources/mssql.md b/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 |  |