[source-postgres] : Remove legacy bad values handling code (#37445)

airbytehq · Apr 22, 2024 · 9b9ec1c · 9b9ec1c
1 parent 01381ae
commit 9b9ec1c
Show file tree

Hide file tree

Showing 8 changed files with 267 additions and 240 deletions.
diff --git a/airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/db/DataTypeUtils.kt b/airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/db/DataTypeUtils.kt
@@ -70,6 +70,30 @@ object DataTypeUtils {
         }
     }
 
+    @JvmStatic
+    fun <T> throwExceptionIfInvalid(valueProducer: DataTypeSupplier<T>): T? {
+        return throwExceptionIfInvalid(valueProducer, Function { _: T? -> true })
+    }
+
+    @JvmStatic
+    fun <T> throwExceptionIfInvalid(
+        valueProducer: DataTypeSupplier<T>,
+        isValidFn: Function<T?, Boolean>
+    ): T? {
+        // Some edge case values (e.g: Infinity, NaN) have no java or JSON equivalent, and will
+        // throw an
+        // exception when parsed. We want to parse those
+        // values as null.
+        // This method reduces error handling boilerplate.
+        try {
+            val value = valueProducer.apply()
+            return if (isValidFn.apply(value)) value
+            else throw SQLException("Given value is not valid.")
+        } catch (e: SQLException) {
+            return null
+        }
+    }
+
     @JvmStatic
     fun toISO8601StringWithMicroseconds(instant: Instant): String {
         val dateWithMilliseconds = dateFormatMillisPattern.format(Date.from(instant))

diff --git a/...cdk/core/src/main/kotlin/io/airbyte/cdk/db/jdbc/AbstractJdbcCompatibleSourceOperations.kt b/...cdk/core/src/main/kotlin/io/airbyte/cdk/db/jdbc/AbstractJdbcCompatibleSourceOperations.kt
@@ -46,6 +46,7 @@ abstract class AbstractJdbcCompatibleSourceOperations<Datatype> :
                 // convert to java types that will convert into reasonable json.
                 copyToJsonField(queryContext, i, jsonNode)
             } catch (e: java.lang.Exception) {
+                jsonNode.putNull(columnName)
                 LOGGER.info(
                     "Failed to serialize column: {}, of type {}, with error {}",
                     columnName,

diff --git a/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties b/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties
@@ -1 +1 @@
-version=0.30.6
+version=0.30.7
diff --git a/airbyte-integrations/connectors/source-postgres/build.gradle b/airbyte-integrations/connectors/source-postgres/build.gradle
@@ -12,7 +12,7 @@ java {
 }
 
 airbyteJavaConnector {
-    cdkVersionRequired = '0.29.13'
+    cdkVersionRequired = '0.30.7'
     features = ['db-sources', 'datastore-postgres']
     useLocalCdk = false
 }

diff --git a/airbyte-integrations/connectors/source-postgres/metadata.yaml b/airbyte-integrations/connectors/source-postgres/metadata.yaml
@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
-  dockerImageTag: 3.3.26
+  dockerImageTag: 3.3.27
   dockerRepository: airbyte/source-postgres
   documentationUrl: https://docs.airbyte.com/integrations/sources/postgres
   githubIssueLabel: source-postgres

diff --git a/...tgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresSourceOperations.java b/...tgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresSourceOperations.java
@@ -347,7 +347,7 @@ private void putBigDecimalArray(final ObjectNode node, final String columnName,
     final ArrayNode arrayNode = Jsons.arrayNode();
     final ResultSet arrayResultSet = resultSet.getArray(colIndex).getResultSet();
     while (arrayResultSet.next()) {
-      final BigDecimal bigDecimal = DataTypeUtils.returnNullIfInvalid(() -> arrayResultSet.getBigDecimal(2));
+      final BigDecimal bigDecimal = DataTypeUtils.throwExceptionIfInvalid(() -> arrayResultSet.getBigDecimal(2));
       if (bigDecimal != null) {
         arrayNode.add(bigDecimal);
       } else {
@@ -361,7 +361,7 @@ private void putBigIntArray(final ObjectNode node, final String columnName, fina
     final ArrayNode arrayNode = Jsons.arrayNode();
     final ResultSet arrayResultSet = resultSet.getArray(colIndex).getResultSet();
     while (arrayResultSet.next()) {
-      final long value = DataTypeUtils.returnNullIfInvalid(() -> arrayResultSet.getLong(2));
+      final long value = DataTypeUtils.throwExceptionIfInvalid(() -> arrayResultSet.getLong(2));
       arrayNode.add(value);
     }
     node.set(columnName, arrayNode);
@@ -371,7 +371,7 @@ private void putDoubleArray(final ObjectNode node, final String columnName, fina
     final ArrayNode arrayNode = Jsons.arrayNode();
     final ResultSet arrayResultSet = resultSet.getArray(colIndex).getResultSet();
     while (arrayResultSet.next()) {
-      arrayNode.add(DataTypeUtils.returnNullIfInvalid(() -> arrayResultSet.getDouble(2), Double::isFinite));
+      arrayNode.add(DataTypeUtils.throwExceptionIfInvalid(() -> arrayResultSet.getDouble(2), Double::isFinite));
     }
     node.set(columnName, arrayNode);
   }
@@ -381,7 +381,8 @@ private void putMoneyArray(final ObjectNode node, final String columnName, final
     final ResultSet arrayResultSet = resultSet.getArray(colIndex).getResultSet();
     while (arrayResultSet.next()) {
       final String moneyValue = parseMoneyValue(arrayResultSet.getString(2));
-      arrayNode.add(DataTypeUtils.returnNullIfInvalid(() -> DataTypeUtils.returnNullIfInvalid(() -> Double.valueOf(moneyValue), Double::isFinite)));
+      arrayNode.add(
+          DataTypeUtils.throwExceptionIfInvalid(() -> DataTypeUtils.throwExceptionIfInvalid(() -> Double.valueOf(moneyValue), Double::isFinite)));
     }
     node.set(columnName, arrayNode);
   }
@@ -612,7 +613,7 @@ protected <T extends PGobject> void putObject(final ObjectNode node,
 
   @Override
   protected void putBigDecimal(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) {
-    final BigDecimal bigDecimal = DataTypeUtils.returnNullIfInvalid(() -> resultSet.getBigDecimal(index));
+    final BigDecimal bigDecimal = DataTypeUtils.throwExceptionIfInvalid(() -> resultSet.getBigDecimal(index));
     if (bigDecimal != null) {
       node.put(columnName, bigDecimal);
     } else {
@@ -633,7 +634,7 @@ protected void putDouble(final ObjectNode node, final String columnName, final R
 
   private void putMoney(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index) throws SQLException {
     final String moneyValue = parseMoneyValue(resultSet.getString(index));
-    node.put(columnName, DataTypeUtils.returnNullIfInvalid(() -> Double.valueOf(moneyValue), Double::isFinite));
+    node.put(columnName, DataTypeUtils.throwExceptionIfInvalid(() -> Double.valueOf(moneyValue), Double::isFinite));
   }
 
   private void putHstoreAsJson(final ObjectNode node, final String columnName, final ResultSet resultSet, final int index)

diff --git a/...tegrations/io/airbyte/integration_tests/sources/CdcWalLogsPostgresSourceDatatypeTest.java b/...tegrations/io/airbyte/integration_tests/sources/CdcWalLogsPostgresSourceDatatypeTest.java
@@ -36,7 +36,7 @@ protected List<AirbyteMessage> runRead(final ConfiguredAirbyteCatalog configured
   @Override
   protected void postSetup() throws Exception {
     final Database database = setupDatabase();
-    for (final TestDataHolder test : getTestDataHolders()) {
+    for (final TestDataHolder test : testDataHolders) {
       database.query(ctx -> {
         ctx.fetch(test.getCreateSqlQuery());
         return null;
@@ -56,7 +56,7 @@ protected void postSetup() throws Exception {
     if (stateAfterFirstSync == null) {
       throw new RuntimeException("stateAfterFirstSync should not be null");
     }
-    for (final TestDataHolder test : getTestDataHolders()) {
+    for (final TestDataHolder test : testDataHolders) {
       database.query(ctx -> {
         test.getInsertSqlQueries().forEach(ctx::fetch);
         return null;