feat: multi-column page skip + BETWEEN filter with WASM range ops

- canSkipPageMultiCol: check ALL filter columns' min/max stats for page skip
  (was only checking first column — missed skips on non-first-column filters)
- Add BETWEEN filter op end-to-end: types, client (.whereBetween()), decode
  (matchesFilter + canSkipPage), WASM range filters, scan-time WASM path
- Add filterFloat64Range, filterInt32Range, filterInt64Range Zig exports
- canSkipPage handles BETWEEN: skip page if max < low or min > high
- WasmAggregateOperator uses multi-column page skip

Author: teamchong

src/client.ts

@@ -169,6 +169,11 @@ export class DataFrame<T extends Row = Row> {
     return this.where(column, op, value);
   }
 
+  /** Filter rows where column value is between low and high (inclusive). */
+  whereBetween(column: string, low: number | bigint, high: number | bigint): DataFrame<T> {
+    return this.derive({ filters: [...this._filters, { column, op: "between", value: [low, high] }] });
+  }
+
   /** Filter rows where a column is not null. */
   whereNotNull(column: string): DataFrame<T> {
     return this.derive({ filters: [...this._filters, { column, op: "is_not_null", value: 0 }] });

src/decode.ts

@@ -25,6 +25,15 @@ export function canSkipPage(page: PageInfo, filters: QueryDescriptor["filters"],
       case "lt":  if (min >= val) return true; break;
       case "lte": if (min > val) return true; break;
       case "eq":  if (val < min || val > max) return true; break;
+      case "between": {
+        if (!Array.isArray(filter.value) || filter.value.length !== 2) break;
+        let lo = filter.value[0];
+        let hi = filter.value[1];
+        if (typeof min === "bigint" && typeof lo === "number") { lo = BigInt(Math.trunc(lo)); hi = BigInt(Math.trunc(hi as number)); }
+        else if (typeof min === "number" && typeof lo === "bigint") { min = BigInt(Math.trunc(min)); max = BigInt(Math.trunc(max as number)); }
+        if (max < lo || min > hi) return true;
+        break;
+      }
     }
   }
   return false;
@@ -383,6 +392,13 @@ export function matchesFilter(
     case "lt":  { const [a, b] = coerceCompare(val, t); return (a as number | bigint | string) < (b as number | bigint | string); }
     case "lte": { const [a, b] = coerceCompare(val, t); return (a as number | bigint | string) <= (b as number | bigint | string); }
     case "in":  return Array.isArray(t) && t.some(v => { const [a, b] = coerceCompare(val, v); return a === b; });
+    case "between": {
+      if (!Array.isArray(t) || t.length !== 2) return false;
+      const [, lo] = coerceCompare(val, t[0]);
+      const [a, hi] = coerceCompare(val, t[1]);
+      return (a as number | bigint | string) >= (lo as number | bigint | string) &&
+             (a as number | bigint | string) <= (hi as number | bigint | string);
+    }
     default:    return true;
   }
 }

src/operators.ts

@@ -41,6 +41,23 @@ export interface FragmentSource {
   readPage(col: ColumnMeta, page: PageInfo): Promise<ArrayBuffer>;
 }
 
+/** Check if a page can be skipped by checking ALL filter columns' min/max stats.
+ *  Returns true if ANY filter column's stats prove zero rows can match. */
+function canSkipPageMultiCol(
+  columns: ColumnMeta[], pageIdx: number, filters: QueryDescriptor["filters"],
+): boolean {
+  if (filters.length === 0) return false;
+  const colMap = new Map(columns.map(c => [c.name, c]));
+  for (const f of filters) {
+    const col = colMap.get(f.column);
+    if (!col) continue;
+    const page = col.pages[pageIdx];
+    if (!page) continue;
+    if (canSkipPage(page, [f], f.column)) return true;
+  }
+  return false;
+}
+
 // ---------------------------------------------------------------------------
 // ScanOperator — reads pages from fragments, decodes, yields batches
 // ---------------------------------------------------------------------------
