plan: add columnChunkAggregation opt-in for column-chunk coalescing

Existing behavior: `parquetPlan` coalesces a row group's column chunks into
one byte range when (a) no `columns` projection is set, AND (b) the group
span is < 32MB. With `columns` set, every chunk became a separate fetch.

That's painful on high-latency stores (R2 / S3): a `metric=all` query over
7 daily parquets with 5 projected cols × ~12 row groups = ~420 range GETs.
On Cloudflare Workers each GET carries non-trivial CPU overhead and the
1102 (CPU exceeded) limit kicks in.

Add `columnChunkAggregation?: number` to `BaseParquetReadOptions`. When
caller passes a value, coalesce up to that threshold even with `columns`
projected — the bounded overfetch (≤ rg span) is a much better trade than
N range GETs.

Default behavior unchanged (32MB when `columns` unset; 0 when set). Pass 0
explicitly to disable coalescing entirely.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: ryan-williams

src/plan.js

@@ -3,8 +3,10 @@ import { parquetSchema } from './metadata.js'
 import { getPhysicalColumns } from './schema.js'
 import { concat } from './utils.js'
 
-// Combine column chunks into a single byte range if less than 32mb
-const columnChunkAggregation = 1 << 25 // 32mb
+// Default: combine column chunks into a single byte range if less than 32mb
+// (only applied when no `columns` projection is set, unless caller opts in via
+// `columnChunkAggregation`).
+const defaultColumnChunkAggregation = 1 << 25 // 32mb
 
 /**
  * @import {AsyncBuffer, ByteRange, ColumnMetaData, GroupPlan, ParquetReadOptions, QueryPlan} from '../src/types.js'
@@ -16,8 +18,13 @@ const columnChunkAggregation = 1 << 25 // 32mb
  * @param {ParquetReadOptions} options
  * @returns {QueryPlan}
  */
-export function parquetPlan({ metadata, rowStart = 0, rowEnd = Infinity, columns, filter }) {
+export function parquetPlan({ metadata, rowStart = 0, rowEnd = Infinity, columns, filter, columnChunkAggregation }) {
   if (!metadata) throw new Error('parquetPlan requires metadata')
+  // When unset, preserve legacy behavior: coalesce only when reading all
+  // columns. When set explicitly, honor the threshold even with `columns`
+  // projected (callers on high-latency stores e.g. R2 prefer a small amount
+  // of overfetch over many tiny range GETs).
+  const aggBytes = columnChunkAggregation ?? (columns ? 0 : defaultColumnChunkAggregation)
   /** @type {GroupPlan[]} */
   const groups = []
   /** @type {ByteRange[]} */
@@ -48,8 +55,8 @@ export function parquetPlan({ metadata, rowStart = 0, rowEnd = Infinity, columns
 
       // map group plan to ranges
       const groupSize = ranges[ranges.length - 1]?.endByte - ranges[0]?.startByte
-      if (!columns && groupSize < columnChunkAggregation) {
-        // full row group
+      if (aggBytes > 0 && groupSize < aggBytes) {
+        // coalesce all selected column chunks of this row group into one fetch
         fetches.push({
           startByte: ranges[0].startByte,
           endByte: ranges[ranges.length - 1].endByte,

src/types.d.ts

@@ -44,6 +44,7 @@ export interface BaseParquetReadOptions {
   utf8?: boolean // decode byte arrays as utf8 strings (default true)
   parsers?: ParquetParsers // custom parsers to decode advanced types
   geoparquet?: boolean // parse geoparquet metadata and set logical type to geometry/geography for geospatial columns (default true)
+  columnChunkAggregation?: number // coalesce a row group's selected column chunks into one byte range when their span is below this many bytes. Default: 32mb when `columns` is unset, 0 (no coalescing) when `columns` is set. Pass an explicit value (e.g. on high-latency stores) to coalesce even with `columns` projected; pass 0 to always issue one fetch per chunk.
 }
 
 interface ArrayRowFormat {

test/plan.test.js

@@ -36,4 +36,73 @@ describe('parquetPlan', () => {
       ],
     })
   })
+
+  describe('columnChunkAggregation option', () => {
+    it('default behavior: columns set → no coalescing (one fetch per chunk)', async () => {
+      const file = await asyncBufferFromFile('test/files/page_indexed.parquet')
+      const metadata = await parquetMetadataAsync(file)
+      const plan = parquetPlan({ file, metadata, columns: ['row', 'quality'] })
+      // Both chunks of each row group emitted as separate fetches.
+      expect(plan.fetches).toEqual([
+        { startByte: 4, endByte: 832 },
+        { startByte: 832, endByte: 1166 },
+        { startByte: 1166, endByte: 1998 },
+        { startByte: 1998, endByte: 2326 },
+      ])
+    })
+
+    it('columns + columnChunkAggregation > groupSize → coalesces within each rg', async () => {
+      const file = await asyncBufferFromFile('test/files/page_indexed.parquet')
+      const metadata = await parquetMetadataAsync(file)
+      const plan = parquetPlan({
+        file, metadata, columns: ['row', 'quality'], columnChunkAggregation: 1 << 25,
+      })
+      // Same shape as default no-columns case: one fetch per row group.
+      expect(plan.fetches).toEqual([
+        { startByte: 4, endByte: 1166 },
+        { startByte: 1166, endByte: 2326 },
+      ])
+    })
+
+    it('columns + columnChunkAggregation < groupSize → no coalescing', async () => {
+      const file = await asyncBufferFromFile('test/files/page_indexed.parquet')
+      const metadata = await parquetMetadataAsync(file)
+      // groupSize per rg is 1162 bytes; threshold of 1000 disables coalescing.
+      const plan = parquetPlan({
+        file, metadata, columns: ['row', 'quality'], columnChunkAggregation: 1000,
+      })
+      expect(plan.fetches).toEqual([
+        { startByte: 4, endByte: 832 },
+        { startByte: 832, endByte: 1166 },
+        { startByte: 1166, endByte: 1998 },
+        { startByte: 1998, endByte: 2326 },
+      ])
+    })
+
+    it('no columns + columnChunkAggregation: 0 → disables default coalescing', async () => {
+      const file = await asyncBufferFromFile('test/files/page_indexed.parquet')
+      const metadata = await parquetMetadataAsync(file)
+      const plan = parquetPlan({ file, metadata, columnChunkAggregation: 0 })
+      expect(plan.fetches).toEqual([
+        { startByte: 4, endByte: 832 },
+        { startByte: 832, endByte: 1166 },
+        { startByte: 1166, endByte: 1998 },
+        { startByte: 1998, endByte: 2326 },
+      ])
+    })
+
+    it('columns subset + coalesce: still one fetch per rg covering subset span', async () => {
+      const file = await asyncBufferFromFile('test/files/page_indexed.parquet')
+      const metadata = await parquetMetadataAsync(file)
+      const plan = parquetPlan({
+        file, metadata, columns: ['quality'], columnChunkAggregation: 1 << 25,
+      })
+      // Only the second chunk of each rg is selected; "coalesce" with one
+      // chunk just returns that chunk's range.
+      expect(plan.fetches).toEqual([
+        { startByte: 832, endByte: 1166 },
+        { startByte: 1998, endByte: 2326 },
+      ])
+    })
+  })
 })