fix: 3 correctness bugs — empty min/max Infinity, MaterializedExecutor missing aggs; deduplicate pipeline/agg/LIKE code (-136 lines)

Correctness:
- partial-agg resolveValue: return 0 for min/max when count=0 (was Infinity/-Infinity)
- WasmAggregateOperator: return null for min/max when count=0
- MaterializedExecutor: add count_distinct, stddev, variance, median, percentile (was silently null)

Code reduction:
- Extract assemblePipeline() shared by buildPipeline/buildEdgePipeline (-60 lines)
- Extract feedAggStates()/initStates() helpers in partial-agg (-45 lines)
- Deduplicate LIKE regex: evaluator.ts reuses compileLikeRegex from decode.ts (-15 lines)
- Deduplicate identityIndices: partial-agg imports from operators (-10 lines)

Author: teamchong

src/client.ts

@@ -942,7 +942,12 @@ export class MaterializedExecutor implements QueryExecutor {
           if (agg.fn === "count") {
             if (agg.column === "*") { out[alias] = bucket.length; }
             else { let cnt = 0; for (const r of bucket) if (r[agg.column] != null) cnt++; out[alias] = cnt; }
+          } else if (agg.fn === "count_distinct") {
+            const seen = new Set<string>();
+            for (const r of bucket) { const v = r[agg.column]; if (v != null) seen.add(String(v)); }
+            out[alias] = seen.size;
           } else {
+            const values: number[] = [];
             let sum = 0, count = 0, min = Infinity, max = -Infinity;
             for (const r of bucket) {
               const v = r[agg.column];
@@ -952,13 +957,37 @@ export class MaterializedExecutor implements QueryExecutor {
               sum += n; count++;
               if (n < min) min = n;
               if (n > max) max = n;
+              if (agg.fn === "stddev" || agg.fn === "variance" || agg.fn === "median" || agg.fn === "percentile") values.push(n);
             }
             if (count === 0) { out[alias] = null; continue; }
             switch (agg.fn) {
               case "sum": out[alias] = sum; break;
               case "avg": out[alias] = sum / count; break;
               case "min": out[alias] = min; break;
               case "max": out[alias] = max; break;
+              case "stddev": {
+                const mean = sum / count;
+                let sq = 0; for (const v of values) sq += (v - mean) ** 2;
+                out[alias] = Math.sqrt(sq / count); break;
+              }
+              case "variance": {
+                const mean = sum / count;
+                let sq = 0; for (const v of values) sq += (v - mean) ** 2;
+                out[alias] = sq / count; break;
+              }
+              case "median": {
+                values.sort((a, b) => a - b);
+                const mid = Math.floor(count / 2);
+                out[alias] = count % 2 ? values[mid] : (values[mid - 1] + values[mid]) / 2; break;
+              }
+              case "percentile": {
+                if (agg.percentileTarget === undefined) { out[alias] = null; break; }
+                values.sort((a, b) => a - b);
+                const p = agg.percentileTarget;
+                const idx = p * (count - 1);
+                const lo = Math.floor(idx), hi = Math.ceil(idx);
+                out[alias] = lo === hi ? values[lo] : values[lo] + (values[hi] - values[lo]) * (idx - lo); break;
+              }
               default: out[alias] = null;
             }
           }

src/decode.ts

@@ -529,7 +529,7 @@ function getInSet(values: readonly (number | bigint | string)[]): Set<number | b
 /** Cache compiled LIKE regexes — avoids re-compilation per row. */
 const likeRegexCache = new Map<string, RegExp>();
 
-function compileLikeRegex(pattern: string): RegExp {
+export function compileLikeRegex(pattern: string): RegExp {
   let cached = likeRegexCache.get(pattern);
   if (cached) return cached;
   // Escape regex metacharacters, then replace SQL wildcards

src/operators.ts

@@ -18,7 +18,7 @@ const _textEncoder = new TextEncoder();
 
 /** Cached identity index arrays to avoid repeated allocations on hot paths. */
 const _identityCache = new Map<number, Uint32Array>();
-function identityIndices(n: number): Uint32Array {
+export function identityIndices(n: number): Uint32Array {
   let cached = _identityCache.get(n);
   if (cached) return cached;
   cached = new Uint32Array(n);
@@ -2056,8 +2056,8 @@ export class WasmAggregateOperator implements Operator {
       switch (agg.fn) {
         case "sum": row[alias] = acc[i].sum; break;
         case "avg": row[alias] = acc[i].count === 0 ? 0 : acc[i].sum / acc[i].count; break;
-        case "min": row[alias] = acc[i].min; break;
-        case "max": row[alias] = acc[i].max; break;
+        case "min": row[alias] = acc[i].count === 0 ? null : acc[i].min; break;
+        case "max": row[alias] = acc[i].count === 0 ? null : acc[i].max; break;
         case "count": row[alias] = acc[i].count; break;
       }
     }
@@ -2902,88 +2902,7 @@ export function buildPipeline(
   }
 
   const scan = new ScanOperator(fragments, query, wasm, /* applyFilters */ true);
-  let pipeline: Operator = scan;
-  const memBudget = options?.memoryBudgetBytes ?? DEFAULT_MEMORY_BUDGET;
-
-  // Filter — skip if ScanOperator already applies filters via WASM SIMD
-  const hasFilters = query.filters.length > 0 || (query.filterGroups && query.filterGroups.length > 0);
-  if (hasFilters && !scan.filtersApplied) {
-    pipeline = new FilterOperator(pipeline, query.filters, query.filterGroups);
-  }
-
-  // SubqueryIn filters (pre-computed value sets)
-  if (query.subqueryIn) {
-    for (const sq of query.subqueryIn) {
-      pipeline = new SubqueryInOperator(pipeline, sq.column, sq.valueSet);
-    }
-  }
-
-  // Computed columns (in-process callbacks)
-  if (query.computedColumns) {
-    pipeline = new ComputedColumnOperator(pipeline, query.computedColumns);
-  }
-
-  // User-injected pipe stages (inserted after filter/computed, before agg/sort)
-  if (query.pipeStages) {
-    for (const stage of query.pipeStages) {
-      pipeline = stage(pipeline);
-    }
-  }
-
-  // Window functions
-  if (query.windows && query.windows.length > 0) {
-    pipeline = new WindowOperator(pipeline, query.windows);
-  }
-
-  // Distinct
-  if (query.distinct) {
-    pipeline = new DistinctOperator(pipeline, query.distinct);
-  }
-
-  const hasAgg = query.aggregates && query.aggregates.length > 0;
-
-  if (hasAgg) {
-    // Aggregate: consumes all filtered rows, produces aggregate output
-    pipeline = new AggregateOperator(pipeline, query);
-
-    // Sort/limit on aggregate output (aggregate output is small — always in-memory)
-    if (query.sortColumn && query.limit !== undefined) {
-      pipeline = new TopKOperator(
-        pipeline, query.sortColumn, query.sortDirection === "desc",
-        query.limit, query.offset ?? 0,
-      );
-    } else if (query.sortColumn) {
-      pipeline = new InMemorySortOperator(
-        pipeline, query.sortColumn, query.sortDirection === "desc",
-        query.offset ?? 0,
-      );
-    } else if (query.offset || query.limit !== undefined) {
-      pipeline = new LimitOperator(pipeline, query.limit ?? Infinity, query.offset ?? 0);
-    }
-  } else if (query.sortColumn) {
-    // Sort path
-    if (query.limit !== undefined) {
-      pipeline = new TopKOperator(
-        pipeline, query.sortColumn, query.sortDirection === "desc",
-        query.limit, query.offset ?? 0,
-      );
-    } else {
-      // Full sort without limit — use external sort to handle TB+ data
-      pipeline = new ExternalSortOperator(
-        pipeline, query.sortColumn, query.sortDirection === "desc",
-        query.offset ?? 0, memBudget, options?.spill,
-      );
-    }
-  } else if (query.offset || query.limit !== undefined) {
-    // No sort — streaming limit with offset
-    pipeline = new LimitOperator(pipeline, query.limit ?? Infinity, query.offset ?? 0);
-  }
-
-  // Project — strip extra columns at the end
-  if (outputColumns.length > 0) {
-    pipeline = new ProjectOperator(pipeline, outputColumns);
-  }
-
+  const pipeline = assemblePipeline(scan, query, outputColumns, scan.filtersApplied, options);
   return { pipeline, scan };
 }
 
@@ -2997,43 +2916,51 @@ export function buildEdgePipeline(
   outputColumns: string[],
   options?: PipelineOptions,
 ): Operator {
-  let pipeline: Operator = scan;
+  const filtersApplied = "filtersApplied" in scan && (scan as { filtersApplied: boolean }).filtersApplied;
+  // Skip projection when aggregation is active — aggregate output columns
+  // (aliases like "count_*", "sum_value") don't match the original table columns,
+  // so projecting to table columns would strip all aggregate results.
+  const hasAgg = query.aggregates && query.aggregates.length > 0;
+  const projectCols = hasAgg ? [] : outputColumns;
+  return assemblePipeline(scan, query, projectCols, filtersApplied, options);
+}
+
+/** Shared pipeline assembly: filter → subquery → computed → pipe → window → distinct → agg → sort → project */
+function assemblePipeline(
+  source: Operator,
+  query: QueryDescriptor,
+  outputColumns: string[],
+  filtersApplied: boolean,
+  options?: PipelineOptions,
+): Operator {
+  let pipeline: Operator = source;
   const memBudget = options?.memoryBudgetBytes ?? DEFAULT_MEMORY_BUDGET;
 
-  // 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;
-  const edgeHasFilters = query.filters.length > 0 || (query.filterGroups && query.filterGroups.length > 0);
-  if (edgeHasFilters && !scanHandlesFilters) {
+  const hasFilters = query.filters.length > 0 || (query.filterGroups && query.filterGroups.length > 0);
+  if (hasFilters && !filtersApplied) {
     pipeline = new FilterOperator(pipeline, query.filters, query.filterGroups);
   }
 
-  // SubqueryIn filters
   if (query.subqueryIn) {
     for (const sq of query.subqueryIn) {
       pipeline = new SubqueryInOperator(pipeline, sq.column, sq.valueSet);
     }
   }
 
-  // Computed columns
   if (query.computedColumns) {
     pipeline = new ComputedColumnOperator(pipeline, query.computedColumns);
   }
 
-  // User-injected pipe stages (inserted after filter/computed, before agg/sort)
   if (query.pipeStages) {
     for (const stage of query.pipeStages) {
       pipeline = stage(pipeline);
     }
   }
 
-  // Window functions
   if (query.windows && query.windows.length > 0) {
     pipeline = new WindowOperator(pipeline, query.windows);
   }
 
-  // Distinct
   if (query.distinct) {
     pipeline = new DistinctOperator(pipeline, query.distinct);
   }
@@ -3071,10 +2998,7 @@ export function buildEdgePipeline(
     pipeline = new LimitOperator(pipeline, query.limit ?? Infinity, query.offset ?? 0);
   }
 
-  // Skip projection when aggregation is active — aggregate output columns
-  // (aliases like "count_*", "sum_value") don't match the original table columns,
-  // so projecting to table columns would strip all aggregate results.
-  if (!hasAgg && outputColumns.length > 0) {
+  if (outputColumns.length > 0) {
     pipeline = new ProjectOperator(pipeline, outputColumns);
   }