Revert "updating the default string to gson object for mongodb to big…

…query da…" (GoogleCloudPlatform#1850)

This reverts commit bf05990.

v2/mongodb-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/mongodb/templates/MongoDbUtils.java

@@ -21,8 +21,6 @@
 import com.google.api.services.bigquery.model.TableRow;
 import com.google.api.services.bigquery.model.TableSchema;
 import com.google.gson.Gson;
-import com.google.gson.JsonObject;
-import com.google.gson.reflect.TypeToken;
 import com.mongodb.client.MongoClient;
 import com.mongodb.client.MongoClients;
 import com.mongodb.client.MongoCollection;
@@ -39,7 +37,6 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
-import java.util.Map;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 import javax.script.Invocable;
@@ -84,7 +81,7 @@ public static TableSchema getTableFieldSchema(
           });
     } else {
       bigquerySchemaFields.add(new TableFieldSchema().setName("id").setType("STRING"));
-      bigquerySchemaFields.add(new TableFieldSchema().setName("source_data").setType("JSON"));
+      bigquerySchemaFields.add(new TableFieldSchema().setName("source_data").setType("STRING"));
     }
     bigquerySchemaFields.add(new TableFieldSchema().setName("timestamp").setType("TIMESTAMP"));
     TableSchema bigquerySchema = new TableSchema().setFields(bigquerySchemaFields);
@@ -148,14 +145,8 @@ public static TableRow getTableSchema(Document document, String userOption) {
     } else {
       LocalDateTime localdate = LocalDateTime.now(ZoneId.of("UTC"));
       String sourceData = GSON.toJson(document);
-      JsonObject sourceDataJsonObject = GSON.toJsonTree(document).getAsJsonObject();
-
-      // Convert to a Map
-      Map<String, Object> sourceDataMap =
-          GSON.fromJson(sourceDataJsonObject, new TypeToken<Map<String, Object>>() {}.getType());
-
       row.set("id", document.get("_id").toString())
-          .set("source_data", sourceDataMap)
+          .set("source_data", sourceData)
           .set("timestamp", localdate.format(TIMEFORMAT));
     }
     return row;
@@ -204,7 +195,7 @@ public static TableSchema getTableFieldSchemaForUDF(
           });
     } else {
       bigquerySchemaFields.add(new TableFieldSchema().setName("id").setType("STRING"));
-      bigquerySchemaFields.add(new TableFieldSchema().setName("source_data").setType("JSON"));
+      bigquerySchemaFields.add(new TableFieldSchema().setName("source_data").setType("STRING"));
     }
 
     bigquerySchemaFields.add(new TableFieldSchema().setName("timestamp").setType("TIMESTAMP"));