feat: add support for blind writes

src/database.ts

@@ -61,8 +61,11 @@ import {
 import {CreateTableCallback, CreateTableResponse, Table} from './table';
 import {
   BatchWriteOptions,
+  CommitCallback,
+  CommitResponse,
   ExecuteSqlRequest,
   MutationGroup,
+  Mutation,
   RunCallback,
   RunResponse,
   RunUpdateCallback,
@@ -3346,6 +3349,91 @@ class Database extends common.GrpcServiceObject {
     return proxyStream as NodeJS.ReadableStream;
   }
 
+  /**
+   * Apply Blind Write of the Mutations
+   *
+   * writeAtLeastOnce(Blind Write) requests are not replay protected, meaning that it may apply mutations more
+   * than once, if the mutations are not idempotent, this may lead to a failure being
+   * reported when the mutation was applied once. Replays of non-idempotent mutations may
+   * have undesirable effects. For example, replays of an insert mutation may produce an
+   * already exists error. For this reason, most users of the library will prefer to use
+   * {@link runTransaction} instead.
+   *
+   * However, {@link writeAtLeastOnce()} requires only a single RPC, whereas {@link runTransaction()}
+   * requires two RPCs (one of which may be performed in advance), and so this method may be
+   * appropriate for latency sensitive and/or high throughput blind writing.
+   *
+   * We recommend structuring your mutation groups to be idempotent to avoid this issue.
+   *
+   * @param {Mutation} [mutation]  Mutations to be applied.
+   * @param {CallOptions} [options] Options object for blind write request.
+   * @param {CommitCallback} [callback] Callback function for blind write request.
+   *
+   * @returns {Promise}
+   *
+   * @example
+   * ```
+   * const {Spanner} = require('@google-cloud/spanner');
+   * const spanner = new Spanner();
+   *
+   * const instance = spanner.instance('my-instance');
+   * const database = instance.database('my-database');
+   * const mutation = new Mutation();
+   * mutation.insert('Singers', {
+   *  SingerId: '1',
+   *  FirstName: 'Marc',
+   *  LastName: 'Richards',
+   *  });
+   * mutation.update('Singers', {
+   *  SingerId: '1',
+   *  FirstName: 'John',
+   *  LastName: 'Richards',
+   *  });
+   *
+   * try {
+   *  const [response, err] = await database.writeAtLeastOnce(mutation, {});
+   *  console.log(response.commitTimestamp);
+   * } catch(err) {
+   *  console.log("Error: ", err);
+   * }
+   * ```
+   */
+  writeAtLeastOnce(mutation: Mutation): Promise<CommitResponse>;
+  writeAtLeastOnce(
+    mutation: Mutation,
+    options: CallOptions
+  ): Promise<CommitResponse>;
+  writeAtLeastOnce(mutation: Mutation, callback: CommitCallback): void;
+  writeAtLeastOnce(
+    mutation: Mutation,
+    optionsOrCallback?: CallOptions | CommitCallback,
+    callback?: CommitCallback
+  ): void | Promise<CommitResponse> {
+    const cb =
+      typeof optionsOrCallback === 'function'
+        ? (optionsOrCallback as CommitCallback)
+        : callback;
+    const options =
+      typeof optionsOrCallback === 'object' && optionsOrCallback
+        ? (optionsOrCallback as CallOptions)
+        : {};
+    this.pool_.getSession((err, session?, transaction?) => {
+      if (err && isSessionNotFoundError(err as grpc.ServiceError)) {
+        cb
+          ? this.writeAtLeastOnce(mutation, cb)
+          : this.writeAtLeastOnce(mutation, options);
+        return;
+      }
+      if (err) {
+        cb!(err as grpc.ServiceError);
+        return;
+      }
+      this._releaseOnEnd(session!, transaction!);
+      transaction?.setQueuedMutations(mutation.proto());
+      return transaction?.commit(options, cb!);
+    });
+  }
+
   /**
    * Create a Session object.
    *
@@ -3674,6 +3762,7 @@ callbackifyAll(Database, {
     'batchCreateSessions',
     'batchTransaction',
     'batchWriteAtLeastOnce',
+    'writeAtLeastOnce',
     'close',
     'createBatchTransaction',
     'createSession',

src/transaction.ts

@@ -1854,6 +1854,17 @@ export class Transaction extends Dml {
     return undefined;
   }
 
+  /**
+   * This method updates the _queuedMutations property of the transaction.
+   *
+   * @public
+   *
+   * @param {spannerClient.spanner.v1.Mutation[]} [mutation]
+   */
+  setQueuedMutations(mutation: spannerClient.spanner.v1.Mutation[]): void {
+    this._queuedMutations = mutation;
+  }
+
   /**
    * @typedef {object} CommitOptions
    * @property {google.spanner.v1.IRequestOptions} requestOptions The request options to include
@@ -2530,6 +2541,110 @@ function buildDeleteMutation(
   return mutation as spannerClient.spanner.v1.Mutation;
 }
 
+/**
+ * Mutation represent a set of changes to be applied atomically to a cloud spanner
+ * database with a {@link Transaction}.
+ * Mutations are used to insert, update, upsert(insert or update), replace, or
+ * delete rows within tables.
+ *
+ * Mutations are added to a {@link Transaction} and are not executed until the
+ * transaction is committed via {@link Transaction#commit}.
+ *
+ * If the transaction is rolled back or encounters an error, the mutations are
+ * discarded.
+ *
+ * @example
+ * ```
+ * const {Spanner, Mutation} = require('@google-cloud/spanner');
+ * const spanner = new Spanner();
+ *
+ * const instance = spanner.instance('my-instance');
+ * const database = instance.database('my-database');
+ *
+ * const mutation = new Mutation();
+ * mutation.insert('Singers', {SingerId: '123', FirstName: 'David'});
+ * mutation.update('Singers', {SingerId: '123', FirstName: 'Marc'});
+ *
+ * try {
+ *  database.writeAtLeastOnce(mutation, (err, res) => {
+ *    console.log("RESPONSE: ", res);
+ *  });
+ * } catch(err) {
+ *  console.log("ERROR: ", err);
+ * }
+ * ```
+ */
+export class Mutation {
+  /**
+   * An array to store the mutations.
+   */
+  private _queuedMutations: spannerClient.spanner.v1.Mutation[];
+
+  /**
+   * Creates a new Mutation object.
+   */
+  constructor() {
+    this._queuedMutations = [];
+  }
+
+  /**
+   * Adds an insert operation to the mutation set.
+   * @param {string} table. The name of the table to insert into.
+   * @param {object|object[]} rows. A single row object or an array of row objects to insert.
+   */
+  insert(table: string, rows: object | object[]): void {
+    this._queuedMutations.push(buildMutation('insert', table, rows));
+  }
+
+  /**
+   * Adds an update operation to the mutation set.
+   * @param {string} table. The name of the table to update.
+   * @param {object|object[]} rows. A single row object or an array of row objects to update.
+   * Each row object must contain the primary key values to indentify the row to update.
+   */
+  update(table: string, rows: object | object[]): void {
+    this._queuedMutations.push(buildMutation('update', table, rows));
+  }
+
+  /**
+   * Adds an upsert operation to the mutation set.
+   * An upsert will insert a new row if it does not exist or update an existing row if it does.
+   * @param {string} table. The name of the table to upsert.
+   * @param {object|object[]} rows. A single row object or an array of row objects to upsert.
+   */
+  upsert(table: string, rows: object | object[]): void {
+    this._queuedMutations.push(buildMutation('insertOrUpdate', table, rows));
+  }
+
+  /**
+   * Adds a replace operation to the mutation set.
+   * A replace operation deletes the existing row (if it exists) and inserts the new row.
+   * @param {string} table. The name of the table to replace.
+   * @param {object|object[]} rows. A single row object or an array of row objects to replace.
+   */
+  replace(table: string, rows: object | object[]): void {
+    this._queuedMutations.push(buildMutation('replace', table, rows));
+  }
+
+  /**
+   * Adds a deleteRows operation to the mutation set.
+   * This operation deletes rows from the specified table based on their primary keys.
+   * @param {string} table. The name of the table to deleteRows from.
+   * @param {key[]} key. An array of key objects, each represeting the primary key of a row to delete.
+   */
+  deleteRows(table: string, keys: Key[]): void {
+    this._queuedMutations.push(buildDeleteMutation(table, keys));
+  }
+
+  /**
+   * Returns the internal representation of the queued mutations as a protobuf message.
+   * @returns {spannerClient.spanner.v1.Mutation[]}. The protobuf message representing the mutations.
+   */
+  proto(): spannerClient.spanner.v1.Mutation[] {
+    return this._queuedMutations;
+  }
+}
+
 /**
  * A group of mutations to be committed together.
  * Related mutations should be placed in a group.

test/database.ts

@@ -40,7 +40,13 @@ import {protos} from '../src';
 import * as inst from '../src/instance';
 import RequestOptions = google.spanner.v1.RequestOptions;
 import EncryptionType = google.spanner.admin.database.v1.RestoreDatabaseEncryptionConfig.EncryptionType;
-import {BatchWriteOptions} from '../src/transaction';
+import {
+  BatchWriteOptions,
+  CommitCallback,
+  CommitOptions,
+  Mutation,
+} from '../src/transaction';
+import {error} from 'is';
 
 let promisified = false;
 const fakePfy = extend({}, pfy, {
@@ -125,17 +131,31 @@ class FakeTable {
 class FakeTransaction extends EventEmitter {
   calledWith_: IArguments;
   _options!: google.spanner.v1.ITransactionOptions;
+  private _queuedMutations: google.spanner.v1.Mutation[];
   constructor(options) {
     super();
     this._options = options;
     this.calledWith_ = arguments;
+    this._queuedMutations = [];
   }
   begin() {}
   end() {}
   runStream(): Transform {
     return through.obj();
   }
   runUpdate() {}
+  setQueuedMutations(mutation) {
+    this._queuedMutations = mutation;
+  }
+  commit(
+    options?: CommitOptions,
+    callback?: CommitCallback
+  ): void | Promise<google.spanner.v1.ICommitResponse> {
+    if (callback) {
+      callback(null, {commitTimestamp: {seconds: 1, nanos: 0}});
+    }
+    return Promise.resolve({commitTimestamp: {seconds: 1, nanos: 0}});
+  }
 }
 
 let fakeTransactionRunner: FakeTransactionRunner;
@@ -762,6 +782,89 @@ describe('Database', () => {
     });
   });
 
+  describe('writeAtLeastOnce', () => {
+    const mutation = new Mutation();
+    mutation.insert('MyTable', {
+      Key: 'k3',
+      Thing: 'xyz',
+    });
+
+    const SESSION = new FakeSession();
+    const RESPONSE = {commitTimestamp: {seconds: 1, nanos: 0}};
+    const TRANSACTION = new FakeTransaction(
+      {} as google.spanner.v1.TransactionOptions.ReadWrite
+    );
+
+    let pool: FakeSessionPool;
+
+    beforeEach(() => {
+      pool = database.pool_;
+      (sandbox.stub(pool, 'getSession') as sinon.SinonStub).callsFake(
+        callback => {
+          callback(null, SESSION, TRANSACTION);
+        }
+      );
+    });
+
+    it('should return any errors getting a session', done => {
+      const fakeErr = new Error('err');
+
+      (pool.getSession as sinon.SinonStub).callsFake(callback =>
+        callback(fakeErr, null, null)
+      );
+
+      database.writeAtLeastOnce(mutation, err => {
+        assert.deepStrictEqual(err, fakeErr);
+        done();
+      });
+    });
+
+    it('should return successful CommitResponse when passing an empty mutation', done => {
+      const fakeMutation = new Mutation();
+      try {
+        database.writeAtLeastOnce(fakeMutation, (err, response) => {
+          assert.ifError(err);
+          assert.deepStrictEqual(
+            response.commitTimestamp,
+            RESPONSE.commitTimestamp
+          );
+        });
+        done();
+      } catch (error) {
+        assert(error instanceof Error);
+      }
+    });
+
+    it('should return an error when passing null mutation', done => {
+      const fakeError = new Error('err');
+      try {
+        database.writeAtLeastOnce(null, (err, res) => {});
+      } catch (err) {
+        (err as grpc.ServiceError).message.includes(
+          "Cannot read properties of null (reading 'proto')"
+        );
+        done();
+      }
+    });
+
+    it('should return CommitResponse on successful write using Callback', done => {
+      database.writeAtLeastOnce(mutation, (err, res) => {
+        assert.deepStrictEqual(err, null);
+        assert.deepStrictEqual(res, RESPONSE);
+        done();
+      });
+    });
+
+    it('should return CommitResponse on successful write using await', async () => {
+      sinon.stub(database, 'writeAtLeastOnce').resolves([RESPONSE]);
+      const [response, err] = await database.writeAtLeastOnce(mutation, {});
+      assert.deepStrictEqual(
+        response.commitTimestamp,
+        RESPONSE.commitTimestamp
+      );
+    });
+  });
+
   describe('close', () => {
     const FAKE_ID = 'a/c/b/d';