Handle errors in a middle of a stream

packages/client-common/src/settings.ts

@@ -1583,6 +1583,7 @@ interface ClickHouseServerSettings {
 
 /** @see https://clickhouse.com/docs/en/interfaces/http */
 interface ClickHouseHTTPSettings {
+  http_write_exception_in_output_format: Bool
   /** Ensures that the entire response is buffered.
    *  In this case, the data that is not stored in memory will be buffered in a temporary server file.
    *  This could help prevent errors that might occur during the streaming of SELECT queries.

packages/client-node/__tests__/integration/node_streaming_errors.test.ts

@@ -0,0 +1,83 @@
+import {
+  ClickHouseError,
+  type ClickHouseSettings,
+} from '@clickhouse/client-common'
+import { createTestClient } from '@test/utils'
+import type { StreamableDataFormat } from '../../src'
+import type { NodeClickHouseClient } from '../../src/client'
+
+describe('[Node.js] Errors during streaming', () => {
+  let client: NodeClickHouseClient
+  beforeAll(() => {
+    client = createTestClient() as NodeClickHouseClient
+  })
+  afterAll(async () => {
+    await client.close()
+  })
+
+  it('should work with CSV', async () => {
+    let err: unknown
+    let rowsCount = 0
+    try {
+      const rs = await runQuery('CSV')
+      const stream = rs.stream()
+      for await (const rows of stream) {
+        rowsCount += rows.length
+      }
+    } catch (e) {
+      err = e
+    }
+    assertStreamFailure(rowsCount, err)
+  })
+
+  it('should work with JSONEachRow', async () => {
+    let err: unknown
+    let rowsCount = 0
+    try {
+      const rs = await runQuery('JSONEachRow', {
+        http_write_exception_in_output_format: 0,
+      })
+      const stream = rs.stream()
+      for await (const rows of stream) {
+        for (const row of rows) {
+          row.json() // should not produce SyntaxError
+          rowsCount++
+        }
+      }
+    } catch (e) {
+      err = e
+    }
+    assertStreamFailure(rowsCount, err)
+  })
+
+  function runQuery<F extends StreamableDataFormat>(
+    format: F,
+    settings?: ClickHouseSettings,
+  ) {
+    return client.query({
+      query: `
+        SELECT
+            number,
+            if(number > 20, throwIf(1, 'Boom'), number) AS value
+        FROM numbers(1000)
+      `,
+      format,
+      clickhouse_settings: {
+        // Forces CH to send more chunks of data (1 row each),
+        // so that we get 200 OK + headers first instead of 500 ISE
+        max_block_size: '1',
+        ...(settings ?? {}),
+      },
+    })
+  }
+
+  function assertStreamFailure(rowsCount: number, err: unknown) {
+    // rows count may vary, but the last row will always contain the error message
+    expect(rowsCount).toBeGreaterThanOrEqual(10)
+    expect(err).toBeInstanceOf(ClickHouseError)
+    expect((err as ClickHouseError).code).toBe('395')
+    expect((err as ClickHouseError).type).toBe(
+      'FUNCTION_THROW_IF_VALUE_IS_NON_ZERO',
+    )
+  }
+})

packages/client-node/src/result_set.ts

@@ -7,8 +7,10 @@ import type {
   Row,
 } from '@clickhouse/client-common'
 import {
+  ClickHouseError,
   isNotStreamableJSONFamily,
   isStreamableJSONFamily,
+  parseError,
   validateStreamFormat,
 } from '@clickhouse/client-common'
 import { Buffer } from 'buffer'
@@ -111,6 +113,7 @@ export class ResultSet<Format extends DataFormat | unknown>
     validateStreamFormat(this.format)
 
     let incompleteChunks: Buffer[] = []
+    let errorRowText: string | undefined
     const logError = this.log_error
     const toRows = new Transform({
       transform(
@@ -156,7 +159,22 @@ export class ResultSet<Format extends DataFormat | unknown>
             } else {
               // to be processed during the first pass for the next chunk
               incompleteChunks.push(chunk.subarray(lastIdx))
-              this.push(rows)
+              // error reporting goes like this:
+              // __exception__\r\n              // - the row before the last one
+              // Code: X. DB::Exception: ...\n  // - the very last row
+              // we are not going to push these rows downstream
+              if (
+                rows.length > 1 &&
+                rows[rows.length - 2].text === errorHeaderMessage
+              ) {
+                errorRowText = rows[rows.length - 1].text
+                // push the remaining rows before the error
+                if (rows.length > 2) {
+                  this.push(rows.slice(0, -2))
+                }
+              } else {
+                this.push(rows)
+              }
             }
             lastIdx = idx + 1 // skipping newline character
           } while (idx !== -1)
@@ -165,6 +183,19 @@ export class ResultSet<Format extends DataFormat | unknown>
         }
         callback()
       },
+      // will be triggered if ClickHouse terminates the connection with an error while streaming
+      destroy(err: Error | null, callback: (error?: Error | null) => void) {
+        if (errorRowText !== undefined) {
+          const maybeLastRowErr = parseError(errorRowText)
+          if (maybeLastRowErr instanceof ClickHouseError) {
+            callback(maybeLastRowErr)
+          }
+        } else if (err !== null) {
+          callback(err)
+        } else {
+          callback()
+        }
+      },
       autoDestroy: true,
       objectMode: true,
     })
@@ -203,3 +234,4 @@ export class ResultSet<Format extends DataFormat | unknown>
 
 const streamAlreadyConsumedMessage = 'Stream has been already consumed'
 const resultSetClosedMessage = 'ResultSet has been closed'
+const errorHeaderMessage = `__exception__\r`