googleapis
diff --git a/‎src/database.ts‎
Lines changed: 93 additions & 0 deletions b/‎src/database.ts‎
Lines changed: 93 additions & 0 deletions
diff --git a/‎src/index.ts‎
Lines changed: 10 additions & 0 deletions b/‎src/index.ts‎
Lines changed: 10 additions & 0 deletions
diff --git a/‎src/transaction.ts‎
Lines changed: 115 additions & 0 deletions b/‎src/transaction.ts‎
Lines changed: 115 additions & 0 deletions
@@ -61,8 +61,11 @@ import {
 import {CreateTableCallback, CreateTableResponse, Table} from './table';
 import {
  BatchWriteOptions,
+ CommitCallback,
+ CommitResponse,
  ExecuteSqlRequest,
  MutationGroup,
+ MutationSet,
  RunCallback,
  RunResponse,
  RunUpdateCallback,
@@ -3346,6 +3349,95 @@ class Database extends common.GrpcServiceObject {
  return proxyStream as NodeJS.ReadableStream;
  }
 
+ /**
+ * Write mutations using a single RPC invocation without replay protection.
+ *
+ * writeAtLeastOnce writes mutations to Spanner using a single Commit RPC.
+ * These 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 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 set to be idempotent to avoid this issue.
+ *
+ * @param {MutationSet} [mutations] Set of 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 mutations = new MutationSet();
+ * mutations.upsert('Singers', {
+ * SingerId: 1,
+ * FirstName: 'Scarlet',
+ * LastName: 'Terry',
+ * });
+ * mutations.upsert('Singers', {
+ * SingerId: 2,
+ * FirstName: 'Marc',
+ * LastName: 'Richards',
+ * });
+ *
+ * try {
+ * const [response, err] = await database.writeAtLeastOnce(mutations, {});
+ * console.log(response.commitTimestamp);
+ * } catch(err) {
+ * console.log("Error: ", err);
+ * }
+ * ```
+ */
+ writeAtLeastOnce(mutations: MutationSet): Promise<CommitResponse>;
+ writeAtLeastOnce(
+ mutations: MutationSet,
+ options: CallOptions
+ ): Promise<CommitResponse>;
+ writeAtLeastOnce(mutations: MutationSet, callback: CommitCallback): void;
+ writeAtLeastOnce(
+ mutations: MutationSet,
+ options: CallOptions,
+ callback: CommitCallback
+ ): void;
+ writeAtLeastOnce(
+ mutations: MutationSet,
+ 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)) {
+ this.writeAtLeastOnce(mutations, options, cb!);
+ return;
+ }
+ if (err) {
+ cb!(err as grpc.ServiceError);
+ return;
+ }
+ this._releaseOnEnd(session!, transaction!);
+ transaction?.setQueuedMutations(mutations.proto());
+ return transaction?.commit(options, cb!);
+ });
+ }
+
  /**
  * Create a Session object.
  *
@@ -3674,6 +3766,7 @@ callbackifyAll(Database, {
  'batchCreateSessions',
  'batchTransaction',
  'batchWriteAtLeastOnce',
+ 'writeAtLeastOnce',
  'close',
  'createBatchTransaction',
  'createSession',
 
@@ -72,6 +72,7 @@ import {SessionPool} from './session-pool';
 import {Table} from './table';
 import {
  MutationGroup,
+ MutationSet,
  PartitionedDml,
  Snapshot,
  Transaction,
@@ -2025,6 +2026,15 @@ export {Transaction};
  */
 export {MutationGroup};
 
+/**
+ * {@link MutationSet} class.
+ *
+ * @name Spanner.MutationSet
+ * @see MutationSet
+ * @type {Constructor}
+ */
+export {MutationSet};
+
 /**
  * @type {object}
  * @property {constructor} DatabaseAdminClient
 
@@ -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;
 }
 
+/**
+ * MutationSet 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 mutations = new MutationSet();
+ * mutations.insert('Singers', {SingerId: '123', FirstName: 'David'});
+ * mutations.update('Singers', {SingerId: '123', FirstName: 'Marc'});
+ *
+ * try {
+ * database.writeAtLeastOnce(mutations, (err, res) => {
+ * console.log("RESPONSE: ", res);
+ * });
+ * } catch(err) {
+ * console.log("ERROR: ", err);
+ * }
+ * ```
+ */
+export class MutationSet {
+ /**
+ * 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.