copy files from hightable and remove dependency

(it was including React in the bundle)

Author: severo

packages/utils/package.json

@@ -36,7 +36,6 @@
     "test": "vitest run"
   },
   "dependencies": {
-    "hightable": "0.6.3",
     "hyparquet": "1.5.0",
     "hyparquet-compressors": "0.1.4"
   },

packages/utils/src/dataframe.ts

@@ -0,0 +1,162 @@
+type WrappedPromise<T> = Promise<T> & {
+  resolved?: T
+  rejected?: Error
+}
+
+/**
+ * A row where each cell is a promise.
+ * The promise must be wrapped with `wrapPromise` so that HighTable can render
+ * the state synchronously.
+ */
+export type AsyncRow = Record<string, WrappedPromise<any>>
+
+/**
+ * A row where each cell is a resolved value.
+ */
+export type Row = Record<string, any>
+
+/**
+ * Streamable row data
+ */
+export interface DataFrame {
+  header: string[]
+  numRows: number
+  // Rows are 0-indexed, excludes the header, end is exclusive
+  rows(start: number, end: number, orderBy?: string): AsyncRow[] | Promise<Row[]>
+  sortable?: boolean
+}
+
+export function resolvableRow(header: string[]): { [key: string]: ResolvablePromise<any> } {
+  return Object.fromEntries(header.map(key => [key, resolvablePromise<any>()]))
+}
+
+/**
+ * Helper method to wrap future rows into AsyncRows.
+ * Helpful when you want to define a DataFrame with simple async fetching of rows.
+ * This function turns future data into a "grid" of wrapped promises.
+ */
+export function asyncRows(rows: AsyncRow[] | Promise<Row[]>, numRows: number, header: string[]): AsyncRow[] {
+  if (Array.isArray(rows)) return rows
+  // Make grid of resolvable promises
+  const wrapped = new Array(numRows).fill(null).map(_ => resolvableRow(header))
+  rows.then(rows => {
+    if (rows.length !== numRows) {
+      console.warn(`Expected ${numRows} rows, got ${rows.length}`)
+    }
+    for (let i = 0; i < rows.length; i++) {
+      const row = rows[i]
+      for (const key of header) {
+        wrapped[i][key].resolve(row[key])
+      }
+    }
+  }).catch(error => {
+    // Reject all promises on error
+    for (let i = 0; i < numRows; i++) {
+      for (const key of header) {
+        wrapped[i][key].reject(error)
+      }
+    }
+  })
+  return wrapped
+}
+
+/**
+ * Wrap a promise to save the resolved value and error.
+ * Note: you can't await on a WrappedPromise, you must use then.
+ */
+export function wrapPromise<T>(promise: Promise<T> | T): WrappedPromise<T> {
+  if (!(promise instanceof Promise)) {
+    promise = Promise.resolve(promise)
+  }
+  const wrapped: WrappedPromise<T> = promise.then(resolved => {
+    wrapped.resolved = resolved
+    return resolved
+  }).catch(rejected => {
+    wrapped.rejected = rejected
+    throw rejected
+  })
+  return wrapped
+}
+
+export type ResolvablePromise<T> = Promise<T> & {
+  resolve: (value: T) => void
+  reject: (error: Error) => void
+}
+
+/**
+ * Create a promise that can be resolved or rejected later.
+ */
+export function resolvablePromise<T>(): ResolvablePromise<T> {
+  let resolve: (value: T) => void
+  let reject: (error: Error) => void
+  const promise = wrapPromise(new Promise<T>((res, rej) => {
+    resolve = res
+    reject = rej
+  })) as ResolvablePromise<T>
+  promise.resolve = resolve!
+  promise.reject = reject!
+  return promise
+}
+
+/**
+ * Wraps a DataFrame to make it sortable.
+ * Requires fetching all rows to sort.
+ */
+export function sortableDataFrame(data: DataFrame): DataFrame {
+  if (data.sortable) return data // already sortable
+  // Fetch all rows and add __index__ column
+  let all: Promise<Row[]>
+  return {
+    ...data,
+    rows(start: number, end: number, orderBy?: string): AsyncRow[] | Promise<Row[]> {
+      if (orderBy) {
+        if (!data.header.includes(orderBy)) {
+          throw new Error(`Invalid orderBy field: ${orderBy}`)
+        }
+        if (!all) {
+          // Fetch all rows and add __index__ column
+          all = awaitRows(data.rows(0, data.numRows))
+            .then(rows => rows.map((row, i) => ({ __index__: i, ...row })))
+        }
+        const sorted = all.then(all => {
+          return all.sort((a, b) => {
+            if (a[orderBy] < b[orderBy]) return -1
+            if (a[orderBy] > b[orderBy]) return 1
+            return 0
+          }).slice(start, end)
+        })
+        return sorted
+      } else {
+        return data.rows(start, end)
+      }
+    },
+    sortable: true,
+  }
+}
+
+/**
+ * Await all promises in an AsyncRow and return resolved row.
+ */
+export function awaitRow(row: AsyncRow): Promise<Row> {
+  return Promise.all(Object.values(row))
+    .then(values => Object.fromEntries(Object.keys(row).map((key, i) => [key, values[i]])))
+}
+
+/**
+ * Await all promises in list of AsyncRows and return resolved rows.
+ */
+export function awaitRows(rows: AsyncRow[] | Promise<Row[]>): Promise<Row[]> {
+  if (rows instanceof Promise) return rows
+  return Promise.all(rows.map(awaitRow))
+}
+
+export function arrayDataFrame(data: Row[]): DataFrame {
+  if (!data.length) return { header: [], numRows: 0, rows: () => Promise.resolve([]) }
+  return {
+    header: Object.keys(data[0]),
+    numRows: data.length,
+    rows(start: number, end: number): Promise<Row[]> {
+      return Promise.resolve(data.slice(start, end))
+    },
+  }
+}

