Revert "Test"

This reverts commit f0e462e.
airbytehq · Oct 2, 2023 · 59991e2 · 59991e2
1 parent f0e462e
commit 59991e2
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 12 deletions.
diff --git a/airbyte-commons/src/main/java/io/airbyte/commons/json/Jsons.java b/airbyte-commons/src/main/java/io/airbyte/commons/json/Jsons.java
@@ -85,7 +85,6 @@ public static <T> T deserialize(final File file, final Class<T> klass) {
     }
   }
 
-
   public static <T> T deserialize(final File file, final TypeReference<T> valueTypeRef) {
     try {
       return OBJECT_MAPPER.readValue(file, valueTypeRef);

diff --git a/...rc/main/java/io/airbyte/integrations/destination/bigquery/BigQueryAsyncStandardFlush.java b/...rc/main/java/io/airbyte/integrations/destination/bigquery/BigQueryAsyncStandardFlush.java
@@ -5,15 +5,11 @@
 package io.airbyte.integrations.destination.bigquery;
 
 import com.google.cloud.bigquery.BigQuery;
-import com.google.cloud.bigquery.InsertAllRequest;
-import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.destination.bigquery.uploader.AbstractBigQueryUploader;
 import io.airbyte.integrations.destination_async.DestinationFlushFunction;
 import io.airbyte.integrations.destination_async.partial_messages.PartialAirbyteMessage;
 import io.airbyte.protocol.models.v0.AirbyteStreamNameNamespacePair;
 import io.airbyte.protocol.models.v0.StreamDescriptor;
-
-import java.util.Map;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
@@ -33,22 +29,19 @@ public BigQueryAsyncStandardFlush(BigQuery bigQuery,
 
   @Override
   public void flush(final StreamDescriptor decs, final Stream<PartialAirbyteMessage> stream) throws Exception {
-    // ConcurrentMap<AirbyteStreamNameNamespacePair, AbstractBigQueryUploader<?>> uploaderMapSupplied = uploaderMap.get();
-    InsertAllRequest.Builder request = InsertAllRequest.newBuilder(decs.getNamespace(), decs.getName());
+    ConcurrentMap<AirbyteStreamNameNamespacePair, AbstractBigQueryUploader<?>> uploaderMapSupplied = uploaderMap.get();
     stream.forEach(aibyteMessage -> {
       try {
-        request.addRow(Jsons.deserialize(aibyteMessage.getSerialized(), Map.class));
-         /*AirbyteStreamNameNamespacePair sd = new AirbyteStreamNameNamespacePair(aibyteMessage.getRecord().getStream(),
+        AirbyteStreamNameNamespacePair sd = new AirbyteStreamNameNamespacePair(aibyteMessage.getRecord().getStream(),
             aibyteMessage.getRecord().getNamespace());
-        uploaderMapSupplied.get(sd).upload(aibyteMessage);*/
+        uploaderMapSupplied.get(sd).upload(aibyteMessage);
       } catch (Exception e) {
         log.error("BQ async standard flush");
         log.error(aibyteMessage.toString());
         throw e;
       }
     });
-    bigQuery.insertAll(request.build());
-    // uploaderMapSupplied.values().forEach(test -> test.closeAfterPush());
+    uploaderMapSupplied.values().forEach(test -> test.closeAfterPush());
   }
 
   @Override
-Original file line number
+Diff line change
@@ Expand Up @@
         }
       }
       public static <T> T deserialize(final File file, final TypeReference<T> valueTypeRef) {
         try {
           return OBJECT_MAPPER.readValue(file, valueTypeRef);
@@ Expand Down @@