fix types in worker

Author: severo

src/lib/workers/parquetWorker.ts

@@ -1,7 +1,7 @@
 import type { ColumnData } from 'hyparquet'
 import { AsyncBuffer, parquetQuery, parquetRead, parquetReadObjects } from 'hyparquet'
 import { compressors } from 'hyparquet-compressors'
-import type { ChunkMessage, ClientMessage, CompleteMessage, PageMessage, ParquetQueryResolveMessage, ParquetReadObjectsResolveMessage, ParquetReadResolveMessage, RejectMessage } from './types.js'
+import type { ChunkMessage, ClientMessage, CompleteMessage, PageMessage, ParquetQueryResolveMessage, ParquetReadObjectsResolveMessage, ParquetReadResolveMessage, RejectMessage, Rows } from './types.js'
 import { fromToAsyncBuffer } from './utils.js'
 
 const cache = new Map<string, Promise<AsyncBuffer>>()
@@ -33,20 +33,20 @@ self.onmessage = async ({ data }: { data: ClientMessage }) => {
   const file = await fromToAsyncBuffer(from, cache)
   try {
     if (kind === 'parquetReadObjects') {
-      const rows = await parquetReadObjects({ ...options, file, compressors, onChunk, onPage })
+      const rows = (await parquetReadObjects({ ...options, rowFormat: 'object', file, compressors, onChunk, onPage })) as Rows
       postParquetReadObjectsResultMessage({ queryId, rows })
     } else if (kind === 'parquetQuery') {
-      const rows = await parquetQuery({ ...options, file, compressors, onComplete, onChunk, onPage })
+      const rows = (await parquetQuery({ ...options, rowFormat: 'object', file, compressors, onComplete, onChunk, onPage })) as Rows
       postParquetQueryResultMessage({ queryId, rows })
     } else {
-      await parquetRead({ ...options, file, compressors, onComplete, onChunk, onPage })
+      await parquetRead({ ...options, rowFormat: 'object', file, compressors, onComplete, onChunk, onPage })
       postParquetReadResultMessage({ queryId })
     }
   } catch (error) {
     postErrorMessage({ error: error as Error, queryId })
   }
 
-  function onComplete(rows: unknown[][]) {
+  function onComplete(rows: Rows) {
     postCompleteMessage({ queryId, rows })
   }
   function onChunk(chunk: ColumnData) {

src/lib/workers/parquetWorkerClient.ts

@@ -7,7 +7,7 @@ import type { ClientMessage, ParquetQueryWorkerOptions, ParquetReadObjectsWorker
 let worker: Worker | undefined
 let nextQueryId = 0
 interface Agent {
-  onComplete?: (rows: Rows) => void
+  onComplete?: ((rows: Rows) => void)
   onChunk?: (chunk: ColumnData) => void
   onPage?: (page: ColumnData) => void
   reject: (error: Error) => void
@@ -73,6 +73,8 @@ function getWorker() {
  * Instead of taking an AsyncBuffer, it takes a AsyncBufferFrom, because it needs
  * to be serialized to the worker. Also: the worker uses hyparquet-compressors and
  * the default parsers.
+ *
+ * Note that it only supports 'rowFormat: object' (the default).
  */
 export function parquetReadWorker(options: ParquetReadWorkerOptions): Promise<void> {
   const { onComplete, onChunk, onPage, from, ...serializableOptions } = options
@@ -91,6 +93,8 @@ export function parquetReadWorker(options: ParquetReadWorkerOptions): Promise<vo
  * Instead of taking an AsyncBuffer, it takes a AsyncBufferFrom, because it needs
  * to be serialized to the worker. Also: the worker uses hyparquet-compressors and
  * the default parsers.
+ *
+ * Note that it only supports 'rowFormat: object' (the default).
  */
 export function parquetReadObjectsWorker(options: ParquetReadObjectsWorkerOptions): Promise<Rows> {
   const { onChunk, onPage, from, ...serializableOptions } = options
@@ -109,6 +113,8 @@ export function parquetReadObjectsWorker(options: ParquetReadObjectsWorkerOption
  * Instead of taking an AsyncBuffer, it takes a AsyncBufferFrom, because it needs
  * to be serialized to the worker. Also: the worker uses hyparquet-compressors and
  * the default parsers.
+ *
+ * Note that it only supports 'rowFormat: object' (the default).
  */
 export function parquetQueryWorker(options: ParquetQueryWorkerOptions): Promise<Rows> {
   const { onComplete, onChunk, onPage, from, ...serializableOptions } = options

src/lib/workers/types.ts

@@ -16,7 +16,8 @@ interface AsyncBufferFromUrl {
 }
 export type AsyncBufferFrom = AsyncBufferFromFile | AsyncBufferFromUrl
 
-export type Rows = unknown[][] | Record<string, unknown>[]
+// Only rowFormat 'object' is supported in the worker
+export type Rows = Record<string, unknown>[]
 
 /**
  * Options for the worker version of parquetRead
@@ -25,9 +26,11 @@ export type Rows = unknown[][] | Record<string, unknown>[]
  * - 'compressors' are not configurable, the worker uses hyparquet-compressors
  * - 'parsers' are not configurable, the worker uses the default parsers
  */
-export interface ParquetReadWorkerOptions extends Omit<ParquetReadOptions, 'compressors' | 'parsers' | 'file' | 'onComplete'> {
-  onComplete?: (rows: Rows) => void // fix for https://github.com/hyparam/hyparquet/issues/28
+export interface ParquetReadWorkerOptions extends Omit<ParquetReadOptions, 'compressors' | 'parsers' | 'file' | 'rowFormat' | 'onComplete'> {
   from: AsyncBufferFrom
+  // rowFormat 'array' is not supported in the worker.
+  rowFormat?: 'object'
+  onComplete?: (rows: Rows) => void
 }
 /**
  * Options for the worker version of parquetReadObjects