remove unused headers and depend on hyparquet

Author: severo

packages/components/src/components/Cell.tsx

@@ -1,9 +1,8 @@
 import { asyncRows } from 'hightable'
-import { parquetMetadataAsync } from 'hyparquet'
+import { asyncBufferFromUrl, parquetMetadataAsync } from 'hyparquet'
 import { useEffect, useState } from 'react'
 import { FileKey, UrlKey } from '../lib/key.ts'
 import { parquetDataFrame } from '../lib/tableProvider.js'
-import { asyncBufferFromUrl } from '../lib/utils.ts'
 import Breadcrumb from './Breadcrumb.tsx'
 import Layout from './Layout.tsx'
 
@@ -37,15 +36,8 @@ export default function CellView({ parsedKey, row, col }: CellProps) {
       try {
         // TODO: handle first row > 100kb
         setProgress(0.25)
-        const asyncBuffer = await asyncBufferFromUrl({
-          url: resolveUrl,
-          headers: {},
-        })
-        const from = {
-          url: resolveUrl,
-          byteLength: asyncBuffer.byteLength,
-          headers: {},
-        }
+        const asyncBuffer = await asyncBufferFromUrl(resolveUrl)
+        const from = { url: resolveUrl, byteLength: asyncBuffer.byteLength }
         setProgress(0.5)
         const metadata = await parquetMetadataAsync(asyncBuffer)
         setProgress(0.75)

packages/components/src/components/viewers/ParquetView.tsx

@@ -1,9 +1,8 @@
 import HighTable, { DataFrame, rowCache } from 'hightable'
-import { parquetMetadataAsync } from 'hyparquet'
+import { asyncBufferFromUrl, parquetMetadataAsync } from 'hyparquet'
 import React, { useCallback, useEffect, useState } from 'react'
 import { FileKey, UrlKey } from '../../lib/key.ts'
 import { parquetDataFrame } from '../../lib/tableProvider.ts'
-import { asyncBufferFromUrl } from '../../lib/utils.ts'
 import { Spinner } from '../Layout.tsx'
 import ContentHeader, { ContentSize } from './ContentHeader.tsx'
 
@@ -35,8 +34,8 @@ export default function ParquetView({ parsedKey, setProgress, setError }: Viewer
     async function loadParquetDataFrame() {
       try {
         setProgress(0.33)
-        const asyncBuffer = await asyncBufferFromUrl({ url: resolveUrl, headers: {} })
-        const from = { url: resolveUrl, byteLength: asyncBuffer.byteLength, headers: {} }
+        const asyncBuffer = await asyncBufferFromUrl(resolveUrl)
+        const from = { url: resolveUrl, byteLength: asyncBuffer.byteLength }
         setProgress(0.66)
         const metadata = await parquetMetadataAsync(asyncBuffer)
         let dataframe = parquetDataFrame(from, metadata)

packages/components/src/lib/tableProvider.ts

@@ -1,14 +1,14 @@
 import { DataFrame, ResolvablePromise, resolvablePromise } from 'hightable'
 import { FileMetaData, parquetSchema } from 'hyparquet'
 import { parquetQueryWorker, parquetSortIndexWorker } from '../workers/parquetWorkerClient.ts'
-import type { AsyncBufferFrom } from '../workers/types.d.ts'
+import type { AsyncBufferFromUrl } from '../workers/types.d.ts'
 
 type ResolvableRow = Record<string, ResolvablePromise<unknown>>;
 
 /**
  * Convert a parquet file into a dataframe.
  */
-export function parquetDataFrame(from: AsyncBufferFrom, metadata: FileMetaData): DataFrame {
+export function parquetDataFrame(from: AsyncBufferFromUrl, metadata: FileMetaData): DataFrame {
   const { children } = parquetSchema(metadata)
   const header = children.map(child => child.element.name)
   const sortCache = new Map<string, Promise<number[]>>()

packages/components/src/lib/utils.ts

@@ -1,57 +1,7 @@
-import { AsyncBuffer } from 'hyparquet'
 
 /**
  * Helper function to join class names
  */
 export function cn(...names: (string | undefined | false)[]): string {
   return names.filter((n) => n).join(' ')
 }
-
-interface AsyncBufferFromUrlOptions {
-  url: string;
-  byteLength?: number;
-  headers?: Record<string, string>;
-}
-
-/**
- * Get the byte length of a URL using a HEAD request.
- *
- * @param {string} url
- * @returns {Promise<number>}
- */
-export async function byteLengthFromUrl(
-  url: globalThis.RequestInfo | URL,
-  init?: globalThis.RequestInit,
-): Promise<number> {
-  return await fetch(url, { ...init, method: 'HEAD' }).then((res) => {
-    if (!res.ok) throw new Error(`fetch head failed ${res.status.toString()}`)
-    const length = res.headers.get('Content-Length')
-    if (!length) throw new Error('missing content length')
-    return parseInt(length)
-  })
-}
-
-export async function asyncBufferFromUrl({
-  url,
-  byteLength,
-  headers,
-}: AsyncBufferFromUrlOptions): Promise<AsyncBuffer> {
-  // byte length from HEAD request
-  byteLength ??= await byteLengthFromUrl(url, { headers })
-  return {
-    byteLength,
-    async slice(start, end) {
-      // fetch byte range from url
-      const endStr = end === undefined ? '' : end - 1
-      const res = await fetch(url, {
-        headers: {
-          ...headers,
-          range: `bytes=${start.toString()}-${endStr.toString()}`,
-        },
-      })
-      if (!res.ok || !res.body)
-        throw new Error(`fetch failed ${res.status.toString()}`)
-      return res.arrayBuffer()
-    },
-  }
-}

packages/components/src/workers/parquetWorker.ts

@@ -1,8 +1,8 @@
-import { AsyncBuffer, ColumnData, parquetQuery } from 'hyparquet'
+import { AsyncBuffer, ColumnData, asyncBufferFromUrl, cachedAsyncBuffer, parquetQuery } from 'hyparquet'
 import { compressors } from 'hyparquet-compressors'
-import { asyncBufferFromUrl } from '../lib/utils.ts'
+// import { asyncBufferFromUrl } from '../lib/utils.ts'
 import type {
-  AsyncBufferFrom,
+  AsyncBufferFromUrl,
   ChunkMessage,
   ErrorMessage,
   IndicesMessage,
@@ -97,66 +97,12 @@ function compare<T>(a: T, b: T): number {
  * Convert AsyncBufferFrom to AsyncBuffer and cache results.
  */
 function asyncBufferFrom(
-  from: AsyncBufferFrom,
+  from: AsyncBufferFromUrl,
 ): Promise<AsyncBuffer> {
   const key = JSON.stringify(from)
   const cached = cache.get(key)
   if (cached) return cached
-  const asyncBuffer = asyncBufferFromUrl(from).then(cachedAsyncBuffer)
+  const asyncBuffer = asyncBufferFromUrl(from.url, from.byteLength).then(cachedAsyncBuffer)
   cache.set(key, asyncBuffer)
   return asyncBuffer
 }
-
-type Awaitable<T> = T | Promise<T>;
-
-function cachedAsyncBuffer(asyncBuffer: AsyncBuffer): AsyncBuffer {
-  const cache = new Map<string, Awaitable<ArrayBuffer>>()
-  const { byteLength } = asyncBuffer
-  return {
-    byteLength,
-    /**
-     * @param {number} start
-     * @param {number} [end]
-     * @returns {Awaitable<ArrayBuffer>}
-     */
-    slice(start: number, end?: number): Awaitable<ArrayBuffer> {
-      const key = cacheKey(start, end, byteLength)
-      const cached = cache.get(key)
-      if (cached) return cached
-      // cache miss, read from file
-      const promise = asyncBuffer.slice(start, end)
-      cache.set(key, promise)
-      return promise
-    },
-  }
-}
-
-/**
- * Returns canonical cache key for a byte range 'start,end'.
- * Normalize int-range and suffix-range requests to the same key.
- *
- * @param {number} start start byte of range
- * @param {number} [end] end byte of range, or undefined for suffix range
- * @param {number} [size] size of file, or undefined for suffix range
- * @returns {string}
- */
-function cacheKey(start: number, end?: number, size?: number): string {
-  if (start < 0) {
-    if (end !== undefined)
-      throw new Error(
-        `invalid suffix range [${start.toString()}, ${end.toString()}]`,
-      )
-    if (size === undefined) return `${start.toString()},`
-    return `${(size + start).toString()},${size.toString()}`
-  } else if (end !== undefined) {
-    if (start > end)
-      throw new Error(
-        `invalid empty range [${start.toString()}, ${end.toString()}]`,
-      )
-    return `${start.toString()},${end.toString()}`
-  } else if (size === undefined) {
-    return `${start.toString()},`
-  } else {
-    return `${start.toString()},${size.toString()}`
-  }
-}

packages/components/src/workers/types.d.ts

@@ -1,41 +1,40 @@
 import { ColumnData, FileMetaData, ParquetReadOptions } from 'hyparquet'
 
 // Serializable constructor for AsyncBuffers
-export interface AsyncBufferFrom {
-    url: string
-    byteLength: number
-    headers?: Record<string, string>
-  }
-  // Same as ParquetReadOptions, but AsyncBufferFrom instead of AsyncBuffer
+export interface AsyncBufferFromUrl {
+  url: string
+  byteLength: number
+}
+// Same as ParquetReadOptions, but AsyncBufferFrom instead of AsyncBuffer
 export interface ParquetReadWorkerOptions extends Omit<ParquetReadOptions, 'file'> {
-    from: AsyncBufferFrom
-    orderBy?: string
-    sortIndex?: boolean
-  }
-  // Row is defined in hightable, but not exported + we change any to unknown
-export type Row = Record<string, unknown>;
+  from: AsyncBufferFromUrl
+  orderBy?: string
+  sortIndex?: boolean
+}
+// Row is defined in hightable, but not exported + we change any to unknown
+export type Row = Record<string, unknown> ;
 
-  interface Message {
-    queryId: number
-  }
+interface Message {
+  queryId: number
+}
 export interface ChunkMessage extends Message {
-    chunk: ColumnData
-  }
+  chunk: ColumnData
+}
 export interface ResultMessage extends Message {
-    result: Row[]
-  }
+  result: Row[]
+}
 export interface IndicesMessage extends Message {
-    indices: number[]
-  }
+  indices: number[]
+}
 export interface ErrorMessage extends Message {
-    error: Error
-  }
+  error: Error
+}
 
 export type ParquetMessage = ChunkMessage | ResultMessage | ErrorMessage
 export type SortParquetMessage = IndicesMessage | ErrorMessage
 
 export interface ParquetSortIndexOptions {
-    metadata: FileMetaData
-    from: AsyncBufferFrom
-    orderBy: string
-  }
+  metadata: FileMetaData
+  from: AsyncBufferFromUrl
+  orderBy: string
+}