perf: cached identity indices, eliminate .map().join() in GROUP BY/DISTINCT keys; fix 5 doc issues

Performance:
- Identity Uint32Array: cache by size instead of allocating per batch per operator (7 sites)
- GROUP BY key: replace .map().join() with direct concatenation (partial-agg.ts, 2 paths)
- DISTINCT key: same .map().join() → direct concatenation (operators.ts, 2 paths)

Docs:
- README: fix Fragment DO pool max (20 → 100), test count (112 → 460+)
- formats.mdx: add missing await on fromCSV
- composability.mdx: add nextColumnar to Operator interface

Author: teamchong

README.md

@@ -251,8 +251,8 @@ npx tsx examples/nextjs-api-route.ts
 - **IVF-PQ vector search** — index-aware routing in Query DO, falls back to flat SIMD search when no index present
 - **Multi-format support** — Lance, Parquet, and Iceberg tables
 - **Local mode** — same API reads Lance/Parquet files from disk or HTTP (Node/Bun)
-- **Fragment DO pool** — fan-out parallel scanning for multi-fragment datasets (max 20 slots per datacenter)
-- **112 unit tests + 26 conformance tests** — unit tests cover footer parsing, column decoding, Parquet/Thrift, merging, aggregates, VIP cache, WASM integration; conformance tests validate every operator against DuckDB at 1M-5M row scale
+- **Fragment DO pool** — fan-out parallel scanning for multi-fragment datasets (max 100 slots per datacenter)
+- **460+ tests** — unit tests cover footer parsing, column decoding, Parquet/Thrift, merging, aggregates, VIP cache, WASM integration, SQL, partition catalog, materialized executor; 110+ conformance tests validate every operator against DuckDB at 1M-5M row scale
 - **CI benchmarks** — head-to-head QueryMode (Miniflare) vs DuckDB (native) on every push, results posted to [GitHub Actions summary](https://github.com/teamchong/querymode/actions/workflows/ci.yml)
 
 ## What doesn't exist yet

docs/src/content/docs/composability.mdx

@@ -10,6 +10,7 @@ Every operator in QueryMode implements one interface:
 ```typescript
 interface Operator {
   next(): Promise<RowBatch | null>
+  nextColumnar?(): Promise<ColumnarBatch | null>  // optional columnar mode
   close(): Promise<void>
 }
 ```

docs/src/content/docs/formats.mdx

@@ -81,7 +81,7 @@ In-memory materialization for small datasets:
 
 ```typescript
 // From CSV string
-const qm = QueryMode.fromCSV(csvString, "my_table")
+const qm = await QueryMode.fromCSV(csvString, "my_table")
 
 // From JSON array
 const qm = QueryMode.fromJSON(jsonArray, "my_table")

src/operators.ts

@@ -15,6 +15,19 @@ import { decodeParquetColumnChunk } from "./parquet-decode.js";
 
 const _textEncoder = new TextEncoder();
 
+/** Cached identity index arrays to avoid repeated allocations on hot paths. */
+const _identityCache = new Map<number, Uint32Array>();
+function identityIndices(n: number): Uint32Array {
+  let cached = _identityCache.get(n);
+  if (cached) return cached;
+  cached = new Uint32Array(n);
+  for (let i = 0; i < n; i++) cached[i] = i;
+  // Keep cache bounded — only cache common page sizes
+  if (_identityCache.size > 16) _identityCache.clear();
+  _identityCache.set(n, cached);
+  return cached;
+}
+
 export type DecodedValue = number | bigint | string | boolean | Float32Array | null;
 import {
   computePartialAgg,
@@ -54,7 +67,7 @@ export interface Operator {
 /** Materialize a ColumnarBatch into Row[] — used at pipeline boundaries. */
 export function materializeRows(batch: ColumnarBatch): Row[] {
   const rows: Row[] = [];
-  const indices = batch.selection ?? Uint32Array.from({ length: batch.rowCount }, (_, i) => i);
+  const indices = batch.selection ?? identityIndices(batch.rowCount);
   for (const idx of indices) {
     const row: Row = {};
     for (const [name, vals] of batch.columns) {
@@ -399,7 +412,7 @@ function scanFilterIndices(
     indices = indices ? wasmIntersect(indices, orResult, wasm) : orResult;
   }
 
-  return indices ?? Uint32Array.from({ length: rowCount }, (_, i) => i);
+  return indices ?? identityIndices(rowCount);
 }
 
 /** Apply AND-connected filters, returning matching row indices. */
@@ -467,7 +480,7 @@ function applyAndFilters(
     }
 
     // JS fallback: evaluate filter on current index set
-    const src = indices ?? Uint32Array.from({ length: rowCount }, (_, i) => i);
+    const src = indices ?? identityIndices(rowCount);
     const kept: number[] = [];
     for (const idx of src) {
       const v = values[idx];
@@ -481,7 +494,7 @@ function applyAndFilters(
     if (indices.length === 0) return indices;
   }
 
-  return indices ?? Uint32Array.from({ length: rowCount }, (_, i) => i);
+  return indices ?? identityIndices(rowCount);
 }
 
 /** Run WASM SIMD filter on decoded numeric values (f64, i32, i64). */
@@ -1112,15 +1125,17 @@ export class DistinctOperator implements Operator {
       const batch = await this.upstream.nextColumnar();
       if (!batch) return null;
 
-      const indices = batch.selection ?? Uint32Array.from({ length: batch.rowCount }, (_, i) => i);
+      const indices = batch.selection ?? identityIndices(batch.rowCount);
       const kept: number[] = [];
       const cols = this.columns.length > 0 ? this.columns : Array.from(batch.columns.keys());
 
       for (const idx of indices) {
-        const key = cols.map(c => {
-          const vals = batch.columns.get(c);
-          return String(vals ? (vals[idx] ?? "") : "");
-        }).join("\x00");
+        let key = "";
+        for (let g = 0; g < cols.length; g++) {
+          if (g > 0) key += "\x00";
+          const vals = batch.columns.get(cols[g]);
+          key += String(vals ? (vals[idx] ?? "") : "");
+        }
         if (!this.seen.has(key)) {
           this.seen.add(key);
           kept.push(idx);
@@ -1139,9 +1154,12 @@ export class DistinctOperator implements Operator {
 
       const unique: Row[] = [];
       for (const row of batch) {
-        const key = this.columns.length > 0
-          ? this.columns.map(c => String(row[c] ?? "")).join("\x00")
-          : Object.keys(row).map(k => String(row[k] ?? "")).join("\x00");
+        let key = "";
+        const keyCols = this.columns.length > 0 ? this.columns : Object.keys(row);
+        for (let g = 0; g < keyCols.length; g++) {
+          if (g > 0) key += "\x00";
+          key += String(row[keyCols[g]] ?? "");
+        }
         if (!this.seen.has(key)) {
           this.seen.add(key);
           unique.push(row);
@@ -1354,7 +1372,7 @@ export class FilterOperator implements Operator {
       if (!batch) return null;
 
       // Apply filters on the columnar batch — narrow the selection vector
-      const srcIndices = batch.selection ?? Uint32Array.from({ length: batch.rowCount }, (_, i) => i);
+      const srcIndices = batch.selection ?? identityIndices(batch.rowCount);
       const kept: number[] = [];
 
       for (const idx of srcIndices) {
@@ -1681,7 +1699,7 @@ export class TopKOperator implements Operator {
       while (true) {
         const batch = await this.upstream.nextColumnar();
         if (!batch) break;
-        const indices = batch.selection ?? Uint32Array.from({ length: batch.rowCount }, (_, i) => i);
+        const indices = batch.selection ?? identityIndices(batch.rowCount);
         const colNames = Array.from(batch.columns.keys());
         const sortVals = batch.columns.get(col);
         for (const idx of indices) {

src/partial-agg.ts

@@ -2,6 +2,17 @@ import type { Row } from "./types.js";
 import type { QueryDescriptor } from "./client.js";
 import type { ColumnarBatch, DecodedValue } from "./operators.js";
 
+const _identityCacheLocal = new Map<number, Uint32Array>();
+function identityIndicesLocal(n: number): Uint32Array {
+  let cached = _identityCacheLocal.get(n);
+  if (cached) return cached;
+  cached = new Uint32Array(n);
+  for (let i = 0; i < n; i++) cached[i] = i;
+  if (_identityCacheLocal.size > 16) _identityCacheLocal.clear();
+  _identityCacheLocal.set(n, cached);
+  return cached;
+}
+
 export interface PartialAgg {
   states: PartialAggState[];
   groups?: Map<string, PartialAggState[]>;
@@ -143,7 +154,11 @@ export function computePartialAgg(
   const groupCols = query.groupBy;
 
   for (const row of rows) {
-    const key = groupCols.map((c) => String(row[c] ?? "")).join("\x00");
+    let key = "";
+    for (let g = 0; g < groupCols.length; g++) {
+      if (g > 0) key += "\x00";
+      key += String(row[groupCols[g]] ?? "");
+    }
     let states = groups.get(key);
     if (!states) {
       states = aggregates.map((agg) =>
@@ -176,7 +191,7 @@ export function computePartialAggColumnar(
   query: QueryDescriptor,
 ): PartialAgg {
   const aggregates = query.aggregates ?? [];
-  const indices = batch.selection ?? Uint32Array.from({ length: batch.rowCount }, (_, i) => i);
+  const indices = batch.selection ?? identityIndicesLocal(batch.rowCount);
 
   if (!query.groupBy || query.groupBy.length === 0) {
     const states = aggregates.map((agg) =>
@@ -207,10 +222,12 @@ export function computePartialAggColumnar(
   const groupCols = query.groupBy;
 
   for (const idx of indices) {
-    const key = groupCols.map((c) => {
-      const vals = batch.columns.get(c);
-      return String(vals ? (vals[idx] ?? "") : "");
-    }).join("\x00");
+    let key = "";
+    for (let g = 0; g < groupCols.length; g++) {
+      if (g > 0) key += "\x00";
+      const vals = batch.columns.get(groupCols[g]);
+      key += String(vals ? (vals[idx] ?? "") : "");
+    }
     let states = groups.get(key);
     if (!states) {
       states = aggregates.map((agg) =>