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

src/database.ts

@@ -142,6 +142,10 @@
   updateMask?: FieldMask | null;
 }
 
+export interface RunPartitionedUpdateOptions extends ExecuteSqlRequest {
+  excludeTxnFromChangeStreams?: boolean;
+}
+
 export type UpdateSchemaCallback = ResourceCallback<
   GaxOperation,
   databaseAdmin.longrunning.IOperation
@@ -1527,7 +1531,7 @@
   ): void;
   async getDatabaseDialect(
     optionsOrCallback?: CallOptions | GetDatabaseDialectCallback,
     cb?: GetDatabaseDialectCallback
   ): Promise<
     | EnumKey<typeof databaseAdmin.spanner.admin.database.v1.DatabaseDialect>
     | undefined
@@ -2092,6 +2096,9 @@
       if (options.optimisticLock) {
         transaction!.useOptimisticLock();
       }
+      if (options.excludeTxnFromChangeStreams) {
+        transaction!.excludeTxnFromChangeStreams();
+      }
       if (!err) {
         this._releaseOnEnd(session!, transaction!);
       }
@@ -2711,13 +2718,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,11 +2741,14 @@
 
   _runPartitionedUpdate(
     session: Session,
-    query: string | ExecuteSqlRequest,
+    query: string | RunPartitionedUpdateOptions,
     callback?: RunUpdateCallback
   ): void | Promise<number> {
     const transaction = session.partitionedDml();
 
+    if (typeof query !== 'string' && query.excludeTxnFromChangeStreams) {
+      transaction.excludeTxnFromChangeStreams();
+    }
     transaction.begin(err => {
       if (err) {
         this.pool_.release(session!);
@@ -3059,6 +3071,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 +3188,9 @@
         if (options.optimisticLock) {
           transaction.useOptimisticLock();
         }
+        if (options.excludeTxnFromChangeStreams) {
+          transaction.excludeTxnFromChangeStreams();
+        }
         const runner = new AsyncTransactionRunner<T>(
           session,
           transaction,

src/table.ts

@@ -46,6 +46,7 @@ export type DropTableCallback = UpdateSchemaCallback;
 
 interface MutateRowsOptions extends CommitOptions {
   requestOptions?: Omit<IRequestOptions, 'requestTag'>;
+  excludeTxnFromChangeStreams?: boolean;
 }
 
 export type DeleteRowsCallback = CommitCallback;
@@ -1073,15 +1074,27 @@ class Table {
   ): void {
     const requestOptions =
       'requestOptions' in options ? options.requestOptions : {};
-    this.database.runTransaction({requestOptions}, (err, transaction) => {
-      if (err) {
-        callback(err);
-        return;
-      }
 
-      transaction![method](this.name, rows as Key[]);
-      transaction!.commit(options, callback);
-    });
+    const excludeTxnFromChangeStreams =
+      'excludeTxnFromChangeStreams' in options
+        ? options.excludeTxnFromChangeStreams
+        : false;
+
+    this.database.runTransaction(
+      {
+        requestOptions: requestOptions,
+        excludeTxnFromChangeStreams: excludeTxnFromChangeStreams,
+      },
+      (err, transaction) => {
+        if (err) {
+          callback(err);
+          return;
+        }
+
+        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

@@ -2470,6 +2470,18 @@ export class Transaction extends Dml {
   useOptimisticLock(): void {
     this._options.readWrite!.readLockMode = ReadLockMode.OPTIMISTIC;
   }
+
+  /**
+   * Use option excludeTxnFromChangeStreams to exclude read/write transactions
+   * from being tracked in change streams.
+   *
+   * Enabling this options to true will effectively disable change stream tracking
+   * for a specified transaction, allowing read/write transaction to operate without being
+   * included in change streams.
+   */
+  excludeTxnFromChangeStreams(): void {
+    this._options.excludeTxnFromChangeStreams = true;
+  }
 }
 
 /*! Developer Documentation
@@ -2503,6 +2515,17 @@ export class PartitionedDml extends Dml {
     super(session);
     this._options = {partitionedDml: options};
   }
+  /**
+   * Use option excludeTxnFromChangeStreams to exclude partitionedDml
+   * queries from being tracked in change streams.
+   *
+   * Enabling this options to true will effectively disable change stream tracking
+   * for a specified partitionedDml query, allowing write queries to operate
+   * without being included in change streams.
+   */
+  excludeTxnFromChangeStreams(): void {
+    this._options.excludeTxnFromChangeStreams = true;
+  }
 
   /**
    * Execute a DML statement and get the affected row count. Unlike

test/database.ts

@@ -2544,7 +2544,8 @@ describe('Database', () => {
 
       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);
     });
 
@@ -2581,6 +2582,24 @@ describe('Database', () => {
       assert.ok(fakeCallback.calledOnce);
     });
 
+    it('should accept excludeTxnFromChangeStreams', () => {
+      const fakeCallback = sandbox.spy();
+
+      database.runPartitionedUpdate(
+        {
+          excludeTxnFromChangeStream: true,
+        },
+        fakeCallback
+      );
+
+      const [query] = runUpdateStub.lastCall.args;
+
+      assert.deepStrictEqual(query, {
+        excludeTxnFromChangeStream: true,
+      });
+      assert.ok(fakeCallback.calledOnce);
+    });
+
     it('should ignore directedReadOptions set for client', () => {
       const fakeCallback = sandbox.spy();