@@ -110,8 +127,7 @@ export class ScanOperator implements Operator {
     const firstCol = frag.columns[0];
     if (!firstCol) return -1;
     for (let i = start; i < firstCol.pages.length; i++) {
-      const page = firstCol.pages[i];
-      if (this.query.vectorSearch || !canSkipPage(page, this.query.filters, firstCol.name)) {
+      if (this.query.vectorSearch || !canSkipPageMultiCol(frag.columns, i, this.query.filters)) {
         return i;
       }
     }
@@ -156,9 +172,8 @@ export class ScanOperator implements Operator {
         const pi = this.pageIdx;
         this.pageIdx++;
 
-        // Page-level skip via min/max stats on first column
-        const page = firstCol.pages[pi];
-        if (!this.query.vectorSearch && canSkipPage(page, this.query.filters, firstCol.name)) {
+        // Page-level skip via min/max stats — check ALL filter columns, not just first
+        if (!this.query.vectorSearch && canSkipPageMultiCol(frag.columns, pi, this.query.filters)) {
           this.pagesSkipped += frag.columns.length;
           continue;
         }
@@ -318,7 +333,7 @@ function scanFilterIndices(
       return new Uint32Array(0);
     }
 
-    const wasmOp = filter.op !== "in" ? filterOpToWasm(filter.op) : -1;
+    const wasmOp = filter.op !== "in" && filter.op !== "between" ? filterOpToWasm(filter.op) : -1;
 
     // Try WASM SIMD path for numeric scalar filters
     if (wasmOp >= 0 && typeof filter.value === "number" &&
@@ -333,6 +348,20 @@ function scanFilterIndices(
       }
     }
 
+    // Try WASM BETWEEN path for numeric range filters
+    if (filter.op === "between" && Array.isArray(filter.value) && filter.value.length === 2 &&
+        typeof filter.value[0] === "number" && typeof filter.value[1] === "number" &&
+        (dtype === "float64" || dtype === "float32" || dtype === "int32" || dtype === "int64")) {
+      const filterResult = wasmFilterRange(
+        values, dtype, filter.value[0], filter.value[1], rowCount, wasm,
+      );
+      if (filterResult) {
+        indices = indices ? wasmIntersect(indices, filterResult, wasm) : filterResult;
+        if (indices.length === 0) return indices;
+        continue;
+      }
+    }
+
     // JS fallback: evaluate filter on current index set
     const src = indices ?? Uint32Array.from({ length: rowCount }, (_, i) => i);
     const kept: number[] = [];
@@ -413,6 +442,60 @@ function wasmFilterNumeric(
   }
 }
 
+/** Run WASM BETWEEN (range) filter on decoded numeric values. */
+function wasmFilterRange(
+  values: DecodedValue[],
+  dtype: string,
+  low: number,
+  high: number,
+  rowCount: number,
+  wasm: WasmEngine,
+): Uint32Array | null {
+  try {
+    wasm.exports.resetHeap();
+
+    if (dtype === "float64" || dtype === "float32") {
+      const dataPtr = wasm.exports.alloc(rowCount * 8);
+      if (!dataPtr) return null;
+      const dst = new Float64Array(wasm.exports.memory.buffer, dataPtr, rowCount);
+      for (let i = 0; i < rowCount; i++) dst[i] = (values[i] as number) ?? 0;
+      const outPtr = wasm.exports.alloc(rowCount * 4);
+      if (!outPtr) return null;
+      const count = wasm.exports.filterFloat64Range(dataPtr, rowCount, low, high, outPtr, rowCount);
+      return new Uint32Array(wasm.exports.memory.buffer.slice(outPtr, outPtr + count * 4));
+    }
+
+    if (dtype === "int64") {
+      const dataPtr = wasm.exports.alloc(rowCount * 8);
+      if (!dataPtr) return null;
+      const dst = new BigInt64Array(wasm.exports.memory.buffer, dataPtr, rowCount);
+      for (let i = 0; i < rowCount; i++) {
+        const v = values[i];
+        dst[i] = typeof v === "bigint" ? v : BigInt((v as number) ?? 0);
+      }
+      const outPtr = wasm.exports.alloc(rowCount * 4);
+      if (!outPtr) return null;
+      const count = wasm.exports.filterInt64Range(dataPtr, rowCount, BigInt(low), BigInt(high), outPtr, rowCount);
+      return new Uint32Array(wasm.exports.memory.buffer.slice(outPtr, outPtr + count * 4));
+    }
+
+    if (dtype === "int32") {
+      const dataPtr = wasm.exports.alloc(rowCount * 4);
+      if (!dataPtr) return null;
+      const dst = new Int32Array(wasm.exports.memory.buffer, dataPtr, rowCount);
+      for (let i = 0; i < rowCount; i++) dst[i] = (values[i] as number) ?? 0;
+      const outPtr = wasm.exports.alloc(rowCount * 4);
+      if (!outPtr) return null;
+      const count = wasm.exports.filterInt32Range(dataPtr, rowCount, low, high, outPtr, rowCount);
+      return new Uint32Array(wasm.exports.memory.buffer.slice(outPtr, outPtr + count * 4));
+    }
+
+    return null;
+  } catch {
+    return null;
+  }
+}
+
 /** Intersect two sorted index arrays using WASM. */
 function wasmIntersect(a: Uint32Array, b: Uint32Array, wasm: WasmEngine): Uint32Array {
   try {
@@ -1296,8 +1379,8 @@ export class WasmAggregateOperator implements Operator {
       const pageCount = firstCol.pages.length;
 
       for (let pi = 0; pi < pageCount; pi++) {
-        // Page-level skip via min/max stats
-        if (canSkipPage(firstCol.pages[pi], filters, firstCol.name)) {
+        // Page-level skip via min/max stats — check all filter columns
+        if (canSkipPageMultiCol(frag.columns, pi, filters)) {
           this.pagesSkipped++;
           continue;
         }

src/types.ts

@@ -85,7 +85,7 @@ export interface TableMeta {
 /** A filter predicate that can be pushed down to page-level skipping */
 export interface FilterOp {
   column: string;
-  op: "eq" | "neq" | "gt" | "gte" | "lt" | "lte" | "in" | "is_null" | "is_not_null";
+  op: "eq" | "neq" | "gt" | "gte" | "lt" | "lte" | "in" | "between" | "is_null" | "is_not_null";
   value: number | bigint | string | (number | bigint | string)[];
 }
 

src/wasm-engine.ts

@@ -67,6 +67,9 @@ export interface WasmExports {
   filterFloat64Buffer(dataPtr: number, len: number, op: number, value: number, outPtr: number, maxOut: number): number;
   filterInt32Buffer(dataPtr: number, len: number, op: number, value: number, outPtr: number, maxOut: number): number;
   filterInt64Buffer(dataPtr: number, len: number, op: number, value: bigint, outPtr: number, maxOut: number): number;
+  filterFloat64Range(dataPtr: number, len: number, low: number, high: number, outPtr: number, maxOut: number): number;
+  filterInt32Range(dataPtr: number, len: number, low: number, high: number, outPtr: number, maxOut: number): number;
+  filterInt64Range(dataPtr: number, len: number, low: bigint, high: bigint, outPtr: number, maxOut: number): number;
   intersectIndices(aPtr: number, aLen: number, bPtr: number, bLen: number, outPtr: number, maxOut: number): number;
   unionIndices(aPtr: number, aLen: number, bPtr: number, bLen: number, outPtr: number, maxOut: number): number;
 

wasm/src/wasm/aggregates.zig

@@ -805,6 +805,48 @@ export fn filterFloat64Range(
     return out_count;
 }
 
+/// Filter int32 with range (BETWEEN): returns indices where low <= val <= high
+export fn filterInt32Range(
+    data_ptr: [*]const i32,
+    len: usize,
+    low: i32,
+    high: i32,
+    out_indices: [*]u32,
+    max_indices: usize,
+) usize {
+    var out_count: usize = 0;
+    for (0..len) |i| {
+        if (out_count >= max_indices) break;
+        const v = data_ptr[i];
+        if (v >= low and v <= high) {
+            out_indices[out_count] = @intCast(i);
+            out_count += 1;
+        }
+    }
+    return out_count;
+}
+
+/// Filter int64 with range (BETWEEN): returns indices where low <= val <= high
+export fn filterInt64Range(
+    data_ptr: [*]const i64,
+    len: usize,
+    low: i64,
+    high: i64,
+    out_indices: [*]u32,
+    max_indices: usize,
+) usize {
+    var out_count: usize = 0;
+    for (0..len) |i| {
+        if (out_count >= max_indices) break;
+        const v = data_ptr[i];
+        if (v >= low and v <= high) {
+            out_indices[out_count] = @intCast(i);
+            out_count += 1;
+        }
+    }
+    return out_count;
+}
+
 /// AND two index arrays (intersection)
 /// Uses O(n+m) sorted merge since filter outputs are always in ascending order
 export fn intersectIndices(