fix: IN/NOT_IN bigint cross-type, WindowSpec.column extraction, null vs empty string in DISTINCT/SET/WINDOW; perf: bitmap bitwise, columnar Map.get() hoist, streaming aggs, lazy IN eval; docs accuracy

Author: teamchong

README.md

@@ -93,7 +93,8 @@ Operation         Operator class              What it does
 ─────────         ──────────────              ────────────
 
 Filtering
-  predicate       FilterOperator              eq, neq, gt, gte, lt, lte, in
+  predicate       FilterOperator              14 ops: eq, neq, gt, gte, lt, lte, in, not_in,
+                                              between, not_between, like, not_like, is_null, is_not_null
   membership      SubqueryInOperator          Semi-join filter against a value set
 
 Projection
@@ -125,7 +126,7 @@ Set operations
 
 Limiting
   limit/offset    LimitOperator               Row limiting with offset
-  sample          (planned)                   Random sampling
+  sample          DataFrame.sample()            Random sampling (Fisher-Yates)
 
 Similarity
   vector near     (planned)                   NEAR topK as composable operator — currently in scan layer
@@ -252,7 +253,7 @@ npx tsx examples/nextjs-api-route.ts
 - **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 100 slots per datacenter)
-- **580+ tests** — unit tests cover footer parsing, column decoding, Parquet/Thrift, merging, aggregates, VIP cache, WASM integration, SQL, partition catalog, materialized executor, toCode decompiler; 110+ conformance tests validate every operator against DuckDB at 1M-5M row scale
+- **600+ tests** — unit tests cover footer parsing, column decoding, Parquet/Thrift, merging, aggregates, VIP cache, WASM integration, SQL, partition catalog, materialized executor, toCode decompiler; 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/architecture.mdx

@@ -179,6 +179,6 @@ Tests run in two runtimes:
 |---------|------|-------|
 | **workerd** (real CF Workers) | Operators, DOs, decode, format parsing | — |
 | **Node** | DuckDB conformance (1M-5M rows), fixture files, SQL, infra | — |
-| **Total** | All runtimes | **580+ tests** |
+| **Total** | All runtimes | **600+ tests** |
 
 Conformance tests validate every operator against DuckDB at scale. CI benchmarks compare QueryMode vs DuckDB on every push.

docs/src/content/docs/dataframe-api.mdx

@@ -120,8 +120,8 @@ df1.except(df2)
 df.where("age", "gt", 25)                 // alias for filter()
 df.whereIn("status", ["active", "trial"]) // shorthand for in
 df.whereNotIn("role", ["banned"])          // shorthand for not_in
-df.whereBetween("score", [80, 100])       // shorthand for between
-df.whereNotBetween("age", [0, 17])        // shorthand for not_between
+df.whereBetween("score", 80, 100)         // shorthand for between
+df.whereNotBetween("age", 0, 17)          // shorthand for not_between
 df.whereLike("name", "%alice%")           // shorthand for like
 df.whereNotLike("email", "%spam%")        // shorthand for not_like
 df.filterIn("region", subquery)           // IN subquery (semi-join)

docs/src/content/docs/deployment.mdx

@@ -108,7 +108,8 @@ pnpm build && wrangler deploy
     { "column": "amount", "op": "gt", "value": 100 }
   ],
   "projections": ["id", "amount", "region"],
