🐛 Fix data type tests in CdcPostgresSourceDatatypeTest

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/decd338e-5647-4c0b-adf4-da0e75f5a750.json

@@ -2,7 +2,7 @@
   "sourceDefinitionId": "decd338e-5647-4c0b-adf4-da0e75f5a750",
   "name": "Postgres",
   "dockerRepository": "airbyte/source-postgres",
-  "dockerImageTag": "0.3.9",
+  "dockerImageTag": "0.3.13",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/postgres",
   "icon": "postgresql.svg"
 }

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

@@ -59,7 +59,7 @@
 - sourceDefinitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
   name: Postgres
   dockerRepository: airbyte/source-postgres
-  dockerImageTag: 0.3.11
+  dockerImageTag: 0.3.13
   documentationUrl: https://docs.airbyte.io/integrations/sources/postgres
   icon: postgresql.svg
   sourceType: database

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

@@ -0,0 +1,61 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.debezium.internals;
+
+import io.airbyte.db.DataTypeUtils;
+import io.debezium.spi.converter.RelationalColumn;
+import java.sql.Timestamp;
+import java.time.Duration;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeParseException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DebeziumConverterUtils {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(DebeziumConverterUtils.class);
+
+  public static String convertDate(Object x) {
+    /**
+     * While building this custom converter we were not sure what type debezium could return cause there
+     * is no mention of it in the documentation. Secondly if you take a look at
+     * {@link io.debezium.connector.mysql.converters.TinyIntOneToBooleanConverter#converterFor(io.debezium.spi.converter.RelationalColumn, io.debezium.spi.converter.CustomConverter.ConverterRegistration)}
+     * method, even it is handling multiple data types but its not clear under what circumstances which
+     * data type would be returned. I just went ahead and handled the data types that made sense.
+     * Secondly, we use LocalDateTime to handle this cause it represents DATETIME datatype in JAVA
+     */
+    if (x instanceof LocalDateTime) {
+      return DataTypeUtils.toISO8601String((LocalDateTime) x);
+    } else if (x instanceof LocalDate) {
+      return DataTypeUtils.toISO8601String((LocalDate) x);
+    } else if (x instanceof Duration) {
+      return DataTypeUtils.toISO8601String((Duration) x);
+    } else if (x instanceof Timestamp) {
+      return DataTypeUtils.toISO8601String(((Timestamp) x).toLocalDateTime());
+    } else if (x instanceof Number) {
+      return DataTypeUtils.toISO8601String(new Timestamp(((Number) x).longValue()).toLocalDateTime());
+    } else if (x instanceof String) {
+      try {
+        return LocalDateTime.parse((String) x).toString();
+      } catch (final DateTimeParseException e) {
+        LOGGER.warn("Cannot convert value '{}' to LocalDateTime type", x);
+        return x.toString();
+      }
+    }
+    LOGGER.warn("Uncovered date class type '{}'. Use default converter", x.getClass().getName());
+    return x.toString();
+  }
+
+  public static Object getDefaultValue(RelationalColumn field) {
+    if (field.isOptional()) {
+      return null;
+    } else if (field.hasDefaultValue()) {
+      return field.defaultValue();
+    }
+    return null;
+  }
+
+}

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

@@ -4,14 +4,8 @@
 
 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.sql.Timestamp;
-import java.time.Duration;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.format.DateTimeParseException;
 import java.util.Arrays;
 import java.util.Properties;
 import org.apache.kafka.connect.data.SchemaBuilder;
