Drain insert response stream in Web version (#213)

CHANGELOG.md

@@ -1,3 +1,9 @@
+## 0.2.7 (Web only)
+
+### Bug fixes
+
+- Drain insert response stream in Web version - required to properly work with `async_insert`, especially in the Cloudflare Workers context.
+
 ## 0.2.6 (Common, Node.js)
 
 ### New features

examples/insert_cloud.ts

@@ -0,0 +1,58 @@
+import { createClient } from '@clickhouse/client' // or '@clickhouse/client-web'
+
+void (async () => {
+  const client = createClient({
+    host: getFromEnv('CLICKHOUSE_HOST'),
+    password: getFromEnv('CLICKHOUSE_PASSWORD'),
+    // See https://clickhouse.com/docs/en/optimize/asynchronous-inserts
+    clickhouse_settings: {
+      async_insert: 1,
+      wait_for_async_insert: 1,
+    },
+  })
+
+  // Create the table if necessary
+  const table = 'async_insert_example'
+  await client.command({
+    query: `
+      CREATE TABLE IF NOT EXISTS ${table}
+      (id UInt32, data String)
+      ENGINE MergeTree
+      ORDER BY id
+    `,
+    // Tell the server to send the response only when the DDL is fully executed
+    clickhouse_settings: {
+      wait_end_of_query: 1,
+    },
+  })
+
+  // Generate some random data for the sake of example...
+  const batch = [...new Array(1_000).keys()].map(() => ({
+    id: Math.floor(Math.random() * 100_000) + 1,
+    data: Math.random().toString(36).slice(2),
+  }))
+
+  await client.insert({
+    table,
+    format: 'JSONEachRow', // or other, depends on your data
+    values: batch,
+  })
+
+  const res = await client
+    .query({
+      query: `SELECT count(*) FROM ${table}`,
+      format: 'JSONEachRow',
+    })
+    .then((r) => r.json())
+
+  console.info(res)
+})()
+
+// Node.js only
+function getFromEnv(key: string) {
+  if (process.env[key]) {
+    return process.env[key]
+  }
+  console.error(`${key} environment variable should be set`)
+  process.exit(1)
+}

packages/client-common/__tests__/integration/insert.test.ts

@@ -137,4 +137,18 @@ describe('insert', () => {
       })
     )
   })
+
+  it('should work with async inserts', async () => {
+    await client.insert({
+      table: tableName,
+      values: jsonValues,
+      format: 'JSONEachRow',
+      // See https://clickhouse.com/docs/en/optimize/asynchronous-inserts
+      clickhouse_settings: {
+        async_insert: 1,
+        wait_for_async_insert: 1,
+      },
+    })
+    await assertJsonValues(client, tableName)
+  })
 })

packages/client-common/src/version.ts

@@ -1 +1 @@
-export default '0.2.6'
+export default '0.2.7'

packages/client-node/src/version.ts

@@ -1 +1 @@
-export default '0.2.6'
+export default '0.2.7'

packages/client-web/src/connection/web_connection.ts

@@ -92,11 +92,14 @@ export class WebConnection implements Connection<ReadableStream> {
       session_id: params.session_id,
       query_id,
     })
-    await this.request({
+    const res = await this.request({
       values: params.values,
       params,
       searchParams,
     })
+    if (res.body !== null) {
+      await res.text() // drain the response (it's empty anyway)
+    }
     return {
       query_id,
     }

packages/client-web/src/version.ts

@@ -1 +1 @@
-export default '0.2.6'
+export default '0.2.7'