Destination bigquery: rerelease 1s1t behind gate

airbyte-integrations/bases/base-java-s3/src/main/java/io/airbyte/integrations/destination/s3/avro/AvroRecordFactory.java

@@ -11,6 +11,7 @@
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import io.airbyte.commons.jackson.MoreMappers;
 import io.airbyte.integrations.base.JavaBaseConstants;
+import io.airbyte.integrations.base.TypingAndDedupingFlag;
 import io.airbyte.protocol.models.v0.AirbyteRecordMessage;
 import java.util.UUID;
 import org.apache.avro.Schema;
@@ -32,8 +33,14 @@ public AvroRecordFactory(final Schema schema, final JsonAvroConverter converter)
 
   public GenericData.Record getAvroRecord(final UUID id, final AirbyteRecordMessage recordMessage) throws JsonProcessingException {
     final ObjectNode jsonRecord = MAPPER.createObjectNode();
-    jsonRecord.put(JavaBaseConstants.COLUMN_NAME_AB_ID, id.toString());
-    jsonRecord.put(JavaBaseConstants.COLUMN_NAME_EMITTED_AT, recordMessage.getEmittedAt());
+    if (TypingAndDedupingFlag.isDestinationV2()) {
+      jsonRecord.put(JavaBaseConstants.COLUMN_NAME_AB_RAW_ID, id.toString());
+      jsonRecord.put(JavaBaseConstants.COLUMN_NAME_AB_EXTRACTED_AT, recordMessage.getEmittedAt());
+      jsonRecord.put(JavaBaseConstants.COLUMN_NAME_AB_LOADED_AT, (Long) null);
+    } else {
+      jsonRecord.put(JavaBaseConstants.COLUMN_NAME_AB_ID, id.toString());
+      jsonRecord.put(JavaBaseConstants.COLUMN_NAME_EMITTED_AT, recordMessage.getEmittedAt());
+    }
     jsonRecord.setAll((ObjectNode) recordMessage.getData());
 
     return converter.convertToGenericDataRecord(WRITER.writeValueAsBytes(jsonRecord), schema);

airbyte-integrations/bases/base-java-s3/src/test/java/io/airbyte/integrations/destination/s3/avro/AvroSerializedBufferTest.java

@@ -10,6 +10,7 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.integrations.base.DestinationConfig;
 import io.airbyte.integrations.destination.record_buffer.BufferStorage;
 import io.airbyte.integrations.destination.record_buffer.FileBuffer;
 import io.airbyte.integrations.destination.record_buffer.InMemoryBuffer;
@@ -27,6 +28,7 @@
 import org.apache.avro.file.SeekableByteArrayInput;
 import org.apache.avro.generic.GenericData.Record;
 import org.apache.avro.generic.GenericDatumReader;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
 public class AvroSerializedBufferTest {
@@ -49,6 +51,11 @@ public class AvroSerializedBufferTest {
       Field.of("nested_column", JsonSchemaType.OBJECT));
   private static final ConfiguredAirbyteCatalog catalog = CatalogHelpers.createConfiguredAirbyteCatalog(STREAM, null, FIELDS);
 
+  @BeforeAll
+  public static void setup() {
+    DestinationConfig.initialize(Jsons.deserialize("{}"));
+  }
+
   @Test
   public void testSnappyAvroWriter() throws Exception {
     final S3AvroFormatConfig config = new S3AvroFormatConfig(Jsons.jsonNode(Map.of("compression_codec", Map.of(

airbyte-integrations/bases/base-java-s3/src/test/java/io/airbyte/integrations/destination/s3/parquet/ParquetSerializedBufferTest.java

@@ -11,6 +11,7 @@
 import com.amazonaws.util.IOUtils;
 import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.integrations.base.DestinationConfig;
 import io.airbyte.integrations.destination.record_buffer.SerializableBuffer;
 import io.airbyte.integrations.destination.s3.S3DestinationConfig;
 import io.airbyte.protocol.models.Field;
@@ -33,6 +34,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.parquet.avro.AvroReadSupport;
 import org.apache.parquet.hadoop.ParquetReader;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
 public class ParquetSerializedBufferTest {
@@ -60,6 +62,11 @@ public class ParquetSerializedBufferTest {
       Field.of("datetime_with_timezone", JsonSchemaType.STRING_TIMESTAMP_WITH_TIMEZONE));
   private static final ConfiguredAirbyteCatalog catalog = CatalogHelpers.createConfiguredAirbyteCatalog(STREAM, null, FIELDS);
 
+  @BeforeAll
+  public static void setup() {
+    DestinationConfig.initialize(Jsons.deserialize("{}"));
+  }
+
   @Test
   public void testUncompressedParquetWriter() throws Exception {
     final S3DestinationConfig config = S3DestinationConfig.getS3DestinationConfig(Jsons.jsonNode(Map.of(

airbyte-integrations/bases/base-java/run_with_normalization.sh

@@ -6,10 +6,34 @@ set -o pipefail
 destination_exit_code=$?
 echo '{"type": "LOG","log":{"level":"INFO","message":"Destination process done (exit code '"$destination_exit_code"')"}}'
 
+# store original args
+args=$@
+
+while [ $# -ne 0 ]; do
+  case "$1" in
+  --config)
+    CONFIG_FILE="$2"
+    shift 2
+    ;;
+  *)
+    # move on
+    shift
+    ;;
+  esac
+done
+
+# restore original args after shifts
+set -- $args
+
+USE_1S1T_FORMAT="false"
+if [[ -s "$CONFIG_FILE" ]]; then
+  USE_1S1T_FORMAT=$(jq -r '.use_1s1t_format' "$CONFIG_FILE")
+fi
+
 if test "$1" != 'write'
 then
   normalization_exit_code=0
-elif test "$NORMALIZATION_TECHNIQUE" = 'LEGACY'
+elif test "$NORMALIZATION_TECHNIQUE" = 'LEGACY' && test "$USE_1S1T_FORMAT" != "true"
 then
   echo '{"type": "LOG","log":{"level":"INFO","message":"Starting in-connector normalization"}}'
   # the args in a write command are `write --catalog foo.json --config bar.json`

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/DestinationConfig.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.base;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.annotations.VisibleForTesting;
+import jakarta.inject.Singleton;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Singleton of destination config for easy lookup of values.
+ */
+@Singleton
+public class DestinationConfig {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(DestinationConfig.class);
+
+  private static DestinationConfig config;
+
+  @VisibleForTesting
+  protected JsonNode root;
+
+  private DestinationConfig() {}
+
+  public static void initialize(final JsonNode root) {
+    if (config == null) {
+      if (root == null) {
+        throw new IllegalArgumentException("Cannot create DestinationConfig from null.");
+      }
+      config = new DestinationConfig();
+      config.root = root;
+    } else {
+      LOGGER.warn("Singleton was already initialized.");
+    }
+  }
+
+  public static DestinationConfig getInstance() {
+    if (config == null) {
+      throw new IllegalStateException("Singleton not initialized.");
+    }
+    return config;
+  }
+
+  public JsonNode getNodeValue(final String key) {
+    final JsonNode node = config.root.get(key);
+    if (node == null) {
+      LOGGER.debug("Cannot find node with key {} ", key);
+    }
+    return node;
+  }
+
+  // string value, otherwise empty string
+  public String getTextValue(final String key) {
+    final JsonNode node = getNodeValue(key);
+    if (node == null || !node.isTextual()) {
+      LOGGER.debug("Cannot retrieve text value for node with key {}", key);
+      return "";
+    }
+    return node.asText();
+  }
+
+  // boolean value, otherwise false
+  public Boolean getBooleanValue(final String key) {
+    final JsonNode node = getNodeValue(key);
+    if (node == null || !node.isBoolean()) {
+      LOGGER.debug("Cannot retrieve boolean value for node with key {}", key);
+      return false;
+    }
+    return node.asBoolean();
+  }
+
+}

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/IntegrationRunner.java

@@ -144,6 +144,8 @@ private void runInternal(final IntegrationConfig parsed) throws Exception {
         case WRITE -> {
           final JsonNode config = parseConfig(parsed.getConfigPath());
           validateConfig(integration.spec().getConnectionSpecification(), config, "WRITE");
+          // save config to singleton
+          DestinationConfig.initialize(config);
           final ConfiguredAirbyteCatalog catalog = parseConfig(parsed.getCatalogPath(), ConfiguredAirbyteCatalog.class);
 
           final Procedure consumeWriteStreamCallable = () -> {

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/JavaBaseConstants.java

@@ -20,4 +20,11 @@ private JavaBaseConstants() {}
   public static final String COLUMN_NAME_EMITTED_AT = "_airbyte_emitted_at";
   public static final String COLUMN_NAME_DATA = "_airbyte_data";
 
+  // destination v2
+  public static final String COLUMN_NAME_AB_RAW_ID = "_airbyte_raw_id";
+  public static final String COLUMN_NAME_AB_LOADED_AT = "_airbyte_loaded_at";
+  public static final String COLUMN_NAME_AB_EXTRACTED_AT = "_airbyte_extracted_at";
+
+  public static final String AIRBYTE_NAMESPACE_SCHEMA = "airbyte";
+
 }

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/TypingAndDedupingFlag.java

@@ -0,0 +1,13 @@
+/*
+ * Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.base;
+
+public class TypingAndDedupingFlag {
+
+  public static final boolean isDestinationV2() {
+    return DestinationConfig.getInstance().getBooleanValue("use_1s1t_format");
+  }
+
+}

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/destination/buffered_stream_consumer/BufferedStreamConsumer.java

@@ -7,6 +7,7 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import io.airbyte.commons.functional.CheckedFunction;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.base.AirbyteMessageConsumer;
@@ -92,7 +93,13 @@ public class BufferedStreamConsumer extends FailureTrackingAirbyteMessageConsume
 
   private Instant nextFlushDeadline;
   private final Duration bufferFlushFrequency;
+  private final String defaultNamespace;
 
+  /**
+   * Feel free to continue using this in non-1s1t destinations - it may be easier to use. However,
+   * 1s1t destinations should prefer the version which accepts a {@code defaultNamespace}.
+   */
+  @Deprecated
   public BufferedStreamConsumer(final Consumer<AirbyteMessage> outputRecordCollector,
                                 final OnStartFunction onStart,
                                 final BufferingStrategy bufferingStrategy,
@@ -105,7 +112,27 @@ public BufferedStreamConsumer(final Consumer<AirbyteMessage> outputRecordCollect
         onClose,
         catalog,
         isValidRecord,
-        Duration.ofMinutes(15));
+        Duration.ofMinutes(15),
+        // This is purely for backwards compatibility. Many older destinations handle this internally.
+        // Starting with Destinations V2, we recommend passing in an explicit namespace.
+        null);
+  }
+
+  public BufferedStreamConsumer(final Consumer<AirbyteMessage> outputRecordCollector,
+                                final OnStartFunction onStart,
+                                final BufferingStrategy bufferingStrategy,
+                                final OnCloseFunction onClose,
+                                final ConfiguredAirbyteCatalog catalog,
+                                final CheckedFunction<JsonNode, Boolean, Exception> isValidRecord,
+                                final String defaultNamespace) {
+    this(outputRecordCollector,
+        onStart,
+        bufferingStrategy,
+        onClose,
+        catalog,
+        isValidRecord,
+        Duration.ofMinutes(15),
+        defaultNamespace);
   }
 
   /*
@@ -119,7 +146,8 @@ public BufferedStreamConsumer(final Consumer<AirbyteMessage> outputRecordCollect
                          final OnCloseFunction onClose,
                          final ConfiguredAirbyteCatalog catalog,
                          final CheckedFunction<JsonNode, Boolean, Exception> isValidRecord,
-                         final Duration flushFrequency) {
+                         final Duration flushFrequency,
+                         final String defaultNamespace) {
     this.outputRecordCollector = outputRecordCollector;
     this.hasStarted = false;
     this.hasClosed = false;
@@ -132,6 +160,7 @@ public BufferedStreamConsumer(final Consumer<AirbyteMessage> outputRecordCollect
     this.bufferingStrategy = bufferingStrategy;
     this.stateManager = new DefaultDestStateLifecycleManager();
     this.bufferFlushFrequency = flushFrequency;
+    this.defaultNamespace = defaultNamespace;
   }
 
   @Override
@@ -157,7 +186,11 @@ protected void acceptTracked(final AirbyteMessage message) throws Exception {
     Preconditions.checkState(hasStarted, "Cannot accept records until consumer has started");
     if (message.getType() == Type.RECORD) {
       final AirbyteRecordMessage record = message.getRecord();
-      final AirbyteStreamNameNamespacePair stream = AirbyteStreamNameNamespacePair.fromRecordMessage(record);
+      if (Strings.isNullOrEmpty(record.getNamespace())) {
+        record.setNamespace(defaultNamespace);
+      }
+      final AirbyteStreamNameNamespacePair stream;
+      stream = AirbyteStreamNameNamespacePair.fromRecordMessage(record);
 
       // if stream is not part of list of streams to sync to then throw invalid stream exception
       if (!streamNames.contains(stream)) {

airbyte-integrations/bases/base-java/src/test/java/io/airbyte/integrations/base/DestinationConfigTest.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.base;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.airbyte.commons.json.Jsons;
+import org.junit.jupiter.api.Test;
+
+public class DestinationConfigTest {
+
+  private static final String JSON = """
+                                     {
+                                       "foo": "bar",
+                                       "baz": true
+                                     }
+                                     """;
+
+  private static final JsonNode NODE = Jsons.deserialize(JSON);
+
+  @Test
+  public void testInitialization() {
+    // bad initialization
+    assertThrows(IllegalArgumentException.class, () -> DestinationConfig.initialize(null));
+    assertThrows(IllegalStateException.class, DestinationConfig::getInstance);
+
+    // good initialization
+    DestinationConfig.initialize(NODE);
+    assertNotNull(DestinationConfig.getInstance());
+    assertEquals(NODE, DestinationConfig.getInstance().root);
+
+    // initializing again doesn't change the config
+    final JsonNode nodeUnused = Jsons.deserialize("{}");
+    DestinationConfig.initialize(nodeUnused);
+    assertEquals(NODE, DestinationConfig.getInstance().root);
+  }
+
+  @Test
+  public void testValues() {
+    DestinationConfig.initialize(NODE);
+
+    assertEquals("bar", DestinationConfig.getInstance().getTextValue("foo"));
+    assertEquals("", DestinationConfig.getInstance().getTextValue("baz"));
+
+    assertFalse(DestinationConfig.getInstance().getBooleanValue("foo"));
+    assertTrue(DestinationConfig.getInstance().getBooleanValue("baz"));
+
+    // non-existent key
+    assertEquals("", DestinationConfig.getInstance().getTextValue("blah"));
+    assertFalse(DestinationConfig.getInstance().getBooleanValue("blah"));
+
+    assertEquals(Jsons.deserialize("\"bar\""), DestinationConfig.getInstance().getNodeValue("foo"));
+    assertEquals(Jsons.deserialize("true"), DestinationConfig.getInstance().getNodeValue("baz"));
+    assertNull(DestinationConfig.getInstance().getNodeValue("blah"));
+  }
+
+}

airbyte-integrations/bases/base-java/src/test/java/io/airbyte/integrations/destination/buffered_stream_consumer/BufferedStreamConsumerTest.java

@@ -358,7 +358,8 @@ private BufferedStreamConsumer getConsumerWithFlushFrequency() {
         onClose,
         CATALOG,
         isValidRecord,
-        Duration.ofSeconds(PERIODIC_BUFFER_FREQUENCY));
+        Duration.ofSeconds(PERIODIC_BUFFER_FREQUENCY),
+        null);
     return flushFrequencyConsumer;
   }