✨ Destination BigQuery - Add v1v2 Migration

airbyte-integrations/bases/base-typing-deduping-test/src/main/java/io/airbyte/integrations/base/destination/typing_deduping/BaseSqlGeneratorIntegrationTest.java

@@ -9,6 +9,7 @@
 import static org.junit.jupiter.api.Assertions.assertAll;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
@@ -21,6 +22,7 @@
 import io.airbyte.protocol.models.v0.SyncMode;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Stream;
 import org.apache.commons.lang3.tuple.Pair;
@@ -122,6 +124,11 @@ public abstract class BaseSqlGeneratorIntegrationTest<DialectTableDefinition> {
    */
   protected abstract void createRawTable(StreamId streamId) throws Exception;
 
+  /**
+   * Creates a raw table in the v1 format
+   */
+  protected abstract void createV1RawTable(StreamId v1RawTable) throws Exception;
+
   /**
    * Create a final table usingi the StreamId's finalTableId. Subclasses are recommended to hardcode
    * the columns from {@link #FINAL_TABLE_COLUMN_NAMES} or {@link #FINAL_TABLE_COLUMN_NAMES_CDC}. The
@@ -132,6 +139,8 @@ public abstract class BaseSqlGeneratorIntegrationTest<DialectTableDefinition> {
 
   protected abstract void insertRawTableRecords(StreamId streamId, List<JsonNode> records) throws Exception;
 
+  protected abstract void insertV1RawTableRecords(StreamId streamId, List<JsonNode> records) throws Exception;
+
   protected abstract void insertFinalTableRecords(boolean includeCdcDeletedAt, StreamId streamId, String suffix, List<JsonNode> records)
       throws Exception;
 
@@ -709,6 +718,30 @@ public void weirdColumnNames() throws Exception {
         dumpFinalTableRecords(streamId, ""));
   }
 
+  @Test
+  public void testV1V2migration() throws Exception {
+    // This is maybe a little hacky, but it avoids having to refactor this entire class and subclasses
+    // for something that is going away
+    StreamId v1RawTableStreamId = new StreamId(null, null, streamId.finalNamespace(), "v1_" + streamId.rawName(), null, null);
+    createV1RawTable(v1RawTableStreamId);
+    insertV1RawTableRecords(v1RawTableStreamId, singletonList(Jsons.jsonNode(Map.of(
+        "_airbyte_ab_id", "v1v2",
+        "_airbyte_emitted_at", "2023-01-01T00:00:00Z",
+        "_airbyte_data", "{\"hello\": \"world\"}"))));
+    final String migration = generator.migrateFromV1toV2(streamId, v1RawTableStreamId.rawNamespace(), v1RawTableStreamId.rawName());
+    destinationHandler.execute(migration);
+    List<JsonNode> v1RawRecords = dumpRawTableRecords(v1RawTableStreamId);
+    List<JsonNode> v2RawRecords = dumpRawTableRecords(streamId);
+    assertAll(
+        () -> assertEquals(1, v1RawRecords.size()),
+        () -> assertEquals(1, v2RawRecords.size()),
+        () -> assertEquals(v1RawRecords.get(0).get("_airbyte_ab_id").asText(), v2RawRecords.get(0).get("_airbyte_raw_id").asText()),
+        () -> assertEquals(Jsons.deserialize(v1RawRecords.get(0).get("_airbyte_data").asText()), v2RawRecords.get(0).get("_airbyte_data")),
+        () -> assertEquals(v1RawRecords.get(0).get("_airbyte_emitted_at").asText(), v2RawRecords.get(0).get("_airbyte_extracted_at").asText()),
+        () -> assertNull(v2RawRecords.get(0).get("_airbyte_loaded_at")));
+
+  }
+
   private void verifyRecords(final String expectedRawRecordsFile,
                              final List<JsonNode> actualRawRecords,
                              final String expectedFinalRecordsFile,

airbyte-integrations/bases/base-typing-deduping/build.gradle

@@ -3,5 +3,6 @@ plugins {
 }
 
 dependencies {
-  implementation libs.airbyte.protocol
+    implementation libs.airbyte.protocol
+    implementation project(path: ':airbyte-integrations:bases:base-java')
 }

airbyte-integrations/bases/base-typing-deduping/src/main/java/io/airbyte/integrations/base/destination/typing_deduping/BaseDestinationV1V2Migrator.java

@@ -0,0 +1,164 @@
+/*
+ * Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.base.destination.typing_deduping;
+
+import static io.airbyte.integrations.base.JavaBaseConstants.LEGACY_RAW_TABLE_COLUMNS;
+import static io.airbyte.integrations.base.JavaBaseConstants.V2_RAW_TABLE_COLUMN_NAMES;
+
+import io.airbyte.protocol.models.v0.DestinationSyncMode;
+import java.util.Collection;
+import java.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class BaseDestinationV1V2Migrator<DialectTableDefinition> implements DestinationV1V2Migrator {
+
+  Logger LOGGER = LoggerFactory.getLogger(BaseDestinationV1V2Migrator.class);
+
+  @Override
+  public void migrateIfNecessary(
+                                 final SqlGenerator sqlGenerator,
+                                 final DestinationHandler destinationHandler,
+                                 final StreamConfig streamConfig)
+      throws TableNotMigratedException, UnexpectedSchemaException {
+    if (shouldMigrate(streamConfig)) {
+      LOGGER.info("Starting v2 Migration for stream {}", streamConfig.id().finalName());
+      migrate(sqlGenerator, destinationHandler, streamConfig);
+    }
+  }
+
+  /**
+   * Determine whether a given stream needs to be migrated from v1 to v2
+   *
+   * @param streamConfig the stream in question
+   * @return whether to migrate the stream
+   */
+  protected boolean shouldMigrate(final StreamConfig streamConfig) {
+    final var v1RawTable = convertToV1RawName(streamConfig);
+    return isMigrationRequiredForSyncMode(streamConfig.destinationSyncMode())
+        && !doesValidV2RawTableAlreadyExist(streamConfig)
+        && doesValidV1RawTableExist(v1RawTable.namespace(), v1RawTable.tableName());
+  }
+
+  /**
+   * Execute sql statements that converts a v1 raw table to a v2 raw table. Leaves the v1 raw table
+   * intact
+   *
+   * @param sqlGenerator the class which generates dialect specific sql statements
+   * @param destinationHandler the class which executes the sql statements
+   * @param streamConfig the stream to migrate the raw table of
+   */
+  public void migrate(final SqlGenerator<DialectTableDefinition> sqlGenerator,
+                      final DestinationHandler<DialectTableDefinition> destinationHandler,
+                      final StreamConfig streamConfig)
+      throws TableNotMigratedException {
+    final var namespacedTableName = convertToV1RawName(streamConfig);
+    final var migrateAndReset = String.join("\n",
+        sqlGenerator.migrateFromV1toV2(streamConfig.id(), namespacedTableName.namespace(),
+            namespacedTableName.tableName()),
+        sqlGenerator.softReset(streamConfig));
+    try {
+      destinationHandler.execute(migrateAndReset);
+    } catch (Exception e) {
+      final var message = "Attempted and failed to migrate stream %s".formatted(streamConfig.id().finalName());
+      throw new TableNotMigratedException(message, e);
+    }
+  }
+
+  /**
+   * Checks the schema of the v1 raw table to ensure it matches the expected format
+   *
+   * @param existingV2AirbyteRawTable the v1 raw table
+   * @return whether the schema is as expected
+   */
+  private boolean doesV1RawTableMatchExpectedSchema(DialectTableDefinition existingV2AirbyteRawTable) {
+
+    return schemaMatchesExpectation(existingV2AirbyteRawTable, LEGACY_RAW_TABLE_COLUMNS);
+  }
+
+  /**
+   * Checks the schema of the v2 raw table to ensure it matches the expected format
+   *
+   * @param existingV2AirbyteRawTable the v2 raw table
+   */
+  private void validateAirbyteInternalNamespaceRawTableMatchExpectedV2Schema(DialectTableDefinition existingV2AirbyteRawTable) {
+    if (!schemaMatchesExpectation(existingV2AirbyteRawTable, V2_RAW_TABLE_COLUMN_NAMES)) {
+      throw new UnexpectedSchemaException("Destination V2 Raw Table does not match expected Schema");
+    }
+  }
+
+  /**
+   * If the sync mode is a full refresh and we overwrite the table then there is no need to migrate
+   *
+   * @param destinationSyncMode destination sync mode
+   * @return whether this is full refresh overwrite
+   */
+  private boolean isMigrationRequiredForSyncMode(final DestinationSyncMode destinationSyncMode) {
+    return !DestinationSyncMode.OVERWRITE.equals(destinationSyncMode);
+  }
+
+  /**
+   * Checks if a valid destinations v2 raw table already exists
+   *
+   * @param streamConfig the raw table to check
+   * @return whether it exists and is in the correct format
+   */
+  private boolean doesValidV2RawTableAlreadyExist(final StreamConfig streamConfig) {
+    if (doesAirbyteInternalNamespaceExist(streamConfig)) {
+      final var existingV2Table = getTableIfExists(streamConfig.id().rawNamespace(), streamConfig.id().rawName());
+      existingV2Table.ifPresent(this::validateAirbyteInternalNamespaceRawTableMatchExpectedV2Schema);
+      return existingV2Table.isPresent();
+    }
+    return false;
+  }
+
+  /**
+   * Checks if a valid v1 raw table already exists
+   *
+   * @param namespace
+   * @param tableName
+   * @return whether it exists and is in the correct format
+   */
+  private boolean doesValidV1RawTableExist(final String namespace, final String tableName) {
+    final var existingV1RawTable = getTableIfExists(namespace, tableName);
+    return existingV1RawTable.isPresent() && doesV1RawTableMatchExpectedSchema(existingV1RawTable.get());
+  }
+
+  /**
+   * Checks to see if Airbyte's internal schema for destinations v2 exists
+   *
+   * @param streamConfig the stream to check
+   * @return whether the schema exists
+   */
+  abstract protected boolean doesAirbyteInternalNamespaceExist(StreamConfig streamConfig);
+
+  /**
+   * Checks a Table's schema and compares it to an expected schema to make sure it matches
+   *
+   * @param existingTable the table to check
+   * @param columns the expected schema
+   * @return whether the existing table schema matches the expectation
+   */
+  abstract protected boolean schemaMatchesExpectation(DialectTableDefinition existingTable, Collection<String> columns);
+
+  /**
+   * Get a reference ta a table if it exists
+   *
+   * @param namespace
+   * @param tableName
+   * @return an optional potentially containing a reference to the table
+   */
+  abstract protected Optional<DialectTableDefinition> getTableIfExists(String namespace, String tableName);
+
+  /**
+   * We use different naming conventions for raw table names in destinations v2, we need a way to map
+   * that back to v1 names
+   *
+   * @param streamConfig the stream in question
+   * @return the valid v1 name and namespace for the same stream
+   */
+  abstract protected NamespacedTableName convertToV1RawName(StreamConfig streamConfig);
+
+}

