feat(bigquery): allow the user to ask for skipping parsing rows when querying (#7848)

Author: rafal-hawrylak

handwritten/bigquery/src/bigquery.ts

@@ -128,6 +128,7 @@ export type Query = JobRequest<bigquery.IJobConfigurationQuery> & {
   pageToken?: string;
   wrapIntegers?: boolean | IntegerTypeCastOptions;
   parseJSON?: boolean;
+  skipParsing?: boolean;
   // Overrides default job creation mode set on the client.
   jobCreationMode?: JobCreationMode;
 };
@@ -2217,6 +2218,7 @@ export class BigQuery extends Service {
         ? {
             wrapIntegers: query.wrapIntegers,
             parseJSON: query.parseJSON,
+            skipParsing: query.skipParsing,
           }
         : {};
     const callback =
@@ -2268,12 +2270,16 @@ export class BigQuery extends Service {
               'formatOptions.useInt64Timestamp':
                 queryReq.formatOptions?.useInt64Timestamp,
             };
-            rows = BigQuery.mergeSchemaWithRows_(res.schema, res.rows, {
-              wrapIntegers: options.wrapIntegers || false,
-              parseJSON: options.parseJSON,
-              listParams,
-            });
-            delete res.rows;
+            if (options.skipParsing) {
+              rows = res.rows;
+            } else {
+              rows = BigQuery.mergeSchemaWithRows_(res.schema, res.rows, {
+                wrapIntegers: options.wrapIntegers || false,
+                parseJSON: options.parseJSON,
+                listParams,
+              });
+              delete res.rows;
+            }
           } catch (e) {
             (callback as SimpleQueryRowsCallback)(e as Error, null, job);
             return;

handwritten/bigquery/src/job.ts

@@ -51,6 +51,7 @@ export type QueryResultsOptions = {
   job?: Job;
   wrapIntegers?: boolean | IntegerTypeCastOptions;
   parseJSON?: boolean;
+  skipParsing?: boolean;
 } & PagedRequest<bigquery.jobs.IGetQueryResultsParams> & {
     /**
      * internal properties
@@ -602,10 +603,15 @@ class Job extends Operation {
             error never makes it to the callback. Instead, pass the error to the
             callback the user provides so that the user can see the error.
              */
-            rows = BigQuery.mergeSchemaWithRows_(resp.schema, resp.rows, {
-              wrapIntegers,
-              parseJSON,
-            });
+            if (options.skipParsing) {
+              rows = resp.rows;
+            } else {
+              rows = BigQuery.mergeSchemaWithRows_(resp.schema, resp.rows, {
+                wrapIntegers,
+                parseJSON,
+              });
+              delete resp.rows;
+            }
           } catch (e) {
             callback!(e as Error, null, null, resp);
             return;
@@ -633,7 +639,6 @@ class Job extends Operation {
           });
           delete nextQuery.startIndex;
         }
-        delete resp.rows;
         callback!(null, rows, nextQuery, resp);
       },
     );

handwritten/bigquery/src/table.ts

