feat: extend WASM compute pushdown to edge pipeline

EdgeScanOperator Parquet path (src/query-do.ts):
- Register decoded columns in WASM and use executeQuery() for SIMD
  filter/sort instead of JS row-by-row assembly
- Mark filtersApplied=true so buildEdgePipeline skips FilterOperator

buildEdgePipeline (src/operators.ts):
- Check for filtersApplied flag on scan operator, skip FilterOperator
  when scan already handles filters via WASM

unionIndices (wasm/src/wasm/aggregates.zig):
- O(n+m) sorted merge (was O(n*m) nested loop with linear search)

Author: teamchong

src/operators.ts

@@ -2162,7 +2162,11 @@ export function buildEdgePipeline(
   let pipeline: Operator = scan;
   const memBudget = options?.memoryBudgetBytes ?? DEFAULT_MEMORY_BUDGET;
 
-  if (query.filters.length > 0) {
+  // Skip FilterOperator if the scan operator already applies filters
+  // (e.g. EdgeScanOperator Lance path uses WASM SQL with WHERE,
+  //  EdgeScanOperator Parquet path now uses WASM executeQuery with filters)
+  const scanHandlesFilters = "filtersApplied" in scan && (scan as { filtersApplied: boolean }).filtersApplied;
+  if (query.filters.length > 0 && !scanHandlesFilters) {
     pipeline = new FilterOperator(pipeline, query.filters);
   }
 

src/query-do.ts

@@ -64,6 +64,8 @@ class EdgeScanOperator implements Operator {
   edgeCacheMisses = 0;
   r2ReadMs = 0;
   wasmExecMs = 0;
+  /** Filters are applied inside WASM executeQuery — skip downstream FilterOperator. */
+  filtersApplied = true;
 
   constructor(
     bucket: R2Bucket,
@@ -328,22 +330,49 @@ class EdgeScanOperator implements Operator {
       decodedColumns.set(col.name, decoded);
     }
 
-    // Assemble rows for this single page
+    // Try WASM SQL path: register decoded columns → executeQuery (SIMD filter/sort)
     let numRows = 0;
     for (const v of decodedColumns.values()) if (v.length > numRows) numRows = v.length;
     if (numRows === 0) return this.next(); // skip empty pages
 
-    const rows: Row[] = [];
-    for (let i = 0; i < numRows; i++) {
-      const row: Row = {};
-      for (const name of colNames) {
-        const vals = decodedColumns.get(name);
-        row[name] = vals && i < vals.length ? vals[i] : null;
+    this.wasmEngine.exports.resetHeap();
+    const fragTable = `__edge_pq_${pi}`;
+    let wasmRegistered = true;
+    for (const col of cols) {
+      const values = decodedColumns.get(col.name);
+      if (!values?.length) continue;
+      if (!this.wasmEngine.registerDecodedColumn(fragTable, col.name, col.dtype, values)) {
+        wasmRegistered = false;
+        break;
       }
-      rows.push(row);
     }
+
+    let rows: Row[];
+    if (wasmRegistered) {
+      const pqQuery: QueryDescriptor = {
+        ...this.query,
+        table: fragTable,
+        sortColumn: undefined, limit: undefined, offset: undefined,
+        aggregates: undefined, join: undefined,
+      };
+      const wasmRows = this.wasmEngine.executeQuery(pqQuery);
+      this.wasmEngine.clearTable(fragTable);
+      rows = wasmRows ?? [];
+    } else {
+      // Fallback: assemble rows in JS
+      rows = [];
+      for (let i = 0; i < numRows; i++) {
+        const row: Row = {};
+        for (const name of colNames) {
+          const vals = decodedColumns.get(name);
+          row[name] = vals && i < vals.length ? vals[i] : null;
+        }
+        rows.push(row);
+      }
+    }
+
     this.wasmExecMs += Date.now() - wasmStart;
-    return rows;
+    return rows.length > 0 ? rows : this.next();
   }
 
   async close(): Promise<void> {

wasm/src/wasm/aggregates.zig

@@ -775,6 +775,7 @@ export fn intersectIndices(
 }
 
 /// OR two index arrays (union, deduplicated)
+/// Uses O(n+m) sorted merge since filter outputs are always in ascending order
 export fn unionIndices(
     a: [*]const u32,
     a_len: usize,
@@ -784,23 +785,39 @@ export fn unionIndices(
     max_out: usize,
 ) usize {
     var out_count: usize = 0;
+    var i: usize = 0;
+    var j: usize = 0;
 
-    // Add all from a
-    for (0..a_len) |i| {
+    while (i < a_len and j < b_len) {
         if (out_count >= max_out) break;
+        if (a[i] == b[j]) {
+            out[out_count] = a[i];
+            out_count += 1;
+            i += 1;
+            j += 1;
+        } else if (a[i] < b[j]) {
+            out[out_count] = a[i];
+            out_count += 1;
+            i += 1;
+        } else {
+            out[out_count] = b[j];
+            out_count += 1;
+            j += 1;
+        }
+    }
+
+    // Drain remaining from a
+    while (i < a_len and out_count < max_out) {
         out[out_count] = a[i];
         out_count += 1;
+        i += 1;
     }
 
-    // Add from b if not already in out
-    outer: for (0..b_len) |i| {
-        if (out_count >= max_out) break;
-        const b_val = b[i];
-        for (0..out_count) |j| {
-            if (out[j] == b_val) continue :outer;
-        }
-        out[out_count] = b_val;
+    // Drain remaining from b
+    while (j < b_len and out_count < max_out) {
+        out[out_count] = b[j];
         out_count += 1;
+        j += 1;
     }
 
     return out_count;