@@ -61,50 +55,15 @@ private void registerText(final RelationalColumn field, final ConverterRegistrat
 
       if (x instanceof byte[]) {
         return new String((byte[]) x);
-      } else
+      } else {
         return x.toString();
+      }
     });
   }
 
   private void registerDate(final RelationalColumn field, final ConverterRegistration<SchemaBuilder> registration) {
-    registration.register(SchemaBuilder.string(), x -> {
-      if (x == null) {
-        if (field.isOptional()) {
-          return null;
-        } else if (field.hasDefaultValue()) {
-          return field.defaultValue();
-        }
-        return null;
-      }
-      /**
-       * While building this custom converter we were not sure what type debezium could return cause there
-       * is no mention of it in the documentation. Secondly if you take a look at
-       * {@link io.debezium.connector.mysql.converters.TinyIntOneToBooleanConverter#converterFor(RelationalColumn, ConverterRegistration)}
-       * method, even it is handling multiple data types but its not clear under what circumstances which
-       * data type would be returned. I just went ahead and handled the data types that made sense.
-       * Secondly, we use LocalDateTime to handle this cause it represents DATETIME datatype in JAVA
-       */
-      if (x instanceof LocalDateTime) {
-        return DataTypeUtils.toISO8601String((LocalDateTime) x);
-      } else if (x instanceof LocalDate) {
-        return DataTypeUtils.toISO8601String((LocalDate) x);
-      } else if (x instanceof Duration) {
-        return DataTypeUtils.toISO8601String((Duration) x);
-      } else if (x instanceof Timestamp) {
-        return DataTypeUtils.toISO8601String(((Timestamp) x).toLocalDateTime());
-      } else if (x instanceof Number) {
-        return DataTypeUtils.toISO8601String(new Timestamp(((Number) x).longValue()).toLocalDateTime());
-      } else if (x instanceof String) {
-        try {
-          return LocalDateTime.parse((String) x).toString();
-        } catch (final DateTimeParseException e) {
-          LOGGER.warn("Cannot convert value '{}' to LocalDateTime type", x);
-          return x.toString();
-        }
-      }
-      LOGGER.warn("Uncovered date class type '{}'. Use default converter", x.getClass().getName());
-      return x.toString();
-    });
+    registration.register(SchemaBuilder.string(),
+        x -> x == null ? DebeziumConverterUtils.getDefaultValue(field) : DebeziumConverterUtils.convertDate(x));
   }
 
 }

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

