Merge pull request data-integrations#1391 from cloudsufi/patch/bq-sou…

…rce-json-support [PLUGIN-1563] Add support in BQ Source to read JSON columns
cloudsufi · Apr 19, 2024 · 54077e6 · 54077e6
2 parents 444fe79 + 8cfba20
commit 54077e6
Show file tree

Hide file tree

Showing 4 changed files with 24 additions and 2 deletions.
diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/util/BigQueryDataParser.java b/src/main/java/io/cdap/plugin/gcp/bigquery/util/BigQueryDataParser.java
@@ -159,7 +159,9 @@ private static void addToRecordBuilder(StructuredRecord.Builder recordBuilder, S
      */
   public static Object convertValue(Field field, FieldValue fieldValue) {
     LegacySQLTypeName type = field.getType();
-    StandardSQLTypeName standardType = type.getStandardType();
+    // Treat JSON as string
+    StandardSQLTypeName standardType = LegacySQLTypeName.valueOf("JSON").equals(type) ?
+    StandardSQLTypeName.STRING : type.getStandardType();
     switch (standardType) {
       case TIME:
         return LocalTime.parse(fieldValue.getStringValue());

diff --git a/src/main/java/io/cdap/plugin/gcp/bigquery/util/BigQueryUtil.java b/src/main/java/io/cdap/plugin/gcp/bigquery/util/BigQueryUtil.java
@@ -292,7 +292,9 @@ public static Schema convertFieldType(Field field, @Nullable FailureCollector co
   public static Schema convertFieldType(Field field, @Nullable FailureCollector collector,
                                         @Nullable String recordPrefix) {
     LegacySQLTypeName type = field.getType();
-    StandardSQLTypeName standardType = type.getStandardType();
+    // Treat JSON as string
+    StandardSQLTypeName standardType = LegacySQLTypeName.valueOf("JSON").equals(type) ?
+    StandardSQLTypeName.STRING : type.getStandardType();
     switch (standardType) {
       case FLOAT64:
         // float is a float64, so corresponding type becomes double

diff --git a/src/test/java/io/cdap/plugin/gcp/bigquery/util/BigQueryDataParserTest.java b/src/test/java/io/cdap/plugin/gcp/bigquery/util/BigQueryDataParserTest.java
@@ -20,6 +20,7 @@
 import com.google.cloud.bigquery.FieldList;
 import com.google.cloud.bigquery.FieldValue;
 import com.google.cloud.bigquery.FieldValueList;
+import com.google.cloud.bigquery.LegacySQLTypeName;
 import com.google.cloud.bigquery.StandardSQLTypeName;
 import com.google.cloud.bigquery.TableResult;
 import com.google.common.io.BaseEncoding;
@@ -159,4 +160,13 @@ private String paddHeaddingZero(String value, int length) {
     }
     return Strings.repeat('0', length - value.length()) + value;
   }
+
+  @Test
+  public void testJsonFieldConversionToString() {
+    Field field = Field.newBuilder("demo", LegacySQLTypeName.valueOf("JSON")).build();
+    String jsonValue = "{\"key\":\"value\"}";
+    FieldValue fieldValue = FieldValue.of(FieldValue.Attribute.PRIMITIVE, jsonValue);
+    Object result = BigQueryDataParser.convertValue(field, fieldValue);
+    Assert.assertEquals(jsonValue, result);
+  }
 }
diff --git a/src/test/java/io/cdap/plugin/gcp/bigquery/util/BigQueryUtilTest.java b/src/test/java/io/cdap/plugin/gcp/bigquery/util/BigQueryUtilTest.java
@@ -17,6 +17,7 @@
 package io.cdap.plugin.gcp.bigquery.util;
 
 import com.google.cloud.bigquery.Field;
+import com.google.cloud.bigquery.LegacySQLTypeName;
 import com.google.cloud.bigquery.StandardSQLTypeName;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.cdap.etl.api.FailureCollector;
@@ -470,4 +471,11 @@ public void testJobLabelWithValueStartingWithCaptialLetter() {
             collector.getValidationFailures().get(0).getMessage());
   }
 
+  @Test
+  public void testConvertFieldTypeJsonToString() {
+      Field field = Field.newBuilder("demo", LegacySQLTypeName.valueOf("JSON")).build();
+      Schema expectedSchema = Schema.of(Schema.Type.STRING);
+      Schema result = BigQueryUtil.convertFieldType(field, null, null);
+      Assert.assertEquals(expectedSchema, result);
+  }
 }