packages/utils/src/rowCache.ts

@@ -0,0 +1,50 @@
+import { AsyncRow, DataFrame, asyncRows } from './dataframe.js'
+
+/**
+ * Wrap a dataframe with cached rows.
+ *
+ * @param df async dataframe to wrap
+ */
+export function rowCache(df: DataFrame): DataFrame {
+  // Row cache is stored as a sorted array of RowGroups, per sort order
+  const caches: {[key: string]: AsyncRow[]} = {}
+
+  let hits = 0
+  let misses = 0
+
+  return {
+    ...df,
+    rows(start: number, end: number, orderBy?: string): AsyncRow[] {
+      // Cache per sort order
+      const cache = caches[orderBy || ''] ||= new Array(df.numRows)
+      const n = hits + misses
+      if (n && !(n % 10)) {
+        console.log(`Cache hits: ${hits} / ${hits + misses} (${(100 * hits / (hits + misses)).toFixed(1)}%)`)
+      }
+
+      // Fetch missing rows in contiguous blocks
+      let blockStart: number | undefined
+      let hasCacheMiss = false
+      for (let i = start; i <= end; i++) {
+        if (i < end && !cache[i]) {
+          if (blockStart === undefined) {
+            blockStart = i
+          }
+        } else if (blockStart !== undefined) {
+          const blockEnd = i
+          const numRows = blockEnd - blockStart
+          const futureRows = asyncRows(df.rows(blockStart, blockEnd, orderBy), numRows, df.header)
+          for (let j = 0; j < blockEnd - blockStart; j++) {
+            cache[blockStart + j] = futureRows[j]
+          }
+          blockStart = undefined
+          hasCacheMiss = true
+        }
+      }
+      if (hasCacheMiss) misses++
+      else hits++
+
+      return cache.slice(start, end)
+    },
+  }
+}

packages/utils/src/tableProvider.ts

@@ -1,5 +1,5 @@
-import { DataFrame, ResolvablePromise, resolvablePromise } from 'hightable'
 import { FileMetaData, parquetSchema } from 'hyparquet'
+import { DataFrame, ResolvablePromise, resolvablePromise } from './dataframe.js'
 import { parquetQueryWorker, parquetSortIndexWorker } from './workers/parquetWorkerClient.js'
 import type { AsyncBufferFrom } from './workers/types.d.ts'
 

packages/utils/test/dataframe.test.ts

