perf: skip flatMap copy for single-shard merge, cache distinct column keys

- merge.ts: single-shard results with no offset/limit now skip the
  flatMap copy entirely — avoids doubling peak memory in the QueryDO
  (128MB ceiling) for the common case of one fragment result.

- operators.ts: DistinctOperator.nextColumnar caches the resolved
  column list (from batch.columns.keys()) across batches instead of
  re-allocating Array.from() on every batch for DISTINCT(*) queries.

Author: teamchong

src/merge.ts

@@ -224,12 +224,17 @@ export function mergeQueryResults(
   // Fallback: mixed or row-based
   for (const p of partials) ensureRows(p);
   let rows: Row[];
-  const allRows = partials.flatMap((p) => p.rows);
   const off = query.offset ?? 0;
-  if (off > 0 || query.limit !== undefined) {
-    rows = allRows.slice(off, query.limit !== undefined ? off + query.limit : undefined);
+  if (partials.length === 1 && off === 0 && query.limit === undefined) {
+    // Single shard, no slicing — avoid flatMap copy
+    rows = partials[0].rows;
   } else {
-    rows = allRows;
+    const allRows = partials.flatMap((p) => p.rows);
+    if (off > 0 || query.limit !== undefined) {
+      rows = allRows.slice(off, query.limit !== undefined ? off + query.limit : undefined);
+    } else {
+      rows = allRows;
+    }
   }
 
   return { rows, rowCount: rows.length, columns, ...baseResult };

src/operators.ts

@@ -1077,6 +1077,7 @@ export class DistinctOperator implements Operator {
   private upstream: Operator;
   private columns: string[];
   private seen = new Set<string>();
+  private _resolvedCols: string[] | null = null;
 
   constructor(upstream: Operator, columns: string[]) {
     this.upstream = upstream;
@@ -1100,7 +1101,7 @@ export class DistinctOperator implements Operator {
 
       const indices = batch.selection ?? identityIndices(batch.rowCount);
       const kept: number[] = [];
-      const cols = this.columns.length > 0 ? this.columns : Array.from(batch.columns.keys());
+      const cols = this.columns.length > 0 ? this.columns : (this._resolvedCols ??= Array.from(batch.columns.keys()));
       // Pre-extract column arrays to avoid Map.get per column per row
       const colArrays: ((number | bigint | string | boolean | null)[] | undefined)[] = new Array(cols.length);
       for (let g = 0; g < cols.length; g++) colArrays[g] = batch.columns.get(cols[g]) ?? undefined;