-  "orderBy": { "column": "amount", "desc": true },
+  "sortColumn": "amount",
+  "sortDirection": "desc",
   "limit": 20
 }
 ```

src/decode.ts

@@ -228,29 +228,16 @@ export function decodePage(
   rowCount = 0,
   dataOffsetInPage?: number,
 ): (number | bigint | string | null)[] {
-  let nulls: Set<number> | null = null;
+  let nullBitmap: Uint8Array | null = null;
+  let nulls: { has(i: number): boolean; size: number } | null = null;
   if (nullCount > 0 && rowCount > 0) {
     const bitmapBytes = Math.ceil(rowCount / 8);
     if (buf.byteLength < bitmapBytes) return [];
-    const bytes = new Uint8Array(buf, 0, bitmapBytes);
-    nulls = new Set<number>();
-    // Fast path: skip bytes that are 0xFF (all valid) — avoids per-bit work
-    for (let b = 0; b < bitmapBytes; b++) {
-      const byte = bytes[b];
-      if (byte === 0xFF) continue; // all 8 bits valid, skip
-      const base = b << 3;
-      if (byte === 0x00) {
-        // all 8 bits null — add them all
-        const end = Math.min(base + 8, rowCount);
-        for (let i = base; i < end; i++) nulls.add(i);
-      } else {
-        // mixed — check each bit
-        const end = Math.min(8, rowCount - base);
-        for (let bit = 0; bit < end; bit++) {
-          if (((byte >> bit) & 1) === 0) nulls.add(base + bit);
-        }
-      }
-    }
+    nullBitmap = new Uint8Array(buf, 0, bitmapBytes);
+    nulls = {
+      has(i: number): boolean { return (nullBitmap![i >> 3] & (1 << (i & 7))) === 0; },
+      get size() { return nullCount; },
+    };
     // Lance v2 uses alignment padding between bitmap and data.
     // dataOffsetInPage gives the exact data start; otherwise strip only bitmap bytes.
     const stripBytes = dataOffsetInPage ?? bitmapBytes;
@@ -483,12 +470,19 @@ export function matchesFilter(
     case "in": {
       if (!Array.isArray(t)) return false;
       const set = getInSet(t);
-      return set.has(val);
+      if (set.has(val)) return true;
+      // bigint/number cross-type: 5n !== 5 for Set.has, so coerce
+      if (typeof val === "bigint") return set.has(Number(val));
+      if (typeof val === "number") return set.has(BigInt(val));
+      return false;
     }
     case "not_in": {
       if (!Array.isArray(t)) return false;
       const set = getInSet(t);
-      return !set.has(val);
+      if (set.has(val)) return false;
+      if (typeof val === "bigint") return !set.has(Number(val));
+      if (typeof val === "number") return !set.has(BigInt(val));
+      return true;
     }
     case "between": {
       if (!Array.isArray(t) || t.length !== 2) return false;

src/index.ts

@@ -21,13 +21,14 @@ export {
   ComputedColumnOperator, HashJoinOperator, ExternalSortOperator,
   InMemorySortOperator, WindowOperator, DistinctOperator, SetOperator,
   LimitOperator, SubqueryInOperator,
-  drainPipeline, buildEdgePipeline,
+  drainPipeline, buildEdgePipeline, FsSpillBackend,
 } from "./operators.js";
 export type { Operator, RowBatch } from "./operators.js";
 export { QueryModeError } from "./errors.js";
 export type { ErrorCode } from "./errors.js";
 export { LocalExecutor } from "./local-executor.js";
 export { bigIntReplacer } from "./decode.js";
+export { R2SpillBackend } from "./r2-spill.js";
 export { createFromJSON, createFromCSV, createDemo } from "./convenience.js";
 export { formatResultSummary, formatExplain, formatBytes } from "./format.js";
 export type { LocalTimingInfo } from "./format.js";

src/operators.ts

@@ -929,7 +929,8 @@ export class WindowOperator implements Operator {
         key = "";
         for (let p = 0; p < win.partitionBy.length; p++) {
           if (p > 0) key += "\x00";
-          key += String(rows[i][win.partitionBy[p]] ?? "");
+          const v = rows[i][win.partitionBy[p]];
+          key += v === null || v === undefined ? "\x01NULL\x01" : String(v);
         }
       } else {
         key = "__all__";
@@ -1141,7 +1142,8 @@ export class DistinctOperator implements Operator {
         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] ?? "") : "");
+          const v = vals ? vals[idx] : null;
+          key += v === null || v === undefined ? "\x01NULL\x01" : String(v);
         }
         if (!this.seen.has(key)) {
           this.seen.add(key);
@@ -1165,7 +1167,8 @@ export class DistinctOperator implements Operator {
         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]] ?? "");
+          const v = row[keyCols[g]];
+          key += v === null || v === undefined ? "\x01NULL\x01" : String(v);
         }
         if (!this.seen.has(key)) {
           this.seen.add(key);
@@ -1205,12 +1208,15 @@ export class SetOperator implements Operator {
     if (mode !== "union_all") this.seen = new Set();
   }
 
+  private _sortedKeys: string[] | null = null;
   private rowKey(row: Row): string {
-    const keys = Object.keys(row).sort();
+    if (!this._sortedKeys) this._sortedKeys = Object.keys(row).sort();
+    const keys = this._sortedKeys;
     let result = "";
     for (let i = 0; i < keys.length; i++) {
       if (i > 0) result += "\x00";
-      result += keys[i] + "=" + String(row[keys[i]] ?? "");
+      const v = row[keys[i]];
+      result += keys[i] + "=" + (v === null || v === undefined ? "\x01NULL\x01" : String(v));
     }
     return result;
   }
@@ -1476,29 +1482,27 @@ export class LimitOperator implements Operator {
       const batch = await this.upstream.nextColumnar();
       if (!batch) return null;
 
-      let indices = batch.selection
-        ? Array.from(batch.selection)
-        : Array.from({ length: batch.rowCount }, (_, i) => i);
+      let sel = batch.selection ?? identityIndices(batch.rowCount);
 
       // Handle offset: skip rows
       if (this.skipped < this.offset) {
         const toSkip = this.offset - this.skipped;
-        if (toSkip >= indices.length) {
-          this.skipped += indices.length;
+        if (toSkip >= sel.length) {
+          this.skipped += sel.length;
           continue;
         }
-        indices = indices.slice(toSkip);
+        sel = sel.subarray(toSkip);
         this.skipped = this.offset;
       }
 
       // Apply limit
-      if (indices.length > this.remaining) {
-        indices = indices.slice(0, this.remaining);
+      if (sel.length > this.remaining) {
+        sel = sel.slice(0, this.remaining);
       }
-      this.remaining -= indices.length;
+      this.remaining -= sel.length;
 
-      if (indices.length > 0) {
-        return { columns: batch.columns, rowCount: batch.rowCount, selection: new Uint32Array(indices) };
+      if (sel.length > 0) {
+        return { columns: batch.columns, rowCount: batch.rowCount, selection: sel };
       }
       return null;
     }

src/partial-agg.ts

@@ -211,14 +211,13 @@ export function computePartialAggColumnar(
     const states = aggregates.map((agg) =>
       initPartialAggState(agg.fn, agg.column, { percentileTarget: agg.percentileTarget }),
     );
+    const aggArrays = aggregates.map(a => a.column === "*" ? null : (batch.columns.get(a.column) ?? null));
     for (const idx of indices) {
       for (let i = 0; i < aggregates.length; i++) {
-        const col = aggregates[i].column;
-        if (col === "*") {
+        if (!aggArrays[i]) {
           states[i].count++;
         } else {
-          const vals = batch.columns.get(col);
-          const val = vals ? vals[idx] : null;
+          const val = aggArrays[i]![idx];
           if (typeof val === "number") {
             updatePartialAgg(states[i], val, val);
           } else if (typeof val === "bigint") {
@@ -238,13 +237,14 @@ export function computePartialAggColumnar(
 
   const groups = new Map<string, PartialAggState[]>();
   const groupCols = query.groupBy;
+  const groupArrays = groupCols.map(c => batch.columns.get(c) ?? null);
+  const aggArrays2 = aggregates.map(a => a.column === "*" ? null : (batch.columns.get(a.column) ?? null));
 
   for (const idx of indices) {
     let key = "";
     for (let g = 0; g < groupCols.length; g++) {
       if (g > 0) key += "\x00";
-      const vals = batch.columns.get(groupCols[g]);
-      const v = vals ? vals[idx] : null;
+      const v = groupArrays[g] ? groupArrays[g]![idx] : null;
       key += v === null || v === undefined ? "\x01NULL\x01" : String(v);
     }
     let states = groups.get(key);
@@ -255,12 +255,10 @@ export function computePartialAggColumnar(
       groups.set(key, states);
     }
     for (let i = 0; i < aggregates.length; i++) {
-      const col = aggregates[i].column;
-      if (col === "*") {
+      if (!aggArrays2[i]) {
         states[i].count++;
       } else {
-        const vals = batch.columns.get(col);
-        const val = vals ? vals[idx] : null;
+        const val = aggArrays2[i]![idx];
         if (typeof val === "number") {
           updatePartialAgg(states[i], val, val);
         } else if (typeof val === "bigint") {

src/sql/compiler.ts

@@ -216,6 +216,15 @@ function compileWindowFunction(expr: SqlExpr & { kind: "call" }, alias?: string)
     alias: alias ?? `${fn}_result`,
   };
 
+  if (expr.args.length > 0) {
+    const colArg = expr.args[0];
+    if (colArg.kind === "column") {
+      spec.column = colArg.name;
+    } else if (colArg.kind === "star") {
+      spec.column = "*";
+    }
+  }
+
   if ((fn === "lag" || fn === "lead") && expr.args.length > 1) {
     const args: WindowSpec["args"] = {};
     if (expr.args[1]?.kind === "value" && expr.args[1].value.type === "integer") {

src/sql/evaluator.ts

@@ -28,8 +28,10 @@ export function evaluateExpr(expr: SqlExpr, row: Row): unknown {
     case "in_list": {
       const val = evaluateExpr(expr.expr, row);
       if (val === null) return null;
-      const values = expr.values.map(v => evaluateExpr(v, row));
-      const found = values.some(v => looseEqual(val, v));
+      let found = false;
+      for (const v of expr.values) {
+        if (looseEqual(val, evaluateExpr(v, row))) { found = true; break; }
+      }
       return expr.negated ? !found : found;
     }