feat: streaming execution engine for TB+ scale queries

Pull-based operator pipeline replaces full materialization:
- ScanOperator, FilterOperator, ProjectOperator, LimitOperator
- AggregateOperator (streaming partial agg, O(groups) memory)
- TopKOperator (ORDER BY + LIMIT via min-heap, O(K) memory)
- ExternalSortOperator (spill to NDJSON temp files, K-way merge)
- HashJoinOperator (.join() API, build/probe with hash map)
- WasmAggregateOperator (SIMD sum/min/max/avg on raw buffers)
- Lance v2 page stats (compute-and-cache min/max for page skipping)

Author: teamchong

src/client.ts

@@ -3,6 +3,7 @@ import type {
   AppendResult,
   ExplainResult,
   FilterOp,
+  JoinDescriptor,
   QueryResult,
   Row,
   VectorSearchParams,
@@ -32,6 +33,7 @@ export class TableQuery<T extends Row = Row> {
   private _aggregates: AggregateOp[] = [];
   private _groupBy: string[] = [];
   private _cacheTTL?: number;
+  private _join?: JoinDescriptor;
   private _executor: QueryExecutor;
 
   constructor(table: string, executor: QueryExecutor) {
@@ -102,6 +104,24 @@ export class TableQuery<T extends Row = Row> {
     return this;
   }
 
+  /**
+   * Join with another table query. Build side (right) is hashed, probe side (left) streams through.
+   *
+   * Usage:
+   *   const orders = qm.table("orders");
+   *   const users = qm.table("users");
+   *   const result = await orders.join(users, { left: "user_id", right: "id" }).exec();
+   */
+  join(other: TableQuery, on: { left: string; right: string }, type?: "inner" | "left"): this {
+    this._join = {
+      right: other.toDescriptor(),
+      leftKey: on.left,
+      rightKey: on.right,
+      type: type ?? "inner",
+    };
+    return this;
+  }
+
   /** Enable query result caching with a TTL in milliseconds. Remote executor only. */
   cache(opts: { ttl: number }): this {
     this._cacheTTL = opts.ttl;
@@ -186,6 +206,7 @@ export class TableQuery<T extends Row = Row> {
       aggregates: this._aggregates.length > 0 ? this._aggregates : undefined,
       groupBy: this._groupBy.length > 0 ? this._groupBy : undefined,
       cacheTTL: this._cacheTTL,
+      join: this._join,
     };
   }
 }
@@ -203,6 +224,7 @@ export interface QueryDescriptor {
   aggregates?: AggregateOp[];
   groupBy?: string[];
   cacheTTL?: number;
+  join?: JoinDescriptor;
 }
 
 /** Interface for query execution backends (local, DO, browser) */

src/index.ts

@@ -25,6 +25,7 @@ export type {
   FragmentInfo,
   FilterOp,
   AggregateOp,
+  JoinDescriptor,
   QueryResult,
   Row,
   VectorSearchParams,
@@ -74,8 +75,8 @@ export class QueryMode {
   }
 
   /** Create an QueryMode client for local use (Node/Bun, reads files from disk or URLs). */
-  static local(wasmModule?: WebAssembly.Module): QueryMode {
-    const executor = new LocalExecutor(wasmModule);
+  static local(opts?: { wasmModule?: WebAssembly.Module; memoryBudgetBytes?: number }): QueryMode {
+    const executor = opts ? new LocalExecutor(opts) : new LocalExecutor();
     return new QueryMode(executor);
   }
 

src/lance-v2.ts

@@ -391,6 +391,150 @@ export function lanceV2ToColumnMeta(colInfos: LanceV2ColumnInfo[]): ColumnMeta[]
   return columns;
 }
 
+/**
+ * Compute min/max statistics for Lance v2 columns by reading page data.
+ * Lance v2 protobuf doesn't store page-level stats, so we compute them
+ * on first access and populate the ColumnMeta pages in place.
+ *
+ * Only computes stats for numeric types (int8-int64, uint8-uint64, float32, float64)
+ * since those benefit most from page skipping.
+ *
+ * @param columns - ColumnMeta array to populate with stats (mutated in place)
+ * @param readPage - Function to read a page's raw bytes given offset and length
+ */
+export async function computeLanceV2Stats(
+  columns: ColumnMeta[],
+  readPage: (byteOffset: number, byteLength: number) => Promise<ArrayBuffer>,
+): Promise<void> {
+  const numericTypes = new Set([
+    "int8", "int16", "int32", "int64",
+    "uint8", "uint16", "uint32", "uint64",
+    "float32", "float64",
+  ]);
+
+  for (const col of columns) {
+    if (!numericTypes.has(col.dtype)) continue;
+
+    for (const page of col.pages) {
+      // Skip if stats already populated
+      if (page.minValue !== undefined) continue;
+
+      try {
+        const buf = await readPage(Number(page.byteOffset), page.byteLength);
+        const stats = computePageStats(buf, col.dtype, page.rowCount, page.nullCount, page.dataOffsetInPage);
+        if (stats) {
+          page.minValue = stats.min;
+          page.maxValue = stats.max;
+        }
+      } catch {
+        // Silently skip — stats are an optimization, not required
+      }
+    }
+  }
+}
+
+/** Compute min/max from a raw page buffer for a numeric dtype. */
+function computePageStats(
+  buf: ArrayBuffer,
+  dtype: string,
+  rowCount: number,
+  nullCount: number,
+  dataOffsetInPage?: number,
+): { min: number | bigint; max: number | bigint } | null {
+  let dataBuf = buf;
+
+  // Strip null bitmap for nullable columns
+  if (nullCount > 0 && rowCount > 0) {
+    const stripBytes = dataOffsetInPage ?? Math.ceil(rowCount / 8);
+    dataBuf = buf.slice(stripBytes);
+  }
+
+  const view = new DataView(dataBuf);
+  const numValues = dataOffsetInPage !== undefined ? rowCount : rowCount - nullCount;
+  if (numValues <= 0) return null;
+
+  // Read null bitmap to skip null positions
+  let nullSet: Set<number> | null = null;
+  if (nullCount > 0 && rowCount > 0) {
+    const bitmapBytes = Math.ceil(rowCount / 8);
+    if (buf.byteLength >= bitmapBytes) {
+      const bytes = new Uint8Array(buf, 0, bitmapBytes);
+      nullSet = new Set<number>();
+      let idx = 0;
+      for (let b = 0; b < bitmapBytes && idx < rowCount; b++) {
+        for (let bit = 0; bit < 8 && idx < rowCount; bit++, idx++) {
+          if (((bytes[b] >> bit) & 1) === 0) nullSet.add(idx);
+        }
+      }
+    }
+  }
+
+  switch (dtype) {
+    case "float64": {
+      let min = Infinity, max = -Infinity;
+      for (let i = 0; i < numValues; i++) {
+        if (nullSet?.has(i)) continue;
+        const v = view.getFloat64(i * 8, true);
+        if (v < min) min = v;
+        if (v > max) max = v;
+      }
+      return min <= max ? { min, max } : null;
+    }
+    case "float32": {
+      let min = Infinity, max = -Infinity;
+      for (let i = 0; i < numValues; i++) {
+        if (nullSet?.has(i)) continue;
+        const v = view.getFloat32(i * 4, true);
+        if (v < min) min = v;
+        if (v > max) max = v;
+      }
+      return min <= max ? { min, max } : null;
+    }
+    case "int64": {
+      let min = BigInt("9223372036854775807"), max = BigInt("-9223372036854775808");
+      for (let i = 0; i < numValues; i++) {
+        if (nullSet?.has(i)) continue;
+        const v = view.getBigInt64(i * 8, true);
+        if (v < min) min = v;
+        if (v > max) max = v;
+      }
+      return min <= max ? { min, max } : null;
+    }
+    case "int32": {
+      let min = 2147483647, max = -2147483648;
+      for (let i = 0; i < numValues; i++) {
+        if (nullSet?.has(i)) continue;
+        const v = view.getInt32(i * 4, true);
+        if (v < min) min = v;
+        if (v > max) max = v;
+      }
+      return min <= max ? { min, max } : null;
+    }
+    case "int16": {
+      let min = 32767, max = -32768;
+      for (let i = 0; i < numValues; i++) {
+        if (nullSet?.has(i)) continue;
+        const v = view.getInt16(i * 2, true);
+        if (v < min) min = v;
+        if (v > max) max = v;
+      }
+      return min <= max ? { min, max } : null;
+    }
+    case "int8": {
+      let min = 127, max = -128;
+      for (let i = 0; i < numValues; i++) {
+        if (nullSet?.has(i)) continue;
+        const v = view.getInt8(i);
+        if (v < min) min = v;
+        if (v > max) max = v;
+      }
+      return min <= max ? { min, max } : null;
+    }
+    default:
+      return null;
+  }
+}
+
 /**
  * Decode Lance v2 utf8 data from a buffer containing [offsets_array | padding | string_data].
  * The offsets array is (rowCount) i64 values representing cumulative end positions.