Destination bigquery: Bump cdk again (airbytehq#38331)

Author: edgao

airbyte-integrations/connectors/destination-bigquery/build.gradle

@@ -3,7 +3,7 @@ plugins {
 }
 
 airbyteJavaConnector {
-    cdkVersionRequired = '0.36.8'
+    cdkVersionRequired = '0.37.1'
     features = [
             'db-destinations',
             'datastore-bigquery',

airbyte-integrations/connectors/destination-bigquery/metadata.yaml

@@ -5,7 +5,7 @@ data:
   connectorSubtype: database
   connectorType: destination
   definitionId: 22f6c74f-5699-40ff-833c-4a879ea40133
-  dockerImageTag: 2.6.2
+  dockerImageTag: 2.6.3
   dockerRepository: airbyte/destination-bigquery
   documentationUrl: https://docs.airbyte.com/integrations/destinations/bigquery
   githubIssueLabel: destination-bigquery

airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryConsumerFactory.kt

@@ -33,9 +33,9 @@ object BigQueryConsumerFactory {
             catalog = catalog,
             bufferManager =
                 BufferManager(
+                    defaultNamespace,
                     (Runtime.getRuntime().maxMemory() * 0.4).toLong(),
                 ),
-            defaultNamespace = Optional.of(defaultNamespace),
         )
     }
 
@@ -59,9 +59,9 @@ object BigQueryConsumerFactory {
             catalog = catalog,
             bufferManager =
                 BufferManager(
+                    defaultNamespace,
                     (Runtime.getRuntime().maxMemory() * 0.5).toLong(),
                 ),
-            defaultNamespace = Optional.of(defaultNamespace),
         )
     }
 }

airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryDestination.kt

@@ -56,7 +56,6 @@ import java.io.ByteArrayInputStream
 import java.io.IOException
 import java.util.*
 import java.util.function.Consumer
-import org.apache.commons.lang3.StringUtils
 
 private val log = KotlinLogging.logger {}
 
@@ -188,7 +187,6 @@ class BigQueryDestination : BaseConnector(), Destination {
     ): SerializedAirbyteMessageConsumer {
         val uploadingMethod = getLoadingMethod(config)
         val defaultNamespace = getDatasetId(config)
-        setDefaultStreamNamespace(catalog, defaultNamespace)
         val disableTypeDedupe = getDisableTypeDedupFlag(config)
         val datasetLocation = getDatasetLocation(config)
         val projectId = config[bqConstants.CONFIG_PROJECT_ID].asText()
@@ -220,6 +218,7 @@ class BigQueryDestination : BaseConnector(), Destination {
         val parsedCatalog =
             parseCatalog(
                 sqlGenerator,
+                defaultNamespace,
                 rawNamespaceOverride.orElse(JavaBaseConstants.DEFAULT_AIRBYTE_INTERNAL_NAMESPACE),
                 catalog,
             )
@@ -307,28 +306,18 @@ class BigQueryDestination : BaseConnector(), Destination {
         )
     }
 
-    private fun setDefaultStreamNamespace(catalog: ConfiguredAirbyteCatalog, namespace: String) {
-        // Set the default originalNamespace on streams with null originalNamespace. This means we
-        // don't
-        // need to repeat this
-        // logic in the rest of the connector.
-        // (record messages still need to handle null namespaces though, which currently happens in
-        // e.g.
-        // AsyncStreamConsumer#accept)
-        // This probably should be shared logic amongst destinations eventually.
-        for (stream in catalog.streams) {
-            if (StringUtils.isEmpty(stream.stream.namespace)) {
-                stream.stream.withNamespace(namespace)
-            }
-        }
-    }
-
     private fun parseCatalog(
         sqlGenerator: BigQuerySqlGenerator,
+        defaultNamespace: String,
         rawNamespaceOverride: String,
         catalog: ConfiguredAirbyteCatalog
     ): ParsedCatalog {
-        val catalogParser = CatalogParser(sqlGenerator, rawNamespaceOverride)
+        val catalogParser =
+            CatalogParser(
+                sqlGenerator,
+                defaultNamespace = defaultNamespace,
+                rawNamespace = rawNamespaceOverride,
+            )
 
         return catalogParser.parseCatalog(catalog)
     }

airbyte-integrations/connectors/destination-bigquery/src/test/java/io/airbyte/integrations/destination/bigquery/typing_deduping/BigQuerySqlGeneratorTest.java

@@ -35,7 +35,7 @@ public void testBuildColumnId() {
 
   @Test
   void columnCollision() {
-    final CatalogParser parser = new CatalogParser(generator);
+    final CatalogParser parser = new CatalogParser(generator, "default_ns");
     assertEquals(
         new StreamConfig(
             new StreamId("bar", "foo", "airbyte_internal", "bar_raw__stream_foo", "bar", "foo"),

docs/integrations/destinations/bigquery.md

@@ -223,6 +223,7 @@ tutorials:
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                                         |
 |:--------|:-----------|:-----------------------------------------------------------|:----------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 2.6.3   | 2024-05-30 | [38331](https://github.com/airbytehq/airbyte/pull/38331)   | Internal code changes in preparation for future feature release                                                                                                 |
 | 2.6.2   | 2024-06-07 | [38764](https://github.com/airbytehq/airbyte/pull/38764)   | Increase message length limit to 50MiB                                                                                                                          |
 | 2.6.1   | 2024-05-29 | [38770](https://github.com/airbytehq/airbyte/pull/38770)   | Internal code change (switch to CDK artifact)                                                                                                                   |
 | 2.6.0   | 2024-05-28 | [38359](https://github.com/airbytehq/airbyte/pull/38359)   | Propagate airbyte_meta from sources; add generation_id column                                                                                                   |