fix: neededNames missing distinct/windows/join/subqueryIn columns across all scan paths

All four scan sites (query-do x2, fragment-do, local-executor x2) now
include window partition/order/target columns, distinct columns, join
leftKey, and subqueryIn columns when computing which columns to fetch.
Previously these were only covered when projections were empty (all
columns). With explicit projections, these referenced columns could be
missing from the scan, causing silent data loss or incorrect results.

Author: teamchong

src/fragment-do.ts

@@ -82,6 +82,14 @@ export class FragmentDO extends DurableObject<Env> {
         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);
+        if (query.distinct) for (const d of query.distinct) neededNames.add(d);
+        if (query.windows) for (const w of query.windows) {
+          if (w.column) neededNames.add(w.column);
+          for (const p of w.partitionBy) neededNames.add(p);
+          for (const o of w.orderBy) neededNames.add(o.column);
+        }
+        if (query.join) { neededNames.add(query.join.leftKey); }
+        if (query.subqueryIn) for (const sq of query.subqueryIn) neededNames.add(sq.column);
       } else {
         neededNames = new Set(effectiveMeta.columns.map(c => c.name));
       }

src/local-executor.ts

@@ -472,13 +472,21 @@ export class LocalExecutor implements QueryExecutor {
       }
     }
 
-    // Step 2: Determine columns to fetch (projection + filter + sort + groupBy + aggregate columns)
+    // Step 2: Determine columns to fetch (projection + all referenced columns)
     const neededColumns = new Set(query.projections.length > 0 ? query.projections : columns.map(c => c.name));
     for (const f of query.filters) neededColumns.add(f.column);
     if (query.filterGroups) for (const g of query.filterGroups) for (const f of g) neededColumns.add(f.column);
     if (query.sortColumn) neededColumns.add(query.sortColumn);
     if (query.groupBy) for (const g of query.groupBy) neededColumns.add(g);
     if (query.aggregates) for (const a of query.aggregates) if (a.column !== "*") neededColumns.add(a.column);
+    if (query.distinct) for (const d of query.distinct) neededColumns.add(d);
+    if (query.windows) for (const w of query.windows) {
+      if (w.column) neededColumns.add(w.column);
+      for (const p of w.partitionBy) neededColumns.add(p);
+      for (const o of w.orderBy) neededColumns.add(o.column);
+    }
+    if (query.join) { neededColumns.add(query.join.leftKey); }
+    if (query.subqueryIn) for (const sq of query.subqueryIn) neededColumns.add(sq.column);
     if (query.vectorSearch) neededColumns.add(query.vectorSearch.column);
     const projectedColumns = columns.filter(c => neededColumns.has(c.name));
 
@@ -928,6 +936,14 @@ export class LocalExecutor implements QueryExecutor {
       if (query.sortColumn) neededCols.add(query.sortColumn);
       if (query.groupBy) for (const g of query.groupBy) neededCols.add(g);
       if (query.aggregates) for (const a of query.aggregates) if (a.column !== "*") neededCols.add(a.column);
+      if (query.distinct) for (const d of query.distinct) neededCols.add(d);
+      if (query.windows) for (const w of query.windows) {
+        if (w.column) neededCols.add(w.column);
+        for (const p of w.partitionBy) neededCols.add(p);
+        for (const o of w.orderBy) neededCols.add(o.column);
+      }
+      if (query.join) { neededCols.add(query.join.leftKey); }
+      if (query.subqueryIn) for (const sq of query.subqueryIn) neededCols.add(sq.column);
       if (query.vectorSearch) neededCols.add(query.vectorSearch.column);
       const projectedColumns = meta.columns.filter(c => neededCols.has(c.name));
       const filePath = meta.r2Key;

src/query-do.ts

@@ -117,6 +117,14 @@ class EdgeScanOperator implements Operator {
       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);
+      if (query.distinct) for (const d of query.distinct) neededNames.add(d);
+      if (query.windows) for (const w of query.windows) {
+        if (w.column) neededNames.add(w.column);
+        for (const p of w.partitionBy) neededNames.add(p);
+        for (const o of w.orderBy) neededNames.add(o.column);
+      }
+      if (query.join) { neededNames.add(query.join.leftKey); }
+      if (query.subqueryIn) for (const sq of query.subqueryIn) neededNames.add(sq.column);
     } else {
       neededNames = new Set(meta.columns.map(c => c.name));
     }
@@ -971,7 +979,7 @@ 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> {
-    // Determine columns to fetch: projections + all referenced by filters/sort/groupBy/aggregates
+    // Determine columns to fetch: projections + all referenced by filters/sort/groupBy/aggregates/windows/distinct/join
     let neededNames: Set<string>;
     if (query.projections.length > 0) {
       neededNames = new Set(query.projections);
@@ -980,6 +988,14 @@ export class QueryDO extends DurableObject<Env> {
       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);
+      if (query.distinct) for (const d of query.distinct) neededNames.add(d);
+      if (query.windows) for (const w of query.windows) {
+        if (w.column) neededNames.add(w.column);
+        for (const p of w.partitionBy) neededNames.add(p);
+        for (const o of w.orderBy) neededNames.add(o.column);
+      }
+      if (query.join) { neededNames.add(query.join.leftKey); }
+      if (query.subqueryIn) for (const sq of query.subqueryIn) neededNames.add(sq.column);
     } else {
       neededNames = new Set(meta.columns.map(c => c.name));
     }