Remove split2, refactor Rows for better performance

__tests__/integration/abort_request.test.ts

@@ -63,7 +63,7 @@ describe('abort request', () => {
         .then(async (rows) => {
           const stream = rows.stream()
           for await (const chunk of stream) {
-            const [[number]] = chunk.json()
+            const [[number]] = chunk.json<[[string]]>()
             // abort when reach number 3
             if (number === '3') {
               controller.abort()
@@ -86,10 +86,10 @@ describe('abort request', () => {
         .then(async function (rows) {
           const stream = rows.stream()
           for await (const chunk of stream) {
-            const [[number]] = chunk.json()
+            const [[number]] = chunk.json<[[string]]>()
             // abort when reach number 3
             if (number === '3') {
-              stream.destroy()
+              await stream.return()
             }
           }
         })

__tests__/integration/select.test.ts

@@ -1,16 +1,19 @@
-import type Stream from 'stream'
 import { type ClickHouseClient, type ResponseJSON, type Row } from '../../src'
 import { createTestClient } from '../utils'
 
-async function rowsValues(stream: Stream.Readable): Promise<any[]> {
+async function rowsValues(
+  stream: AsyncGenerator<Row, unknown>
+): Promise<any[]> {
   const result: any[] = []
   for await (const chunk of stream) {
     result.push((chunk as Row).json())
   }
   return result
 }
 
-async function rowsText(stream: Stream.Readable): Promise<string[]> {
+async function rowsText(
+  stream: AsyncGenerator<Row, unknown>
+): Promise<string[]> {
   const result: string[] = []
   for await (const chunk of stream) {
     result.push((chunk as Row).text())
@@ -47,30 +50,28 @@ describe('select', () => {
   })
 
   describe('consume the response only once', () => {
-    async function assertAlreadyConsumed$<T>(fn: () => Promise<T>) {
+    async function assertAlreadyConsumed<T>(fn: () => Promise<T>) {
       await expect(fn()).rejects.toMatchObject(
         expect.objectContaining({
           message: 'Stream has been already consumed',
         })
       )
     }
-    function assertAlreadyConsumed<T>(fn: () => T) {
-      expect(fn).toThrow(
-        expect.objectContaining({
-          message: 'Stream has been already consumed',
-        })
-      )
-    }
+
     it('should consume a JSON response only once', async () => {
       const rows = await client.query({
         query: 'SELECT * FROM system.numbers LIMIT 1',
         format: 'JSONEachRow',
       })
       expect(await rows.json()).toEqual([{ number: '0' }])
       // wrap in a func to avoid changing inner "this"
-      await assertAlreadyConsumed$(() => rows.json())
-      await assertAlreadyConsumed$(() => rows.text())
-      await assertAlreadyConsumed(() => rows.stream())
+      await assertAlreadyConsumed(() => rows.json())
+      await assertAlreadyConsumed(() => rows.text())
+      await assertAlreadyConsumed(async () => {
+        for await (const r of rows.stream()) {
+          r.text()
+        }
+      })
     })
 
     it('should consume a text response only once', async () => {
@@ -80,9 +81,13 @@ describe('select', () => {
       })
       expect(await rows.text()).toEqual('0\n')
       // wrap in a func to avoid changing inner "this"
-      await assertAlreadyConsumed$(() => rows.json())
-      await assertAlreadyConsumed$(() => rows.text())
-      await assertAlreadyConsumed(() => rows.stream())
+      await assertAlreadyConsumed(() => rows.json())
+      await assertAlreadyConsumed(() => rows.text())
+      await assertAlreadyConsumed(async () => {
+        for await (const r of rows.stream()) {
+          r.text()
+        }
+      })
     })
 
     it('should consume a stream response only once', async () => {
@@ -96,9 +101,13 @@ describe('select', () => {
       }
       expect(result).toEqual('0')
       // wrap in a func to avoid changing inner "this"
-      await assertAlreadyConsumed$(() => rows.json())
-      await assertAlreadyConsumed$(() => rows.text())
-      await assertAlreadyConsumed(() => rows.stream())
+      await assertAlreadyConsumed(() => rows.json())
+      await assertAlreadyConsumed(() => rows.text())
+      await assertAlreadyConsumed(async () => {
+        for await (const r of rows.stream()) {
+          r.text()
+        }
+      })
     })
   })
 
@@ -328,35 +337,14 @@ describe('select', () => {
         format: 'JSON',
       })
       try {
-        expect(() => result.stream()).toThrowError(
+        await expect(async () => result.stream().next()).rejects.toThrowError(
           'JSON format is not streamable'
         )
       } finally {
         result.close()
       }
     })
 
-    it('can pause response stream', async () => {
-      const result = await client.query({
-        query: 'SELECT number FROM system.numbers LIMIT 10000',
-        format: 'CSV',
-      })
-
-      const stream = result.stream()
-
-      let last = null
-      let i = 0
-      for await (const chunk of stream) {
-        last = chunk.text()
-        i++
-        if (i % 1000 === 0) {
-          stream.pause()
-          setTimeout(() => stream.resume(), 100)
-        }
-      }
-      expect(last).toBe('9999')
-    })
-
     describe('text()', () => {
       it('returns stream of rows in CSV format', async () => {
         const result = await client.query({

__tests__/integration/streaming_e2e.test.ts

@@ -1,7 +1,4 @@
-import Fs from 'fs'
-import Path from 'path'
 import Stream from 'stream'
-import split from 'split2'
 import { type ClickHouseClient } from '../../src'
 import { createTestClient, guid } from '../utils'
 import { createSimpleTable } from './fixtures/simple_table'
@@ -26,33 +23,32 @@ describe('streaming e2e', () => {
     await client.close()
   })
 
-  it('should stream a file', async () => {
-    // contains id as numbers in JSONCompactEachRow format ["0"]\n["1"]\n...
-    const filename = Path.resolve(
-      __dirname,
-      './fixtures/streaming_e2e_data.ndjson'
-    )
-
-    await client.insert({
-      table: tableName,
-      values: Fs.createReadStream(filename).pipe(
-        // should be removed when "insert" accepts a stream of strings/bytes
-        split((row: string) => JSON.parse(row))
-      ),
-      format: 'JSONCompactEachRow',
-    })
-
-    const response = await client.query({
-      query: `SELECT * from ${tableName}`,
-      format: 'JSONCompactEachRow',
-    })
-
-    const actual: string[] = []
-    for await (const row of response.stream()) {
-      actual.push(row.json())
-    }
-    expect(actual).toEqual(expected)
-  })
+  // it('should stream a file', async () => {
+  //   // contains id as numbers in JSONCompactEachRow format ["0"]\n["1"]\n...
+  //   const filename = Path.resolve(
+  //     __dirname,
+  //     './fixtures/streaming_e2e_data.ndjson'
+  //   )
+  //   await client.insert({
+  //     table: tableName,
+  //     values: Fs.createReadStream(filename).pipe(
+  //       // should be removed when "insert" accepts a stream of strings/bytes
+  //       split((row: string) => JSON.parse(row))
+  //     ),
+  //     format: 'JSONCompactEachRow',
+  //   })
+  //
+  //   const response = await client.query({
+  //     query: `SELECT * from ${tableName}`,
+  //     format: 'JSONCompactEachRow',
+  //   })
+  //
+  //   const actual: string[] = []
+  //   for await (const row of response.stream()) {
+  //     actual.push(row.json())
+  //   }
+  //   expect(actual).toEqual(expected)
+  // })
 
   it('should stream a stream created in-place', async () => {
     await client.insert({

__tests__/unit/rows.test.ts

@@ -1,4 +1,6 @@
-import { Row, Rows } from '../../src'
+import type { Row } from '../../src'
+import { Rows } from '../../src'
+import * as Stream from 'stream'
 import { Readable } from 'stream'
 
 describe('rows', () => {
@@ -27,29 +29,38 @@ describe('rows', () => {
     const rows = makeRows()
     const stream = rows.stream()
 
-    expect(stream.readableEnded).toBeFalsy()
-
     const result = []
     for await (const row of stream) {
-      result.push((row as Row).json())
+      result.push(row.json())
     }
 
     expect(result).toEqual(expectedJson)
-    expect(stream.readableEnded).toBeTruthy()
+    expect((await stream.next()).done).toBeTruthy()
 
-    expect(() => rows.stream()).toThrowError(err)
+    await expect(async () => {
+      for await (const r of rows.stream()) {
+        r.text()
+      }
+    }).rejects.toThrowError(err)
     await expect(rows.json()).rejects.toThrowError(err)
     await expect(rows.text()).rejects.toThrowError(err)
   })
 
   it('should be able to call Row.text and Row.json multiple times', async () => {
-    const chunk = '{"foo":"bar"}'
-    const obj = { foo: 'bar' }
-    const row = new Row(chunk, 'JSON')
-    expect(row.text()).toEqual(chunk)
-    expect(row.text()).toEqual(chunk)
-    expect(row.json()).toEqual(obj)
-    expect(row.json()).toEqual(obj)
+    const rows = new Rows(
+      Stream.Readable.from([Buffer.from('{"foo":"bar"}\n')]),
+      'JSONEachRow'
+    )
+    const singleRows: Row[] = []
+    for await (const r of rows.stream()) {
+      singleRows.push(r)
+    }
+    expect(singleRows).toHaveLength(1)
+    const [row] = singleRows
+    expect(row.text()).toEqual('{"foo":"bar"}')
+    expect(row.text()).toEqual('{"foo":"bar"}')
+    expect(row.json()).toEqual({ foo: 'bar' })
+    expect(row.json()).toEqual({ foo: 'bar' })
   })
 
   function makeRows() {

__tests__/unit/schema_select_result.test.ts

@@ -27,7 +27,10 @@ describe('schema select result', () => {
       .spyOn(client, 'query')
       .mockResolvedValueOnce(
         new Rows(
-          Readable.from(['{"valid":"json"}\n', 'invalid_json}\n']),
+          Readable.from([
+            Buffer.from('{"valid":"json"}\n'),
+            Buffer.from('invalid_json}\n'),
+          ]),
           'JSONEachRow'
         )
       )

examples/insert_file_stream_ndjson.ts

@@ -1,47 +1,47 @@
-import { createClient } from '@clickhouse/client'
-import Path from 'path'
-import Fs from 'fs'
-import split from 'split2'
-
-void (async () => {
-  const client = createClient()
-  const tableName = 'insert_file_stream_ndjson'
-  await client.exec({
-    query: `DROP TABLE IF EXISTS ${tableName}`,
-  })
-  await client.exec({
-    query: `
-      CREATE TABLE ${tableName} (id UInt64)
-      ENGINE MergeTree()
-      ORDER BY (id)
-    `,
-  })
-
-  // contains id as numbers in JSONCompactEachRow format ["0"]\n["0"]\n...
-  // see also: NDJSON format
-  const filename = Path.resolve(
-    process.cwd(),
-    './examples/resources/data.ndjson'
-  )
-
-  await client.insert({
-    table: tableName,
-    values: Fs.createReadStream(filename).pipe(
-      split((row: string) => JSON.parse(row))
-    ),
-    format: 'JSONCompactEachRow',
-  })
-
-  const rows = await client.query({
-    query: `SELECT * from ${tableName}`,
-    format: 'JSONEachRow',
-  })
-
-  // or just `rows.text()` / `rows.json()`
-  // to consume the entire response at once
-  for await (const row of rows.stream()) {
-    console.log(row.json())
-  }
-
-  await client.close()
-})()
+// import { createClient } from '@clickhouse/client'
+// import Path from 'path'
+// import Fs from 'fs'
+// import split from 'split2'
+//
+// void (async () => {
+//   const client = createClient()
+//   const tableName = 'insert_file_stream_ndjson'
+//   await client.exec({
+//     query: `DROP TABLE IF EXISTS ${tableName}`,
+//   })
+//   await client.exec({
+//     query: `
+//       CREATE TABLE ${tableName} (id UInt64)
+//       ENGINE MergeTree()
+//       ORDER BY (id)
+//     `,
+//   })
+//
+//   // contains id as numbers in JSONCompactEachRow format ["0"]\n["0"]\n...
+//   // see also: NDJSON format
+//   const filename = Path.resolve(
+//     process.cwd(),
+//     './examples/resources/data.ndjson'
+//   )
+//
+//   await client.insert({
+//     table: tableName,
+//     values: Fs.createReadStream(filename).pipe(
+//       split((row: string) => JSON.parse(row))
+//     ),
+//     format: 'JSONCompactEachRow',
+//   })
+//
+//   const rows = await client.query({
+//     query: `SELECT * from ${tableName}`,
+//     format: 'JSONEachRow',
+//   })
+//
+//   // or just `rows.text()` / `rows.json()`
+//   // to consume the entire response at once
+//   for await (const row of rows.stream()) {
+//     console.log(row.json())
+//   }
+//
+//   await client.close()
+// })()