feat: provide append() methods that accept com.google.gson objects

Author: agrawal-siddharth

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java

@@ -679,9 +679,7 @@ public long getInflightWaitSeconds() {
     return inflightWaitSec.longValue();
   }
 
-  /**
-   * @return a unique Id for the writer.
-   */
+  /** @return a unique Id for the writer. */
   public String getWriterId() {
     return writerId;
   }

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/JsonStreamWriter.java

@@ -21,6 +21,7 @@
 import com.google.api.gax.core.ExecutorProvider;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.rpc.TransportChannelProvider;
+import com.google.gson.JsonArray;
 import com.google.protobuf.Descriptors;
 import java.io.IOException;
 import java.time.Duration;
@@ -45,9 +46,7 @@ public class JsonStreamWriter implements AutoCloseable {
    * @param builder The Builder object for the JsonStreamWriter
    */
   private JsonStreamWriter(SchemaAwareStreamWriter.Builder<Object> builder)
-      throws Descriptors.DescriptorValidationException,
-          IllegalArgumentException,
-          IOException,
+      throws Descriptors.DescriptorValidationException, IllegalArgumentException, IOException,
           InterruptedException {
     this.schemaAwareStreamWriter = builder.build();
   }
@@ -83,13 +82,46 @@ public ApiFuture<AppendRowsResponse> append(JSONArray jsonArr, long offset)
     return this.schemaAwareStreamWriter.append(jsonArr, offset);
   }
 
