Bigquery / Bulk Load CDK: implement T+D, truncate refresh, etc (#57578)

# Conflicts:
#	airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/message/DestinationMessage.kt
#	airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/formatter/BigQueryRecordFormatter.kt
#	airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/BigQuerySqlGenerator.kt
#	airbyte-integrations/connectors/destination-bigquery/src/test-integration/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryDestinationTestUtils.kt
#	airbyte-integrations/connectors/destination-bigquery/src/test-integration/kotlin/io/airbyte/integrations/destination/bigquery/BigqueryWriteTest.kt
#	airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/BigqueryDestinationHandlerTest.kt

Author: edgao

airbyte-cdk/bulk/toolkits/load-db/src/main/kotlin/io/airbyte/cdk/load/orchestration/db/direct_load_table/DirectLoadSqlGenerator.kt

@@ -0,0 +1,7 @@
+/*
+ * Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.cdk.load.orchestration.db.direct_load_table
+
+interface DirectLoadSqlGenerator {}

airbyte-cdk/bulk/toolkits/load-db/src/main/kotlin/io/airbyte/cdk/load/orchestration/db/direct_load_table/DirectLoadTableOperations.kt

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.cdk.load.orchestration.db.direct_load_table
+
+import io.airbyte.cdk.load.command.DestinationStream
+import io.airbyte.cdk.load.orchestration.db.DatabaseHandler
+import io.airbyte.cdk.load.orchestration.db.TableName
+
+class DirectLoadTableOperations(
+    private val sqlGenerator: DirectLoadSqlGenerator,
+    private val databaseHandler: DatabaseHandler,
+) {
+    fun createTable(
+        stream: DestinationStream,
+        finalTableName: TableName,
+        suffix: String,
+        replace: Boolean
+    ) {
+        databaseHandler.execute(TODO())
+    }
+
+    fun alterTable(
+        stream: DestinationStream,
+        finalTableName: TableName,
+    ) {
+        // TODO we should figure out some reasonable abstraction for diffing existing+expected
+        //  table schema
+        TODO()
+    }
+
+    fun overwriteFinalTable(
+        finalTableName: TableName,
+        suffix: String,
+    ) {
+        TODO()
+    }
+}

airbyte-cdk/bulk/toolkits/load-db/src/main/kotlin/io/airbyte/cdk/load/orchestration/db/direct_load_table/DirectLoadTableStreamLoader.kt

@@ -0,0 +1,26 @@
+/*
+ * Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.cdk.load.orchestration.db.direct_load_table
+
+import io.airbyte.cdk.load.command.DestinationStream
+import io.airbyte.cdk.load.state.StreamProcessingFailed
+import io.airbyte.cdk.load.write.StreamLoader
+
+class DirectLoadTableStreamLoader(
+    override val stream: DestinationStream,
+    val tableOperations: DirectLoadTableOperations,
+) : StreamLoader {
+    override suspend fun start() {
+        // TODO
+        //   * create table if not exists
+        //   * alter table if needed
+        //   * truncate refresh setup
+    }
+
+    override suspend fun close(hadNonzeroRecords: Boolean, streamFailure: StreamProcessingFailed?) {
+        // TODO
+        //   * commit truncate refresh
+    }
+}

airbyte-cdk/bulk/toolkits/load-db/src/main/kotlin/io/airbyte/cdk/load/orchestration/db/direct_load_table/DirectLoadTableWriter.kt

@@ -0,0 +1,17 @@
+/*
+ * Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.cdk.load.orchestration.db.direct_load_table
+
+import io.airbyte.cdk.load.command.DestinationStream
+import io.airbyte.cdk.load.write.DestinationWriter
+import io.airbyte.cdk.load.write.StreamLoader
+
+class DirectLoadTableWriter(
+    private val tableOperations: DirectLoadTableOperations,
+) : DestinationWriter {
+    override fun createStreamLoader(stream: DestinationStream): StreamLoader {
+        return DirectLoadTableStreamLoader(stream, tableOperations)
+    }
+}