bigquery: allow AVRO format (#1380)

Author: stephenplusplus

lib/bigquery/table.js

@@ -54,6 +54,18 @@ var streamRouter = require('../common/stream-router.js');
  */
 var util = require('../common/util.js');
 
+/**
+ * The file formats accepted by BigQuery.
+ *
+ * @type {object}
+ * @private
+ */
+var FORMATS = {
+  avro: 'AVRO',
+  csv: 'CSV',
+  json: 'NEWLINE_DELIMITED_JSON'
+};
+
 /*! Developer Documentation
  *
  * @param {module:bigquery/dataset} dataset - Dataset instance.
@@ -293,7 +305,7 @@ Table.mergeSchemaWithRows_ = function(schema, rows) {
  * table.copy(yourTable, metadata, function(err, job, apiResponse) {});
  */
 Table.prototype.copy = function(destination, metadata, callback) {
-  var that = this;
+  var self = this;
 
   if (!(destination instanceof Table)) {
     throw new Error('Destination must be a Table object.');
@@ -331,7 +343,7 @@ Table.prototype.copy = function(destination, metadata, callback) {
       return;
     }
 
-    var job = that.bigQuery.job(resp.jobReference.jobId);
+    var job = self.bigQuery.job(resp.jobReference.jobId);
     job.metadata = resp;
 
     callback(null, job, resp);
@@ -361,8 +373,8 @@ Table.prototype.createReadStream = function() {
 };
 
 /**
- * Load data into your table from a readable stream of JSON or CSV-formatted
- * data.
+ * Load data into your table from a readable stream of JSON, CSV, or
+ * AVRO data.
  *
  * @resource [Jobs: insert API Documentation]{@link https://cloud.google.com/bigquery/docs/reference/v2/jobs/insert}
  *
@@ -404,21 +416,17 @@ Table.prototype.createReadStream = function() {
  *   .on('complete', function(job) {});
  */
 Table.prototype.createWriteStream = function(metadata) {
-  var that = this;
+  var self = this;
 
   metadata = metadata || {};
 
-  var fileTypeMap = {
-    csv: 'CSV',
-    json: 'NEWLINE_DELIMITED_JSON'
-  };
-  var fileTypes = Object.keys(fileTypeMap).map(function(key) {
-    return fileTypeMap[key];
+  var fileTypes = Object.keys(FORMATS).map(function(key) {
+    return FORMATS[key];
   });
 
   if (is.string(metadata)) {
     metadata = {
-      sourceFormat: fileTypeMap[metadata.toLowerCase()]
+      sourceFormat: FORMATS[metadata.toLowerCase()]
     };
   }
 
@@ -428,9 +436,9 @@ Table.prototype.createWriteStream = function(metadata) {
 
   extend(true, metadata, {
     destinationTable: {
-      projectId: that.bigQuery.projectId,
-      datasetId: that.dataset.id,
-      tableId: that.id
+      projectId: self.bigQuery.projectId,
+      datasetId: self.dataset.id,
+      tableId: self.id
     }
   });
 
@@ -443,7 +451,7 @@ Table.prototype.createWriteStream = function(metadata) {
 
   dup.once('writing', function() {
     util.makeWritableStream(dup, {
-      makeAuthenticatedRequest: that.bigQuery.makeAuthenticatedRequest,
+      makeAuthenticatedRequest: self.bigQuery.makeAuthenticatedRequest,
       metadata: {
         configuration: {
           load: metadata
@@ -452,11 +460,11 @@ Table.prototype.createWriteStream = function(metadata) {
       request: {
         uri: format('{base}/{projectId}/jobs', {
           base: 'https://www.googleapis.com/upload/bigquery/v2/projects',
-          projectId: that.bigQuery.projectId
+          projectId: self.bigQuery.projectId
         })
       }
     }, function(data) {
-      var job = that.bigQuery.job(data.jobReference.jobId);
+      var job = self.bigQuery.job(data.jobReference.jobId);
       job.metadata = data;
 
       dup.emit('complete', job);
@@ -523,19 +531,13 @@ Table.prototype.createWriteStream = function(metadata) {
  * ], options, function(err, job, apiResponse) {});
  */
 Table.prototype.export = function(destination, options, callback) {
-  var that = this;
+  var self = this;
 
   if (is.fn(options)) {
     callback = options;
     options = {};
   }
 
-  var formats = {
-    avro: 'AVRO',
-    csv: 'CSV',
-    json: 'NEWLINE_DELIMITED_JSON'
-  };
-
   extend(true, options, {
     destinationUris: arrify(destination).map(function(dest) {
       if (!(dest instanceof File)) {
@@ -545,8 +547,8 @@ Table.prototype.export = function(destination, options, callback) {
       // If no explicit format was provided, attempt to find a match from the
       // file's extension. If no match, don't set, and default upstream to CSV.
       var format = path.extname(dest.name).substr(1).toLowerCase();
-      if (!options.destinationFormat && !options.format && formats[format]) {
-        options.destinationFormat = formats[format];
+      if (!options.destinationFormat && !options.format && FORMATS[format]) {
+        options.destinationFormat = FORMATS[format];
       }
 
       return 'gs://' + dest.bucket.name + '/' + dest.name;
@@ -556,8 +558,8 @@ Table.prototype.export = function(destination, options, callback) {
   if (options.format) {
     options.format = options.format.toLowerCase();
 
-    if (formats[options.format]) {
-      options.destinationFormat = formats[options.format];
+    if (FORMATS[options.format]) {
+      options.destinationFormat = FORMATS[options.format];
       delete options.format;
     } else {
       throw new Error('Destination format not recognized: ' + options.format);
@@ -591,7 +593,7 @@ Table.prototype.export = function(destination, options, callback) {
       return;
     }
 
-    var job = that.bigQuery.job(resp.jobReference.jobId);
+    var job = self.bigQuery.job(resp.jobReference.jobId);
     job.metadata = resp;
 
     callback(null, job, resp);
@@ -658,7 +660,7 @@ Table.prototype.export = function(destination, options, callback) {
  *   });
  */
 Table.prototype.getRows = function(options, callback) {
-  var that = this;
+  var self = this;
 
   if (is.fn(options)) {
     callback = options;
@@ -684,9 +686,9 @@ Table.prototype.getRows = function(options, callback) {
       });
     }
 
-    if (resp.rows && resp.rows.length > 0 && !that.metadata.schema) {
+    if (resp.rows && resp.rows.length > 0 && !self.metadata.schema) {
       // We don't know the schema for this table yet. Do a quick stat.
-      that.getMetadata(function(err, metadata, apiResponse) {
+      self.getMetadata(function(err, metadata, apiResponse) {
         if (err) {
           onComplete(err, null, null, apiResponse);
           return;
@@ -707,7 +709,7 @@ Table.prototype.getRows = function(options, callback) {
       return;
     }
 
-    rows = Table.mergeSchemaWithRows_(that.metadata.schema, rows || []);
+    rows = Table.mergeSchemaWithRows_(self.metadata.schema, rows || []);
 
     callback(null, rows, nextQuery, resp);
   }
@@ -720,16 +722,17 @@ Table.prototype.getRows = function(options, callback) {
  * asynchronously. If you would like instantaneous access to your data, insert
  * it using {module:bigquery/table#insert}.
  *
- * Note: Only JSON and CSV source files are supported. The file type will be
- * inferred by the given file's extension. If you wish to override this, you
- * must provide a `metadata` object.
+ * Note: The file type will be inferred by the given file's extension. If you
+ * wish to override this, you must provide `metadata.format`.
  *
  * @resource [Jobs: insert API Documentation]{@link https://cloud.google.com/bigquery/docs/reference/v2/jobs/insert}
  *
  * @param {string|module:storage/file} source - The source file to import.
  * @param {object=} metadata - Metadata to set with the load operation. The
  *     metadata object should be in the format of the
  *     [`configuration.load`](http://goo.gl/BVcXk4) property of a Jobs resource.
+ * @param {string} metadata.format - The format the data being imported is in.
+ *     Allowed options are "CSV", "JSON", or "AVRO".
  * @param {function} callback - The callback function.
  * @param {?error} callback.err - An error returned while making this request
  * @param {module:bigquery/job} callback.job - The job used to import your data.
@@ -772,7 +775,7 @@ Table.prototype.getRows = function(options, callback) {
  * ], function(err, job, apiResponse) {});
  */
 Table.prototype.import = function(source, metadata, callback) {
-  var that = this;
+  var self = this;
 
   if (is.fn(metadata)) {
     callback = metadata;
@@ -782,17 +785,18 @@ Table.prototype.import = function(source, metadata, callback) {
   callback = callback || util.noop;
   metadata = metadata || {};
 
-  var formats = {
-    csv: 'CSV',
-    json: 'NEWLINE_DELIMITED_JSON'
-  };
+  var format = metadata.sourceFormat || metadata.format;
+  if (format) {
+    metadata.sourceFormat = FORMATS[format.toLowerCase()];
+    delete metadata.format;
+  }
 
   if (is.string(source)) {
     // A path to a file was given. If a sourceFormat wasn't specified, try to
     // find a match from the file's extension.
-    var format = formats[path.extname(source).substr(1).toLowerCase()];
-    if (!metadata.sourceFormat && format) {
-      metadata.sourceFormat = format;
+    var detectedFormat = FORMATS[path.extname(source).substr(1).toLowerCase()];
+    if (!metadata.sourceFormat && detectedFormat) {
+      metadata.sourceFormat = detectedFormat;
     }
 
     // Read the file into a new write stream.
@@ -826,7 +830,7 @@ Table.prototype.import = function(source, metadata, callback) {
       // If no explicit format was provided, attempt to find a match from
       // the file's extension. If no match, don't set, and default upstream
       // to CSV.
-      var format = formats[path.extname(src.name).substr(1).toLowerCase()];
+      var format = FORMATS[path.extname(src.name).substr(1).toLowerCase()];
       if (!metadata.sourceFormat && format) {
         body.configuration.load.sourceFormat = format;
       }
@@ -845,7 +849,7 @@ Table.prototype.import = function(source, metadata, callback) {
       return;
     }
 
-    var job = that.bigQuery.job(resp.jobReference.jobId);
+    var job = self.bigQuery.job(resp.jobReference.jobId);
     job.metadata = resp;
 
     callback(null, job, resp);
@@ -1028,7 +1032,7 @@ Table.prototype.query = function(query, callback) {
  * table.setMetadata(metadata, function(err, metadata, apiResponse) {});
  */
 Table.prototype.setMetadata = function(metadata, callback) {
-  var that = this;
+  var self = this;
 
   if (metadata.name) {
     metadata.friendlyName = metadata.name;
@@ -1049,7 +1053,7 @@ Table.prototype.setMetadata = function(metadata, callback) {
       return;
     }
 
-    that.metadata = resp;
+    self.metadata = resp;
 
     callback(null, resp);
   });