airbyte-integrations/bases/base-typing-deduping/src/main/java/io/airbyte/integrations/base/destination/typing_deduping/CollectionUtils.java

@@ -34,6 +34,13 @@ public static boolean containsIgnoreCase(final Collection<String> collection, fi
    * @return whether all searchTerms are in the searchCollection
    */
   public static boolean containsAllIgnoreCase(final Collection<String> searchCollection, final Collection<String> searchTerms) {
+    if (searchTerms.isEmpty()) {
+      // There isn't a good behavior for an empty collection. Without this check, an empty collection
+      // would always return
+      // true, but it feels misleading to say that the searchCollection does "contain all" when
+      // searchTerms is empty
+      throw new IllegalArgumentException("Search Terms collection may not be empty");
+    }
     return searchTerms.stream().allMatch(term -> containsIgnoreCase(searchCollection, term));
   }
 

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

@@ -19,7 +19,7 @@
  * <p>
  * In a typical sync, destinations should call the methods:
  * <ol>
- * <li>{@link #prepareFinalTables()} once at the start of the sync</li>
+ * <li>{@link #prepareTables()} once at the start of the sync</li>
  * <li>{@link #typeAndDedupe(String, String)} as needed throughout the sync</li>
  * <li>{@link #commitFinalTables()} once at the end of the sync</li>
  * </ol>
@@ -35,15 +35,19 @@ public class DefaultTyperDeduper<DialectTableDefinition> implements TyperDeduper
 
   private final SqlGenerator<DialectTableDefinition> sqlGenerator;
   private final DestinationHandler<DialectTableDefinition> destinationHandler;
+
+  private final DestinationV1V2Migrator<DialectTableDefinition> v1V2Migrator;
   private final ParsedCatalog parsedCatalog;
   private Set<StreamId> overwriteStreamsWithTmpTable;
 
   public DefaultTyperDeduper(SqlGenerator<DialectTableDefinition> sqlGenerator,
                              DestinationHandler<DialectTableDefinition> destinationHandler,
-                             ParsedCatalog parsedCatalog) {
+                             ParsedCatalog parsedCatalog,
+                             DestinationV1V2Migrator<DialectTableDefinition> v1V2Migrator) {
     this.sqlGenerator = sqlGenerator;
     this.destinationHandler = destinationHandler;
     this.parsedCatalog = parsedCatalog;
+    this.v1V2Migrator = v1V2Migrator;
   }
 
   /**
@@ -52,7 +56,7 @@ public DefaultTyperDeduper(SqlGenerator<DialectTableDefinition> sqlGenerator,
    * empty) we write to a temporary final table, and swap it into the true final table at the end of
    * the sync. This is to prevent user downtime during a sync.
    */
-  public void prepareFinalTables() throws Exception {
+  public void prepareTables() throws Exception {
     if (overwriteStreamsWithTmpTable != null) {
       throw new IllegalStateException("Tables were already prepared.");
     }
@@ -63,6 +67,8 @@ public void prepareFinalTables() throws Exception {
     // Also, for OVERWRITE streams, decide if we're writing directly to the final table, or into an
     // _airbyte_tmp table.
     for (StreamConfig stream : parsedCatalog.streams()) {
+      // Migrate the Raw Tables if this is the first v2 sync after a v1 sync
+      v1V2Migrator.migrateIfNecessary(sqlGenerator, destinationHandler, stream);
       final Optional<DialectTableDefinition> existingTable = destinationHandler.findExistingTable(stream.id());
       if (existingTable.isPresent()) {
         // The table already exists. Decide whether we're writing to it directly, or using a tmp table.

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

@@ -0,0 +1,25 @@
+/*
+ * Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.base.destination.typing_deduping;
+
+public interface DestinationV1V2Migrator<DialectTableDefinition> {
+
+  /**
+   * This is the primary entrypoint to this interface
+   * <p>
+   * Determine whether a migration is necessary for a given stream and if so, migrate the raw table
+   * and rebuild the final table with a soft reset
+   *
+   * @param sqlGenerator the class to use to generate sql
+   * @param destinationHandler the handler to execute the sql statements
+   * @param streamConfig the stream to assess migration needs
+   */
+  void migrateIfNecessary(
+                          final SqlGenerator<DialectTableDefinition> sqlGenerator,
+                          final DestinationHandler<DialectTableDefinition> destinationHandler,
+                          final StreamConfig streamConfig)
+      throws TableNotMigratedException, UnexpectedSchemaException;
+
+}

airbyte-integrations/bases/base-typing-deduping/src/main/java/io/airbyte/integrations/base/destination/typing_deduping/NamespacedTableName.java

@@ -0,0 +1,10 @@
+/*
+ * Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.base.destination.typing_deduping;
+
+// yet another namespace, name combo class
+public record NamespacedTableName(String namespace, String tableName) {
+
+}

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

@@ -0,0 +1,17 @@
+/*
+ * Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.base.destination.typing_deduping;
+
+public class NoOpDestinationV1V2Migrator<DialectTableDefinition> implements DestinationV1V2Migrator<DialectTableDefinition> {
+
+  @Override
+  public void migrateIfNecessary(final SqlGenerator<DialectTableDefinition> sqlGenerator,
+                                 final DestinationHandler<DialectTableDefinition> destinationHandler,
+                                 final StreamConfig streamConfig)
+      throws TableNotMigratedException, UnexpectedSchemaException {
+    // Do nothing
+  }
+
+}

airbyte-integrations/bases/base-typing-deduping/src/main/java/io/airbyte/integrations/base/destination/typing_deduping/NoopTyperDeduper.java

@@ -7,7 +7,7 @@
 public class NoopTyperDeduper implements TyperDeduper {
 
   @Override
-  public void prepareFinalTables() throws Exception {
+  public void prepareTables() throws Exception {
 
   }
 

airbyte-integrations/bases/base-typing-deduping/src/main/java/io/airbyte/integrations/base/destination/typing_deduping/SqlGenerator.java

@@ -76,4 +76,15 @@ public interface SqlGenerator<DialectTableDefinition> {
    */
   String overwriteFinalTable(StreamId stream, String finalSuffix);
 
+  /**
+   * Creates a sql query which will create a v2 raw table from the v1 raw table, then performs a soft
+   * reset.
+   *
+   * @param streamId the stream to migrate
+   * @param namespace
+   * @param tableName
+   * @return a string containing the necessary sql to migrate
+   */
+  String migrateFromV1toV2(StreamId streamId, String namespace, String tableName);
+
 }

airbyte-integrations/bases/base-typing-deduping/src/main/java/io/airbyte/integrations/base/destination/typing_deduping/TableNotMigratedException.java

@@ -14,4 +14,8 @@ public TableNotMigratedException(String message) {
     super(message);
   }
 
+  public TableNotMigratedException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
 }