feat: BETWEEN in WasmAggregateOperator + two-phase aggregate I/O

- WasmAggregateOperator handles BETWEEN via filterFloat64Range/Int32/Int64Range
- canUseWasmAggregate allows "between" filter op on numeric columns
- Two-phase I/O in WasmAggregateOperator: read only filter columns first,
  skip aggregate column I/O when filter eliminates all rows on a page

Author: teamchong

src/operators.ts

@@ -1373,26 +1373,26 @@ export class WasmAggregateOperator implements Operator {
     for (const frag of this.fragments) {
       const colMap = new Map(frag.columns.map(c => [c.name, c]));
 
-      // Collect all columns needed (aggregate + filter)
-      const neededCols = new Set<string>();
-      for (const agg of aggregates) if (agg.column !== "*") neededCols.add(agg.column);
-      for (const f of filters) neededCols.add(f.column);
+      // Separate filter columns from aggregate-only columns
+      const filterColNames = new Set(filters.map(f => f.column));
+      const aggOnlyColNames = new Set<string>();
+      for (const agg of aggregates) {
+        if (agg.column !== "*" && !filterColNames.has(agg.column)) aggOnlyColNames.add(agg.column);
+      }
 
-      // Process page by page
       const firstCol = frag.columns[0];
       if (!firstCol) continue;
       const pageCount = firstCol.pages.length;
 
       for (let pi = 0; pi < pageCount; pi++) {
-        // Page-level skip via min/max stats — check all filter columns
         if (canSkipPageMultiCol(frag.columns, pi, filters)) {
           this.pagesSkipped++;
           continue;
         }
 
-        // Read needed columns for this page
+        // Phase 1: Read filter columns (+ any aggregate columns that are also filter columns)
         const pageBuffers = new Map<string, ArrayBuffer>();
-        for (const colName of neededCols) {
+        for (const colName of filterColNames) {
           const col = colMap.get(colName);
           if (!col || !col.pages[pi]) continue;
           const buf = await frag.readPage(col, col.pages[pi]);
@@ -1468,6 +1468,18 @@ export class WasmAggregateOperator implements Operator {
           matchCount = currentIndices.length;
         }
 
+        // Phase 2: Read aggregate-only columns (skipped if filter eliminated all rows)
+        if (!hasFilters || matchCount > 0) {
+          for (const colName of aggOnlyColNames) {
+            if (pageBuffers.has(colName)) continue; // already read (also a filter column)
+            const col = colMap.get(colName);
+            if (!col || !col.pages[pi]) continue;
+            const buf = await frag.readPage(col, col.pages[pi]);
+            this.bytesRead += buf.byteLength;
+            pageBuffers.set(colName, buf);
+          }
+        }
+
         // Aggregate per column
         for (let ai = 0; ai < aggregates.length; ai++) {
           const agg = aggregates[ai];