bigquery handles airbyte_meta/sync_id/generation_id

airbytehq · May 20, 2024 · d414927 · d414927
1 parent 0651902
commit d414927
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 1 deletion.
diff --git a/...query/src/main/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryDestination.kt b/...query/src/main/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryDestination.kt
@@ -287,7 +287,7 @@ class BigQueryDestination : BaseConnector(), Destination {
                         bigQueryGcsStorageOperations,
                         initialStatus,
                         FileUploadFormat.CSV,
-                        V2_WITHOUT_META,
+                        V2_WITH_GENERATION,
                         disableTD
                     )
                 },

diff --git a/...ntegrations/destination/bigquery/migrators/BigqueryAirbyteMetaAndGenerationIdMigration.kt b/...ntegrations/destination/bigquery/migrators/BigqueryAirbyteMetaAndGenerationIdMigration.kt
@@ -0,0 +1,27 @@
+package io.airbyte.integrations.destination.bigquery.migrators
+
+import io.airbyte.integrations.base.destination.typing_deduping.DestinationHandler
+import io.airbyte.integrations.base.destination.typing_deduping.DestinationInitialStatus
+import io.airbyte.integrations.base.destination.typing_deduping.StreamConfig
+import io.airbyte.integrations.base.destination.typing_deduping.migrators.Migration
+import io.github.oshai.kotlinlogging.KotlinLogging
+
+class BigqueryAirbyteMetaAndGenerationIdMigration: Migration<BigQueryDestinationState> {
+    private val logger = KotlinLogging.logger {}
+
+    override fun migrateIfNecessary(
+        destinationHandler: DestinationHandler<BigQueryDestinationState>,
+        stream: StreamConfig,
+        state: DestinationInitialStatus<BigQueryDestinationState>
+    ): Migration.MigrationResult<BigQueryDestinationState> {
+        if (!state.initialRawTableStatus.rawTableExists) {
+            // The raw table doesn't exist. No migration necessary. Update the state.
+            logger.info {
+                "Skipping generation_id migration for ${stream.id.originalNamespace}.${stream.id.originalName} because the raw table doesn't exist"
+            }
+            return Migration.MigrationResult(state.destinationState, false)
+        }
+
+        TODO("Not yet implemented")
+    }
+}