From beedcee68741a107c764cfe85c69da8da6dbbcc0 Mon Sep 17 00:00:00 2001
From: Dede Indrapurna <dede@touchten.com>
Date: Mon, 7 Sep 2015 16:50:17 +0700
Subject: [PATCH] - Added support for RECORD type column, nested RECORD, and
 RECORD inside array. - Added "underscore-deep-extend" dependency. - Updated
 tests for the new functions. - Added .idea & *.iml in gitignore.

---
 .gitignore       |   4 +
 lib/gcp.js       | 129 +++++++++++++++++++++++++-------
 package.json     |   3 +-
 test/gcp_test.js | 189 ++++++++++++++++++++++++++++++++++++++++++++---
 4 files changed, 287 insertions(+), 38 deletions(-)

diff --git a/.gitignore b/.gitignore
index f59d45b..68dd8c2 100644
--- a/.gitignore
+++ b/.gitignore
@@ -29,3 +29,7 @@ node_modules
 
 # Working directory
 temp
+
+# IDE files
+.idea
+*.iml
\ No newline at end of file
diff --git a/lib/gcp.js b/lib/gcp.js
index b51388b..f11ae5d 100644
--- a/lib/gcp.js
+++ b/lib/gcp.js
@@ -3,11 +3,14 @@
 var util = require('util');
 var path = require('path');
 var _ = require('underscore');
+var underscoreDeepExtend = require('underscore-deep-extend');
 var gcloud = require('gcloud');
 var moment = require('moment');
 var ObjectID = require('mongodb').ObjectID;
 var through2 = require('through2');
 
+_.mixin({deepExtend: underscoreDeepExtend(_)});
+
 function GCP(opts) {
   this.projectSettings = {
     projectId: opts.project
@@ -154,6 +157,8 @@ GCP.BigQueryTable = {
           return item !== null;
         });
       }
+    } else if(v === Object(v)) {
+      return v;
     } else {
       return JSON.stringify(v);
     }
@@ -166,8 +171,12 @@ GCP.BigQueryTable = {
       return 'FLOAT';
     } else if (typeof v === 'boolean') {
       return 'BOOLEAN';
+    } else if(v instanceof ObjectID) {
+      return 'STRING';
     } else if (util.isArray(v) && v[0]) {
-      return this.type(v[0]);
+      return GCP.BigQueryTable.type(v[0]);
+    } else if (v === Object(v)) {
+      return 'RECORD';
     }
     return 'STRING';
   },
@@ -181,44 +190,101 @@ GCP.BigQueryTable = {
   },
 
   column: function (k, v) {
-    return {
-      name: GCP.safeName(k),
-      type: GCP.BigQueryTable.type(v),
-      mode: GCP.BigQueryTable.mode(v)
-    };
+    var colType = GCP.BigQueryTable.type(v);
+
+    if(colType === 'RECORD')
+      return {
+        name: GCP.safeName(k),
+        type: colType,
+        mode: GCP.BigQueryTable.mode(v),
+        fields: {}
+      };
+    else
+      return {
+        name: GCP.safeName(k),
+        type: colType,
+        mode: GCP.BigQueryTable.mode(v)
+      };
+
   },
 
   createJSONStream: function (fields, autoSchemaDetection) {
     var schema = {};
 
     return through2.obj(function (data, encoding, next) {
-      data.id = data._id;
-      delete data._id;
-
-      var out = {};
-      var keys = Object.keys(data);
-      var len = keys.length;
-      var i, k, v, col;
-
-      for (i = 0; i < len; ++i) {
-        k = keys[i];
-        v = data[k];
-
-        if (v !== null) {
-          col = GCP.safeName(k);
-          out[col] = GCP.BigQueryTable.value(v);
-          if (autoSchemaDetection && !schema[col]) {
-            schema[col] = GCP.BigQueryTable.column(k, v);
-          }
-        }
-      }
-      this.push(JSON.stringify(out) + '\n');
+      var out = GCP.BigQueryTable.parseJSONLine(Object.keys(data), data, autoSchemaDetection);
+      schema = _.deepExtend(schema, out.schema);
+      this.push(JSON.stringify(out.record) + '\n');
       next();
     }).on('end', function () {
       GCP.BigQueryTable.convertSchemaFields(schema, fields);
     });
   },
 
