Move parsing to web worker

platypii · platypii · commit 5f16b3ba6a09 · 2024-10-14T15:33:32.000-07:00
diff --git a/.gitattributes b/.gitattributes
@@ -1,3 +1,2 @@
-public/build/app.js -diff
 *.js.map -diff
 *.min.js -diff
diff --git a/package.json b/package.json
@@ -36,9 +36,9 @@
     "@testing-library/react": "16.0.1",
     "@types/node": "22.7.5",
     "@types/react": "18.3.11",
-    "@types/react-dom": "18.3.0",
+    "@types/react-dom": "18.3.1",
     "@vitejs/plugin-react": "4.3.2",
-    "@vitest/coverage-v8": "2.1.2",
+    "@vitest/coverage-v8": "2.1.3",
     "eslint": "9.12.0",
     "globals": "15.11.0",
     "jsdom": "25.0.1",
@@ -48,7 +48,7 @@
     "rollup-plugin-postcss": "4.0.2",
     "tslib": "2.7.0",
     "typescript": "5.6.3",
-    "typescript-eslint": "8.8.1",
-    "vitest": "2.1.2"
+    "typescript-eslint": "8.9.0",
+    "vitest": "2.1.3"
   }
 }
diff --git a/public/build/app.min.js b/public/build/app.min.js
diff --git a/public/build/app.min.js.map b/public/build/app.min.js.map
diff --git a/public/build/worker.min.js b/public/build/worker.min.js
diff --git a/public/build/worker.min.js.map b/public/build/worker.min.js.map
diff --git a/rollup.config.js b/rollup.config.js
@@ -33,4 +33,19 @@ export default [
       }),
     ],
   },
+  // web worker
+  {
+    input: 'src/workers/parquetWorker.ts',
+    output: {
+      file: 'public/build/worker.min.js',
+      format: 'umd',
+      sourcemap: true,
+    },
+    plugins: [
+      commonjs(),
+      resolve({ browser: true }),
+      terser(),
+      typescript(),
+    ],
+  },
 ]
diff --git a/src/components/Cell.tsx b/src/components/Cell.tsx
@@ -1,7 +1,7 @@
 import React, { useEffect, useState } from 'react'
-import { asyncBufferFrom, parquetDataFrame } from '../tableProvider.js'
-import Highlight from './Highlight.js'
+import { parquetDataFrame } from '../tableProvider.js'
 import Layout from './Layout.js'
