[Breaking] remove sortability from parquetDataFrame (#310)

* update dependencies

* update code to hightable 0.19.0

* update code related to hightable 0.19.1

* fix types in worker

* parquetDataFrame returns an unsortable dataframe

* upgrade deps

---------

Co-authored-by: Kenny Daniel <[email protected]>

Author: severo

package.json

@@ -55,7 +55,7 @@
     "watch:url": "NODE_ENV=development nodemon bin/cli.js https://hyperparam.blob.core.windows.net/hyperparam/starcoderdata-js-00000-of-00065.parquet"
   },
   "dependencies": {
-    "hightable": "0.18.5",
+    "hightable": "0.19.4",
     "hyparquet": "1.18.0",
     "hyparquet-compressors": "1.1.1",
     "icebird": "0.3.0",

src/components/Cell/Cell.tsx

@@ -42,11 +42,11 @@ export default function CellView({ source, row, col }: CellProps) {
         setProgress(0.75)
         const df = parquetDataFrame(from, metadata)
 
-        const columnName = df.header[col]
+        const columnName = df.columnDescriptors[col]?.name
         if (columnName === undefined) {
           throw new Error(`Column name missing at index col=${col}`)
         }
-        await df.fetch({ rowStart: row, rowEnd: row + 1, columns: [columnName] })
+        await df.fetch?.({ rowStart: row, rowEnd: row + 1, columns: [columnName] })
         const cell = df.getCell({ row, column: columnName })
         const text = cell === undefined ? UNLOADED_CELL_PLACEHOLDER : stringify(cell.value)
         setText(text)

src/components/CellPanel/CellPanel.tsx

@@ -58,7 +58,7 @@ export default function CellPanel({ df, row, col, setProgress, setError, onClose
       try {
         setProgress(0.5)
 
-        const columnName = df.header[col]
+        const columnName = df.columnDescriptors[col]?.name
         if (columnName === undefined) {
           throw new Error(`Column name missing at index col=${col}`)
         }
@@ -67,7 +67,7 @@ export default function CellPanel({ df, row, col, setProgress, setError, onClose
           fillContent(undefined)
           return
         }
-        await df.fetch({ rowStart: row, rowEnd: row + 1, columns: [columnName] })
+        await df.fetch?.({ rowStart: row, rowEnd: row + 1, columns: [columnName] })
         cell = df.getCell({ row, column: columnName })
         if (cell === undefined) {
           throw new Error(`Cell at row=${row}, column=${columnName} is undefined`)
@@ -85,7 +85,7 @@ export default function CellPanel({ df, row, col, setProgress, setError, onClose
 
   const headers = <>
     <SlideCloseButton onClick={onClose} />
-    <span>column: {df.header[col]}</span>
+    <span>column: {df.columnDescriptors[col]?.name}</span>
     <span>row: {row + 1}</span>
   </>
 

src/components/ParquetView/ParquetView.tsx

@@ -82,7 +82,7 @@ export default function ParquetView({ source, setProgress, setError }: ViewerPro
       if (cell?.col === col && cell.row === row) {
         return undefined
       }
-      const columnName = content?.dataframe.header[col]
+      const columnName = content?.dataframe.columnDescriptors[col]?.name
       if (columnName === undefined || !content?.dataframe.getCell({ row, column: columnName })) {
         // don't open the cell panel until it has loaded
         return undefined

src/lib/tableProvider.ts

@@ -1,4 +1,4 @@
-import { DataFrame, DataFrameEvents, ResolvedValue, UnsortableDataFrame, createEventTarget, sortableDataFrame } from 'hightable'
+import { DataFrame, DataFrameEvents, ResolvedValue, checkSignal, createEventTarget, validateFetchParams, validateGetCellParams, validateGetRowNumberParams } from 'hightable'
 import type { ColumnData } from 'hyparquet'
 import { FileMetaData, ParquetReadOptions, parquetSchema } from 'hyparquet'
 import { parquetReadWorker } from './workers/parquetWorkerClient.js'
@@ -20,13 +20,16 @@ interface VirtualRowGroup {
 
 /**
  * Convert a parquet file into a dataframe.
+ *
+ * It fetches data on demand in chunks of 1000 rows within each row group.
+ * It's not sortable. You can use sortableDataFrame from hightable to make it sortable.
  */
-export function parquetDataFrame(from: AsyncBufferFrom, metadata: FileMetaData, options?: Pick<ParquetReadOptions, 'utf8'>): DataFrame {
+export function parquetDataFrame(from: AsyncBufferFrom, metadata: FileMetaData, options?: Pick<ParquetReadOptions, 'utf8'>): DataFrame<{parquet: FileMetaData}> {
   const { children } = parquetSchema(metadata)
-  const header = children.map(child => child.element.name)
+  const columnDescriptors = children.map(child => ({ name: child.element.name }))
   const eventTarget = createEventTarget<DataFrameEvents>()
 
-  const cellCache = new Map<string, ResolvedValue<unknown>[]>(header.map(name => [name, []]))
+  const cellCache = new Map<string, ResolvedValue<unknown>[]>(columnDescriptors.map(({ name }) => [name, []]))
 
   // virtual row groups are up to 1000 rows within row group boundaries
   const groups: VirtualRowGroup[] = []
@@ -39,7 +42,7 @@ export function parquetDataFrame(from: AsyncBufferFrom, metadata: FileMetaData,
       groups.push({
         groupStart,
         groupEnd,
-        state: new Map(header.map(name => [name, { kind: 'unfetched' }])),
+        state: new Map(columnDescriptors.map(({ name }) => [name, { kind: 'unfetched' }])),
       })
       groupStart = groupEnd
     }
@@ -84,22 +87,21 @@ export function parquetDataFrame(from: AsyncBufferFrom, metadata: FileMetaData,
 
   const numRows = Number(metadata.num_rows)
 
-  const unsortableDataFrame: UnsortableDataFrame = {
-    header,
+  const unsortableDataFrame: DataFrame<{parquet: FileMetaData}> = {
+    columnDescriptors,
     numRows,
-    metadata,
+    metadata: { parquet: metadata },
     eventTarget,
-    getRowNumber({ row }) {
-      validateRow({ row, data: { numRows } })
+    getRowNumber({ row, orderBy }) {
+      validateGetRowNumberParams({ row, orderBy, data: { numRows, columnDescriptors } })
       return { value: row }
     },
-    getCell({ row, column }) {
-      validateRow({ row, data: { numRows } })
-      validateColumn({ column, data: { header } })
+    getCell({ row, column, orderBy }) {
+      validateGetCellParams({ row, column, orderBy, data: { numRows, columnDescriptors } })
       return cellCache.get(column)?.[row]
     },
     fetch: async ({ rowStart, rowEnd, columns, signal }) => {
-      validateFetchParams({ rowStart, rowEnd, columns, data: { numRows, header } })
+      validateFetchParams({ rowStart, rowEnd, columns, data: { numRows, columnDescriptors } })
       checkSignal(signal)
 
       if (!columns || columns.length === 0) {
@@ -126,29 +128,5 @@ export function parquetDataFrame(from: AsyncBufferFrom, metadata: FileMetaData,
     },
   }
 
-  return sortableDataFrame(unsortableDataFrame)
-}
-
-function validateFetchParams({ rowStart, rowEnd, columns, data: { numRows, header } }: {rowStart: number, rowEnd: number, columns?: string[], data: Pick<DataFrame, 'numRows' | 'header'>}): void {
-  if (rowStart < 0 || rowEnd > numRows || !Number.isInteger(rowStart) || !Number.isInteger(rowEnd) || rowStart > rowEnd) {
-    throw new Error(`Invalid row range: ${rowStart} - ${rowEnd}, numRows: ${numRows}`)
-  }
-  if (columns?.some(column => !header.includes(column))) {
-    throw new Error(`Invalid columns: ${columns.join(', ')}. Available columns: ${header.join(', ')}`)
-  }
-}
-function validateRow({ row, data: { numRows } }: {row: number, data: Pick<DataFrame, 'numRows'>}): void {
-  if (row < 0 || row >= numRows || !Number.isInteger(row)) {
-    throw new Error(`Invalid row index: ${row}, numRows: ${numRows}`)
-  }
-}
-function validateColumn({ column, data: { header } }: {column: string, data: Pick<DataFrame, 'header'>}): void {
-  if (!header.includes(column)) {
-    throw new Error(`Invalid column: ${column}. Available columns: ${header.join(', ')}`)
-  }
-}
-function checkSignal(signal?: AbortSignal): void {
-  if (signal?.aborted) {
-    throw new DOMException('The operation was aborted.', 'AbortError')
-  }
+  return unsortableDataFrame
 }

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, onChunk, onPage })
       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[][] | Record<string, 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