fix: window function wrong column, partition catalog false negatives

- Add optional `column` field to WindowSpec for lag/lead/aggregate target
  column. Falls back to orderBy[0].column for backwards compat.
- Fix partition catalog eq/in lookups on range data (min≠max) — these
  could falsely prune fragments containing matching rows. Now returns
  null (skip catalog) for non-exact-partition data, falling back to
  min/max pruning which is always correct.
- Persist exactPartition flag in catalog serialization
- Add 3 tests for range-data catalog safety

Author: teamchong

src/operators.ts

@@ -973,10 +973,11 @@ export class WindowOperator implements Operator {
       case "lag": {
         const offset = win.args?.offset ?? 1;
         const defaultVal = win.args?.default_ ?? null;
+        const lagCol = win.column ?? (orderBy.length > 0 ? orderBy[0].column : "");
         for (let i = 0; i < indices.length; i++) {
           const srcIdx = i - offset;
-          if (srcIdx >= 0 && srcIdx < indices.length && orderBy.length > 0) {
-            rows[indices[i]][alias] = rows[indices[srcIdx]][orderBy[0].column] as Row[string];
+          if (srcIdx >= 0 && srcIdx < indices.length && lagCol) {
+            rows[indices[i]][alias] = rows[indices[srcIdx]][lagCol] as Row[string];
           } else {
             rows[indices[i]][alias] = defaultVal as Row[string];
           }
@@ -987,10 +988,11 @@ export class WindowOperator implements Operator {
       case "lead": {
         const offset = win.args?.offset ?? 1;
         const defaultVal = win.args?.default_ ?? null;
+        const leadCol = win.column ?? (orderBy.length > 0 ? orderBy[0].column : "");
         for (let i = 0; i < indices.length; i++) {
           const srcIdx = i + offset;
-          if (srcIdx >= 0 && srcIdx < indices.length && orderBy.length > 0) {
-            rows[indices[i]][alias] = rows[indices[srcIdx]][orderBy[0].column] as Row[string];
+          if (srcIdx >= 0 && srcIdx < indices.length && leadCol) {
+            rows[indices[i]][alias] = rows[indices[srcIdx]][leadCol] as Row[string];
           } else {
             rows[indices[i]][alias] = defaultVal as Row[string];
           }
@@ -999,7 +1001,7 @@ export class WindowOperator implements Operator {
       }
 
       case "sum": case "avg": case "min": case "max": case "count": {
-        const col = orderBy.length > 0 ? orderBy[0].column : "";
+        const col = win.column ?? (orderBy.length > 0 ? orderBy[0].column : "");
         this.applyAggregateWindow(rows, indices, win, col);
         break;
       }

src/partition-catalog.test.ts

@@ -161,6 +161,41 @@ describe("PartitionCatalog", () => {
     });
   });
 
+  describe("prune — non-exact partition (range data)", () => {
+    it("returns null for eq on range data (prevents false negatives)", () => {
+      // min=1, max=100 — value 50 exists but isn't indexed
+      const fragments = new Map<number, TableMeta>([
+        makeFragmentMeta(1, "id", 1, 100),
+        makeFragmentMeta(2, "id", 101, 200),
+      ]);
+      const catalog = PartitionCatalog.fromFragments("id", fragments);
+      const result = catalog.prune([{ column: "id", op: "eq", value: 50 }]);
+      // Must return null (can't safely prune), not [] (no fragments)
+      expect(result).toBeNull();
+    });
+
+    it("returns null for in on range data", () => {
+      const fragments = new Map<number, TableMeta>([
+        makeFragmentMeta(1, "id", 1, 100),
+        makeFragmentMeta(2, "id", 101, 200),
+      ]);
+      const catalog = PartitionCatalog.fromFragments("id", fragments);
+      const result = catalog.prune([{ column: "id", op: "in", value: [50, 150] }]);
+      expect(result).toBeNull();
+    });
+
+    it("neq still works on range data (conservative — includes all non-excluded)", () => {
+      const fragments = new Map<number, TableMeta>([
+        makeFragmentMeta(1, "id", 1, 100),
+        makeFragmentMeta(2, "id", 101, 200),
+      ]);
+      const catalog = PartitionCatalog.fromFragments("id", fragments);
+      // neq is safe even for range data — worst case is slightly over-inclusive
+      const result = catalog.prune([{ column: "id", op: "neq", value: 1 }]);
+      expect(result).not.toBeNull();
+    });
+  });
+
   describe("prune — non-partition column", () => {
     it("returns null when filter is on a different column", () => {
       const catalog = buildCatalog();

src/partition-catalog.ts

@@ -33,6 +33,9 @@ export class PartitionCatalog {
   private allFragmentIds: number[] = [];
   /** O(1) dedup during construction/registration. */
   private allFragmentIdSet = new Set<number>();
+  /** True when every indexed page has min===max (Hive-style partitioning).
+   *  Only exact-partition catalogs are safe for eq/in lookups. */
+  private exactPartition = true;
 
   constructor(column: string) {
     this.column = column;
@@ -53,6 +56,11 @@ export class PartitionCatalog {
       for (const page of col.pages) {
         if (page.minValue !== undefined) values.add(String(page.minValue));
         if (page.maxValue !== undefined) values.add(String(page.maxValue));
+        // If min !== max on any page, this isn't exact-partition data
+        if (page.minValue !== undefined && page.maxValue !== undefined &&
+            String(page.minValue) !== String(page.maxValue)) {
+          catalog.exactPartition = false;
+        }
       }
 
       for (const v of values) {
@@ -116,10 +124,14 @@ export class PartitionCatalog {
   private matchFilter(filter: FilterOp): number[] | null {
     switch (filter.op) {
       case "eq": {
+        // Only safe for exact-partition data (min===max per page).
+        // For range data, a value between min and max wouldn't be indexed.
+        if (!this.exactPartition) return null;
         const entry = this.index.get(String(filter.value));
         return entry ?? [];
       }
       case "in": {
+        if (!this.exactPartition) return null;
         if (!Array.isArray(filter.value)) return null;
         const ids = new Set<number>();
         for (const v of filter.value) {
@@ -149,19 +161,21 @@ export class PartitionCatalog {
   }
 
   /** Serialize to a plain object for DO durable storage. */
-  serialize(): { column: string; index: Record<string, number[]>; allFragmentIds: number[] } {
+  serialize(): { column: string; index: Record<string, number[]>; allFragmentIds: number[]; exactPartition: boolean } {
     const index: Record<string, number[]> = {};
     for (const [key, ids] of this.index) index[key] = ids;
-    return { column: this.column, index, allFragmentIds: this.allFragmentIds };
+    return { column: this.column, index, allFragmentIds: this.allFragmentIds, exactPartition: this.exactPartition };
   }
 
   /** Restore from serialized form. */
-  static deserialize(data: { column: string; index: Record<string, number[]>; allFragmentIds: number[] }): PartitionCatalog {
+  static deserialize(data: { column: string; index: Record<string, number[]>; allFragmentIds: number[]; exactPartition?: boolean }): PartitionCatalog {
     const catalog = new PartitionCatalog(data.column);
     for (const [key, ids] of Object.entries(data.index)) {
       catalog.index.set(key, ids);
     }
     catalog.allFragmentIds = data.allFragmentIds;
+    catalog.allFragmentIdSet = new Set(data.allFragmentIds);
+    catalog.exactPartition = data.exactPartition ?? true; // backwards-compat: old catalogs assumed exact
     return catalog;
   }
 

src/types.ts

@@ -102,6 +102,8 @@ export interface AggregateOp {
 /** Window function specification (serializable — no callbacks) */
 export interface WindowSpec {
   fn: "row_number" | "rank" | "dense_rank" | "lag" | "lead" | "sum" | "avg" | "min" | "max" | "count";
+  /** Target column for lag/lead/sum/avg/min/max/count. Falls back to orderBy[0].column if not set. */
+  column?: string;
   args?: { offset?: number; default_?: unknown };
   partitionBy: string[];
   orderBy: { column: string; direction: "asc" | "desc" }[];