From 423cce4d13c95fd7bcdfd3c6dcb9b8813b10e0c1 Mon Sep 17 00:00:00 2001 From: Renato Cron Date: Tue, 24 Mar 2026 16:51:40 -0300 Subject: [PATCH 1/2] feat(ResultSet): add rawStream() method to access underlying stream without row parsing Exposes the raw decompressed stream from ResultSet, useful when you need to pipe CSV/TSV data directly to a file or another stream without the overhead of row-by-row parsing via the Transform pipeline. This is particularly valuable for bulk data export scenarios (e.g., loading ClickHouse CSV into DuckDB via temp files) where .query() provides HTTP gzip compression (~15x smaller transfer) but the parsed .stream() adds unnecessary overhead. Implemented for both Node.js (Stream.Readable) and Web (ReadableStream). Co-Authored-By: Claude Opus 4.6 (1M context) --- packages/client-common/src/result.ts | 16 ++++++++++++ .../__tests__/unit/node_result_set.test.ts | 26 +++++++++++++++++++ packages/client-node/src/result_set.ts | 8 ++++++ packages/client-web/src/result_set.ts | 6 +++++ 4 files changed, 56 insertions(+) diff --git a/packages/client-common/src/result.ts b/packages/client-common/src/result.ts index 8f3bf1213..a83140910 100644 --- a/packages/client-common/src/result.ts +++ b/packages/client-common/src/result.ts @@ -141,6 +141,22 @@ export interface BaseResultSet { */ stream(): ResultStream + /** + * Returns the raw underlying stream without any row parsing or transformation. + * + * Useful when you need the raw bytes (e.g., CSV or TSV data) + * and want to pipe them directly to a file or another stream + * without the overhead of row-by-row parsing. + * + * The stream is already decompressed if HTTP compression is enabled. + * + * Should be called only once. + * + * The method should throw if the underlying stream was already consumed + * by calling the other methods. + */ + rawStream(): Stream + /** Close the underlying stream. */ close(): void diff --git a/packages/client-node/__tests__/unit/node_result_set.test.ts b/packages/client-node/__tests__/unit/node_result_set.test.ts index 7fbf4b00c..135fed67e 100644 --- a/packages/client-node/__tests__/unit/node_result_set.test.ts +++ b/packages/client-node/__tests__/unit/node_result_set.test.ts @@ -61,6 +61,32 @@ describe('[Node.js] ResultSet', () => { await expect(rs.text()).rejects.toEqual(err) }) + it('should return the raw stream without row parsing', async () => { + const rs = makeResultSet(getDataStream(), 'CSVWithNames') + const raw = rs.rawStream() + const chunks: Buffer[] = [] + for await (const chunk of raw) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)) + } + expect(Buffer.concat(chunks).toString()).toEqual(expectedText) + }) + + it('should throw if rawStream is called after text', async () => { + const rs = makeResultSet(getDataStream()) + await rs.text() + expect(() => rs.rawStream()).toThrow(errMsg) + }) + + it('should throw if text is called after rawStream', async () => { + const rs = makeResultSet(getDataStream()) + const raw = rs.rawStream() + // consume it + for await (const _ of raw) { + /* noop */ + } + await expect(rs.text()).rejects.toEqual(err) + }) + it('should be able to call Row.text and Row.json multiple times', async () => { const rs = makeResultSet( Stream.Readable.from([Buffer.from('{"foo":"bar"}\n')]), diff --git a/packages/client-node/src/result_set.ts b/packages/client-node/src/result_set.ts index d78bdd5b2..9aa27acdb 100644 --- a/packages/client-node/src/result_set.ts +++ b/packages/client-node/src/result_set.ts @@ -212,6 +212,14 @@ export class ResultSet< return pipeline as any } + /** See {@link BaseResultSet.rawStream}. */ + rawStream(): Stream.Readable { + if (this._stream.readableEnded) { + throw Error(streamAlreadyConsumedMessage) + } + return this._stream + } + /** See {@link BaseResultSet.close}. */ close() { this._stream.destroy(new Error(resultSetClosedMessage)) diff --git a/packages/client-web/src/result_set.ts b/packages/client-web/src/result_set.ts index 901be8df3..042ac7758 100644 --- a/packages/client-web/src/result_set.ts +++ b/packages/client-web/src/result_set.ts @@ -183,6 +183,12 @@ export class ResultSet< return pipeline as any } + /** See {@link BaseResultSet.rawStream}. */ + rawStream(): ReadableStream { + this.markAsConsumed() + return this._stream + } + async close(): Promise { this.markAsConsumed() await this._stream.cancel() From 638e020aeecfa33af974f96b6128c252c087aee4 Mon Sep 17 00:00:00 2001 From: Renato Cron Date: Tue, 24 Mar 2026 17:02:07 -0300 Subject: [PATCH 2/2] fix: add consumed flag to Node ResultSet, make close() idempotent on Web Addresses review feedback: - Node ResultSet now uses an explicit _consumed flag (like Web) instead of relying solely on readableEnded, ensuring mutual exclusion between rawStream()/text()/json()/stream() is enforced immediately. - Web close() no longer throws if called after a consumer method, allowing safe cleanup after rawStream() or stream(). - Added test for calling text()/stream() immediately after rawStream() without draining. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../__tests__/unit/node_result_set.test.ts | 15 ++++++---- packages/client-node/src/result_set.ts | 29 +++++++++---------- packages/client-web/src/result_set.ts | 3 +- 3 files changed, 24 insertions(+), 23 deletions(-) diff --git a/packages/client-node/__tests__/unit/node_result_set.test.ts b/packages/client-node/__tests__/unit/node_result_set.test.ts index 135fed67e..545e4be14 100644 --- a/packages/client-node/__tests__/unit/node_result_set.test.ts +++ b/packages/client-node/__tests__/unit/node_result_set.test.ts @@ -77,16 +77,19 @@ describe('[Node.js] ResultSet', () => { expect(() => rs.rawStream()).toThrow(errMsg) }) - it('should throw if text is called after rawStream', async () => { + it('should throw if text is called immediately after rawStream', async () => { const rs = makeResultSet(getDataStream()) - const raw = rs.rawStream() - // consume it - for await (const _ of raw) { - /* noop */ - } + rs.rawStream() + // text() should throw immediately — not only after the stream ends await expect(rs.text()).rejects.toEqual(err) }) + it('should throw if stream is called after rawStream', async () => { + const rs = makeResultSet(getDataStream()) + rs.rawStream() + expect(() => rs.stream()).toThrow(errMsg) + }) + it('should be able to call Row.text and Row.json multiple times', async () => { const rs = makeResultSet( Stream.Readable.from([Buffer.from('{"foo":"bar"}\n')]), diff --git a/packages/client-node/src/result_set.ts b/packages/client-node/src/result_set.ts index 9aa27acdb..d77658b40 100644 --- a/packages/client-node/src/result_set.ts +++ b/packages/client-node/src/result_set.ts @@ -67,6 +67,7 @@ export class ResultSet< private readonly exceptionTag: string | undefined = undefined private readonly log_error: (error: Error) => void private readonly jsonHandling: JSONHandling + private _consumed = false constructor( private _stream: Stream.Readable, @@ -91,20 +92,22 @@ export class ResultSet< } } - /** See {@link BaseResultSet.text}. */ - async text(): Promise { - if (this._stream.readableEnded) { + private markAsConsumed() { + if (this._consumed || this._stream.readableEnded) { throw Error(streamAlreadyConsumedMessage) } + this._consumed = true + } + + /** See {@link BaseResultSet.text}. */ + async text(): Promise { + this.markAsConsumed() return (await getAsText(this._stream)).toString() } /** See {@link BaseResultSet.json}. */ async json(): Promise> { - if (this._stream.readableEnded) { - throw Error(streamAlreadyConsumedMessage) - } - // JSONEachRow, etc. + // JSONEachRow, etc. — delegates to stream() which marks as consumed if (isStreamableJSONFamily(this.format as DataFormat)) { const result: T[] = [] const stream = this.stream() @@ -117,6 +120,7 @@ export class ResultSet< } // JSON, JSONObjectEachRow, etc. if (isNotStreamableJSONFamily(this.format as DataFormat)) { + this.markAsConsumed() const text = await getAsText(this._stream) return this.jsonHandling.parse(text) } @@ -126,12 +130,7 @@ export class ResultSet< /** See {@link BaseResultSet.stream}. */ stream(): ResultStream[]>> { - // If the underlying stream has already ended by calling `text` or `json`, - // Stream.pipeline will create a new empty stream - // but without "readableEnded" flag set to true - if (this._stream.readableEnded) { - throw Error(streamAlreadyConsumedMessage) - } + this.markAsConsumed() validateStreamFormat(this.format) @@ -214,9 +213,7 @@ export class ResultSet< /** See {@link BaseResultSet.rawStream}. */ rawStream(): Stream.Readable { - if (this._stream.readableEnded) { - throw Error(streamAlreadyConsumedMessage) - } + this.markAsConsumed() return this._stream } diff --git a/packages/client-web/src/result_set.ts b/packages/client-web/src/result_set.ts index 042ac7758..82b5048f5 100644 --- a/packages/client-web/src/result_set.ts +++ b/packages/client-web/src/result_set.ts @@ -190,7 +190,8 @@ export class ResultSet< } async close(): Promise { - this.markAsConsumed() + // close() should be safe to call even after a consumer method (rawStream, stream, etc.) + this.isAlreadyConsumed = true await this._stream.cancel() }