fix: 11 correctness bugs — page skip alignment, WASM bounds, filterGroups, cache ordering

teamchong · teamchong · commit 845538d36383 · 2026-03-09T22:08:01.000-04:00
Critical fixes:
- Per-column page skip misaligned rows across columns (query-do, fragment-do)
  Now uses uniform page-level skip: same decision for all columns per page
- parseWasmResult bounds check compared relative dp vs absolute end (wasm-engine)
- queryToSql silently dropped OR filterGroups (wasm-engine)
- Fragment-do page data out of order with partially warm cache
  Now tracks results by index and assembles in original order
- Missing filter/sort/groupBy/aggregate columns when projections set (query-do)
  Now computes neededNames from all query references
- Grace hash partition path allowed NULL=NULL matches (operators)
  Now skips NULL keys in both build and probe, matching in-memory path
- rewriteAggregatesAsColumns missed BETWEEN/IN/CASE/CAST nodes (evaluator)
  HAVING with these constructs silently dropped all groups
- Partial filter pushdown left stale filters on rollback (compiler)
- BigInt multi-column ORDER BY used string comparison (executor)
- count() fast path ignored join/vectorSearch/setOperation (local-executor)
- Cache key missing distinct/windows/computedColumns/setOp fields (query-do, local-executor)
diff --git a/src/fragment-do.ts b/src/fragment-do.ts
@@ -85,40 +85,57 @@ export class FragmentDO extends DurableObject<Env> {
         }
       }
 
