parquetMetadata move file into options object

Author: platypii

CHANGELOG.md

@@ -6,6 +6,7 @@
  - Rename `onPage.columnName: string` to `onPage.pathInSchema: string[]`
  - Remove `rowFormat` and always return rows as objects
  - Change `byteLengthFromUrl` to accept options object
+ - Move `file` parameter to options object in `parquetMetadata`
 
 ## [1.23.3]
  - Fix `readColumn` truncation in struct columns (#148)

README.md

@@ -74,7 +74,7 @@ You can read just the metadata, including schema and data statistics using the `
 import { parquetMetadata, parquetSchema } from 'hyparquet'
 
 const file = await asyncBufferFromUrl({ url })
-const metadata = await parquetMetadata(file)
+const metadata = await parquetMetadata({ file })
 // Get total number of rows (convert bigint to number)
 const numRows = Number(metadata.num_rows)
 // Get nested table schema

src/metadata.js

@@ -31,16 +31,15 @@ function decode(/** @type {Uint8Array} */ value) {
  * This ensures that we either make one 512kb initial request for the metadata,
  * or a second request for up to the metadata size.
  *
- * @param {AsyncBuffer} asyncBuffer parquet file contents
- * @param {MetadataOptions & { initialFetchSize?: number }} options initial fetch size in bytes (default 512kb)
+ * @param {MetadataOptions & { file: AsyncBuffer, initialFetchSize?: number }} options parquet file and metadata options
  * @returns {Promise<FileMetaData>} parquet metadata object
  */
-export async function parquetMetadata(asyncBuffer, { parsers, initialFetchSize = defaultInitialFetchSize, geoparquet = true } = {}) {
-  if (!asyncBuffer || !(asyncBuffer.byteLength >= 0)) throw new Error('parquet expected AsyncBuffer')
+export async function parquetMetadata({ file, parsers, initialFetchSize = defaultInitialFetchSize, geoparquet = true }) {
+  if (!file || !(file.byteLength >= 0)) throw new Error('parquet expected AsyncBuffer')
 
   // fetch last bytes (footer) of the file
-  const footerOffset = Math.max(0, asyncBuffer.byteLength - initialFetchSize)
-  const footerBuffer = await asyncBuffer.slice(footerOffset, asyncBuffer.byteLength)
+  const footerOffset = Math.max(0, file.byteLength - initialFetchSize)
+  const footerBuffer = await file.slice(footerOffset, file.byteLength)
 
   // Check for parquet magic number "PAR1"
   const footerView = new DataView(footerBuffer)
@@ -51,38 +50,36 @@ export async function parquetMetadata(asyncBuffer, { parsers, initialFetchSize =
   // Parquet files store metadata at the end of the file
   // Metadata length is 4 bytes before the last PAR1
   const metadataLength = footerView.getUint32(footerBuffer.byteLength - 8, true)
-  if (metadataLength > asyncBuffer.byteLength - 8) {
-    throw new Error(`parquet metadata length ${metadataLength} exceeds available buffer ${asyncBuffer.byteLength - 8}`)
+  if (metadataLength > file.byteLength - 8) {
+    throw new Error(`parquet metadata length ${metadataLength} exceeds available buffer ${file.byteLength - 8}`)
   }
 
   // check if metadata size fits inside the initial fetch
   if (metadataLength + 8 > initialFetchSize) {
     // fetch the rest of the metadata
-    const metadataOffset = asyncBuffer.byteLength - metadataLength - 8
-    const metadataBuffer = await asyncBuffer.slice(metadataOffset, footerOffset)
+    const metadataOffset = file.byteLength - metadataLength - 8
+    const metadataBuffer = await file.slice(metadataOffset, footerOffset)
     // combine initial fetch with the new slice
     const combinedBuffer = new ArrayBuffer(metadataLength + 8)
     const combinedView = new Uint8Array(combinedBuffer)
     combinedView.set(new Uint8Array(metadataBuffer))
     combinedView.set(new Uint8Array(footerBuffer), footerOffset - metadataOffset)
-    return parquetMetadataSync(combinedBuffer, { parsers, geoparquet })
+    return parquetMetadataSync({ file: combinedBuffer, parsers, geoparquet })
   } else {
     // parse metadata from the footer
-    return parquetMetadataSync(footerBuffer, { parsers, geoparquet })
+    return parquetMetadataSync({ file: footerBuffer, parsers, geoparquet })
   }
 }
 
 /**
  * Read parquet metadata from a buffer synchronously.
  *
- * @import {KeyValue} from '../src/types.d.ts'
- * @param {ArrayBuffer} arrayBuffer parquet file footer
- * @param {MetadataOptions} options metadata parsing options
+ * @param {MetadataOptions & { file: ArrayBuffer }} options parquet file and metadata options
  * @returns {FileMetaData} parquet metadata object
  */
-export function parquetMetadataSync(arrayBuffer, { parsers, geoparquet = true } = {}) {
-  if (!(arrayBuffer instanceof ArrayBuffer)) throw new Error('parquet expected ArrayBuffer')
-  const view = new DataView(arrayBuffer)
+export function parquetMetadataSync({ file, parsers, geoparquet = true }) {
+  if (!(file instanceof ArrayBuffer)) throw new Error('parquet expected ArrayBuffer')
+  const view = new DataView(file)
 
   // Use default parsers if not given
   parsers = { ...DEFAULT_PARSERS, ...parsers }
@@ -287,7 +284,7 @@ function timeUnit(unit) {
 /**
  * Convert column statistics based on column type.
  *
- * @import {AsyncBuffer, FileMetaData, LogicalType, MetadataOptions, MinMaxType, ParquetParsers, SchemaElement, SchemaTree, Statistics, TimeUnit} from '../src/types.d.ts'
+ * @import {AsyncBuffer, FileMetaData, KeyValue, LogicalType, MetadataOptions, MinMaxType, ParquetParsers, SchemaElement, SchemaTree, Statistics, TimeUnit} from '../src/types.d.ts'
  * @param {any} stats
  * @param {SchemaElement} schema
  * @param {ParquetParsers} parsers

src/query.js

@@ -17,7 +17,7 @@ export async function parquetQuery(options) {
   if (!options.file || !(options.file.byteLength >= 0)) {
     throw new Error('parquet expected AsyncBuffer')
   }
-  options.metadata ??= await parquetMetadata(options.file, options)
+  options.metadata ??= await parquetMetadata(options)
 
   const { metadata, rowStart = 0, columns, orderBy, filter } = options
   if (rowStart < 0) throw new Error('parquet rowStart must be positive')
@@ -102,14 +102,13 @@ export async function parquetQuery(options) {
  * @returns {Promise<(Record<string, any> & {__index__: number})[]>}
  */
 async function parquetReadRows(options) {
-  const { file, rows } = options
-  options.metadata ??= await parquetMetadata(file, options)
+  options.metadata ??= await parquetMetadata(options)
   const { row_groups: rowGroups } = options.metadata
   // Compute row groups to fetch
   const groupIncluded = Array(rowGroups.length).fill(false)
   let groupStart = 0
   const groupEnds = rowGroups.map(group => groupStart += Number(group.num_rows))
-  for (const index of rows) {
+  for (const index of options.rows) {
     const groupIndex = groupEnds.findIndex(end => index < end)
     groupIncluded[groupIndex] = true
   }

src/read.js

@@ -21,7 +21,7 @@ import { concat, flatten } from './utils.js'
  */
 export async function parquetRead(options) {
   // load metadata if not provided
-  options.metadata ??= await parquetMetadata(options.file, options)
+  options.metadata ??= await parquetMetadata(options)
 
   const { rowStart = 0, rowEnd, columns, onChunk, onComplete, filter, filterStrict = true } = options
 
@@ -143,7 +143,7 @@ export async function parquetReadColumn(options) {
   if (options.columns?.length !== 1) {
     throw new Error('parquetReadColumn expected columns: [columnName]')
   }
-  options.metadata ??= await parquetMetadata(options.file, options)
+  options.metadata ??= await parquetMetadata(options)
   const asyncGroups = parquetReadAsync(options)
 
   // assemble struct columns

test/column.test.js

@@ -16,7 +16,7 @@ describe('readColumn', () => {
     const testFile = 'test/files/float16_nonzeros_and_nans.parquet'
     const file = await asyncBufferFromFile(testFile)
     const arrayBuffer = await file.slice(0)
-    const metadata = await parquetMetadata(arrayBuffer)
+    const metadata = await parquetMetadata({ file })
 
     const column = metadata.row_groups[0].columns[0]
     if (!column.meta_data) throw new Error(`No column metadata for ${testFile}`)
@@ -47,7 +47,7 @@ describe('readColumn', () => {
     const testFile = 'test/files/datapage_v2.snappy.parquet'
     const file = await asyncBufferFromFile(testFile)
     const arrayBuffer = await file.slice(0)
-    const metadata = await parquetMetadata(arrayBuffer)
+    const metadata = await parquetMetadata({ file })
 
     const column = metadata.row_groups[0].columns[1] // second column
     if (!column.meta_data) throw new Error(`No column metadata for ${testFile}`)

test/indexes.test.js

@@ -10,16 +10,16 @@ describe('readColumnIndex', () => {
   const columnIndexesFiles = fs.readdirSync('test/files').filter(f => f.endsWith('.column_indexes.json'))
   const parquetFiles = columnIndexesFiles.map(f => f.replace(/.column_indexes.json$/i, '.parquet'))
 
-  parquetFiles.forEach((file, i) => {
-    it(`parse column indexes from ${file}`, async () => {
-      const arrayBuffer = await readFileToArrayBuffer(`test/files/${file}`)
-      const metadata = await parquetMetadata(arrayBuffer)
+  parquetFiles.forEach((parquetFile, i) => {
+    it(`parse column indexes from ${parquetFile}`, async () => {
+      const file = await readFileToArrayBuffer(`test/files/${parquetFile}`)
+      const metadata = await parquetMetadata({ file })
 
       const result = metadata.row_groups.map((rowGroup) => rowGroup.columns.map((column) => {
         if (column.column_index_offset === undefined || column.column_index_length === undefined) return null
         const columnIndexOffset = Number(column.column_index_offset)
         const columnIndexLength = Number(column.column_index_length)
-        const columnIndexArrayBuffer = arrayBuffer.slice(columnIndexOffset, columnIndexOffset + columnIndexLength)
+        const columnIndexArrayBuffer = file.slice(columnIndexOffset, columnIndexOffset + columnIndexLength)
         const columnIndexReader = { view: new DataView(columnIndexArrayBuffer), offset: 0 }
         const schemaPath = getSchemaPath(metadata.schema, column.meta_data?.path_in_schema ?? [])
         return readColumnIndex(columnIndexReader, schemaPath.at(-1)?.element || { name: '' })
@@ -34,16 +34,16 @@ describe('readOffsetIndex', () => {
   const offsetIndexesFiles = fs.readdirSync('test/files').filter(f => f.endsWith('.offset_indexes.json'))
   const parquetFiles = offsetIndexesFiles.map(f => f.replace(/.offset_indexes.json$/i, '.parquet'))
 
-  parquetFiles.forEach((file, i) => {
-    it(`parse offset indexes from ${file}`, async () => {
-      const arrayBuffer = await readFileToArrayBuffer(`test/files/${file}`)
-      const metadata = await parquetMetadata(arrayBuffer)
+  parquetFiles.forEach((parquetFile, i) => {
+    it(`parse offset indexes from ${parquetFile}`, async () => {
+      const file = await readFileToArrayBuffer(`test/files/${parquetFile}`)
+      const metadata = await parquetMetadata({ file })
 
       const result = metadata.row_groups.map((rowGroup) => rowGroup.columns.map((column) => {
         if (column.offset_index_offset === undefined || column.offset_index_length === undefined) return null
         const offsetIndexOffset = Number(column.offset_index_offset)
         const offsetIndexLength = Number(column.offset_index_length)
-        const offsetIndexArrayBuffer = arrayBuffer.slice(offsetIndexOffset, offsetIndexOffset + offsetIndexLength)
+        const offsetIndexArrayBuffer = file.slice(offsetIndexOffset, offsetIndexOffset + offsetIndexLength)
         const offsetIndexReader = { view: new DataView(offsetIndexArrayBuffer), offset: 0 }
         return readOffsetIndex(offsetIndexReader)
       }))

test/metadata.test.js

@@ -12,42 +12,43 @@ describe('parquetMetadata', () => {
     it(`parse metadata from ${file}`, async () => {
       const asyncBuffer = await asyncBufferFromFile(`test/files/${file}`)
       const arrayBuffer = await asyncBuffer.slice(0)
-      const result = toJson(parquetMetadataSync(arrayBuffer))
+      const result = toJson(parquetMetadataSync({ file: arrayBuffer }))
       const base = file.replace('.parquet', '')
       const expected = fileToJson(`test/files/${base}.metadata.json`)
       expect(result, JSON.stringify(result, null, 2)).toEqual(expected)
     })
   })
 
   it('throws for arrayBuffer undefined', () => {
+    const file = undefined
     // @ts-expect-error testing invalid input
-    expect(() => parquetMetadataSync(undefined)).toThrow('parquet expected ArrayBuffer')
+    expect(() => parquetMetadataSync({ file })).toThrow('parquet expected ArrayBuffer')
   })
 
   it('throws for a too short file', () => {
-    const arrayBuffer = new ArrayBuffer(0)
-    expect(() => parquetMetadataSync(arrayBuffer)).toThrow('parquet file is too short')
+    const file = new ArrayBuffer(0)
+    expect(() => parquetMetadataSync({ file })).toThrow('parquet file is too short')
   })
 
   it('throws for invalid metadata length', () => {
-    const arrayBuffer = new ArrayBuffer(12)
-    const view = new DataView(arrayBuffer)
+    const file = new ArrayBuffer(12)
+    const view = new DataView(file)
     view.setUint32(0, 0x31524150, true) // magic number PAR1
     view.setUint32(4, 1000, true) // 1000 bytes exceeds buffer
     view.setUint32(8, 0x31524150, true) // magic number PAR1
-    expect(() => parquetMetadataSync(arrayBuffer))
+    expect(() => parquetMetadataSync({ file }))
       .toThrow('parquet metadata length 1000 exceeds available buffer 4')
   })
 
   it('throws for invalid magic number', () => {
-    const arrayBuffer = new ArrayBuffer(8)
-    expect(() => parquetMetadataSync(arrayBuffer))
+    const file = new ArrayBuffer(8)
+    expect(() => parquetMetadataSync({ file }))
       .toThrow('parquet file invalid (footer != PAR1)')
   })
 
-  it('throws for invalid metadata length', () => {
+  it('throws for invalid metadata length 2', () => {
     const { buffer } = new Uint8Array([255, 255, 255, 255, 80, 65, 82, 49])
-    expect(() => parquetMetadataSync(buffer))
+    expect(() => parquetMetadataSync({ file: buffer }))
       .toThrow('parquet metadata length 4294967295 exceeds available buffer 0')
   })
 })
@@ -56,29 +57,29 @@ describe('parquetMetadataAsync', () => {
   files.forEach(file => {
     it(`parse metadata async from ${file}`, async () => {
       const asyncBuffer = await asyncBufferFromFile(`test/files/${file}`)
-      const result = await parquetMetadata(asyncBuffer)
+      const result = await parquetMetadata({ file: asyncBuffer })
       const base = file.replace('.parquet', '')
       const expected = fileToJson(`test/files/${base}.metadata.json`)
       expect(toJson(result)).toEqual(expected)
     })
   })
 
   it('throws for asyncBuffer undefined', async () => {
-    const arrayBuffer = undefined
+    const file = undefined
     // @ts-expect-error testing invalid input
-    await expect(parquetMetadata(arrayBuffer)).rejects
+    await expect(parquetMetadata({ file })).rejects
       .toThrow('parquet expected AsyncBuffer')
   })
 
   it('throws for invalid magic number', async () => {
     const { buffer } = new Uint8Array([255, 255, 255, 255, 255, 255, 255, 255])
-    await expect(parquetMetadata(buffer)).rejects
+    await expect(parquetMetadata({ file: buffer })).rejects
       .toThrow('parquet file invalid (footer != PAR1)')
   })
 
   it('throws for invalid metadata length', async () => {
     const { buffer } = new Uint8Array([255, 255, 255, 255, 80, 65, 82, 49])
-    await expect(parquetMetadata(buffer)).rejects
+    await expect(parquetMetadata({ file: buffer })).rejects
       .toThrow('parquet metadata length 4294967295 exceeds available buffer 0')
   })
 })

test/plan.test.js

@@ -6,7 +6,7 @@ import { parquetPlan } from '../src/plan.js'
 describe('parquetPlan', () => {
   it('generates a query plan', async () => {
     const file = await asyncBufferFromFile('test/files/offset_indexed.parquet')
-    const metadata = await parquetMetadata(file)
+    const metadata = await parquetMetadata({ file })
     const plan = parquetPlan({ file, metadata })
     expect(plan).toMatchObject({
       metadata,

test/query.test.js

@@ -203,7 +203,7 @@ describe('parquetQuery', () => {
   it('reads data efficiently with filter', async () => {
     const originalFile = await asyncBufferFromFile('test/files/alpha.parquet')
     // don't count metadata reads
-    const metadata = await parquetMetadata(originalFile)
+    const metadata = await parquetMetadata({ file: originalFile })
     const file = countingBuffer(await asyncBufferFromFile('test/files/alpha.parquet'))
     // Query for rows where id = 'kk'
     const rows = await parquetQuery({ file, metadata, filter: { id: { $eq: 'kk' } } })
@@ -216,7 +216,7 @@ describe('parquetQuery', () => {
   it('reads data efficiently with filter and sort', async () => {
     const originalFile = await asyncBufferFromFile('test/files/alpha.parquet')
     // don't count metadata reads
-    const metadata = await parquetMetadata(originalFile)
+    const metadata = await parquetMetadata({ file: originalFile })
     const file = countingBuffer(await asyncBufferFromFile('test/files/alpha.parquet'))
     const rows = await parquetQuery({ file, metadata, filter: { id: { $gt: 'xx' } }, orderBy: 'id' } )
     expect(rows[0]).toEqual({ id: 'xy' })