+  parseJSONLine: function(keys, data, autoSchemaDetection){
+    var record = {};
+    var schema = {};
+    var k, v, col;
+
+    for(var i=0; i<keys.length; i++){
+      k = keys[i];
+      v = data[k];
+
+      if (k !== '__v' && v !== null) {
+        col = GCP.safeName(k);
+        var colType = GCP.BigQueryTable.type(v);
+        var colMode = GCP.BigQueryTable.mode(v);
+        var subRecord;
+
+        if(autoSchemaDetection && !schema[col]) schema[col] = GCP.BigQueryTable.column(k, v);
+
+        if(colMode === 'REPEATED'){
+          var j;
+          record[col] = [];
+
+          if(colType === 'RECORD'){
+            for (j=0; j<v.length; j++){
+              if(!_.isEmpty(v[j])) {
+                subRecord = GCP.BigQueryTable.parseJSONLine(
+                    Object.keys(v[j]),
+                    v[j],
+                    autoSchemaDetection
+                );
+                record[col].push(subRecord.record);
+                if(autoSchemaDetection) schema[col].fields = _.deepExtend(schema[col].fields, subRecord.schema);
+              }
+            }
+          }
+          else{
+            for (j=0; j<v.length; j++){
+              record[col].push(GCP.BigQueryTable.value(v[j]));
+            }
+          }
+
+          if(_.isEmpty(record[col])){
+            record = _.omit(record, col);
+            if(autoSchemaDetection) schema = _.omit(schema, col);
+          }
+        }
+        else if(colType === 'RECORD'){
+          if(!_.isEmpty(v)) {
+            subRecord = GCP.BigQueryTable.parseJSONLine(Object.keys(v), v, autoSchemaDetection);
+            record[col] = subRecord.record;
+            if(autoSchemaDetection) schema[col].fields = subRecord.schema;
+          }
+        }
+        else{
+          record[col] = GCP.BigQueryTable.value(v);
+        }
+      }
+    }
+
+    return {
+      record: record,
+      schema: schema
+    };
+  },
+
   convertSchemaFields: function (schema, fields) {
     var keys = Object.keys(schema);
     keys.sort(function(a, b) {
@@ -231,7 +297,14 @@ GCP.BigQueryTable = {
       }
     });
     keys.forEach(function (key) {
-      fields.push(schema[key]);
+      if(schema[key].fields){
+        var val = _.clone(schema[key]);
+        val.fields = [];
+        GCP.BigQueryTable.convertSchemaFields(schema[key].fields, val.fields);
+        fields.push(val);
+      }
+      else
+        fields.push(schema[key]);
     });
   }
 };
diff --git a/package.json b/package.json
index f40eb44..bf7d992 100644
--- a/package.json
+++ b/package.json
@@ -14,7 +14,8 @@
     "moment": "^2.8.4",
     "mongodb": "^2.0.40",
     "through2": "^0.6.5",
