feat(spanner): add support for change streams transaction exclusion option

src/database.ts

@@ -62,6 +62,7 @@
 import {
   ExecuteSqlRequest,
   RunCallback,
+  runPartitionedUpdateOptions,
   RunResponse,
   RunUpdateCallback,
   Snapshot,
@@ -1527,7 +1528,7 @@
   ): void;
   async getDatabaseDialect(
     optionsOrCallback?: CallOptions | GetDatabaseDialectCallback,
     cb?: GetDatabaseDialectCallback
   ): Promise<
     | EnumKey<typeof databaseAdmin.spanner.admin.database.v1.DatabaseDialect>
     | undefined
@@ -2092,6 +2093,9 @@
       if (options.optimisticLock) {
         transaction!.useOptimisticLock();
       }
+      if (options.excludeTxnFromChangeStreams) {
+        transaction!.excludeTxnFromChangeStreams();
+      }
       if (!err) {
         this._releaseOnEnd(session!, transaction!);
       }
@@ -2711,13 +2715,15 @@
    * @param {RunUpdateCallback} [callback] Callback function.
    * @returns {Promise<RunUpdateResponse>}
    */
-  runPartitionedUpdate(query: string | ExecuteSqlRequest): Promise<[number]>;
   runPartitionedUpdate(
-    query: string | ExecuteSqlRequest,
+    query: string | runPartitionedUpdateOptions
+  ): Promise<[number]>;
+  runPartitionedUpdate(
+    query: string | runPartitionedUpdateOptions,
     callback?: RunUpdateCallback
   ): void;
   runPartitionedUpdate(
-    query: string | ExecuteSqlRequest,
+    query: string | runPartitionedUpdateOptions,
     callback?: RunUpdateCallback
   ): void | Promise<[number]> {
     this.pool_.getSession((err, session) => {
@@ -2732,33 +2738,44 @@
 
   _runPartitionedUpdate(
     session: Session,
-    query: string | ExecuteSqlRequest,
+    query: string | runPartitionedUpdateOptions,
     callback?: RunUpdateCallback
   ): void | Promise<number> {
     const transaction = session.partitionedDml();
 
-    transaction.begin(err => {
-      if (err) {
-        this.pool_.release(session!);
-        callback!(err, 0);
-        return;
-      }
+    if (typeof query === 'string') {
+      query = {sql: query} as runPartitionedUpdateOptions;
+    }
+
+    query = Object.assign({}, query) as runPartitionedUpdateOptions;
 
-      transaction.runUpdate(query, (err, updateCount) => {
+    transaction.begin(
+      {
+        excludeTxnFromChangeStreams: query.excludeTxnFromChangeStreams,
+      },
+      err => {
         if (err) {
-          if (err.code !== grpc.status.ABORTED) {
-            this.pool_.release(session!);
-            callback!(err, 0);
-            return;
-          }
-          this._runPartitionedUpdate(session, query, callback);
-        } else {
           this.pool_.release(session!);
-          callback!(null, updateCount);
+          callback!(err, 0);
           return;
         }
-      });
-    });
+
+        transaction.runUpdate(query, (err, updateCount) => {
+          if (err) {
+            if (err.code !== grpc.status.ABORTED) {
+              this.pool_.release(session!);
+              callback!(err, 0);
+              return;
+            }
+            this._runPartitionedUpdate(session, query, callback);
+          } else {
+            this.pool_.release(session!);
+            callback!(null, updateCount);
+            return;
+          }
+        });
+      }
+    );
   }
 
   /**
@@ -3059,6 +3076,9 @@
       if (options.optimisticLock) {
         transaction!.useOptimisticLock();
       }
+      if (options.excludeTxnFromChangeStreams) {
+        transaction!.excludeTxnFromChangeStreams();
+      }
 
       const release = this.pool_.release.bind(this.pool_, session!);
       const runner = new TransactionRunner(
@@ -3173,6 +3193,9 @@
         if (options.optimisticLock) {
           transaction.useOptimisticLock();
         }
+        if (options.excludeTxnFromChangeStreams) {
+          transaction.excludeTxnFromChangeStreams();
+        }
         const runner = new AsyncTransactionRunner<T>(
           session,
           transaction,

src/table.ts

@@ -1079,6 +1079,15 @@ class Table {
         return;
       }
 
+      const excludeTxnFromChangeStreams =
+        'excludeTxnFromChangeStreams' in options
+          ? options.excludeTxnFromChangeStreams
+          : false;
+
+      if (excludeTxnFromChangeStreams) {
+        transaction!.excludeTxnFromChangeStreams();
+      }
+
       transaction![method](this.name, rows as Key[]);
       transaction!.commit(options, callback);
     });

src/transaction-runner.ts

@@ -45,6 +45,7 @@ export interface RunTransactionOptions {
   timeout?: number;
   requestOptions?: Pick<IRequestOptions, 'transactionTag'>;
   optimisticLock?: boolean;
+  excludeTxnFromChangeStreams?: boolean;
 }
 
 /**
@@ -204,6 +205,9 @@ export abstract class Runner<T> {
     if (this.options.optimisticLock) {
       transaction.useOptimisticLock();
     }
+    if (this.options.excludeTxnFromChangeStreams) {
+      transaction.excludeTxnFromChangeStreams();
+    }
     if (this.attempts > 0) {
       await transaction.begin();
     }

src/transaction.ts

@@ -103,11 +103,16 @@ export interface RequestOptions {
   columnsMetadata?: object;
 }
 
+export interface BeginTransactionOptions extends CallOptions {
+  excludeTxnFromChangeStreams?: boolean;
+}
+
 export interface CommitOptions {
   requestOptions?: Pick<IRequestOptions, 'priority'>;
   returnCommitStats?: boolean;
   maxCommitDelay?: spannerClient.protobuf.IDuration;
   gaxOptions?: CallOptions;
+  excludeTxnFromChangeStreams?: true;
 }
 
 export interface Statement {
@@ -127,6 +132,10 @@ export interface ExecuteSqlRequest extends Statement, RequestOptions {
   directedReadOptions?: google.spanner.v1.IDirectedReadOptions;
 }
 
+export interface runPartitionedUpdateOptions extends ExecuteSqlRequest {
+  excludeTxnFromChangeStreams?: boolean;
+}
+
 export interface KeyRange {
   startClosed?: Value[];
   startOpen?: Value[];
@@ -388,11 +397,14 @@ export class Snapshot extends EventEmitter {
    *   });
    * ```
    */
-  begin(gaxOptions?: CallOptions): Promise<BeginResponse>;
+  begin(gaxOptions?: BeginTransactionOptions): Promise<BeginResponse>;
   begin(callback: BeginTransactionCallback): void;
-  begin(gaxOptions: CallOptions, callback: BeginTransactionCallback): void;
   begin(
-    gaxOptionsOrCallback?: CallOptions | BeginTransactionCallback,
+    gaxOptions: BeginTransactionOptions,
+    callback: BeginTransactionCallback
+  ): void;
+  begin(
+    gaxOptionsOrCallback?: BeginTransactionOptions | BeginTransactionCallback,
     cb?: BeginTransactionCallback
   ): void | Promise<BeginResponse> {
     const gaxOpts =
@@ -402,6 +414,19 @@ export class Snapshot extends EventEmitter {
 
     const session = this.session.formattedName_!;
     const options = this._options;
+
+    const excludeTxnFromChangeStreams =
+      'excludeTxnFromChangeStreams' in gaxOpts
+        ? gaxOpts.excludeTxnFromChangeStreams
+        : false;
+
+    if (
+      excludeTxnFromChangeStreams ||
+      this._options.excludeTxnFromChangeStreams === true
+    ) {
+      options.excludeTxnFromChangeStreams = true;
+    }
+
     const reqOpts: spannerClient.spanner.v1.IBeginTransactionRequest = {
       session,
       options,
@@ -1935,7 +1960,7 @@ export class Transaction extends Dml {
     } else if (!this._useInRunner) {
       reqOpts.singleUseTransaction = this._options;
     } else {
-      this.begin().then(() => this.commit(options, callback), callback);
+      this.begin(options).then(() => this.commit(options, callback), callback);
       return;
     }
 
@@ -2470,6 +2495,10 @@ export class Transaction extends Dml {
   useOptimisticLock(): void {
     this._options.readWrite!.readLockMode = ReadLockMode.OPTIMISTIC;
   }
+
+  excludeTxnFromChangeStreams(): void {
+    this._options.excludeTxnFromChangeStreams = true;
+  }
 }
 
 /*! Developer Documentation

test/database.ts

@@ -2485,7 +2485,7 @@ describe('Database', () => {
 
       beginStub = (
         sandbox.stub(fakePartitionedDml, 'begin') as sinon.SinonStub
-      ).callsFake(callback => callback(null));
+      ).callsFake((object, callback) => callback(null));
 
       runUpdateStub = (
         sandbox.stub(fakePartitionedDml, 'runUpdate') as sinon.SinonStub
@@ -2523,7 +2523,7 @@ describe('Database', () => {
     it('should return any begin errors', done => {
       const fakeError = new Error('err');
 
-      beginStub.callsFake(callback => callback(fakeError));
+      beginStub.callsFake((object, callback) => callback(fakeError));
 
       const releaseStub = (
         sandbox.stub(fakePool, 'release') as sinon.SinonStub
@@ -2543,8 +2543,8 @@ describe('Database', () => {
       database.runPartitionedUpdate(QUERY, fakeCallback);
 
       const [query] = runUpdateStub.lastCall.args;
-
-      assert.strictEqual(query, QUERY);
+      assert.strictEqual(query.sql, QUERY.sql);
+      assert.deepStrictEqual(query.params, QUERY.params);
       assert.ok(fakeCallback.calledOnce);
     });
 
@@ -2607,6 +2607,28 @@ describe('Database', () => {
       });
       assert.ok(fakeCallback.calledOnce);
     });
+
+    it('should ignore excludeTxnFromChangeStreams set for client', () => {
+      const fakeCallback = sandbox.spy();
+
+      database.parent.parent = {
+        excludeTxnFromChangeStream: true,
+      };
+
+      database.runPartitionedUpdate(
+        {
+          excludeTxnFromChangeStream: true,
+        },
+        fakeCallback
+      );
+
+      const [query] = runUpdateStub.lastCall.args;
+
+      assert.deepStrictEqual(query, {
+        excludeTxnFromChangeStream: true,
+      });
+      assert.ok(fakeCallback.calledOnce);
+    });
   });
 
   describe('runTransaction', () => {