Support toJSON() on read methods.

src/database.js

@@ -862,6 +862,9 @@ Database.prototype.runStream = function(query, options) {
     };
   }
 
+  delete reqOpts.toJSON;
+  delete reqOpts.toJSONOptions;
+
   function makeRequest(resumeToken) {
     return self.pool_.requestStream({
       client: 'SpannerClient',
@@ -870,7 +873,10 @@ Database.prototype.runStream = function(query, options) {
     });
   }
 
-  return new PartialResultStream(makeRequest);
+  return new PartialResultStream(makeRequest, {
+    toJSON: query.toJSON,
+    toJSONOptions: query.toJSONOptions,
+  });
 };
 
 /**

src/partial-result-stream.js

@@ -19,6 +19,8 @@
 var checkpointStream = require('checkpoint-stream');
 var chunk = require('lodash.chunk');
 var eventsIntercept = require('events-intercept');
+var exec = require('methmeth');
+var extend = require('extend');
 var is = require('is');
 var mergeStream = require('merge-stream');
 var split = require('split-array-stream');
@@ -43,10 +45,12 @@ var RowBuilder = require('./row-builder.js');
  *     receive one argument, `resumeToken`, which should be used however is
  *     necessary to send to the API for additional requests.
  */
-function partialResultStream(requestFn) {
+function partialResultStream(requestFn, options) {
   var lastResumeToken;
   var activeRequestStream;
 
+  options = extend({toJSON: false}, options);
+
   // mergeStream allows multiple streams to be connected into one. This is good
   // if we need to retry a request and pipe more data to the user's stream.
   var requestsStream = mergeStream();
@@ -104,6 +108,12 @@ function partialResultStream(requestFn) {
 
       rowChunks = [];
 
+      if (options.toJSON) {
+        formattedRows = formattedRows.map(
+          exec('toJSON', options.toJSONOptions)
+        );
+      }
+
       split(formattedRows, userStream, function() {
         next();
       });

src/transaction-request.js

@@ -244,6 +244,9 @@ TransactionRequest.prototype.createReadStream = function(table, query) {
     query
   );
 
+  delete reqOpts.toJSON;
+  delete reqOpts.toJSONOptions;
+
   if (this.transaction && this.id) {
     reqOpts.transaction = {
       id: this.id,
@@ -293,7 +296,10 @@ TransactionRequest.prototype.createReadStream = function(table, query) {
     });
   }
 
-  return new PartialResultStream(makeRequest);
+  return new PartialResultStream(makeRequest, {
+    toJSON: query.toJSON,
+    toJSONOptions: query.toJSONOptions,
+  });
 };
 
 /**

system-test/spanner.js

@@ -946,6 +946,76 @@ describe('Spanner', function() {
       );
     });
 
+    it('should automatically convert to JSON', function(done) {
+      var id = generateName('id');
+      var name = generateName('name');
+
+      table.insert(
+        {
+          SingerId: id,
+          Name: name,
+        },
+        function(err) {
+          assert.ifError(err);
+
+          var rows = [];
+
+          table
+            .createReadStream({
+              keys: [id],
+              columns: ['SingerId', 'name'],
+              toJSON: true,
+            })
+            .on('error', done)
+            .on('data', function(row) {
+              rows.push(row);
+            })
+            .on('end', function() {
+              assert.deepEqual(rows, [
+                {
+                  SingerId: id,
+                  Name: name,
+                },
+              ]);
+
+              done();
+            });
+        }
+      );
+    });
+
+    it('should automatically convert to JSON with options', function(done) {
+      var id = generateName('id');
+
+      table.insert(
+        {
+          SingerId: id,
+          Int: 8,
+        },
+        function(err) {
+          assert.ifError(err);
+
+          var rows = [];
+
+          table
+            .createReadStream({
+              keys: [id],
+              columns: ['SingerId', 'Int'],
+              toJSON: true,
+              toJSONOptions: {wrapNumbers: true},
+            })
+            .on('error', done)
+            .on('data', function(row) {
+              rows.push(row);
+            })
+            .on('end', function() {
+              assert.strictEqual(rows[0].Int.value, '8');
+              done();
+            });
+        }
+      );
+    });
+
     it('should insert and delete a row', function(done) {
       var id = generateName('id');
       var name = generateName('name');

test/database.js

@@ -623,6 +623,34 @@ describe('Database', function() {
       assert(stream instanceof FakePartialResultStream);
     });
 
+    it('should pass toJSON, toJSONOptions to PartialResultStream', function() {
+      var query = extend({}, QUERY);
+      query.toJSON = {};
+      query.toJSONOptions = {};
+
+      var stream = database.runStream(query);
+      assert.deepStrictEqual(stream.calledWith_[1], {
+        toJSON: query.toJSON,
+        toJSONOptions: query.toJSONOptions,
+      });
+    });
+
+    it('should not pass toJSON, toJSONOptions to request', function(done) {
+      database.pool_.requestStream = function(config) {
+        assert.strictEqual(config.reqOpts.toJSON, undefined);
+        assert.strictEqual(config.reqOpts.toJSONOptions, undefined);
+        done();
+      };
+
+      var query = extend({}, QUERY);
+      query.toJSON = {};
+      query.toJSONOptions = {};
+
+      var stream = database.runStream(query);
+      var makeRequestFn = stream.calledWith_[0];
+      makeRequestFn();
+    });
+
     it('should assign a resumeToken to the request', function(done) {
       var resumeToken = 'resume-token';
 

test/partial-result-stream.js

@@ -244,6 +244,33 @@ describe('PartialResultStream', function() {
       );
     });
 
+    it('should return the formatted row as JSON', function(done) {
+      var options = {
+        toJSON: true,
+        toJSONOptions: {},
+      };
+
+      var partialResultStream = partialResultStreamModule(function() {
+        return fakeRequestStream;
+      }, options);
+
+      var formattedRow = {
+        toJSON: function(options_) {
+          assert.strictEqual(options_, options.toJSONOptions);
+          done();
+        },
+      };
+
+      partialResultStreamModule.formatRow_ = function() {
+        return formattedRow;
+      };
+
+      fakeRequestStream.push(RESULT_WITH_TOKEN);
+      fakeRequestStream.push(null);
+
+      partialResultStream.on('error', done).resume();
+    });
+
     it('should separately emit formatted rows', function(done) {
       var formattedRows = [{}, {}];
 

test/transaction-request.js

@@ -536,6 +536,36 @@ describe('TransactionRequest', function() {
         var makeRequestFn = stream.calledWith_[0];
         makeRequestFn(resumeToken);
       });
+
+      it('should accept toJSON and toJSONOptions', function() {
+        var query = {
+          toJSON: {},
+          toJSONOptions: {},
+        };
+
+        var stream = transactionRequest.createReadStream(TABLE, query);
+        var streamOptions = stream.calledWith_[1];
+
+        assert.strictEqual(streamOptions.toJSON, query.toJSON);
+        assert.strictEqual(streamOptions.toJSONOptions, query.toJSONOptions);
+      });
+
+      it('should delete toJSON, toJSONOptions from reqOpts', function(done) {
+        var query = {
+          toJSON: {},
+          toJSONOptions: {},
+        };
+
+        transactionRequest.requestStream = function(config) {
+          assert.strictEqual(config.reqOpts.toJSON, undefined);
+          assert.strictEqual(config.reqOpts.toJSONOptions, undefined);
+          done();
+        };
+
+        var stream = transactionRequest.createReadStream(TABLE, query);
+        var makeRequestFn = stream.calledWith_[0];
+        makeRequestFn();
+      });
     });
   });