Added support for RECORD

.gitignore

@@ -29,3 +29,7 @@ node_modules
 
 # Working directory
 temp
+
+# IDE files
+.idea
+*.iml

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

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",