Destinations CDK: Refactor T+D to gather required world state upfront (…

…airbytehq#35342)

Signed-off-by: Gireesh Sreepathi <[email protected]>

airbyte-cdk/java/airbyte-cdk/core/src/main/java/io/airbyte/cdk/integrations/util/ConnectorExceptionUtil.java

@@ -10,6 +10,7 @@
 import io.airbyte.cdk.integrations.base.errors.messages.ErrorMessage;
 import io.airbyte.commons.exceptions.ConfigErrorException;
 import io.airbyte.commons.exceptions.ConnectionErrorException;
+import io.airbyte.commons.functional.Either;
 import java.sql.SQLException;
 import java.sql.SQLSyntaxErrorException;
 import java.util.Collection;
@@ -85,6 +86,17 @@ public static <T extends Throwable> void logAllAndThrowFirst(final String initia
     }
   }
 
+  public static <T extends Throwable, Result> List<Result> getResultsOrLogAndThrowFirst(final String initialMessage,
+                                                                                        final List<Either<? extends T, Result>> eithers)
+      throws T {
+    List<? extends T> throwables = eithers.stream().filter(Either::isLeft).map(Either::getLeft).toList();
+    if (!throwables.isEmpty()) {
+      logAllAndThrowFirst(initialMessage, throwables);
+    }
+    // No need to filter on isRight since isLeft will throw before reaching this line.
+    return eithers.stream().map(Either::getRight).toList();
+  }
+
   private static boolean isConfigErrorException(Throwable e) {
     return e instanceof ConfigErrorException;
   }

airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties

@@ -1 +1 @@
-version=0.22.1
+version=0.23.0

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/java/io/airbyte/cdk/integrations/destination/jdbc/AbstractJdbcDestination.java

@@ -252,9 +252,7 @@ private void assertCustomParametersDontOverwriteDefaultParameters(final Map<Stri
 
   protected abstract JdbcSqlGenerator getSqlGenerator();
 
-  protected JdbcDestinationHandler getDestinationHandler(final String databaseName, final JdbcDatabase database) {
-    return new JdbcDestinationHandler(databaseName, database);
-  }
+  protected abstract JdbcDestinationHandler getDestinationHandler(final String databaseName, final JdbcDatabase database);
 
   /**
    * "database" key at root of the config json, for any other variants in config, override this
@@ -318,14 +316,14 @@ private TyperDeduper getV2TyperDeduper(final JsonNode config, final ConfiguredAi
     final String databaseName = getDatabaseName(config);
     final var migrator = new JdbcV1V2Migrator(namingResolver, database, databaseName);
     final NoopV2TableMigrator v2TableMigrator = new NoopV2TableMigrator();
-    final DestinationHandler<TableDefinition> destinationHandler = getDestinationHandler(databaseName, database);
+    final DestinationHandler destinationHandler = getDestinationHandler(databaseName, database);
     final boolean disableTypeDedupe = !config.has(DISABLE_TYPE_DEDUPE) || config.get(DISABLE_TYPE_DEDUPE).asBoolean(false);
     final TyperDeduper typerDeduper;
     if (disableTypeDedupe) {
-      typerDeduper = new NoOpTyperDeduperWithV1V2Migrations<>(sqlGenerator, destinationHandler, parsedCatalog, migrator, v2TableMigrator);
+      typerDeduper = new NoOpTyperDeduperWithV1V2Migrations(sqlGenerator, destinationHandler, parsedCatalog, migrator, v2TableMigrator);
     } else {
       typerDeduper =
-          new DefaultTyperDeduper<>(sqlGenerator, destinationHandler, parsedCatalog, migrator, v2TableMigrator);
+          new DefaultTyperDeduper(sqlGenerator, destinationHandler, parsedCatalog, migrator, v2TableMigrator);
     }
     return typerDeduper;
   }

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/java/io/airbyte/cdk/integrations/destination/jdbc/ColumnDefinition.java

@@ -4,8 +4,13 @@
 
 package io.airbyte.cdk.integrations.destination.jdbc;
 
-import java.sql.SQLType;
-
-public record ColumnDefinition(String name, String type, SQLType sqlType, int columnSize) {
+/**
+ * Jdbc destination column definition representation
+ *
+ * @param name
+ * @param type
+ * @param columnSize
+ */
+public record ColumnDefinition(String name, String type, int columnSize, boolean isNullable) {
 
 }

airbyte-cdk/java/airbyte-cdk/db-destinations/src/main/java/io/airbyte/cdk/integrations/destination/jdbc/TableDefinition.java

@@ -7,7 +7,7 @@
 import java.util.LinkedHashMap;
 
 /**
- * Jdbc destination table definition representation
+ * Jdbc destination table definition representation with a map of column names to column definitions
  *
  * @param columns
  */