fix: aggregation pipeline, offset pagination, and query schema

- Fix buildEdgePipeline stripping aggregate results via ProjectOperator
  (count/sum/avg returned [{}] instead of aggregated values)
- Add aggregation support to scanPages Parquet path (with limit)
- Fix offset not applied in applyJsPostProcessing
- Add orderBy alias to query schema (maps to sortColumn/sortDirection)
- Expand aggregate schema to all 10 functions (was only 5)
- Add percentileTarget to aggregate schema
- Add 33 HTTP API + schema validation tests (165 total workerd tests)

Author: teamchong

src/http-api.test.ts

@@ -0,0 +1,207 @@
+/**
+ * HTTP API + query schema tests.
+ *
+ * Part 1: HTTP routing tests using SELF.fetch for stateless endpoints only
+ * (health, 404 — no DO interaction, no storage mutation).
+ *
+ * Part 2: Schema validation tests using parseAndValidateQuery directly
+ * (covers orderBy alias, aggregate fns, select alias, groupBy, etc.)
+ *
+ * Data-dependent query tests are in operators-conformance.test.ts (at scale
+ * against DuckDB) and convenience.test.ts (in-memory).
+ */
+import { describe, it, expect } from "vitest";
+import { SELF } from "cloudflare:test";
+import { parseAndValidateQuery } from "./query-schema.js";
+
+// ── HTTP routing (stateless endpoints only) ─────────────────────────────────
+
+describe("HTTP routing", () => {
+  it("GET /health returns ok", async () => {
+    const res = await SELF.fetch("https://fake-host/health");
+    expect(res.status).toBe(200);
+    expect(res.headers.get("content-type")).toBe("application/json");
+    expect(res.headers.get("x-querymode-request-id")).toBeDefined();
+    const body = await res.json() as Record<string, unknown>;
+    expect(body.status).toBe("ok");
+    expect(body.service).toBe("querymode");
+    expect(body.timestamp).toBeDefined();
+  });
+
+  it("GET /nonexistent returns 404", async () => {
+    const res = await SELF.fetch("https://fake-host/nonexistent");
+    expect(res.status).toBe(404);
+    const text = await res.text();
+    expect(text).toContain("/query");
+  });
+});
+
+// ── Schema validation ─────────────────────────────────────────────────────
+
+describe("parseAndValidateQuery", () => {
+  it("parses minimal query", () => {
+    const q = parseAndValidateQuery({ table: "events" });
+    expect(q.table).toBe("events");
+    expect(q.filters).toEqual([]);
+    expect(q.projections).toEqual([]);
+  });
+
+  it("parses filter with all ops", () => {
+    for (const op of ["eq", "neq", "gt", "gte", "lt", "lte", "in"]) {
+      const value = op === "in" ? [1, 2] : 42;
+      const q = parseAndValidateQuery({
+        table: "t",
+        filters: [{ column: "x", op, value }],
+      });
+      expect(q.filters[0].op).toBe(op);
+    }
+  });
+
+  it("parses sortColumn + sortDirection", () => {
+    const q = parseAndValidateQuery({
+      table: "t",
+      sortColumn: "amount",
+      sortDirection: "desc",
+    });
+    expect(q.sortColumn).toBe("amount");
+    expect(q.sortDirection).toBe("desc");
+  });
+
+  it("parses orderBy alias → sortColumn/sortDirection", () => {
+    const q = parseAndValidateQuery({
+      table: "t",
+      orderBy: { column: "amount", desc: true },
+    });
+    expect(q.sortColumn).toBe("amount");
+    expect(q.sortDirection).toBe("desc");
+  });
+
+  it("orderBy without desc defaults to asc", () => {
+    const q = parseAndValidateQuery({
+      table: "t",
+      orderBy: { column: "x" },
+    });
+    expect(q.sortColumn).toBe("x");
+    expect(q.sortDirection).toBe("asc");
+  });
+
+  it("sortColumn takes precedence over orderBy", () => {
+    const q = parseAndValidateQuery({
+      table: "t",
+      sortColumn: "a",
+      sortDirection: "desc",
+      orderBy: { column: "b" },
+    });
+    expect(q.sortColumn).toBe("a");
+    expect(q.sortDirection).toBe("desc");
+  });
+
+  it("parses select alias → projections", () => {
+    const q = parseAndValidateQuery({
+      table: "t",
+      select: ["id", "name"],
+    });
+    expect(q.projections).toEqual(["id", "name"]);
+  });
+
+  it("projections takes precedence over select", () => {
+    const q = parseAndValidateQuery({
+      table: "t",
+      projections: ["a"],
+      select: ["b"],
+    });
+    expect(q.projections).toEqual(["a"]);
+  });
+
+  it("parses limit + offset", () => {
+    const q = parseAndValidateQuery({ table: "t", limit: 10, offset: 5 });
+    expect(q.limit).toBe(10);
+    expect(q.offset).toBe(5);
+  });
+
+  it("parses groupBy", () => {
+    const q = parseAndValidateQuery({
+      table: "t",
+      groupBy: ["region", "category"],
+    });
+    expect(q.groupBy).toEqual(["region", "category"]);
+  });
+
+  it("parses cacheTTL", () => {
+    const q = parseAndValidateQuery({ table: "t", cacheTTL: 60 });
+    expect(q.cacheTTL).toBe(60);
+  });
+
+  // Aggregate functions
+  for (const fn of ["sum", "avg", "min", "max", "count", "count_distinct", "stddev", "variance", "median", "percentile"]) {
+    it(`accepts aggregate fn="${fn}"`, () => {
+      const q = parseAndValidateQuery({
+        table: "t",
+        aggregates: [{ fn, column: "x" }],
+      });
+      expect(q.aggregates).toHaveLength(1);
+      expect(q.aggregates![0].fn).toBe(fn);
+    });
+  }
+
+  it("accepts percentileTarget", () => {
+    const q = parseAndValidateQuery({
+      table: "t",
+      aggregates: [{ fn: "percentile", column: "x", percentileTarget: 0.95 }],
+    });
+    expect(q.aggregates![0].percentileTarget).toBe(0.95);
+  });
+
+  it("accepts aggregate alias", () => {
+    const q = parseAndValidateQuery({
+      table: "t",
+      aggregates: [{ fn: "sum", column: "amount", alias: "total" }],
+    });
+    expect(q.aggregates![0].alias).toBe("total");
+  });
+
+  // Rejection cases
+  it("rejects empty table name", () => {
+    expect(() => parseAndValidateQuery({ table: "" })).toThrow(/table/i);
+  });
+
+  it("rejects missing table", () => {
+    expect(() => parseAndValidateQuery({})).toThrow();
+  });
+
+  it("rejects invalid filter op", () => {
+    expect(() => parseAndValidateQuery({
+      table: "t",
+      filters: [{ column: "x", op: "INVALID", value: 1 }],
+    })).toThrow();
+  });
+
+  it("rejects invalid aggregate fn", () => {
+    expect(() => parseAndValidateQuery({
+      table: "t",
+      aggregates: [{ fn: "BOGUS", column: "x" }],
+    })).toThrow();
+  });
+
+  it("rejects empty filter column", () => {
+    expect(() => parseAndValidateQuery({
+      table: "t",
+      filters: [{ column: "", op: "eq", value: 1 }],
+    })).toThrow();
+  });
+
+  it("rejects negative limit", () => {
+    expect(() => parseAndValidateQuery({ table: "t", limit: -1 })).toThrow();
+  });
+
+  it("rejects negative offset", () => {
+    expect(() => parseAndValidateQuery({ table: "t", offset: -5 })).toThrow();
+  });
+
+  it("rejects percentileTarget out of range", () => {
+    expect(() => parseAndValidateQuery({
+      table: "t",
+      aggregates: [{ fn: "percentile", column: "x", percentileTarget: 1.5 }],
+    })).toThrow();
+  });
+});