-      // Build byte ranges for each page, skipping via min/max stats
+      // Build byte ranges for each page, skipping uniformly across all columns.
+      // A page is skipped only if any AND filter eliminates it — same for all columns.
+      const maxPages = cols.reduce((m, c) => Math.max(m, c.pages.length), 0);
+      const keptPageIndices: number[] = [];
+      for (let pi = 0; pi < maxPages; pi++) {
+        let skip = false;
+        if (!query.vectorSearch) {
+          for (const f of query.filters) {
+            const col = cols.find(c => c.name === f.column);
+            if (!col) continue;
+            const page = col.pages[pi];
+            if (!page) continue;
+            if (canSkipPage(page, [f], f.column)) { skip = true; break; }
+          }
+        }
+        if (skip) { totalPagesSkipped += cols.length; continue; }
+        keptPageIndices.push(pi);
+      }
+
       const ranges: { column: string; offset: number; length: number }[] = [];
       for (const col of cols) {
-        for (const page of col.pages) {
-          if (!query.vectorSearch && canSkipPage(page, query.filters, col.name)) {
-            totalPagesSkipped++;
-            continue;
-          }
+        for (const pi of keptPageIndices) {
+          const page = col.pages[pi];
+          if (!page) continue;
           ranges.push({ column: col.name, offset: Number(page.byteOffset), length: page.byteLength });
         }
       }
 
-      // Cache-before-fetch: check WASM buffer pool for each range
-      const columnData = new Map<string, ArrayBuffer[]>();
-      const uncachedRanges: typeof ranges = [];
+      // Cache-before-fetch: check WASM buffer pool for each range.
+      // Track results by index to preserve page ordering when cache is partially warm.
+      const rangeResults = new Array<ArrayBuffer | null>(ranges.length).fill(null);
+      const uncachedRanges: { range: typeof ranges[0]; idx: number }[] = [];
 
-      for (const r of ranges) {
+      for (let ri = 0; ri < ranges.length; ri++) {
+        const r = ranges[ri];
         const cacheKey = `${r2Key}:${r.offset}:${r.length}`;
         const cached = this.wasmEngine.cacheGet(cacheKey);
         if (cached) {
           totalCacheHits++;
-          const arr = columnData.get(r.column) ?? [];
-          arr.push(cached);
-          columnData.set(r.column, arr);
+          rangeResults[ri] = cached;
         } else {
           totalCacheMisses++;
-          uncachedRanges.push(r);
+          uncachedRanges.push({ range: r, idx: ri });
         }
       }
 
       // Fetch uncached ranges from R2 with retry + timeout
       const r2Start = Date.now();
       if (uncachedRanges.length > 0) {
-        const coalesced = coalesceRanges(uncachedRanges, autoCoalesceGap(uncachedRanges));
+        const fetchRanges = uncachedRanges.map(u => u.range);
+        const coalesced = coalesceRanges(fetchRanges, autoCoalesceGap(fetchRanges));
         const fetched = await fetchBounded(
           coalesced.map(c => () =>
             withRetry(() =>
@@ -135,20 +152,33 @@ export class FragmentDO extends DurableObject<Env> {
           ),
           8,
         );
+        // Build a lookup from (column, offset, length) → fetched slice
+        const fetchedMap = new Map<string, ArrayBuffer>();
         for (const f of fetched) {
           if (!f) continue;
           totalBytesRead += f.data.byteLength;
           for (const sub of f.ranges) {
             const slice = f.data.slice(sub.offset - f.offset, sub.offset - f.offset + sub.length);
-            const arr = columnData.get(sub.column) ?? [];
-            arr.push(slice);
-            columnData.set(sub.column, arr);
-
-            // Populate cache for next time
+            fetchedMap.set(`${sub.column}:${sub.offset}:${sub.length}`, slice);
             const cacheKey = `${r2Key}:${sub.offset}:${sub.length}`;
             this.wasmEngine.cacheSet(cacheKey, slice);
           }
         }
+        // Fill in uncached slots in order
+        for (const u of uncachedRanges) {
+          const key = `${u.range.column}:${u.range.offset}:${u.range.length}`;
+          rangeResults[u.idx] = fetchedMap.get(key) ?? null;
+        }
+      }
+
+      // Assemble columnData in correct page order
+      const columnData = new Map<string, ArrayBuffer[]>();
+      for (let ri = 0; ri < ranges.length; ri++) {
+        const buf = rangeResults[ri];
+        if (!buf) continue;
+        const arr = columnData.get(ranges[ri].column) ?? [];
+        arr.push(buf);
+        columnData.set(ranges[ri].column, arr);
       }
       totalR2ReadMs += Date.now() - r2Start;
 
@@ -161,7 +191,7 @@ export class FragmentDO extends DurableObject<Env> {
         .map(col => ({
           name: col.name, dtype: col.dtype, listDim: col.listDimension,
           pages: columnData.get(col.name)!,
-          pageInfos: col.pages,
+          pageInfos: keptPageIndices.map(pi => col.pages[pi]).filter(Boolean),
         }));
       if (!this.wasmEngine.registerColumns(fragTable, fragColEntries)) {
         throw new Error(`WASM OOM: failed to register columns for fragment "${r2Key}"`);
diff --git a/src/local-executor.ts b/src/local-executor.ts
@@ -182,7 +182,7 @@ export class LocalExecutor implements QueryExecutor {
   /** Count matching rows. No-filter case uses metadata only (zero I/O). */
   async count(query: QueryDescriptor): Promise<number> {
     const meta = await this.getOrLoadMeta(query.table);
-    if (query.filters.length === 0 && !query.filterGroups?.length && !query.aggregates?.length && !query.groupBy?.length && !query.distinct) {
+    if (query.filters.length === 0 && !query.filterGroups?.length && !query.aggregates?.length && !query.groupBy?.length && !query.distinct && !query.join && !query.vectorSearch && !query.setOperation && !query.subqueryIn && !query.computedColumns?.length) {
       return meta.columns[0]?.pages.reduce((s, p) => s + p.rowCount, 0) ?? 0;
     }
     // With filters: fall through to aggregate path
@@ -387,6 +387,10 @@ export class LocalExecutor implements QueryExecutor {
     if (query.groupBy) for (const g of query.groupBy) feed(g);
     if (query.distinct) feed("distinct");
     if (query.version !== undefined) feed(`v${query.version}`);
+    if (query.windows) for (const w of query.windows) { feed(w.fn); feed(w.alias); feed(w.column ?? ""); }
+    if (query.computedColumns) for (const cc of query.computedColumns) feed(cc.name);
+    if (query.setOperation) { feed(query.setOperation.type); feed(query.setOperation.table); }
+    if (query.subqueryIn) { feed(query.subqueryIn.column); feed(query.subqueryIn.table); }
     return `qr:${query.table}:${(h >>> 0).toString(36)}`;
   }
 
diff --git a/src/operators.ts b/src/operators.ts
@@ -2368,10 +2368,13 @@ export class HashJoinOperator implements Operator {
 
   /** Process one partition: load right → build map → probe left → emit matches. */
   private async processPartition(partIdx: number): Promise<Row[]> {
-    // Load right partition into hash map
+    // Load right partition into hash map (skip NULL keys — SQL NULL != NULL)
     const rightMap = new Map<string, Row[]>();
+    const rightNullRows: Row[] = [];
     for await (const row of this.streamPartition(this.rightPartitionIds[partIdx])) {
-      const key = this.toJoinKey(row[this.rightKey]);
+      const val = row[this.rightKey];
+      if (val === null || val === undefined) { rightNullRows.push(row); continue; }
+      const key = this.toJoinKey(val);
       const bucket = rightMap.get(key);
       if (bucket) bucket.push(row);
       else rightMap.set(key, [row]);
@@ -2380,10 +2383,15 @@ export class HashJoinOperator implements Operator {
     // Track matched right rows for right/full joins
     const matched = (this.joinType === "right" || this.joinType === "full") ? new Set<string>() : null;
 
-    // Probe with left partition
+    // Probe with left partition (skip NULL keys)
     const result: Row[] = [];
     for await (const leftRow of this.streamPartition(this.leftPartitionIds[partIdx])) {
-      const key = this.toJoinKey(leftRow[this.leftKey]);
+      const leftVal = leftRow[this.leftKey];
+      if (leftVal === null || leftVal === undefined) {
+        if (this.joinType === "left" || this.joinType === "full") result.push({ ...leftRow });
+        continue;
+      }
+      const key = this.toJoinKey(leftVal);
       const rightRows = rightMap.get(key);
       if (rightRows) {
         for (let i = 0; i < rightRows.length; i++) {
@@ -2395,7 +2403,7 @@ export class HashJoinOperator implements Operator {
       }
     }
 
-    // Emit unmatched right rows for right/full joins
+    // Emit unmatched right rows for right/full joins (including null-keyed right rows)
     if (matched) {
       for (const [key, rows] of rightMap) {
         for (let i = 0; i < rows.length; i++) {
@@ -2404,6 +2412,10 @@ export class HashJoinOperator implements Operator {
           }
         }
       }
+      // NULL-keyed right rows never match — always unmatched
+      for (const row of rightNullRows) {
+        result.push({ ...row });
+      }
     }
 
     return result;
diff --git a/src/query-do.ts b/src/query-do.ts
@@ -95,37 +95,46 @@ class EdgeScanOperator implements Operator {
     const query = this.query;
     const meta = this.meta;
 
-    let cols = query.projections.length > 0
-      ? meta.columns.filter(c => query.projections.includes(c.name))
-      : meta.columns;
-
-    if (query.vectorSearch) {
-      const vc = query.vectorSearch.column;
-      if (!cols.some(c => c.name === vc)) {
-        const ec = meta.columns.find(c => c.name === vc);
-        if (ec) cols = [...cols, ec];
-      }
+    // Determine columns to fetch: projections + all columns referenced by filters/sort/groupBy/aggregates
+    let neededNames: Set<string>;
+    if (query.projections.length > 0) {
+      neededNames = new Set(query.projections);
+      for (const f of query.filters) neededNames.add(f.column);
+      if (query.filterGroups) for (const g of query.filterGroups) for (const f of g) neededNames.add(f.column);
+      if (query.sortColumn) neededNames.add(query.sortColumn);
+      if (query.groupBy) for (const g of query.groupBy) neededNames.add(g);
+      if (query.aggregates) for (const a of query.aggregates) if (a.column !== "*") neededNames.add(a.column);
+    } else {
+      neededNames = new Set(meta.columns.map(c => c.name));
     }
+    if (query.vectorSearch) neededNames.add(query.vectorSearch.column);
+
+    let cols = meta.columns.filter(c => neededNames.has(c.name));
     this.cols = cols;
 
-    for (const col of cols) {
-      const keptPages: typeof col.pages = [];
-      for (const page of col.pages) {
-        if (!query.vectorSearch && canSkipPage(page, query.filters, col.name)) {
-          this.pagesSkipped++;
-          continue;
+    // Determine which pages to keep — must be uniform across all columns to avoid row misalignment.
+    // A page is skipped only if any AND filter eliminates it (same logic as canSkipPageMultiCol).
+    const maxPages = cols.reduce((m, c) => Math.max(m, c.pages.length), 0);
+    const keptPageIndices: number[] = [];
+    for (let pi = 0; pi < maxPages; pi++) {
+      let skip = false;
+      if (!query.vectorSearch) {
+        for (const f of query.filters) {
+          const col = cols.find(c => c.name === f.column);
+          if (!col) continue;
+          const page = col.pages[pi];
+          if (!page) continue;
+          if (canSkipPage(page, [f], f.column)) { skip = true; break; }
         }
-        keptPages.push(page);
       }
-      this.columnPageInfos.set(col.name, keptPages);
+      if (skip) { this.pagesSkipped += cols.length; continue; }
+      keptPageIndices.push(pi);
     }
 
-    // Count total pages from first column
-    const firstCol = cols[0];
-    if (firstCol) {
-      const keptPages = this.columnPageInfos.get(firstCol.name) ?? firstCol.pages;
-      this.pageCount = keptPages.length;
+    for (const col of cols) {
+      this.columnPageInfos.set(col.name, keptPageIndices.map(pi => col.pages[pi]).filter(Boolean));
     }
+    this.pageCount = keptPageIndices.length;
   }
 
   /**
@@ -670,6 +679,11 @@ export class QueryDO extends DurableObject<Env> {
     if (query.offset !== undefined) feed(String(query.offset));
     if (query.aggregates) for (const a of query.aggregates) { feed(a.fn); feed(a.column); if (a.alias) feed(a.alias); }
     if (query.groupBy) for (const g of query.groupBy) feed(g);
+    if (query.distinct) feed("distinct");
+    if (query.windows) for (const w of query.windows) { feed(w.fn); feed(w.alias); feed(w.column ?? ""); }
+    if (query.computedColumns) for (const cc of query.computedColumns) feed(cc.name);
+    if (query.setOperation) { feed(query.setOperation.type); feed(query.setOperation.table); }
+    if (query.subqueryIn) { feed(query.subqueryIn.column); feed(query.subqueryIn.table); }
     return `qr:${query.table}:${(h >>> 0).toString(36)}`;
   }
 
@@ -922,31 +936,52 @@ export class QueryDO extends DurableObject<Env> {
 
   /** Scan only the needed pages from R2 via coalesced Range reads, with cache-before-fetch. */
   private async scanPages(query: QueryDescriptor, meta: TableMeta, t0: number): Promise<QueryResult> {
-    let cols = query.projections.length > 0
-      ? meta.columns.filter(c => query.projections.includes(c.name))
-      : meta.columns;
-
-    if (query.vectorSearch) {
-      const vc = query.vectorSearch.column;
-      if (!cols.some(c => c.name === vc)) {
-        const ec = meta.columns.find(c => c.name === vc);
-        if (ec) cols = [...cols, ec];
-      }
+    // Determine columns to fetch: projections + all referenced by filters/sort/groupBy/aggregates
+    let neededNames: Set<string>;
+    if (query.projections.length > 0) {
+      neededNames = new Set(query.projections);
+      for (const f of query.filters) neededNames.add(f.column);
+      if (query.filterGroups) for (const g of query.filterGroups) for (const f of g) neededNames.add(f.column);
+      if (query.sortColumn) neededNames.add(query.sortColumn);
+      if (query.groupBy) for (const g of query.groupBy) neededNames.add(g);
+      if (query.aggregates) for (const a of query.aggregates) if (a.column !== "*") neededNames.add(a.column);
+    } else {
+      neededNames = new Set(meta.columns.map(c => c.name));
     }
+    if (query.vectorSearch) neededNames.add(query.vectorSearch.column);
+
+    let cols = meta.columns.filter(c => neededNames.has(c.name));
 
-    // Build per-page ranges, applying page-level skip.
-    // Track non-skipped page infos per column so buffer indices stay aligned.
+    // Build per-page ranges, applying page-level skip uniformly across all columns.
+    // A page is skipped only if any AND filter eliminates it — same decision for all columns
+    // to prevent row misalignment when different columns have different page counts.
     const ranges: { column: string; offset: number; length: number }[] = [];
     const columnPageInfos = new Map<string, typeof cols[0]["pages"]>();
     let pagesSkipped = 0;
+
+    const maxPages = cols.reduce((m, c) => Math.max(m, c.pages.length), 0);
+    const keptPageIndices: number[] = [];
+    for (let pi = 0; pi < maxPages; pi++) {
+      let skip = false;
+      if (!query.vectorSearch) {
+        for (const f of query.filters) {
+          const col = cols.find(c => c.name === f.column);
+          if (!col) continue;
+          const page = col.pages[pi];
+          if (!page) continue;
+          if (canSkipPage(page, [f], f.column)) { skip = true; break; }
+        }
+      }
+      if (skip) { pagesSkipped += cols.length; continue; }
+      keptPageIndices.push(pi);
+    }
+
     for (const col of cols) {
-      const keptPages: typeof col.pages = [];
-      for (const page of col.pages) {
-        if (!query.vectorSearch && canSkipPage(page, query.filters, col.name)) { pagesSkipped++; continue; }
-        keptPages.push(page);
+      const keptPages = keptPageIndices.map(pi => col.pages[pi]).filter(Boolean);
+      columnPageInfos.set(col.name, keptPages);
+      for (const page of keptPages) {
         ranges.push({ column: col.name, offset: Number(page.byteOffset), length: page.byteLength });
       }
-      columnPageInfos.set(col.name, keptPages);
     }
 
     // 3-tier cache hierarchy:
diff --git a/src/sql/compiler.ts b/src/sql/compiler.ts
@@ -71,8 +71,11 @@ export function compileFull(stmt: SelectStmt): SqlCompileResult {
   let whereExpr: SqlExpr | undefined;
   let filterGroups: FilterOp[][] | undefined;
   if (stmt.where) {
+    const savedLen = filters.length;
     const fullyFlattened = tryFlattenFilters(stmt.where, filters);
     if (!fullyFlattened) {
+      // Roll back any partially-pushed filters to avoid double-filtering
+      filters.length = savedLen;
       // Try OR decomposition: if top-level is OR, flatten each branch independently
       const orGroups = tryFlattenOrGroups(stmt.where);
       if (orGroups) {
diff --git a/src/sql/evaluator.ts b/src/sql/evaluator.ts
@@ -222,6 +222,31 @@ export function rewriteAggregatesAsColumns(expr: SqlExpr): SqlExpr {
       };
     case "unary":
       return { kind: "unary", op: expr.op, operand: rewriteAggregatesAsColumns(expr.operand) };
+    case "between":
+      return {
+        kind: "between",
+        expr: rewriteAggregatesAsColumns(expr.expr),
+        low: rewriteAggregatesAsColumns(expr.low),
+        high: rewriteAggregatesAsColumns(expr.high),
+      };
+    case "in_list":
+      return {
+        kind: "in_list", negated: expr.negated,
+        expr: rewriteAggregatesAsColumns(expr.expr),
+        values: expr.values.map(rewriteAggregatesAsColumns),
+      };
+    case "case_expr":
+      return {
+        kind: "case_expr",
+        operand: expr.operand ? rewriteAggregatesAsColumns(expr.operand) : undefined,
+        whenClauses: expr.whenClauses.map(w => ({
+          condition: rewriteAggregatesAsColumns(w.condition),
+          result: rewriteAggregatesAsColumns(w.result),
+        })),
+        elseResult: expr.elseResult ? rewriteAggregatesAsColumns(expr.elseResult) : undefined,
+      };
+    case "cast":
+      return { kind: "cast", expr: rewriteAggregatesAsColumns(expr.expr), targetType: expr.targetType };
     default:
       return expr;
   }
diff --git a/src/sql/executor.ts b/src/sql/executor.ts
diff --git a/src/wasm-engine.ts b/src/wasm-engine.ts