@@ -114,6 +114,7 @@ export type TableRowValue = string | TableRow;
 export type GetRowsOptions = PagedRequest<bigquery.tabledata.IListParams> & {
   wrapIntegers?: boolean | IntegerTypeCastOptions;
   parseJSON?: boolean;
+  skipParsing?: boolean;
 };
 
 export type JobLoadMetadata = JobRequest<bigquery.IJobConfigurationLoad> & {
@@ -1873,12 +1874,20 @@ class Table extends ServiceObject {
         the callback. Instead, pass the error to the callback the user provides
         so that the user can see the error.
          */
-        rows = BigQuery.mergeSchemaWithRows_(this.metadata.schema, rows || [], {
-          wrapIntegers,
-          selectedFields,
-          parseJSON,
-          listParams: qs,
-        });
+        if (options.skipParsing) {
+          rows = rows || [];
+        } else {
+          rows = BigQuery.mergeSchemaWithRows_(
+            this.metadata.schema,
+            rows || [],
+            {
+              wrapIntegers,
+              selectedFields,
+              parseJSON,
+              listParams: qs,
+            },
+          );
+        }
       } catch (err) {
         callback!(err as Error | null, null, null, resp);
         return;

handwritten/bigquery/system-test/bigquery.ts

@@ -354,6 +354,27 @@ describe('BigQuery', () => {
     assert.strictEqual(res.totalRows, '100');
   });
 
+  it('should query with skipParsing', async () => {
+    const [rows] = await bigquery.query({
+      query,
+      skipParsing: true,
+    });
+    assert.strictEqual(rows.length, 100);
+    // Raw rows have an 'f' property containing the fields
+    assert.ok(rows[0].f);
+    assert.strictEqual(typeof rows[0].f[0].v, 'string');
+  });
+
+  it('should query with skipParsing via Dataset.query', async () => {
+    const [rows] = await dataset.query({
+      query,
+      skipParsing: true,
+    });
+    assert.strictEqual(rows.length, 100);
+    assert.ok(rows[0].f);
+    assert.strictEqual(typeof rows[0].f[0].v, 'string');
+  });
+
   it('should query without jobs.query and return all PagedResponse as positional parameters', async () => {
     // force jobs.getQueryResult instead of fast query path
     const jobId = generateName('job');
@@ -575,8 +596,9 @@ describe('BigQuery', () => {
       const QUERY = `SELECT * FROM \`${table.id}\``;
       // eslint-disable-next-line @typescript-eslint/no-var-requires
       const SCHEMA = require('../../system-test/data/schema.json');
-      const TEST_DATA_FILE =
-        require.resolve('../../system-test/data/location-test-data.json');
+      const TEST_DATA_FILE = require.resolve(
+        '../../system-test/data/location-test-data.json',
+      );
 
       before(async () => {
         // create a dataset in a certain location will cascade the location
@@ -879,8 +901,9 @@ describe('BigQuery', () => {
   });
 
   describe('BigQuery/Table', () => {
-    const TEST_DATA_JSON_PATH =
-      require.resolve('../../system-test/data/kitten-test-data.json');
+    const TEST_DATA_JSON_PATH = require.resolve(
+      '../../system-test/data/kitten-test-data.json',
+    );
 
     it('should have created the correct schema', () => {
       assert.deepStrictEqual(table.metadata.schema.fields, SCHEMA);

handwritten/bigquery/test/bigquery.ts

@@ -3243,6 +3243,74 @@ describe('BigQuery', () => {
       });
     });
 
+    it('should delete res.rows if skipParsing is false', done => {
+      const rawRows = [{f: [{v: 'hi'}]}];
+      const resp = {
+        jobComplete: true,
+        schema: {
+          fields: [{name: 'name', type: 'STRING'}],
+        },
+        rows: rawRows,
+      };
+
+      const job = {
+        getQueryResults: (options: {}, callback: Function) => {
+          callback(null, [], null, resp);
+        },
+      };
+
+      bq.runJobsQuery = (reqOpts: {}, callback: Function) => {
+        callback(null, job, resp);
+      };
+
+      bq.query(
+        {
+          query: 'SELECT * FROM table',
+          skipParsing: false,
+        },
+        (err: Error, rows: {}[], nextQuery: {}, response: any) => {
+          assert.ifError(err);
+          // the job Complete callback returned the resp
+          assert.deepStrictEqual(response.rows, undefined);
+          done();
+        },
+      );
+    });
+
+    it('should skip parsing if skipParsing is true', done => {
+      const rawRows = [{f: [{v: 'hi'}]}];
+      const resp = {
+        jobComplete: true,
+        schema: {
+          fields: [{name: 'name', type: 'STRING'}],
+        },
+        rows: rawRows,
+      };
+
+      const job = {
+        getQueryResults: (options: QueryResultsOptions, callback: Function) => {
+          callback(null, options._cachedRows, null, options._cachedResponse);
+        },
+      };
+
+      bq.runJobsQuery = (reqOpts: {}, callback: Function) => {
+        callback(null, job, resp);
+      };
+
+      bq.query(
+        {
+          query: 'SELECT * FROM table',
+          skipParsing: true,
+        },
+        (err: Error, rows: {}[], nextQuery: {}, response: any) => {
+          assert.ifError(err);
+          assert.strictEqual(rows, rawRows);
+          assert.deepStrictEqual(response.rows, rawRows);
+          done();
+        },
+      );
+    });
+
     it('should call job#getQueryResults with query options', done => {
       let queryResultsOpts = {};
       const fakeJob = {

handwritten/bigquery/test/dataset.ts

@@ -1010,6 +1010,20 @@ describe('BigQuery/Dataset', () => {
       ds.query(query);
     });
 
+    it('should pass along skipParsing', done => {
+      const query = {
+        query: 'SELECT * FROM allthedata',
+        skipParsing: true,
+      };
+
+      ds.bigQuery.query = (opts: _root.Query) => {
+        assert.strictEqual(opts.skipParsing, true);
+        done();
+      };
+
+      ds.query(query);
+    });
+
     it('should pass along options', done => {
       // eslint-disable-next-line @typescript-eslint/no-explicit-any
       ds.bigQuery.query = (opts: any) => {

handwritten/bigquery/test/job.ts

@@ -403,6 +403,57 @@ describe('BigQuery/Job', () => {
       job.getQueryResults(options, assert.ifError);
     });
 
+    it('should skip parsing if skipParsing is true', done => {
+      const response = {
+        schema: {},
+        rows: [{f: [{v: 'hi'}]}],
+      };
+
+      BIGQUERY.request = (
+        reqOpts: DecorateRequestOptions,
+        callback: Function,
+      ) => {
+        callback(null, response);
+      };
+
+      const mergeStub = sandbox.stub(BigQuery, 'mergeSchemaWithRows_');
+
+      job.getQueryResults({skipParsing: true}, (err: Error, rows: {}[]) => {
+        assert.ifError(err);
+        assert.strictEqual(rows, response.rows);
+        assert.strictEqual(mergeStub.called, false);
+        done();
+      });
+    });
+
+    it('should not delete resp.rows if skipParsing is true', done => {
+      const options: QueryResultsOptions = {
+        skipParsing: true,
+      };
+
+      const rawRows = [{f: [{v: 'hi'}]}];
+      const resp = {
+        jobComplete: true,
+        rows: rawRows,
+        schema: {
+          fields: [{name: 'name', type: 'STRING'}],
+        },
+      };
+
+      job.bigQuery.request = (reqOpts: {}, callback: Function) => {
+        callback(null, resp);
+      };
+
+      job.getQueryResults(
+        options,
+        (err: Error, rows: {}, nextQuery: {}, response: any) => {
+          assert.ifError(err);
+          assert.deepStrictEqual(response.rows, rawRows);
+          done();
+        },
+      );
+    });
+
     it('should return the query when the job is not complete', done => {
       BIGQUERY.request = (
         reqOpts: DecorateRequestOptions,
@@ -447,6 +498,32 @@ describe('BigQuery/Job', () => {
       );
     });
 
+    it('should delete resp.rows if skipParsing is false by default', done => {
+      const options: QueryResultsOptions = {};
+
+      const rawRows = [{f: [{v: 'hi'}]}];
+      const resp = {
+        jobComplete: true,
+        rows: rawRows,
+        schema: {
+          fields: [{name: 'name', type: 'STRING'}],
+        },
+      };
+
+      job.bigQuery.request = (reqOpts: {}, callback: Function) => {
+        callback(null, resp);
+      };
+
+      job.getQueryResults(
+        options,
+        (err: Error, rows: {}, nextQuery: {}, response: any) => {
+          assert.ifError(err);
+          assert.deepStrictEqual(response.rows, undefined);
+          done();
+        },
+      );
+    });
+
     it('should populate nextQuery when more results exist', done => {
       job.getQueryResults(
         options,

handwritten/bigquery/test/table.ts

@@ -2189,6 +2189,27 @@ describe('BigQuery/Table', () => {
       });
     });
 
+    it('should skip parsing if skipParsing is true', done => {
+      const rows = [{f: [{v: 'stephen'}]}];
+      const schema = {fields: [{name: 'name', type: 'string'}]};
+      table.metadata = {schema};
+
+      table.request = (reqOpts: DecorateRequestOptions, callback: Function) => {
+        callback(null, {rows});
+      };
+
+      sandbox.restore();
+      const mergeStub = sandbox.stub(BigQuery, 'mergeSchemaWithRows_');
+
+      table.getRows({skipParsing: true}, (err: Error, rows_: {}[], nextQuery: {}, apiResponse: any) => {
+        assert.ifError(err);
+        assert.strictEqual(rows_, rows);
+        assert.strictEqual(mergeStub.called, false);
+        assert.deepStrictEqual(apiResponse.rows, rows);
+        done();
+      });
+    });
+
     it('should pass nextQuery if pageToken is returned', done => {
       const options = {a: 'b', c: 'd'};
       const pageToken = 'token';
@@ -3025,6 +3046,20 @@ describe('BigQuery/Table', () => {
 
       table.query('a', 'b');
     });
+
+    it('should pass skipParsing through to datasetInstance.query()', done => {
+      const query = {
+        query: 'a',
+        skipParsing: true,
+      };
+      table.dataset.query = (a: {}, b: {}) => {
+        assert.deepStrictEqual(a, query);
+        assert.strictEqual(b, 'b');
+        done();
+      };
+
+      table.query(query, 'b');
+    });
   });
 
   describe('setMetadata', () => {