Skip to content
This repository was archived by the owner on Mar 4, 2026. It is now read-only.

Commit f337089

Browse files
olavloiteolavloite
authored
fix: only reset pending value with resume token (#2000)
Fixes #1959
1 parent 937a7a1 commit f337089

3 files changed

Lines changed: 205 additions & 10 deletions

File tree

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,3 +12,4 @@ system-test/*key.json
1212
.DS_Store
1313
package-lock.json
1414
__pycache__
15+
.idea

‎src/partial-result-stream.ts‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -244,8 +244,7 @@ export class PartialResultStream extends Transform implements ResultEvents {
244244
}
245245
}
246246

247-
_clearPendingValues() {
248-
this._values = [];
247+
_resetPendingValues() {
249248
if (this._pendingValueForResume) {
250249
this._pendingValue = this._pendingValueForResume;
251250
} else {
@@ -484,7 +483,9 @@ export function partialResultStream(
484483
});
485484
};
486485
const makeRequest = (): void => {
487-
partialRSStream._clearPendingValues();
486+
if (is.defined(lastResumeToken) && lastResumeToken.length > 0) {
487+
partialRSStream._resetPendingValues();
488+
}
488489
lastRequestStream = requestFn(lastResumeToken);
489490
lastRequestStream.on('end', endListener);
490491
requestsStream.add(lastRequestStream);

‎test/spanner.ts‎

Lines changed: 200 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -120,7 +120,6 @@ describe('Spanner with mock server', () => {
120120
}
121121
);
122122
});
123-
server.start();
124123
spannerMock.putStatementResult(
125124
selectSql,
126125
mock.StatementResult.resultSet(mock.createSimpleResultSet())
@@ -3695,10 +3694,10 @@ describe('Spanner with mock server', () => {
36953694
});
36963695

36973696
it('should return all values from PartialResultSet with chunked string value', async () => {
3698-
for (const includeResumeToken in [true, false]) {
3697+
for (const includeResumeToken of [true, false]) {
36993698
// eslint-disable-next-line @typescript-eslint/no-explicit-any
37003699
let errorOnIndexes: any;
3701-
for (errorOnIndexes in [[], [0], [1], [0, 1]]) {
3700+
for (errorOnIndexes of [[], [0], [1], [0, 1]]) {
37023701
const sql = 'SELECT * FROM TestTable';
37033702
const prs1 = PartialResultSet.create({
37043703
resumeToken: includeResumeToken
@@ -3747,10 +3746,10 @@ describe('Spanner with mock server', () => {
37473746
});
37483747

37493748
it('should return all values from PartialResultSet with chunked string value in an array', async () => {
3750-
for (const includeResumeToken in [true, false]) {
3749+
for (const includeResumeToken of [true, false]) {
37513750
// eslint-disable-next-line @typescript-eslint/no-explicit-any
37523751
let errorOnIndexes: any;
3753-
for (errorOnIndexes in [[], [0], [1], [0, 1]]) {
3752+
for (errorOnIndexes of [[], [0], [1], [0, 1]]) {
37543753
const sql = 'SELECT * FROM TestTable';
37553754
const prs1 = PartialResultSet.create({
37563755
resumeToken: includeResumeToken
@@ -3800,10 +3799,10 @@ describe('Spanner with mock server', () => {
38003799
});
38013800

38023801
it('should return all values from PartialResultSet with chunked list value', async () => {
3803-
for (const includeResumeToken in [true, false]) {
3802+
for (const includeResumeToken of [true, false]) {
38043803
// eslint-disable-next-line @typescript-eslint/no-explicit-any
38053804
let errorOnIndexes: any;
3806-
for (errorOnIndexes in [[], [0], [1], [0, 1]]) {
3805+
for (errorOnIndexes of [[], [0], [1], [0, 1]]) {
38073806
const sql = 'SELECT * FROM TestTable';
38083807
const prs1 = PartialResultSet.create({
38093808
resumeToken: includeResumeToken
@@ -4047,6 +4046,200 @@ describe('Spanner with mock server', () => {
40474046
}
40484047
});
40494048

4049+
it('should clear pending values if the last partial result did not have a resume token and was not a complete row', async () => {
4050+
const sql = 'SELECT * FROM TestTable';
4051+
const prs1 = PartialResultSet.create({
4052+
resumeToken: undefined,
4053+
metadata: createMultiColumnMetadata(),
4054+
values: [
4055+
{stringValue: 'id1.1'},
4056+
{stringValue: 'id1.2'},
4057+
{stringValue: '100'},
4058+
],
4059+
chunkedValue: false,
4060+
});
4061+
const prs2 = PartialResultSet.create({
4062+
resumeToken: undefined,
4063+
values: [
4064+
{boolValue: true},
4065+
{boolValue: true},
4066+
{numberValue: 0.5},
4067+
{stringValue: 'id2.1'},
4068+
{stringValue: 'id2.2'},
4069+
],
4070+
chunkedValue: false,
4071+
});
4072+
const prs3 = PartialResultSet.create({
4073+
resumeToken: undefined,
4074+
values: [
4075+
{stringValue: '200'},
4076+
{boolValue: true},
4077+
{boolValue: true},
4078+
{numberValue: 0.5},
4079+
],
4080+
});
4081+
// Let the stream return UNAVAILABLE on index 1 (so the second PartialResultSet).
4082+
setupResultsAndErrors(sql, [prs1, prs2, prs3], [1]);
4083+
const database = newTestDatabase();
4084+
try {
4085+
const [rows] = (await database.run({
4086+
sql,
4087+
json: true,
4088+
})) as Json[][];
4089+
verifyQueryResult(rows);
4090+
} finally {
4091+
await database.close();
4092+
}
4093+
});
4094+
4095+
it('should not clear pending values if the last partial result had a resume token and was not a complete row', async () => {
4096+
for (const errorIndexes of [[1], [2]]) {
4097+
const sql = 'SELECT * FROM TestTable';
4098+
const prs1 = PartialResultSet.create({
4099+
resumeToken: Buffer.from('00000000'),
4100+
metadata: createMultiColumnMetadata(),
4101+
values: [
4102+
{stringValue: 'id1.1'},
4103+
{stringValue: 'id1.2'},
4104+
{stringValue: '100'},
4105+
],
4106+
chunkedValue: false,
4107+
});
4108+
const prs2 = PartialResultSet.create({
4109+
resumeToken: undefined,
4110+
values: [
4111+
{boolValue: true},
4112+
{boolValue: true},
4113+
{numberValue: 0.5},
4114+
{stringValue: 'id2.1'},
4115+
{stringValue: 'id2.2'},
4116+
],
4117+
chunkedValue: false,
4118+
});
4119+
const prs3 = PartialResultSet.create({
4120+
resumeToken: undefined,
4121+
values: [
4122+
{stringValue: '200'},
4123+
{boolValue: true},
4124+
{boolValue: true},
4125+
{numberValue: 0.5},
4126+
],
4127+
});
4128+
setupResultsAndErrors(sql, [prs1, prs2, prs3], errorIndexes);
4129+
const database = newTestDatabase();
4130+
try {
4131+
const [rows] = (await database.run({
4132+
sql,
4133+
json: true,
4134+
})) as Json[][];
4135+
verifyQueryResult(rows);
4136+
} finally {
4137+
await database.close();
4138+
}
4139+
}
4140+
});
4141+
4142+
it('should not clear pending values if the last partial result was chunked and had a resume token', async () => {
4143+
for (const errorIndexes of [[2]]) {
4144+
const sql = 'SELECT * FROM TestTable';
4145+
const prs1 = PartialResultSet.create({
4146+
resumeToken: Buffer.from('00000000'),
4147+
metadata: createMultiColumnMetadata(),
4148+
values: [
4149+
{stringValue: 'id1.1'},
4150+
{stringValue: 'id1.2'},
4151+
{stringValue: '100'},
4152+
],
4153+
chunkedValue: true,
4154+
});
4155+
const prs2 = PartialResultSet.create({
4156+
resumeToken: undefined,
4157+
values: [
4158+
// The previous value was chunked, but it is still perfectly possible that it actually contained
4159+
// the entire value. So in this case the actual value was '100'.
4160+
{stringValue: ''},
4161+
{boolValue: true},
4162+
{boolValue: true},
4163+
{numberValue: 0.5},
4164+
{stringValue: 'id2.1'},
4165+
{stringValue: 'id2.2'},
4166+
],
4167+
chunkedValue: false,
4168+
});
4169+
const prs3 = PartialResultSet.create({
4170+
resumeToken: undefined,
4171+
values: [
4172+
{stringValue: '200'},
4173+
{boolValue: true},
4174+
{boolValue: true},
4175+
{numberValue: 0.5},
4176+
],
4177+
});
4178+
setupResultsAndErrors(sql, [prs1, prs2, prs3], errorIndexes);
4179+
const database = newTestDatabase();
4180+
try {
4181+
const [rows] = (await database.run({
4182+
sql,
4183+
json: true,
4184+
})) as Json[][];
4185+
verifyQueryResult(rows);
4186+
} finally {
4187+
await database.close();
4188+
}
4189+
}
4190+
});
4191+
4192+
function verifyQueryResult(rows: Json[]) {
4193+
assert.strictEqual(rows.length, 2);
4194+
assert.strictEqual(rows[0].col1, 'id1.1');
4195+
assert.strictEqual(rows[0].col2, 'id1.2');
4196+
assert.strictEqual(rows[0].col3, 100);
4197+
assert.strictEqual(rows[0].col4, true);
4198+
assert.strictEqual(rows[0].col5, true);
4199+
assert.strictEqual(rows[0].col6, 0.5);
4200+
4201+
assert.strictEqual(rows[1].col1, 'id2.1');
4202+
assert.strictEqual(rows[1].col2, 'id2.2');
4203+
assert.strictEqual(rows[1].col3, 200);
4204+
assert.strictEqual(rows[1].col4, true);
4205+
assert.strictEqual(rows[1].col5, true);
4206+
assert.strictEqual(rows[1].col6, 0.5);
4207+
}
4208+
4209+
function createMultiColumnMetadata() {
4210+
const fields = [
4211+
protobuf.StructType.Field.create({
4212+
name: 'col1',
4213+
type: protobuf.Type.create({code: protobuf.TypeCode.STRING}),
4214+
}),
4215+
protobuf.StructType.Field.create({
4216+
name: 'col2',
4217+
type: protobuf.Type.create({code: protobuf.TypeCode.STRING}),
4218+
}),
4219+
protobuf.StructType.Field.create({
4220+
name: 'col3',
4221+
type: protobuf.Type.create({code: protobuf.TypeCode.INT64}),
4222+
}),
4223+
protobuf.StructType.Field.create({
4224+
name: 'col4',
4225+
type: protobuf.Type.create({code: protobuf.TypeCode.BOOL}),
4226+
}),
4227+
protobuf.StructType.Field.create({
4228+
name: 'col5',
4229+
type: protobuf.Type.create({code: protobuf.TypeCode.BOOL}),
4230+
}),
4231+
protobuf.StructType.Field.create({
4232+
name: 'col6',
4233+
type: protobuf.Type.create({code: protobuf.TypeCode.FLOAT64}),
4234+
}),
4235+
];
4236+
return new protobuf.ResultSetMetadata({
4237+
rowType: new protobuf.StructType({
4238+
fields,
4239+
}),
4240+
});
4241+
}
4242+
40504243
function createMetadata() {
40514244
const fields = [
40524245
protobuf.StructType.Field.create({

0 commit comments

Comments
 (0)