maybe move everything to new classes?

Author: edgao

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

@@ -23,7 +23,6 @@
 import io.airbyte.cdk.integrations.destination.NamingConventionTransformer;
 import io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.JdbcDestinationHandler;
 import io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.JdbcSqlGenerator;
-import io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.JdbcV1V2Migrator;
 import io.airbyte.cdk.integrations.destination_async.partial_messages.PartialAirbyteMessage;
 import io.airbyte.cdk.integrations.destination_async.partial_messages.PartialAirbyteRecordMessage;
 import io.airbyte.commons.exceptions.ConnectionErrorException;
@@ -34,9 +33,9 @@
 import io.airbyte.integrations.base.destination.typing_deduping.DestinationHandler;
 import io.airbyte.integrations.base.destination.typing_deduping.NoOpTyperDeduperWithV1V2Migrations;
 import io.airbyte.integrations.base.destination.typing_deduping.NoopTyperDeduper;
-import io.airbyte.integrations.base.destination.typing_deduping.NoopV2TableMigrator;
 import io.airbyte.integrations.base.destination.typing_deduping.ParsedCatalog;
 import io.airbyte.integrations.base.destination.typing_deduping.TyperDeduper;
+import io.airbyte.integrations.base.destination.typing_deduping.migrators.Migration;
 import io.airbyte.integrations.base.destination.typing_deduping.migrators.MinimumDestinationState;
 import io.airbyte.protocol.models.v0.AirbyteConnectionStatus;
 import io.airbyte.protocol.models.v0.AirbyteConnectionStatus.Status;