-  public String getStreamName() {
-    return this.schemaAwareStreamWriter.getStreamName();
+  private JSONArray gsonToOrgJSON(JsonArray jsonArr) {
+    return new JSONArray(jsonArr.toString());
+  }
+
+  /**
+   * Writes a JsonArray that contains JsonObjects to the BigQuery table by first converting the JSON
+   * data to protobuf messages, then using StreamWriter's append() to write the data at current end
+   * of stream. If there is a schema update, the current StreamWriter is closed. A new StreamWriter
+   * is created with the updated TableSchema.
+   *
+   * @param jsonArr The JSON array that contains JsonObjects to be written
+   * @return {@code ApiFuture<AppendRowsResponse>} returns an AppendRowsResponse message wrapped in
+   *     an ApiFuture
+   */
+  public ApiFuture<AppendRowsResponse> append(JsonArray jsonArr)
+      throws IOException, Descriptors.DescriptorValidationException {
+    return this.append(jsonArr, -1);
   }
 
   /**
-   * @return A unique Id for this writer.
+   * Writes a JsonArray that contains JsonObjects to the BigQuery table by first converting the JSON
+   * data to protobuf messages, then using StreamWriter's append() to write the data at the
+   * specified offset. If there is a schema update, the current StreamWriter is closed. A new
+   * StreamWriter is created with the updated TableSchema.
+   *
+   * @param jsonArr The JSON array that contains JSONObjects to be written
+   * @param offset Offset for deduplication
+   * @return {@code ApiFuture<AppendRowsResponse>} returns an AppendRowsResponse message wrapped in
+   *     an ApiFuture
    */
+  public ApiFuture<AppendRowsResponse> append(JsonArray jsonArr, long offset)
+      throws IOException, Descriptors.DescriptorValidationException {
+    return this.append(gsonToOrgJSON(jsonArr), offset);
+  }
+
+  public String getStreamName() {
+    return this.schemaAwareStreamWriter.getStreamName();
+  }
+
+  /** @return A unique Id for this writer. */
   public String getWriterId() {
     return this.schemaAwareStreamWriter.getWriterId();
   }
@@ -123,9 +155,7 @@ public long getInflightWaitSeconds() {
     return this.schemaAwareStreamWriter.getInflightWaitSeconds();
   }
 
-  /**
-   * @return the missing value interpretation map used for the writer.
-   */
+  /** @return the missing value interpretation map used for the writer. */
   public Map<String, AppendRowsRequest.MissingValueInterpretation>
       getMissingValueInterpretationMap() {
     return this.schemaAwareStreamWriter.getMissingValueInterpretationMap();
@@ -213,9 +243,7 @@ public boolean isClosed() {
     return this.schemaAwareStreamWriter.isClosed();
   }
 
-  /**
-   * @return if user explicitly closed the writer.
-   */
+  /** @return if user explicitly closed the writer. */
   public boolean isUserClosed() {
     return this.schemaAwareStreamWriter.isUserClosed();
   }
@@ -444,9 +472,7 @@ public Builder setMissingValueInterpretationMap(
      * @return JsonStreamWriter
      */
     public JsonStreamWriter build()
-        throws Descriptors.DescriptorValidationException,
-            IllegalArgumentException,
-            IOException,
+        throws Descriptors.DescriptorValidationException, IllegalArgumentException, IOException,
             InterruptedException {
       return new JsonStreamWriter(this.schemaAwareStreamWriterBuilder);
     }

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/SchemaAwareStreamWriter.java

@@ -261,16 +261,12 @@ ApiFuture<AppendRowsResponse> appendWithUniqueId(
     }
   }
 
-  /**
-   * @return The name of the write stream associated with this writer.
-   */
+  /** @return The name of the write stream associated with this writer. */
   public String getStreamName() {
     return this.streamName;
   }
 
-  /**
-   * @return A unique Id for this writer.
-   */
+  /** @return A unique Id for this writer. */
   public String getWriterId() {
     return streamWriter.getWriterId();
   }
@@ -304,9 +300,7 @@ public long getInflightWaitSeconds() {
     return streamWriter.getInflightWaitSeconds();
   }
 
-  /**
-   * @return the missing value interpretation map used for the writer.
-   */
+  /** @return the missing value interpretation map used for the writer. */
   public Map<String, AppendRowsRequest.MissingValueInterpretation>
       getMissingValueInterpretationMap() {
     return streamWriter.getMissingValueInterpretationMap();
@@ -442,9 +436,7 @@ public boolean isClosed() {
     return this.streamWriter.isClosed();
   }
 
-  /**
-   * @return if user explicitly closed the writer.
-   */
+  /** @return if user explicitly closed the writer. */
   public boolean isUserClosed() {
     return this.streamWriter.isUserClosed();
   }
@@ -730,9 +722,7 @@ public Builder setEnableOpenTelemetry(boolean enableOpenTelemetry) {
      * @return SchemaAwareStreamWriter
      */
     public SchemaAwareStreamWriter<T> build()
-        throws DescriptorValidationException,
-            IllegalArgumentException,
-            IOException,
+        throws DescriptorValidationException, IllegalArgumentException, IOException,
             InterruptedException {
       return new SchemaAwareStreamWriter<>(this);
     }

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/StreamWriter.java

@@ -508,37 +508,27 @@ public long getInflightWaitSeconds() {
     return singleConnectionOrConnectionPool.getInflightWaitSeconds(this);
   }
 
-  /**
-   * @return a unique Id for the writer.
-   */
+  /** @return a unique Id for the writer. */
   public String getWriterId() {
     return singleConnectionOrConnectionPool.getWriterId(writerId);
   }
 
-  /**
-   * @return name of the Stream that this writer is working on.
-   */
+  /** @return name of the Stream that this writer is working on. */
   public String getStreamName() {
     return streamName;
   }
 
-  /**
-   * @return the passed in user schema.
-   */
+  /** @return the passed in user schema. */
   public ProtoSchema getProtoSchema() {
     return writerSchema;
   }
 
-  /**
-   * @return the location of the destination.
-   */
+  /** @return the location of the destination. */
   public String getLocation() {
     return location;
   }
 
-  /**
-   * @return the missing value interpretation map used for the writer.
-   */
+  /** @return the missing value interpretation map used for the writer. */
   public Map<String, AppendRowsRequest.MissingValueInterpretation>
       getMissingValueInterpretationMap() {
     return missingValueInterpretationMap;
@@ -559,9 +549,7 @@ public boolean isClosed() {
     }
   }
 
-  /**
-   * @return if user explicitly closed the writer.
-   */
+  /** @return if user explicitly closed the writer. */
   public boolean isUserClosed() {
     return userClosed.get();
   }
@@ -610,9 +598,7 @@ public static void setMaxRequestCallbackWaitTime(Duration waitTime) {
     ConnectionWorker.MAXIMUM_REQUEST_CALLBACK_WAIT_TIME = waitTime;
   }
 
-  /**
-   * @return the default stream name associated with tableName
-   */
+  /** @return the default stream name associated with tableName */
   public static String getDefaultStreamName(TableName tableName) {
     return tableName + defaultStreamMatching;
   }

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/JsonStreamWriter.java

@@ -58,9 +58,7 @@ public class JsonStreamWriter implements AutoCloseable {
    * @param builder The Builder object for the JsonStreamWriter
    */
   private JsonStreamWriter(Builder builder)
-      throws Descriptors.DescriptorValidationException,
-          IllegalArgumentException,
-          IOException,
+      throws Descriptors.DescriptorValidationException, IllegalArgumentException, IOException,
           InterruptedException {
     this.client = builder.client;
     this.descriptor =
@@ -333,9 +331,7 @@ public Builder setTraceId(String traceId) {
      * @return JsonStreamWriter
      */
     public JsonStreamWriter build()
-        throws Descriptors.DescriptorValidationException,
-            IllegalArgumentException,
-            IOException,
+        throws Descriptors.DescriptorValidationException, IllegalArgumentException, IOException,
             InterruptedException {
       return new JsonStreamWriter(this);
     }

google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/JsonStreamWriterTest.java

@@ -38,6 +38,8 @@
 import com.google.cloud.bigquery.storage.v1.Exceptions.AppendSerializationError;
 import com.google.cloud.bigquery.storage.v1.TableFieldSchema.Mode;
 import com.google.common.collect.ImmutableMap;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Descriptors.DescriptorValidationException;
 import com.google.protobuf.Int64Value;
@@ -251,6 +253,49 @@ public void testSingleAppendSimpleJson() throws Exception {
     }
   }
 
+  @Test
+  public void testSingleAppendSimpleGson() throws Exception {
+    FooType expectedProto = FooType.newBuilder().setFoo("allen").build();
+    JsonObject foo = new JsonObject();
+    foo.addProperty("foo", "allen");
+    JsonArray jsonArr = new JsonArray();
+    jsonArr.add(foo);
+
+    try (JsonStreamWriter writer =
+        getTestJsonStreamWriterBuilder(TEST_STREAM, TABLE_SCHEMA)
+            .setTraceId("test:empty")
+            .build()) {
+
+      testBigQueryWrite.addResponse(
+          AppendRowsResponse.newBuilder()
+              .setAppendResult(
+                  AppendRowsResponse.AppendResult.newBuilder().setOffset(Int64Value.of(0)).build())
+              .build());
+
+      ApiFuture<AppendRowsResponse> appendFuture = writer.append(jsonArr);
+      assertEquals(0L, appendFuture.get().getAppendResult().getOffset().getValue());
+      appendFuture.get();
+      assertEquals(
+          1,
+          testBigQueryWrite
+              .getAppendRequests()
+              .get(0)
+              .getProtoRows()
+              .getRows()
+              .getSerializedRowsCount());
+      assertEquals(
+          testBigQueryWrite
+              .getAppendRequests()
+              .get(0)
+              .getProtoRows()
+              .getRows()
+              .getSerializedRows(0),
+          expectedProto.toByteString());
+      assertEquals(
+          "java-jsonwriter test:empty", testBigQueryWrite.getAppendRequests().get(0).getTraceId());
+    }
+  }
+
   @Test
   public void testFlexibleColumnAppend() throws Exception {
     TableFieldSchema field =

google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/it/ITBigQueryBigDecimalByteStringEncoderTest.java

@@ -102,9 +102,7 @@ public static void afterClass() {
 
   @Test
   public void TestBigDecimalEncoding()
-      throws IOException,
-          InterruptedException,
-          ExecutionException,
+      throws IOException, InterruptedException, ExecutionException,
           Descriptors.DescriptorValidationException {
     TableName parent = TableName.of(ServiceOptions.getDefaultProjectId(), DATASET, TABLE);
     TableFieldSchema TEST_NUMERIC_ZERO =

google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/it/ITBigQueryTimeEncoderTest.java

@@ -109,9 +109,7 @@ public static void afterClass() {
 
   @Test
   public void TestTimeEncoding()
-      throws IOException,
-          InterruptedException,
-          ExecutionException,
+      throws IOException, InterruptedException, ExecutionException,
           Descriptors.DescriptorValidationException {
     TableName parent = TableName.of(ServiceOptions.getDefaultProjectId(), DATASET, TABLE);
     TableFieldSchema TEST_STRING =