From 5f1c3e780297297fab176e3496b7b425f30d8696 Mon Sep 17 00:00:00 2001 From: Alka Trivedi Date: Thu, 27 Aug 2026 19:48:44 +0530 Subject: [PATCH 1/3] perf: optimize stream pipeline by eliminating events-intercept --- handwritten/spanner/package.json | 3 --- .../spanner/src/partial-result-stream.ts | 27 +++++++++---------- 2 files changed, 12 insertions(+), 18 deletions(-) diff --git a/handwritten/spanner/package.json b/handwritten/spanner/package.json index 717f82b15383..2f3b01b2b4e6 100644 --- a/handwritten/spanner/package.json +++ b/handwritten/spanner/package.json @@ -73,13 +73,11 @@ "big.js": "^7.0.0", "checkpoint-stream": "^0.1.2", "duplexify": "^4.1.3", - "events-intercept": "^2.0.0", "extend": "^3.0.2", "google-auth-library": "^10.0.0-rc.1", "google-gax": "5.0.6", "grpc-gcp": "^1.0.1", "lodash.snakecase": "^4.1.1", - "merge-stream": "^2.0.0", "p-queue": "^6.0.2", "protobufjs": "^7.4.0", "retry-request": "^8.0.0", @@ -98,7 +96,6 @@ "@types/extend": "^3.0.4", "@types/is": "^0.0.25", "@types/lodash.snakecase": "^4.1.9", - "@types/merge-stream": "^2.0.0", "@types/mocha": "^10.0.10", "@types/mv": "^2.1.4", "@types/ncp": "^2.0.8", diff --git a/handwritten/spanner/src/partial-result-stream.ts b/handwritten/spanner/src/partial-result-stream.ts index 2a1533c5419a..a75066e4f1d9 100644 --- a/handwritten/spanner/src/partial-result-stream.ts +++ b/handwritten/spanner/src/partial-result-stream.ts @@ -16,8 +16,6 @@ import {GrpcService} from './common-grpc/service'; import * as checkpointStream from 'checkpoint-stream'; -import * as eventsIntercept from 'events-intercept'; -import mergeStream = require('merge-stream'); import {common as p} from 'protobufjs'; import {Readable, Transform} from 'stream'; import * as streamEvents from 'stream-events'; @@ -578,17 +576,18 @@ export function partialResultStream( const maxQueued = 10; let lastResumeToken: ResumeToken; let lastRequestStream: Readable; + let errorListener: (err: grpc.ServiceError) => void; const startTime = Date.now(); const timeout = options?.gaxOptions?.timeout ?? Infinity; - // mergeStream allows multiple streams to be connected into one. This is good; + // requestsStream allows multiple streams to be connected into one. This is good; // if we need to retry a request and pipe more data to the user's stream. // We also add an additional stream that can be used to flush any remaining // items in the checkpoint stream that have been received, and that did not // contain a resume token. - const requestsStream = mergeStream(); + const requestsStream = new stream.PassThrough({objectMode: true}); const flushStream = new stream.PassThrough({objectMode: true}); - requestsStream.add(flushStream); + flushStream.pipe(requestsStream, {end: false}); const partialRSStream = new PartialResultStream(options); const userStream = streamEvents(partialRSStream); // We keep track of the number of PartialResultSets that did not include a @@ -626,7 +625,11 @@ export function partialResultStream( } lastRequestStream = requestFn(lastResumeToken); lastRequestStream.on('end', endListener); - requestsStream.add(lastRequestStream); + errorListener = (err: grpc.ServiceError) => { + setImmediate(() => retry(err)); + }; + lastRequestStream.on('error', errorListener); + lastRequestStream.pipe(requestsStream, {end: false}); }; const retry = (err: grpc.ServiceError): void => { @@ -659,6 +662,9 @@ export function partialResultStream( if (lastRequestStream) { lastRequestStream.removeListener('end', endListener); + if (errorListener) { + lastRequestStream.removeListener('error', errorListener); + } lastRequestStream.destroy(); } // Delay the retry until all the values that are already in the stream @@ -674,15 +680,6 @@ export function partialResultStream( }; userStream.once('reading', makeRequest); - eventsIntercept.patch(requestsStream); - - // need types for events-intercept - // eslint-disable-next-line @typescript-eslint/no-explicit-any - (requestsStream as any).intercept('error', err => - // Retry __after__ all pending data has been processed to ensure that the - // checkpoint stream is reset at the correct position. - setImmediate(() => retry(err)), - ); return ( requestsStream From 6f9cacae78817c488ea79a1f7986173b488f39b9 Mon Sep 17 00:00:00 2001 From: Alka Trivedi Date: Fri, 28 Aug 2026 11:11:00 +0530 Subject: [PATCH 2/3] gemini review comments --- handwritten/spanner/src/partial-result-stream.ts | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/handwritten/spanner/src/partial-result-stream.ts b/handwritten/spanner/src/partial-result-stream.ts index a75066e4f1d9..923c65ad3389 100644 --- a/handwritten/spanner/src/partial-result-stream.ts +++ b/handwritten/spanner/src/partial-result-stream.ts @@ -17,7 +17,7 @@ import {GrpcService} from './common-grpc/service'; import * as checkpointStream from 'checkpoint-stream'; import {common as p} from 'protobufjs'; -import {Readable, Transform} from 'stream'; +import { PassThrough, Readable, Transform } from 'stream'; import * as streamEvents from 'stream-events'; import {grpc, CallOptions} from 'google-gax'; import {DeadlineError, isRetryableInternalError} from './transaction-runner'; @@ -25,7 +25,6 @@ import {DeadlineError, isRetryableInternalError} from './transaction-runner'; import {codec, JSONOptions, Json, Field, Value} from './codec'; import {protos} from '@google-cloud/spanner-api'; import google = protos.google; -import * as stream from 'stream'; import {isDefined, isEmpty, isString} from './helper'; const originalDecode = codec.decode; @@ -585,9 +584,9 @@ export function partialResultStream( // We also add an additional stream that can be used to flush any remaining // items in the checkpoint stream that have been received, and that did not // contain a resume token. - const requestsStream = new stream.PassThrough({objectMode: true}); - const flushStream = new stream.PassThrough({objectMode: true}); - flushStream.pipe(requestsStream, {end: false}); + const requestsStream = new PassThrough({ objectMode: true }); + const flushStream = new PassThrough({ objectMode: true }); + flushStream.pipe(requestsStream); const partialRSStream = new PartialResultStream(options); const userStream = streamEvents(partialRSStream); // We keep track of the number of PartialResultSets that did not include a @@ -616,7 +615,6 @@ export function partialResultStream( // then push `null` to end the stream. flushStream.push({resumeToken: '_'}); flushStream.push(null); - requestsStream.end(); }); }; const makeRequest = (): void => { From a4c89d2e252d47229f69e700f4035ba512efe1ef Mon Sep 17 00:00:00 2001 From: Alka Trivedi Date: Fri, 28 Aug 2026 12:14:57 +0530 Subject: [PATCH 3/3] refactor --- handwritten/spanner/src/partial-result-stream.ts | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/handwritten/spanner/src/partial-result-stream.ts b/handwritten/spanner/src/partial-result-stream.ts index 923c65ad3389..b78b09ff5efd 100644 --- a/handwritten/spanner/src/partial-result-stream.ts +++ b/handwritten/spanner/src/partial-result-stream.ts @@ -17,7 +17,7 @@ import {GrpcService} from './common-grpc/service'; import * as checkpointStream from 'checkpoint-stream'; import {common as p} from 'protobufjs'; -import { PassThrough, Readable, Transform } from 'stream'; +import {PassThrough, Readable, Transform} from 'stream'; import * as streamEvents from 'stream-events'; import {grpc, CallOptions} from 'google-gax'; import {DeadlineError, isRetryableInternalError} from './transaction-runner'; @@ -584,8 +584,8 @@ export function partialResultStream( // We also add an additional stream that can be used to flush any remaining // items in the checkpoint stream that have been received, and that did not // contain a resume token. - const requestsStream = new PassThrough({ objectMode: true }); - const flushStream = new PassThrough({ objectMode: true }); + const requestsStream = new PassThrough({objectMode: true}); + const flushStream = new PassThrough({objectMode: true}); flushStream.pipe(requestsStream); const partialRSStream = new PartialResultStream(options); const userStream = streamEvents(partialRSStream); @@ -660,9 +660,8 @@ export function partialResultStream( if (lastRequestStream) { lastRequestStream.removeListener('end', endListener); - if (errorListener) { - lastRequestStream.removeListener('error', errorListener); - } + lastRequestStream.removeAllListeners('error'); + lastRequestStream.on('error', () => {}); // Prevent unhandled exception crash lastRequestStream.destroy(); } // Delay the retry until all the values that are already in the stream