wip

Author: edgao

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

@@ -33,9 +33,10 @@ object BigQueryConsumerFactory {
             catalog = catalog,
             bufferManager =
                 BufferManager(
+                    defaultNamespace,
                     (Runtime.getRuntime().maxMemory() * 0.4).toLong(),
                 ),
-            defaultNamespace = Optional.of(defaultNamespace),
+            defaultNamespace = defaultNamespace,
         )
     }
 
@@ -59,9 +60,10 @@ object BigQueryConsumerFactory {
             catalog = catalog,
             bufferManager =
                 BufferManager(
+                    defaultNamespace,
                     (Runtime.getRuntime().maxMemory() * 0.5).toLong(),
                 ),
-            defaultNamespace = Optional.of(defaultNamespace),
+            defaultNamespace = 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 {}
 
@@ -186,7 +185,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()
@@ -218,6 +216,7 @@ class BigQueryDestination : BaseConnector(), Destination {
         val parsedCatalog =
             parseCatalog(
                 sqlGenerator,
+                defaultNamespace,
                 rawNamespaceOverride.orElse(JavaBaseConstants.DEFAULT_AIRBYTE_INTERNAL_NAMESPACE),
                 catalog,
             )
@@ -300,28 +299,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"),