Merge pull request #1402 from cloudsufi/cherry-pick-8cfba2024019705cebc8864492b73d6883f11134

[🍒][PLUGIN-1563] Add support in BQ Source to read JSON columns

Author: itsankit-google

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());

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

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);
+  }
 }

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);
+  }
 }