From 8cfba2024019705cebc8864492b73d6883f11134 Mon Sep 17 00:00:00 2001
From: psainics <pushpender.saini@cloudsufi.com>
Date: Thu, 14 Mar 2024 14:49:10 +0530
Subject: [PATCH] Support Json

---
 .../plugin/gcp/bigquery/util/BigQueryDataParser.java   |  4 +++-
 .../io/cdap/plugin/gcp/bigquery/util/BigQueryUtil.java |  4 +++-
 .../gcp/bigquery/util/BigQueryDataParserTest.java      | 10 ++++++++++
 .../plugin/gcp/bigquery/util/BigQueryUtilTest.java     |  8 ++++++++
 4 files changed, 24 insertions(+), 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
index 2a6c074608..d8f1f80873 100644
--- 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
index fbcdb6398a..05939df878 100644
--- 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
index f7829bdf91..b75415f135 100644
--- 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
index 41d9c73a46..a09048e2d9 100644
--- 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);
+  }
 }