src/operators.ts

@@ -2020,7 +2020,10 @@ export function buildEdgePipeline(
     pipeline = new LimitOperator(pipeline, query.limit ?? Infinity, query.offset ?? 0);
   }
 
-  if (outputColumns.length > 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) {
     pipeline = new ProjectOperator(pipeline, outputColumns);
   }
 

src/query-do.ts

@@ -17,6 +17,7 @@ import {
   FilterOperator, HashJoinOperator, ProjectOperator,
   DEFAULT_MEMORY_BUDGET,
 } from "./operators.js";
+import { computePartialAgg, finalizePartialAgg } from "./partial-agg.js";
 import { VipCache } from "./vip-cache.js";
 import { parseLanceV2Columns } from "./lance-v2.js";
 import { parseAndValidateQuery } from "./query-schema.js";
@@ -435,7 +436,10 @@ export class QueryDO extends DurableObject<Env> {
         return va < vb ? -dir : va > vb ? dir : 0;
       });
     }
-    if (query.limit && query.limit > 0) result = result.slice(0, query.limit);
+    const offset = query.offset ?? 0;
+    if (offset > 0 || (query.limit && query.limit > 0)) {
+      result = result.slice(offset, query.limit ? offset + query.limit : undefined);
+    }
     return result;
   }
 