-    "underscore": "^1.7.0"
+    "underscore": "^1.7.0",
+    "underscore-deep-extend": "0.0.5"
   },
   "devDependencies": {
     "mkdirp": "^0.5.0",
diff --git a/test/gcp_test.js b/test/gcp_test.js
index 38a9b97..aea2ddf 100644
--- a/test/gcp_test.js
+++ b/test/gcp_test.js
@@ -1,5 +1,7 @@
 var assert = require('assert');
 var GCP = require('../lib/gcp');
+var moment = require('moment');
+var _ = require('underscore');
 var ObjectID = require('mongodb').ObjectID;
 
 describe('GCP', function () {
@@ -71,18 +73,18 @@ describe('GCP', function () {
         });
       });
 
-      it ('should return array when value is array that include object', function () {
+      it ('should return array of objects when value is array that include object', function () {
         var ary = [{key: 'val1'}, {key: 'val2'}];
         var ret = GCP.BigQueryTable.value(ary);
 
         ary.forEach(function (item, i) {
-          assert.equal(ret[i], JSON.stringify(ary[i]));
+          assert.equal(ret[i], Object(ary[i]));
         });
       });
 
-      it ('should return JSON string when value is object', function () {
+      it ('should return object when value is object', function () {
         var obj = { key: 'val' };
-        assert.equal(GCP.BigQueryTable.value(obj), JSON.stringify(obj));
+        assert.equal(GCP.BigQueryTable.value(obj), Object(obj));
       });
     });
 
@@ -117,9 +119,9 @@ describe('GCP', function () {
         assert.equal(GCP.BigQueryTable.type(ary), 'STRING');
       });
 
-      it ('should return STRING when value is object', function () {
+      it ('should return RECORD when value is object', function () {
         var obj = { key: 'val' };
-        assert.equal(GCP.BigQueryTable.type(obj), 'STRING');
+        assert.equal(GCP.BigQueryTable.type(obj), 'RECORD');
       });
     });
 
@@ -143,6 +145,28 @@ describe('GCP', function () {
         assert.equal(ret.type, 'STRING');
         assert.equal(ret.mode, 'NULLABLE');
       });
+
+      it ('should also include fields if val type is RECORD', function() {
+        var key = 'abc';
+        var val = {a: 123, b: "bbb", c: new Date()};
+        var ret = GCP.BigQueryTable.column(key, val);
+
+        assert.equal(ret.name, 'abc');
+        assert.equal(ret.type, 'RECORD');
+        assert.equal(ret.mode, 'NULLABLE');
+        assert.equal(ret.fields, Object(ret.fields));
+      });
+
+      it ('should be able to detect array value', function (){
+        var key = 'a';
+        var val = [{'one': 1, 'two': 2}, {'one': 11, 'two': 22}];
+        var ret = GCP.BigQueryTable.column(key, val);
+
+        assert.equal(ret.name, 'a');
+        assert.equal(ret.type, 'RECORD');
+        assert.equal(ret.mode, 'REPEATED');
+        assert.equal(ret.fields, Object(ret.fields));
+      });
     });
 
     describe('#convertSchemaFields', function () {
@@ -156,17 +180,71 @@ describe('GCP', function () {
         var schema = {
           abc: {
             "name": "abc",
-            "type": "STRING"
+            "type": "RECORD",
+            "mode": "NULLABLE",
+            "fields": {
+              y: {
+                "name": "y",
+                "type": "DATE",
+                "mode": "NULLABLE"
+              },
+              z: {
+                "name": "z",
+                "type": "STRING",
+                "mode": "NULLABLE"
+              },
+              x: {
+                "name": "x",
+                "type": "RECORD",
+                "mode": "REQUIRED",
+                "fields": {
+                  zz: {
+                    "name": "zz",
+                    "type": "BOOLEAN",
+                    "mode": "NULLABLE"
+                  },
+                  id: {
+                    "name": "id",
+                    "type": "INTEGER",
+                    "mode": "REQUIRED"
+                  },
+                  aa: {
+                    "name": "aa",
+                    "type": "STRING",
+                    "mode": "REPEATED"
+                  }
+                }
+              }
+            }
           },
 
           id: {
             "name": "id",
-            "type": "STRING"
+            "type": "STRING",
+            "mode": "REQUIRED"
           },
 
           def: {
             "name": "def",
-            "type": "STRING"
+            "type": "RECORD",
+            "mode": "REPEATED",
+            "fields": {
+              a: {
+                "name": "a",
+                "type": "INTEGER",
+                "mode": "NULLABLE"
+              },
+              b: {
+                "name": "b",
+                "type": "FLOAT",
+                "mode": "REQUIRED"
+              },
+              c: {
+                "name": "c",
+                "type": "STRING",
+                "mode": "REQUIRED"
+              }
+            }
           }
         };
 
@@ -176,7 +254,100 @@ describe('GCP', function () {
         assert.equal(fields[0].name, 'id');
         assert.equal(fields[1].name, 'abc');
         assert.equal(fields[2].name, 'def');
+        assert.equal(fields[1].fields.length, 3);
+        assert.equal(fields[1].fields[0].name, 'x');
+        assert.equal(fields[1].fields[0].fields.length, 3);
+        assert.equal(fields[1].fields[0].fields[0].name, 'id');
+        assert.equal(fields[1].fields[0].fields[1].name, 'aa');
+        assert.equal(fields[1].fields[0].fields[2].name, 'zz');
+        assert.equal(fields[1].fields[1].name, 'y');
+        assert.equal(fields[1].fields[2].name, 'z');
+        assert.equal(fields[2].fields.length, 3);
+        assert.equal(fields[2].fields[0].name, 'a');
+        assert.equal(fields[2].fields[1].name, 'b');
+        assert.equal(fields[2].fields[2].name, 'c');
       });
     });
+
+    describe ('#parseJSONLines', function(){
+      it ('should parse recursive JSON objects', function() {
+        var data = {
+          "_id" : ObjectID("5553566950c403ef04f55d76"),
+          "player" : {
+            "status" : "active",
+            "email" : "test@email.com",
+            "photoURL" : "https://google.com",
+            "name" : "Just a test name",
+            "_id" : ObjectID("5551c5ee7a19fd8a05ebf6cf")
+          },
+          "app" : {
+            "company" : ObjectID("00000000000000000000000a"),
+            "facebookAppId" : "1234567890",
+            "num" : 1,
+            "status" : "active",
+            "countries": [],
+            "_id" : ObjectID("55487a238a5458da3701669e")
+          },
+          "created" : {
+            "at" : moment("2015-05-13T13:49:29.907+0000").toDate(),
+            "ref" : "Player",
+            "user" : ObjectID("5551c5ee7a19fd8a05ebf6cf")
+          },
+          "rewardIds" : [
+              12345,
+              67890,
+              -4444
+          ],
+          "localPrice" : 100,
+          "survey" : {
+            "questions" : [
+              "Your mobile number",
+              "What is this?",
+              "Have you try this?"
+            ],
+            "answers" : [
+              {
+                "answer" : [
+                  "08571234567"
+                ]
+              },
+              {
+                "answer" : [
+                  "Reward Voucher"
+                ]
+              },
+              {
+                "answer" : [
+                  "Nope"
+                ]
+              }
+            ],
+            "_id" : ObjectID("554c9e240e6899890a3d0d28")
+          },
+          "code" : "REWARDCODE",
+          "shippingAddress" : null,
+          "status" : "claimed",
+          "__v" : 0
+        };
+
+        var ret = GCP.BigQueryTable.parseJSONLine(Object.keys(data), data, true);
+
+        assert.equal(ret.record, Object(ret.record));
+        assert.equal(ret.schema, Object(ret.schema));
+        assert.equal(ret.record.id, "5553566950c403ef04f55d76");
+        assert.equal(ret.schema.id.type, "STRING");
+        assert.equal(ret.record.app.countries, null);
+        assert.equal(ret.schema.app.fields.countries, null);
+        assert.equal(ret.record.created.at, "2015-05-13 13:49:29");
+        assert.equal(ret.schema.created.fields.at.type, "TIMESTAMP");
+        assert.equal(ret.record.rewardids[0], 12345);
+        assert.equal(ret.schema.rewardids.mode, "REPEATED");
+        assert.equal(ret.record.survey.answers.length, 3);
+        assert.equal(ret.record.survey.answers[0].answer[0], "08571234567");
+        assert.equal(ret.schema.survey.fields.answers.fields.answer.type, "STRING");
+        assert.equal(ret.record.__v, null);
+        assert.equal(ret.schema.__v, null);
+      })
+    });
   });
 });