[beam] branch master updated: Merge pull request #18374 from [BEAM-13945] Roll forward JSON support for BQIO

[pa...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

pabloem pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new b366d525f06 Merge pull request #18374 from [BEAM-13945] Roll forward JSON support for BQIO
b366d525f06 is described below

commit b366d525f06b4e95c7493d2a2908e14f4b86ea3c
Author: Pablo Estrada <pa...@users.noreply.github.com>
AuthorDate: Fri Jun 3 16:50:48 2022 -0400

    Merge pull request #18374 from [BEAM-13945] Roll forward JSON support for BQIO
    
    * Revert "Revert "Merge pull request #17492 from [BEAM-13945] (FIX) Update Java BQ connector to support new JSON type ""
    
    This reverts commit 688d1836614ebd25268142286307afa4804ba290.
    
    * cherry-pick @johnjcasey fix
---
 .../beam/sdk/io/gcp/bigquery/BigQueryIO.java       | 38 +++++++++++++++++++---
 1 file changed, 33 insertions(+), 5 deletions(-)

diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
index 463a5c80f87..aca23dc7f62 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
@@ -40,6 +40,10 @@ import com.google.cloud.bigquery.storage.v1.CreateReadSessionRequest;
 import com.google.cloud.bigquery.storage.v1.DataFormat;
 import com.google.cloud.bigquery.storage.v1.ReadSession;
 import com.google.cloud.bigquery.storage.v1.ReadStream;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
 import java.io.IOException;
 import java.util.Collections;
 import java.util.List;
@@ -2921,12 +2925,12 @@ public class BigQueryIO {
               "useAvroLogicalTypes can only be set with Avro output.");
         }
 
+        // Batch load jobs currently support JSON data insertion only with CSV files
         if (getJsonSchema() != null && getJsonSchema().isAccessible()) {
-          // Batch load jobs currently support JSON data insertion only with CSV files
-          checkArgument(
-              !getJsonSchema().get().contains("JSON"),
-              "Found JSON type in TableSchema. JSON data insertion is currently "
-                  + "not supported with batch loads.");
+          JsonElement schema = JsonParser.parseString(getJsonSchema().get());
+          if (!schema.getAsJsonObject().keySet().isEmpty()) {
+            validateNoJsonTypeInSchema(schema);
+          }
         }
 
         BatchLoads<DestinationT, T> batchLoads =
@@ -3009,6 +3013,30 @@ public class BigQueryIO {
       }
     }
 
+    private void validateNoJsonTypeInSchema(JsonElement schema) {
+      JsonElement fields = schema.getAsJsonObject().get("fields");
+      if (!fields.isJsonArray() || fields.getAsJsonArray().isEmpty()) {
+        return;
+      }
+
+      JsonArray fieldArray = fields.getAsJsonArray();
+
+      for (int i = 0; i < fieldArray.size(); i++) {
+        JsonObject field = fieldArray.get(i).getAsJsonObject();
+        checkArgument(
+            !field.get("type").getAsString().equals("JSON"),
+            "Found JSON type in TableSchema. JSON data insertion is currently "
+                + "not supported with 'FILE_LOADS' write method. This is supported with the "
+                + "other write methods, however. For more information, visit: "
+                + "https://cloud.google.com/bigquery/docs/reference/standard-sql/"
+                + "json-data#ingest_json_data");
+
+        if (field.get("type").getAsString().equals("STRUCT")) {
+          validateNoJsonTypeInSchema(field);
+        }
+      }
+    }
+
     @Override
     public void populateDisplayData(DisplayData.Builder builder) {
       super.populateDisplayData(builder);