@@ -1041,9 +1045,19 @@ export class QueryDO extends DurableObject<Env> {
       let rows = this.applyJsPostProcessing(
         this.assembleRowsFromColumns(decodedColumns, colNames), query,
       );
+
+      // Handle aggregation in the scanPages Parquet path
+      if (query.aggregates && query.aggregates.length > 0) {
+        const partial = computePartialAgg(rows, query);
+        rows = finalizePartialAgg(partial, query);
+      }
+
       const wasmExecMs = Date.now() - wasmStart;
+      const outputCols = (query.aggregates && query.aggregates.length > 0)
+        ? Object.keys(rows[0] ?? {})
+        : colNames;
       return {
-        rows, rowCount: rows.length, columns: colNames,
+        rows, rowCount: rows.length, columns: outputCols,
         bytesRead, pagesSkipped, durationMs: Date.now() - t0,
         r2ReadMs, wasmExecMs, cacheHits, cacheMisses,
         edgeCacheHits, edgeCacheMisses,

src/query-schema.ts

@@ -18,9 +18,10 @@ const filterOpSchema = z.object({
 });
 
 const aggregateOpSchema = z.object({
-  fn: z.enum(["sum", "avg", "min", "max", "count"]),
+  fn: z.enum(["sum", "avg", "min", "max", "count", "count_distinct", "stddev", "variance", "median", "percentile"]),
   column: z.string().min(1, "Aggregate column name cannot be empty"),
   alias: z.string().optional(),
+  percentileTarget: z.number().min(0).max(1).optional(),
 });
 
 const vectorSearchSchema = z.object({
@@ -39,6 +40,7 @@ export const queryDescriptorSchema = z.object({
   select: z.array(z.string()).optional(), // alias for projections
   sortColumn: z.string().optional(),
   sortDirection: z.enum(["asc", "desc"]).optional(),
+  orderBy: z.object({ column: z.string(), desc: z.boolean().optional() }).optional(), // alias for sortColumn/sortDirection
   limit: z.number().int().positive().optional(),
   offset: z.number().int().nonnegative().optional(),
   vectorSearch: vectorSearchSchema.optional(),
@@ -62,7 +64,7 @@ export function parseAndValidateQuery(body: unknown): {
   limit?: number;
   offset?: number;
   vectorSearch?: { column: string; queryVector: number[] | Float32Array; topK: number };
-  aggregates?: { fn: "sum" | "avg" | "min" | "max" | "count"; column: string; alias?: string }[];
+  aggregates?: { fn: "sum" | "avg" | "min" | "max" | "count" | "count_distinct" | "stddev" | "variance" | "median" | "percentile"; column: string; alias?: string; percentileTarget?: number }[];
   groupBy?: string[];
   cacheTTL?: number;
 } {
@@ -80,12 +82,16 @@ export function parseAndValidateQuery(body: unknown): {
     ? data.projections
     : (data.select ?? []);
 
+  // Merge `orderBy` alias into `sortColumn` / `sortDirection`
+  const sortColumn = data.sortColumn ?? data.orderBy?.column;
+  const sortDirection = data.sortDirection ?? (data.orderBy?.desc ? "desc" : data.orderBy ? "asc" : undefined);
+
   return {
     table: data.table,
     filters: data.filters,
     projections,
-    sortColumn: data.sortColumn,
-    sortDirection: data.sortDirection,
+    sortColumn,
+    sortDirection,
     limit: data.limit,
     offset: data.offset,
     vectorSearch: data.vectorSearch,