Add additional check to skip v1v2 migration

Author: gisripa

airbyte-cdk/java/airbyte-cdk/core/src/main/java/io/airbyte/cdk/integrations/base/JavaBaseConstants.java

@@ -5,6 +5,7 @@
 package io.airbyte.cdk.integrations.base;
 
 import java.util.List;
+import java.util.Set;
 
 public final class JavaBaseConstants {
 
@@ -34,7 +35,7 @@ private JavaBaseConstants() {}
 
   // Meta was introduced later, so to avoid triggering raw table soft-reset in v1->v2
   // use this column list.
-  public static final List<String> V2_RAW_TABLE_COLUMN_NAMES_WITHOUT_META = List.of(
+  public static final Set<String> V2_RAW_TABLE_COLUMN_NAMES_WITHOUT_META = Set.of(
       COLUMN_NAME_AB_RAW_ID,
       COLUMN_NAME_AB_EXTRACTED_AT,
       COLUMN_NAME_AB_LOADED_AT,

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

@@ -5,6 +5,7 @@
 package io.airbyte.integrations.base.destination.typing_deduping;
 
 import static io.airbyte.cdk.integrations.base.JavaBaseConstants.LEGACY_RAW_TABLE_COLUMNS;
+import static io.airbyte.cdk.integrations.base.JavaBaseConstants.V2_RAW_TABLE_COLUMN_NAMES;
 import static io.airbyte.cdk.integrations.base.JavaBaseConstants.V2_RAW_TABLE_COLUMN_NAMES_WITHOUT_META;
 
 import io.airbyte.protocol.models.v0.DestinationSyncMode;
@@ -89,7 +90,10 @@ private boolean doesV1RawTableMatchExpectedSchema(final DialectTableDefinition e
    * @param existingV2AirbyteRawTable the v2 raw table
    */
   private void validateAirbyteInternalNamespaceRawTableMatchExpectedV2Schema(final DialectTableDefinition existingV2AirbyteRawTable) {
-    if (!schemaMatchesExpectation(existingV2AirbyteRawTable, V2_RAW_TABLE_COLUMN_NAMES_WITHOUT_META)) {
+    // Account for the fact that the meta column was added later, so skip the rebuilding of the raw
+    // table.
+    if (!(schemaMatchesExpectation(existingV2AirbyteRawTable, V2_RAW_TABLE_COLUMN_NAMES_WITHOUT_META) ||
+        schemaMatchesExpectation(existingV2AirbyteRawTable, V2_RAW_TABLE_COLUMN_NAMES))) {
       throw new UnexpectedSchemaException("Destination V2 Raw Table does not match expected Schema");
     }
   }

airbyte-cdk/java/airbyte-cdk/typing-deduping/src/test/java/io/airbyte/integrations/base/destination/typing_deduping/DestinationV1V2MigratorTest.java

@@ -5,6 +5,7 @@
 package io.airbyte.integrations.base.destination.typing_deduping;
 
 import static io.airbyte.cdk.integrations.base.JavaBaseConstants.LEGACY_RAW_TABLE_COLUMNS;
+import static io.airbyte.cdk.integrations.base.JavaBaseConstants.V2_RAW_TABLE_COLUMN_NAMES;
 import static io.airbyte.cdk.integrations.base.JavaBaseConstants.V2_RAW_TABLE_COLUMN_NAMES_WITHOUT_META;
 import static org.mockito.ArgumentMatchers.any;
 
@@ -97,6 +98,7 @@ public static BaseDestinationV1V2Migrator makeMockMigrator(final boolean v2Names
     Mockito.when(migrator.doesAirbyteInternalNamespaceExist(any())).thenReturn(v2NamespaceExists);
     final var existingTable = v2TableExists ? Optional.of("v2_raw") : Optional.empty();
     Mockito.when(migrator.getTableIfExists("raw", "raw_table")).thenReturn(existingTable);
+    Mockito.when(migrator.schemaMatchesExpectation("v2_raw", V2_RAW_TABLE_COLUMN_NAMES)).thenReturn(false);
     Mockito.when(migrator.schemaMatchesExpectation("v2_raw", V2_RAW_TABLE_COLUMN_NAMES_WITHOUT_META)).thenReturn(v2RawSchemaMatches);
 
     Mockito.when(migrator.convertToV1RawName(any())).thenReturn(new NamespacedTableName("v1_raw_namespace", "v1_raw_table"));