feat: inline BeginTransaction with first statement #1692
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -216,6 +216,8 @@ export type CommitCallback = | |
| export class Snapshot extends EventEmitter { | ||
| protected _options!: spannerClient.spanner.v1.ITransactionOptions; | ||
| protected _seqno = 1; | ||
| protected _idWaiter: Readable; | ||
| protected _inlineBeginStarted; | ||
| id?: Uint8Array | string; | ||
| ended: boolean; | ||
| metadata?: spannerClient.spanner.v1.ITransaction; | ||
|
|
@@ -289,6 +291,11 @@ export class Snapshot extends EventEmitter { | |
| this.resourceHeader_ = { | ||
| [CLOUD_RESOURCE_HEADER]: (this.session.parent as Database).formattedName_, | ||
| }; | ||
| this._idWaiter = new Readable({ | ||
| read() { | ||
| }, | ||
| }); | ||
| this._inlineBeginStarted = false; | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -378,17 +385,7 @@ export class Snapshot extends EventEmitter { | |
| callback!(err, resp); | ||
| return; | ||
| } | ||
| this._update(resp); | ||
| callback!(null, resp); | ||
| } | ||
| ); | ||
|
|
@@ -573,6 +570,8 @@ export class Snapshot extends EventEmitter { | |
|
|
||
| if (this.id) { | ||
| transaction.id = this.id as Uint8Array; | ||
| } else if (typeof this._options.readWrite !== 'undefined') { | ||
| transaction.begin = this._options; | ||
|
There was a problem hiding this comment. Would you mind adding a test that verifies the following:
Step 3 should use the transaction ID that was returned by step 1, and not include a (https://github.com/googleapis/nodejs-spanner/pull/1253/files is a very old implementation for the same. Some of the tests there might be re-usable.) There was a problem hiding this comment. Added. thanks a lot for this comment: it helped to find a bug in the code |
||
| } else { | ||
| transaction.singleUse = this._options; | ||
| } | ||
|
|
@@ -612,10 +611,14 @@ export class Snapshot extends EventEmitter { | |
| }); | ||
| }; | ||
|
|
||
| return partialResultStream(this._wrapWithIdWaiter(makeRequest), { | ||
| json, | ||
| jsonOptions, | ||
| maxResumeRetries, | ||
| }).on('response', response => { | ||
|
There was a problem hiding this comment. What would happen if the call fails? cc: @olavloite There was a problem hiding this comment. if the call fails, the transaction is created explicitly with "BeginTransaction" rpc call. There was a problem hiding this comment. Is this done at the backend or do we explicitly call begin transaction from the client side? There was a problem hiding this comment. from the client side |
||
| if (response.metadata && response.metadata!.transaction && !this.id) { | ||
| this._update(response.metadata!.transaction); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
|
|
@@ -909,6 +912,9 @@ export class Snapshot extends EventEmitter { | |
| .on('response', response => { | ||
| if (response.metadata) { | ||
| metadata = response.metadata; | ||
| if (metadata.transaction && !this.id) { | ||
| this._update(metadata.transaction); | ||
| } | ||
| } | ||
| }) | ||
| .on('data', row => rows.push(row)) | ||
|
|
@@ -1034,6 +1040,8 @@ export class Snapshot extends EventEmitter { | |
| const transaction: spannerClient.spanner.v1.ITransactionSelector = {}; | ||
| if (this.id) { | ||
|
There was a problem hiding this comment. This does not take into account that a transaction could in theory have multiple requests in flight at the same time. If a transaction starts out with sending two Consider the following test case (the it('should only inline one begin transaction', async () => {
const database = newTestDatabase();
await database.runTransactionAsync(async tx => {
const rowCount1 = getRowCountFromStreamingSql(tx, {sql: selectSql});
const rowCount2 = getRowCountFromStreamingSql(tx, {sql: selectSql});
await Promise.all([rowCount1, rowCount2]);
await tx.commit();
});
await database.close();
let request = spannerMock.getRequests().find(val => {
return (val as v1.ExecuteSqlRequest).sql;
}) as v1.ExecuteSqlRequest;
assert.ok(request, 'no ExecuteSqlRequest found');
assert.deepStrictEqual(request.transaction!.begin!.readWrite, {});
assert.strictEqual(request.sql, selectSql);
request = spannerMock
.getRequests()
.slice()
.reverse()
.find(val => {
return (val as v1.ExecuteSqlRequest).sql;
}) as v1.ExecuteSqlRequest;
assert.ok(request, 'no ExecuteSqlRequest found');
assert.strictEqual(request.sql, selectSql);
assert.ok(request.transaction!.id, 'TransactionID is not set.');
const beginTxnRequest = spannerMock.getRequests().find(val => {
return (val as v1.BeginTransactionRequest).options?.readWrite;
}) as v1.BeginTransactionRequest;
assert.ok(!beginTxnRequest, 'beginTransaction was called');
});There was a problem hiding this comment. You're absolutely right. Added a lock that prevents a multiple inline begin transaction at the same time. Thanks! |
||
| transaction.id = this.id as Uint8Array; | ||
| } else if (typeof this._options.readWrite !== 'undefined') { | ||
| transaction.begin = this._options; | ||
| } else { | ||
| transaction.singleUse = this._options; | ||
| } | ||
|
|
@@ -1059,7 +1067,7 @@ export class Snapshot extends EventEmitter { | |
| }; | ||
|
|
||
| const makeRequest = (resumeToken?: ResumeToken): Readable => { | ||
| if (!reqOpts || (this.id && !reqOpts.transaction.id)) { | ||
| try { | ||
| sanitizeRequest(); | ||
| } catch (e) { | ||
|
|
@@ -1078,10 +1086,14 @@ export class Snapshot extends EventEmitter { | |
| }); | ||
| }; | ||
|
|
||
| return partialResultStream(this._wrapWithIdWaiter(makeRequest), { | ||
| json, | ||
| jsonOptions, | ||
| maxResumeRetries, | ||
| }).on('response', response => { | ||
| if (response.metadata && response.metadata!.transaction && !this.id) { | ||
| this._update(response.metadata!.transaction); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
|
|
@@ -1226,6 +1238,48 @@ export class Snapshot extends EventEmitter { | |
|
|
||
| return {params, paramTypes}; | ||
| } | ||
|
|
||
| /** | ||
| * Update transaction properties from the response. | ||
| * | ||
| * @private | ||
| * | ||
| * @param {spannerClient.spanner.v1.ITransaction} resp Response object. | ||
| */ | ||
| private _update(resp: spannerClient.spanner.v1.ITransaction): void { | ||
| const {id, readTimestamp} = resp; | ||
|
|
||
| this.id = id!; | ||
| this.metadata = resp; | ||
|
|
||
| if (readTimestamp) { | ||
| this.readTimestampProto = readTimestamp; | ||
| this.readTimestamp = new PreciseDate(readTimestamp as DateStruct); | ||
| } | ||
| this._idWaiter.emit('notify'); | ||
| } | ||
|
|
||
| /** | ||
| * Wrap `makeRequest` function with the lock to make sure the inline begin | ||
| * transaction can happen only once. | ||
| * | ||
| * @param makeRequest | ||
| * @private | ||
| */ | ||
| private _wrapWithIdWaiter( | ||
| makeRequest: (resumeToken?: ResumeToken) => Readable): | ||
| (resumeToken?: ResumeToken) => Readable { | ||
| if (this.id || typeof this._options.readWrite === 'undefined') { | ||
| return makeRequest; | ||
| } | ||
| if (!this._inlineBeginStarted) { | ||
| this._inlineBeginStarted = true; | ||
| return makeRequest; | ||
| } | ||
| return (resumeToken?: ResumeToken): Readable => | ||
| this._idWaiter.once('notify', () => | ||
| this._idWaiter.wrap(makeRequest(resumeToken))); | ||
| } | ||
| } | ||
|
|
||
| /*! Developer Documentation | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Should this rather be
else if(this._options.readWrite)There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
yes, fixed, thanks