Fix parquetQueryWorker for parallel requests.

Adds a queryId to the message protocol to match requests to results.

Author: platypii

package.json

@@ -23,7 +23,7 @@
   },
   "dependencies": {
     "highlight.js": "11.10.0",
-    "hightable": "0.5.3",
+    "hightable": "0.5.4",
     "hyparquet": "1.5.0",
     "hyparquet-compressors": "0.1.4"
   },
@@ -34,7 +34,7 @@
     "@rollup/plugin-terser": "0.4.4",
     "@rollup/plugin-typescript": "12.1.1",
     "@testing-library/react": "16.0.1",
-    "@types/node": "22.7.5",
+    "@types/node": "22.7.6",
     "@types/react": "18.3.11",
     "@types/react-dom": "18.3.1",
     "@vitejs/plugin-react": "4.3.2",
@@ -48,7 +48,7 @@
     "rollup-plugin-postcss": "4.0.2",
     "tslib": "2.8.0",
     "typescript": "5.6.3",
-    "typescript-eslint": "8.9.0",
+    "typescript-eslint": "8.10.0",
     "vitest": "2.1.3"
   }
 }

public/build/app.min.js

@@ -11,6 +11,7 @@ export function parquetDataFrame(from: AsyncBufferFrom, metadata: FileMetaData):
     header: children.map(child => child.element.name),
     numRows: Number(metadata.num_rows),
     rows(rowStart: number, rowEnd: number, orderBy?: string) {
+      console.log('parquetDataFrame.rows', { rowStart, rowEnd, orderBy })
       return parquetQueryWorker({ asyncBuffer: from, rowStart, rowEnd, orderBy })
     },
     sortable: true,

public/build/app.min.js.map

@@ -3,14 +3,14 @@ import { compressors } from 'hyparquet-compressors'
 import { asyncBufferFrom } from './parquetWorkerClient.js'
 
 self.onmessage = async ({ data }) => {
-  const { metadata, asyncBuffer, rowStart, rowEnd, orderBy } = data
+  const { metadata, asyncBuffer, rowStart, rowEnd, orderBy, queryId } = data
   const file = await asyncBufferFrom(asyncBuffer)
   try {
     const result = await parquetQuery({
       metadata, file, rowStart, rowEnd, orderBy, compressors,
     })
-    self.postMessage({ result })
+    self.postMessage({ result, queryId })
   } catch (error) {
-    self.postMessage({ error })
+    self.postMessage({ error, queryId })
   }
 }

public/build/worker.min.js

@@ -17,6 +17,8 @@ interface ParquetReadWorkerOptions {
 }
 
 let worker: Worker | undefined
+let nextQueryId = 0
+const pending = new Map<number, { resolve: (value: any) => void, reject: (error: any) => void }>()
 
 /**
  * Presents almost the same interface as parquetRead, but runs in a worker.
@@ -28,21 +30,24 @@ export function parquetQueryWorker({
   metadata, asyncBuffer, rowStart, rowEnd, orderBy }: ParquetReadWorkerOptions
 ): Promise<Record<string, any>[]> {
   return new Promise((resolve, reject) => {
+    const queryId = nextQueryId++
+    pending.set(queryId, { resolve, reject })
     // Create a worker
     if (!worker) {
       worker = new Worker(new URL('worker.min.js', import.meta.url))
-    }
-    worker.onmessage = ({ data }) => {
-      // Convert postmessage data to callbacks
-      if (data.error) {
-        reject(data.error)
-      } else if (data.result) {
-        resolve(data.result)
-      } else {
-        reject(new Error('Unexpected message from worker'))
+      worker.onmessage = ({ data }) => {
+        const { resolve, reject } = pending.get(data.queryId)!
+        // Convert postmessage data to callbacks
+        if (data.error) {
+          reject(data.error)
+        } else if (data.result) {
+          resolve(data.result)
+        } else {
+          reject(new Error('Unexpected message from worker'))
+        }
       }
     }
-    worker.postMessage({ metadata, asyncBuffer, rowStart, rowEnd, orderBy })
+    worker.postMessage({ queryId, metadata, asyncBuffer, rowStart, rowEnd, orderBy })
   })
 }
 
@@ -53,8 +58,8 @@ export async function asyncBufferFrom(from: AsyncBufferFrom): Promise<AsyncBuffe
   const key = JSON.stringify(from)
   const cached = cache.get(key)
   if (cached) return cached
-  const asyncBuffer = asyncBufferFromUrl(from.url, from.byteLength)
-  cache.set(key, asyncBuffer.then(cachedAsyncBuffer))
+  const asyncBuffer = asyncBufferFromUrl(from.url, from.byteLength).then(cachedAsyncBuffer)
+  cache.set(key, asyncBuffer)
   return asyncBuffer
 }
 const cache = new Map<string, Promise<AsyncBuffer>>()