@@ -0,0 +1,132 @@
+/*
+ * 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.Arrays;
+import java.util.Properties;
+import org.apache.kafka.connect.data.SchemaBuilder;
+import org.postgresql.util.PGInterval;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PostgresConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(PostgresConverter.class);
+
+  private final String[] DATE_TYPES = {"DATE", "DATETIME", "TIME", "TIMETZ", "INTERVAL", "TIMESTAMP"};
+  private final String[] BIT_TYPES = {"BIT", "VARBIT"};
+  private final String[] MONEY_ITEM_TYPE = {"MONEY"};
+  private final String[] GEOMETRICS_TYPES = {"BOX", "CIRCLE", "LINE", "LSEG", "POINT", "POLYGON", "PATH"};
+  private final String[] TEXT_TYPES = {"VARCHAR", "VARBINARY", "BLOB", "TEXT", "LONGTEXT", "TINYTEXT", "MEDIUMTEXT", "INVENTORY_ITEM", "TSVECTOR"};
+
+  @Override
+  public void configure(Properties props) {}
+
+  @Override
+  public void converterFor(RelationalColumn field, ConverterRegistration<SchemaBuilder> registration) {
+    if (Arrays.stream(DATE_TYPES).anyMatch(s -> s.equalsIgnoreCase(field.typeName()))) {
+      registerDate(field, registration);
+    } else if (Arrays.stream(TEXT_TYPES).anyMatch(s -> s.equalsIgnoreCase(field.typeName()))
+        || Arrays.stream(GEOMETRICS_TYPES).anyMatch(s -> s.equalsIgnoreCase(field.typeName()))
+        || Arrays.stream(BIT_TYPES).anyMatch(s -> s.equalsIgnoreCase(field.typeName()))) {
+      registerText(field, registration);
+    } else if (Arrays.stream(MONEY_ITEM_TYPE).anyMatch(s -> s.equalsIgnoreCase(field.typeName()))) {
+      registerMoney(field, registration);
+    }
+  }
+
+  private void registerText(RelationalColumn field, ConverterRegistration<SchemaBuilder> registration) {
+    registration.register(SchemaBuilder.string(), x -> {
+      if (x == null) {
+        return DebeziumConverterUtils.getDefaultValue(field);
+      }
+
+      if (x instanceof byte[]) {
+        return new String((byte[]) x);
+      } else {
+        return x.toString();
+      }
+    });
+  }
+
+  private void registerDate(RelationalColumn field, ConverterRegistration<SchemaBuilder> registration) {
+    registration.register(SchemaBuilder.string(), x -> {
+      if (x == null) {
+        return DebeziumConverterUtils.getDefaultValue(field);
+      } else if (x instanceof PGInterval) {
+        return convertInterval((PGInterval) x);
+      } else {
+        return DebeziumConverterUtils.convertDate(x);
+      }
+    });
+  }
+
+  private String convertInterval(PGInterval pgInterval) {
+    StringBuilder resultInterval = new StringBuilder();
+    formatDateUnit(resultInterval, pgInterval.getYears(), " year ");
+    formatDateUnit(resultInterval, pgInterval.getMonths(), " mons ");
+    formatDateUnit(resultInterval, pgInterval.getDays(), " days ");
+
+    formatTimeValues(resultInterval, pgInterval);
+    return resultInterval.toString();
+  }
+
+  private void registerMoney(RelationalColumn field, ConverterRegistration<SchemaBuilder> registration) {
+    registration.register(SchemaBuilder.string(), x -> {
+      if (x == null) {
+        return DebeziumConverterUtils.getDefaultValue(field);
+      } else if (x instanceof Double) {
+        BigDecimal result = BigDecimal.valueOf((Double) x);
+        if (result.compareTo(new BigDecimal("999999999999999")) == 1
+            || result.compareTo(new BigDecimal("-999999999999999")) == -1) {
+          return null;
+        }
+        return result.toString();
+      } else {
+        return x.toString();
+      }
+    });
+  }
+
+  private void formatDateUnit(StringBuilder resultInterval, int dateUnit, String s) {
+    if (dateUnit != 0) {
+      resultInterval
+          .append(dateUnit)
+          .append(s);
+    }
+  }
+
+  private void formatTimeValues(StringBuilder resultInterval, PGInterval pgInterval) {
+    if (isNegativeTime(pgInterval)) {
+      resultInterval.append("-");
+    }
+    // TODO check if value more or less than Integer.MIN_VALUE Integer.MAX_VALUE,
+    int hours = Math.abs(pgInterval.getHours());
+    int minutes = Math.abs(pgInterval.getMinutes());
+    int seconds = Math.abs(pgInterval.getWholeSeconds());
+    resultInterval.append(addFirstDigit(hours));
+    resultInterval.append(hours);
+    resultInterval.append(":");
+    resultInterval.append(addFirstDigit(minutes));
+    resultInterval.append(minutes);
+    resultInterval.append(":");
+    resultInterval.append(addFirstDigit(seconds));
+    resultInterval.append(seconds);
+  }
+
+  private String addFirstDigit(int hours) {
+    return hours <= 9 ? "0" : "";
+  }
+
+  private boolean isNegativeTime(PGInterval pgInterval) {
+    return pgInterval.getHours() < 0
+        || pgInterval.getMinutes() < 0
+        || pgInterval.getWholeSeconds() < 0;
+  }
+
+}

airbyte-integrations/connectors/source-postgres/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.11
+LABEL io.airbyte.version=0.3.13
 LABEL io.airbyte.name=airbyte/source-postgres

airbyte-integrations/connectors/source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresCdcProperties.java

@@ -20,6 +20,9 @@ static Properties getDebeziumProperties(final JsonNode config) {
 
     props.setProperty("publication.autocreate.mode", "disabled");
 
+    props.setProperty("converters", "datetime");
+    props.setProperty("datetime.type", "io.airbyte.integrations.debezium.internals.PostgresConverter");
+
     return props;
   }
 

airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/CdcPostgresSourceAcceptanceTest.java

@@ -70,6 +70,7 @@ protected void setupEnvironment(final TestDestinationEnv environment) throws Exc
         .put("username", container.getUsername())
         .put("password", container.getPassword())
         .put("replication_method", replicationMethod)
+        .put("ssl", false)
         .build());
 
     final Database database = Databases.createDatabase(