@@ -0,0 +1,140 @@
+import { describe, expect, it } from 'vitest'
+import {
+  DataFrame, Row, arrayDataFrame, awaitRows, resolvablePromise, sortableDataFrame, wrapPromise,
+} from '../src/dataframe.js'
+
+function wrapObject(obj: Record<string, any>): Row {
+  return Object.fromEntries(
+    Object.entries(obj).map(([key, value]) => [key, wrapPromise(value)])
+  )
+}
+
+describe('resolvablePromise', () => {
+  it('should resolve with a value', async () => {
+    const promise = resolvablePromise<number>()
+    promise.resolve(42)
+    expect(await promise).toBe(42)
+  })
+  it('should reject with an error', async () => {
+    const promise = resolvablePromise<number>()
+    promise.reject(new Error('Failed'))
+    await expect(promise).rejects.toThrow('Failed')
+  })
+})
+
+describe('awaitRows', () => {
+  it('should resolve with a row', async () => {
+    const row = wrapObject({ id: 1, name: 'Alice', age: 30, __index__: 0 })
+    const result = await awaitRows([row])
+    expect(result).toEqual([{ id: 1, name: 'Alice', age: 30, __index__: 0 }])
+  })
+})
+
+describe('sortableDataFrame', () => {
+  const data = [
+    { id: 3, name: 'Charlie', age: 25 },
+    { id: 1, name: 'Alice', age: 30 },
+    { id: 2, name: 'Bob', age: 20 },
+    { id: 4, name: 'Dani', age: 20 },
+  ]
+
+  const dataFrame: DataFrame = {
+    header: ['id', 'name', 'age'],
+    numRows: data.length,
+    rows(start: number, end: number): Row[] {
+      // Return the slice of data between start and end indices
+      return data.slice(start, end).map(wrapObject)
+    },
+    sortable: false,
+  }
+
+  const sortableDf = sortableDataFrame(dataFrame)
+
+  it('should set sortable to true', () => {
+    expect(sortableDf.sortable).toBe(true)
+  })
+
+  it('should preserve header and numRows', () => {
+    expect(sortableDf.header).toEqual(dataFrame.header)
+    expect(sortableDf.numRows).toBe(dataFrame.numRows)
+  })
+
+  it('should return unsorted data when orderBy is not provided', async () => {
+    const rows = await awaitRows(sortableDf.rows(0, 3))
+    expect(rows).toEqual([
+      { id: 3, name: 'Charlie', age: 25 },
+      { id: 1, name: 'Alice', age: 30 },
+      { id: 2, name: 'Bob', age: 20 },
+    ])
+  })
+
+  it('should return data sorted by column "age"', async () => {
+    const rows = await awaitRows(sortableDf.rows(0, 4, 'age'))
+    expect(rows).toEqual([
+      { id: 2, name: 'Bob', age: 20, __index__: 2 },
+      { id: 4, name: 'Dani', age: 20, __index__: 3 },
+      { id: 3, name: 'Charlie', age: 25, __index__: 0 },
+      { id: 1, name: 'Alice', age: 30, __index__: 1 },
+    ])
+  })
+
+  it('should slice the sorted data correctly', async () => {
+    const rows = await awaitRows(sortableDf.rows(1, 3, 'id'))
+    expect(rows).toEqual([
+      { id: 2, name: 'Bob', age: 20, __index__: 2 },
+      { id: 3, name: 'Charlie', age: 25, __index__: 0 },
+    ])
+  })
+
+  it('returns self if already sortable', () => {
+    const sortableDf2 = sortableDataFrame(sortableDf)
+    expect(sortableDf2).toBe(sortableDf)
+  })
+
+  it('should throw for invalid orderBy field', () => {
+    expect(() => sortableDf.rows(0, 3, 'invalid'))
+      .toThrowError('Invalid orderBy field: invalid')
+  })
+})
+
+describe('arrayDataFrame', () => {
+  const testData = [
+    { id: 1, name: 'Alice', age: 30 },
+    { id: 2, name: 'Bob', age: 25 },
+    { id: 3, name: 'Charlie', age: 35 },
+  ]
+
+  it('should create a DataFrame with correct header and numRows', () => {
+    const df = arrayDataFrame(testData)
+    expect(df.header).toEqual(['id', 'name', 'age'])
+    expect(df.numRows).toBe(3)
+  })
+
+  it('should handle empty data array', () => {
+    const df = arrayDataFrame([])
+    expect(df.header).toEqual([])
+    expect(df.numRows).toBe(0)
+    expect(df.rows(0, 1)).resolves.toEqual([])
+  })
+
+  it('should return correct rows for given range', async () => {
+    const df = arrayDataFrame(testData)
+    const rows = await df.rows(0, 2)
+    expect(rows).toEqual([
+      { id: 1, name: 'Alice', age: 30 },
+      { id: 2, name: 'Bob', age: 25 },
+    ])
+  })
+
+  it('should handle start index equal to end index', async () => {
+    const df = arrayDataFrame(testData)
+    const rows = await df.rows(1, 1)
+    expect(rows).toEqual([])
+  })
+
+  it('should return all rows when end index exceeds array length', async () => {
+    const df = arrayDataFrame(testData)
+    const rows = await df.rows(0, 10)
+    expect(rows).toEqual(testData)
+  })
+})