diff --git a/handwritten/spanner/package.json b/handwritten/spanner/package.json
index 717f82b1538..2f3b01b2b4e 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 2a1533c5419..b78b09ff5ef 100644
--- a/handwritten/spanner/src/partial-result-stream.ts
+++ b/handwritten/spanner/src/partial-result-stream.ts
@@ -16,10 +16,8 @@
 
 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 {PassThrough, Readable, Transform} from 'stream';
 import * as streamEvents from 'stream-events';
 import {grpc, CallOptions} from 'google-gax';
 import {DeadlineError, isRetryableInternalError} from './transaction-runner';
@@ -27,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;
@@ -578,17 +575,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 flushStream = new stream.PassThrough({objectMode: true});
-  requestsStream.add(flushStream);
+  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
@@ -617,7 +615,6 @@ export function partialResultStream(
       // then push `null` to end the stream.
       flushStream.push({resumeToken: '_'});
       flushStream.push(null);
-      requestsStream.end();
     });
   };
   const makeRequest = (): void => {
@@ -626,7 +623,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 +660,8 @@ export function partialResultStream(
 
     if (lastRequestStream) {
       lastRequestStream.removeListener('end', endListener);
+      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
@@ -674,15 +677,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