+import { asyncBufferFromUrl, parquetMetadataAsync } from 'hyparquet'
 
 enum LoadingState {
   NotLoaded,
@@ -36,10 +36,12 @@ export default function CellView() {
       try {
         // TODO: handle first row > 100kb
         setProgress(0.25)
-        const asyncBuffer = await asyncBufferFrom(url)
+        const asyncBuffer = await asyncBufferFromUrl(url)
+        const from = { url, byteLength: asyncBuffer.byteLength }
         setProgress(0.5)
-        const df = await parquetDataFrame(asyncBuffer)
+        const metadata = await parquetMetadataAsync(asyncBuffer)
         setProgress(0.75)
+        const df = await parquetDataFrame(from, metadata)
         const rows = await df.rows(row, row + 1)
         const colName = df.header[col]
         const cell = rows[0][colName]
diff --git a/src/components/viewers/ParquetView.tsx b/src/components/viewers/ParquetView.tsx
@@ -1,8 +1,9 @@
 import HighTable, { DataFrame, rowCache } from 'hightable'
 import React, { useCallback, useEffect, useState } from 'react'
-import { asyncBufferFrom, parquetDataFrame } from '../../tableProvider.js'
+import { parquetDataFrame } from '../../tableProvider.js'
 import { Spinner } from '../Layout.js'
 import ContentHeader from './ContentHeader.js'
+import { asyncBufferFromUrl, FileMetaData, parquetMetadataAsync } from 'hyparquet'
 
 enum LoadingState {
   NotLoaded,
@@ -27,6 +28,7 @@ interface Content {
 export default function ParquetView({ file, setProgress, setError }: ViewerProps) {
   const [loading, setLoading] = useState<LoadingState>(LoadingState.NotLoaded)
   const [content, setContent] = useState<Content>()
+  const [metadata, setMetadata] = useState<FileMetaData>()
 
   const isUrl = file.startsWith('http://') || file.startsWith('https://')
   const url = isUrl ? file : '/api/store/get?key=' + file
@@ -35,9 +37,12 @@ export default function ParquetView({ file, setProgress, setError }: ViewerProps
     async function loadParquetDataFrame() {
       try {
         setProgress(0.33)
-        const asyncBuffer = await asyncBufferFrom(url)
+        const asyncBuffer = await asyncBufferFromUrl(url)
+        const from = { url, byteLength: asyncBuffer.byteLength }
         setProgress(0.66)
-        let dataframe = await parquetDataFrame(asyncBuffer)
+        const metadata = await parquetMetadataAsync(asyncBuffer)
+        setMetadata(metadata)
+        let dataframe = await parquetDataFrame(from, metadata)
         dataframe = rowCache(dataframe)
         const fileSize = asyncBuffer.byteLength
         setContent({ dataframe, fileSize })
diff --git a/src/tableProvider.ts b/src/tableProvider.ts
@@ -1,27 +1,18 @@
 import type { DataFrame } from 'hightable'
-import {
-  AsyncBuffer, SchemaTree, parquetMetadataAsync, parquetQuery, parquetSchema,
-} from 'hyparquet'
-import { compressors } from 'hyparquet-compressors'
+import { AsyncBuffer, FileMetaData, parquetSchema } from 'hyparquet'
 import { readableStreamToArrayBuffer } from './streamConverters.js'
+import { AsyncBufferFrom, parquetQueryWorker } from './workers/parquetWorkerClient.js'
 
 /**
- * Construct a dataframe from a parquet file asynchronously.
+ * Convert a parquet file into a dataframe.
  */
-export async function parquetDataFrame(asyncBuffer: AsyncBuffer): Promise<DataFrame> {
-  // load parquet metadata
-  const metadata = await parquetMetadataAsync(asyncBuffer)
-
-  // construct dataframe
-  const { children }: SchemaTree = parquetSchema(metadata)
-  // TODO: concat child names for like-like columns?
-  const header = children.map(child => child.element.name)
-  const numRows = Number(metadata.num_rows)
+export function parquetDataFrame(from: AsyncBufferFrom, metadata: FileMetaData): DataFrame {
+  const { children } = parquetSchema(metadata)
   return {
-    header,
-    numRows,
-    rows(rowStart?: number, rowEnd?: number, orderBy?: string) {
-      return parquetQuery({ metadata, compressors, file: asyncBuffer, rowStart, rowEnd, orderBy })
+    header: children.map(child => child.element.name),
+    numRows: Number(metadata.num_rows),
+    rows(rowStart: number, rowEnd: number, orderBy?: string) {
+      return parquetQueryWorker({ asyncBuffer: from, rowStart, rowEnd, orderBy })
     },
     sortable: true,
   }
diff --git a/src/workers/parquetWorker.ts b/src/workers/parquetWorker.ts
@@ -0,0 +1,15 @@
+import { asyncBufferFromUrl, parquetQuery } from 'hyparquet'
+import { compressors } from 'hyparquet-compressors'
+
+self.onmessage = async ({ data }) => {
+  const { metadata, asyncBuffer, rowStart, rowEnd, orderBy } = data
+  const file = await asyncBufferFromUrl(asyncBuffer.url)
+  try {
+    const result = await parquetQuery({
+      metadata, file, rowStart, rowEnd, orderBy, compressors,
+    })
+    self.postMessage({ result })
+  } catch (error) {
+    self.postMessage({ error })
+  }
+}
diff --git a/src/workers/parquetWorkerClient.ts b/src/workers/parquetWorkerClient.ts
@@ -0,0 +1,47 @@
+import type { FileMetaData } from 'hyparquet'
+
+// Serializable constructor for AsyncBuffers
+export interface AsyncBufferFrom {
+  url: string
+  byteLength: number
+}
+
+// Same as ParquetReadOptions, but AsyncBufferFrom instead of AsyncBuffer
+interface ParquetReadWorkerOptions {
+  asyncBuffer: AsyncBufferFrom
+  metadata?: FileMetaData // parquet metadata, will be parsed if not provided
+  columns?: number[] // columns to read, all columns if undefined
+  rowStart?: number // inclusive
+  rowEnd?: number // exclusive
+  orderBy?: string // column to sort by
+}
+
+let worker: Worker | undefined
+
+/**
+ * Presents almost the same interface as parquetRead, but runs in a worker.
+ * This is useful for reading large parquet files without blocking the main thread.
+ * Instead of taking an AsyncBuffer, it takes a FileContent, because it needs
+ * to be serialized to the worker.
+ */
+export function parquetQueryWorker({
+  metadata, asyncBuffer, rowStart, rowEnd, orderBy }: ParquetReadWorkerOptions
+): Promise<Record<string, any>[]> {
+  return new Promise((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.postMessage({ metadata, asyncBuffer, rowStart, rowEnd, orderBy })
+  })
+}

Original file line number	Diff line number	Diff line change
`@@ -1,3 +1,2 @@`
`1`		`-public/build/app.js -diff`
`2`	`1`	`*.js.map -diff`
`3`	`2`	`*.min.js -diff`