@@ -54,7 +53,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public abstract class AbstractJdbcDestination extends JdbcConnector implements Destination {
+public abstract class AbstractJdbcDestination<DestinationState extends MinimumDestinationState>
+    extends JdbcConnector implements Destination {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(AbstractJdbcDestination.class);
 
@@ -254,10 +254,20 @@ private void assertCustomParametersDontOverwriteDefaultParameters(final Map<Stri
 
   protected abstract JdbcSqlGenerator getSqlGenerator();
 
-  protected abstract JdbcDestinationHandler<? extends MinimumDestinationState> getDestinationHandler(final String databaseName,
+  protected abstract JdbcDestinationHandler<DestinationState> getDestinationHandler(final String databaseName,
                                                                                                      final JdbcDatabase database,
                                                                                                      final String rawTableSchema);
 
+  /**
+   * Provide any migrations that the destination needs to run. Most destinations will need to provide an instande of
+   * {@link io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.JdbcV1V2Migrator} at minimum.
+   */
+  protected abstract List<Migration<DestinationState>> getMigrations(
+      final NamingConventionTransformer namingResolver,
+      final JdbcDatabase database,
+      final String databaseName,
+      final JdbcSqlGenerator sqlGenerator);
+
   /**
    * "database" key at root of the config json, for any other variants in config, override this
    * method.
@@ -319,17 +329,15 @@ private TyperDeduper getV2TyperDeduper(final JsonNode config, final ConfiguredAi
         .orElse(new CatalogParser(sqlGenerator))
         .parseCatalog(catalog);
     final String databaseName = getDatabaseName(config);
-    final var migrator = new JdbcV1V2Migrator(namingResolver, database, databaseName);
-    final NoopV2TableMigrator v2TableMigrator = new NoopV2TableMigrator();
-    final DestinationHandler<? extends MinimumDestinationState> destinationHandler =
+    final DestinationHandler<DestinationState> destinationHandler =
         getDestinationHandler(databaseName, database, rawNamespaceOverride.orElse(DEFAULT_AIRBYTE_INTERNAL_NAMESPACE));
+    final List<Migration<DestinationState>> migrations = getMigrations(namingResolver, database, databaseName, sqlGenerator);
     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, List.of());
+      typerDeduper = new NoOpTyperDeduperWithV1V2Migrations<>(sqlGenerator, destinationHandler, parsedCatalog, migrations);
     } else {
-      typerDeduper =
-          new DefaultTyperDeduper<>(sqlGenerator, destinationHandler, parsedCatalog, migrator, v2TableMigrator, List.of());
+      typerDeduper = new DefaultTyperDeduper<>(sqlGenerator, destinationHandler, parsedCatalog, migrations);
     }
     return typerDeduper;
   }

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

@@ -10,7 +10,9 @@
 import io.airbyte.commons.exceptions.SQLRuntimeException;
 import io.airbyte.integrations.base.destination.typing_deduping.BaseDestinationV1V2Migrator;
 import io.airbyte.integrations.base.destination.typing_deduping.NamespacedTableName;
+import io.airbyte.integrations.base.destination.typing_deduping.SqlGenerator;
 import io.airbyte.integrations.base.destination.typing_deduping.StreamConfig;
+import io.airbyte.integrations.base.destination.typing_deduping.migrators.MinimumDestinationState;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.Collection;
@@ -21,13 +23,18 @@
  * Largely based on
  * {@link io.airbyte.integrations.destination.snowflake.typing_deduping.SnowflakeV1V2Migrator}.
  */
-public class JdbcV1V2Migrator extends BaseDestinationV1V2Migrator<TableDefinition> {
+public abstract class JdbcV1V2Migrator<DestinationState extends MinimumDestinationState>
+    extends BaseDestinationV1V2Migrator<TableDefinition, DestinationState> {
 
   private final NamingConventionTransformer namingConventionTransformer;
   private final JdbcDatabase database;
   private final String databaseName;
 
-  public JdbcV1V2Migrator(final NamingConventionTransformer namingConventionTransformer, final JdbcDatabase database, final String databaseName) {
+  public JdbcV1V2Migrator(final NamingConventionTransformer namingConventionTransformer,
+                          final SqlGenerator sqlGenerator,
+                          final JdbcDatabase database,
+                          final String databaseName) {
+    super(sqlGenerator);
     this.namingConventionTransformer = namingConventionTransformer;
     this.database = database;
     this.databaseName = databaseName;
@@ -72,5 +79,4 @@ protected NamespacedTableName convertToV1RawName(final StreamConfig streamConfig
         this.namingConventionTransformer.getIdentifier(streamConfig.id().originalNamespace()),
         tableName);
   }
-
 }

airbyte-cdk/java/airbyte-cdk/db-destinations/src/test/java/io/airbyte/cdk/integrations/destination/jdbc/AbstractJdbcDestinationTest.java

@@ -11,12 +11,17 @@
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.cdk.db.jdbc.JdbcDatabase;
 import io.airbyte.cdk.db.jdbc.JdbcUtils;
+import io.airbyte.cdk.integrations.destination.NamingConventionTransformer;
 import io.airbyte.cdk.integrations.destination.StandardNameTransformer;
 import io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.JdbcDestinationHandler;
 import io.airbyte.cdk.integrations.destination.jdbc.typing_deduping.JdbcSqlGenerator;
 import io.airbyte.commons.exceptions.ConfigErrorException;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.integrations.base.destination.typing_deduping.migrators.Migration;
+import io.airbyte.integrations.base.destination.typing_deduping.migrators.MinimumDestinationState;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import org.junit.jupiter.api.Test;
 
@@ -146,6 +151,11 @@ protected JdbcDestinationHandler getDestinationHandler(String databaseName, Jdbc
       return null;
     }
 
+    @Override
+    protected List<Migration<?>> getMigrations(NamingConventionTransformer namingResolver, JdbcDatabase database, String databaseName, JdbcSqlGenerator sqlGenerator) {
+      return Collections.emptyList();
+    }
+
   }
 
 }

airbyte-cdk/java/airbyte-cdk/typing-deduping/src/main/java/io/airbyte/integrations/base/destination/typing_deduping/DefaultTyperDeduper.java

@@ -57,9 +57,6 @@ public class DefaultTyperDeduper<DestinationState extends MinimumDestinationStat
 
   private final SqlGenerator sqlGenerator;
   private final DestinationHandler<DestinationState> destinationHandler;
-
-  private final DestinationV1V2Migrator v1V2Migrator;
-  private final V2TableMigrator v2TableMigrator;
   private final List<Migration<DestinationState>> migrations;
   private final ParsedCatalog parsedCatalog;
   private Set<StreamId> overwriteStreamsWithTmpTable;
@@ -82,14 +79,10 @@ public class DefaultTyperDeduper<DestinationState extends MinimumDestinationStat
   public DefaultTyperDeduper(final SqlGenerator sqlGenerator,
                              final DestinationHandler<DestinationState> destinationHandler,
                              final ParsedCatalog parsedCatalog,
-                             final DestinationV1V2Migrator v1V2Migrator,
-                             final V2TableMigrator v2TableMigrator,
                              final List<Migration<DestinationState>> migrations) {
     this.sqlGenerator = sqlGenerator;
     this.destinationHandler = destinationHandler;
     this.parsedCatalog = parsedCatalog;
-    this.v1V2Migrator = v1V2Migrator;
-    this.v2TableMigrator = v2TableMigrator;
     this.migrations = migrations;
     this.initialRawTableStateByStream = new ConcurrentHashMap<>();
     this.streamsWithSuccessfulSetup = ConcurrentHashMap.newKeySet(parsedCatalog.streams().size());
@@ -99,30 +92,13 @@ public DefaultTyperDeduper(final SqlGenerator sqlGenerator,
         new BasicThreadFactory.Builder().namingPattern(TYPE_AND_DEDUPE_THREAD_NAME).build());
   }
 
-  public DefaultTyperDeduper(
-                             final SqlGenerator sqlGenerator,
-                             final DestinationHandler<DestinationState> destinationHandler,
-                             final ParsedCatalog parsedCatalog,
-                             final DestinationV1V2Migrator v1V2Migrator,
-                             final List<Migration<DestinationState>> migrations) {
-    this(sqlGenerator, destinationHandler, parsedCatalog, v1V2Migrator, new NoopV2TableMigrator(), migrations);
-  }
-
   @Override
   public void prepareSchemasAndRawTables() throws Exception {
     // Technically kind of weird to call this here, but it's the best place we have.
     // Ideally, we'd create just airbyte_internal here, and defer creating the final table schemas
     // until prepareFinalTables... but it doesn't really matter.
     TyperDeduperUtil.prepareSchemas(sqlGenerator, destinationHandler, parsedCatalog);
 
-    TyperDeduperUtil.executeWeirdMigrations(
-        executorService,
-        sqlGenerator,
-        destinationHandler,
-        v1V2Migrator,
-        v2TableMigrator,
-        parsedCatalog);
-
     destinationInitialStates = TyperDeduperUtil.executeRawTableMigrations(
         executorService,
         destinationHandler,

airbyte-cdk/java/airbyte-cdk/typing-deduping/src/main/java/io/airbyte/integrations/base/destination/typing_deduping/DestinationV1V2Migrator.java

@@ -28,8 +28,6 @@
 @Slf4j
 public class NoOpTyperDeduperWithV1V2Migrations<DestinationState extends MinimumDestinationState> implements TyperDeduper {
 
-  private final DestinationV1V2Migrator v1V2Migrator;
-  private final V2TableMigrator v2TableMigrator;
   private final List<Migration<DestinationState>> migrations;
   private final ExecutorService executorService;
   private final ParsedCatalog parsedCatalog;
@@ -39,14 +37,10 @@ public class NoOpTyperDeduperWithV1V2Migrations<DestinationState extends Minimum
   public NoOpTyperDeduperWithV1V2Migrations(final SqlGenerator sqlGenerator,
                                             final DestinationHandler<DestinationState> destinationHandler,
                                             final ParsedCatalog parsedCatalog,
-                                            final DestinationV1V2Migrator v1V2Migrator,
-                                            final V2TableMigrator v2TableMigrator,
                                             final List<Migration<DestinationState>> migrations) {
     this.sqlGenerator = sqlGenerator;
     this.destinationHandler = destinationHandler;
     this.parsedCatalog = parsedCatalog;
-    this.v1V2Migrator = v1V2Migrator;
-    this.v2TableMigrator = v2TableMigrator;
     this.migrations = migrations;
     this.executorService = Executors.newFixedThreadPool(getCountOfTypeAndDedupeThreads(),
         new BasicThreadFactory.Builder().namingPattern(TYPE_AND_DEDUPE_THREAD_NAME).build());
@@ -56,14 +50,6 @@ public NoOpTyperDeduperWithV1V2Migrations(final SqlGenerator sqlGenerator,
   public void prepareSchemasAndRawTables() throws Exception {
     TyperDeduperUtil.prepareSchemas(sqlGenerator, destinationHandler, parsedCatalog);
 
-    TyperDeduperUtil.executeWeirdMigrations(
-        executorService,
-        sqlGenerator,
-        destinationHandler,
-        v1V2Migrator,
-        v2TableMigrator,
-        parsedCatalog);
-
     List<DestinationInitialState<DestinationState>> destinationInitialStates = TyperDeduperUtil.executeRawTableMigrations(
         executorService,
         destinationHandler,

airbyte-cdk/java/airbyte-cdk/typing-deduping/src/main/java/io/airbyte/integrations/base/destination/typing_deduping/NoOpDestinationV1V2Migrator.java

@@ -78,39 +78,6 @@ class TyperDeduperUtil {
       return currentStates
     }
 
-    /**
-     * The legacy-style migrations (V1V2Migrator, V2TableMigrator) need to run before we gather
-     * initial state, because they're dumb and weird.
-     * (specifically: SnowflakeV2TableMigrator inspects the final tables and triggers a soft reset
-     * directly within the migration).
-     * TODO: Migrate these migrations to the new migration system.
-     * This will also reduce the number of times we need to query DB metadata, since (a) we can rely
-     * on the gatherInitialState values, and (b) we can add a DestinationState field for these migrations.
-     * It also enables us to not trigger multiple soft resets in a single sync.
-     */
-    @JvmStatic
-    fun <DestinationState> executeWeirdMigrations(
-        executorService: ExecutorService,
-        sqlGenerator: SqlGenerator,
-        destinationHandler: DestinationHandler<DestinationState>,
-        v1V2Migrator: DestinationV1V2Migrator,
-        v2TableMigrator: V2TableMigrator,
-        parsedCatalog: ParsedCatalog
-    ) {
-      val futures = parsedCatalog.streams.map {
-        CompletableFuture.supplyAsync(
-            {
-              v1V2Migrator.migrateIfNecessary(sqlGenerator, destinationHandler, it)
-              v2TableMigrator.migrateIfNecessary(it)
-            },
-            executorService
-        )
-      }
-      getResultsOrLogAndThrowFirst(
-          "The following exceptions were thrown attempting to run migrations:\n",
-          CompletableFutures.allOf(futures.toList()).toCompletableFuture().join())
-    }
-
     /**
      * Extracts all the "raw" and "final" schemas identified in the [parsedCatalog] and ensures they
      * exist in the Destination Database.