feat(objectql,spec): per-aggregation filter on engine.aggregate — drivers refuse instead of silently dropping (#10413 contract half)

.changeset/aggregate-per-aggregation-filter.md

@@ -0,0 +1,34 @@
+---
+"@objectstack/spec": minor
+"@objectstack/objectql": minor
+"@objectstack/driver-sql": patch
+"@objectstack/driver-turso": patch
+"@objectstack/driver-mongodb": patch
+"@objectstack/driver-memory": patch
+---
+
+`engine.aggregate` honours a per-aggregation `filter` (#10576, the contract
+half of #10413). `AggregationNodeSchema.filter` — declared since #4286 but
+marked experimental and enforced by nothing — is now live with SQL
+`FILTER (WHERE …)` semantics: the predicate narrows the SOURCE rows that one
+aggregation reads while sibling aggregations in the same call keep seeing
+every row of the group, so a measure-scoped filter (`stage: 'closed_won'`)
+can finally reach the engine instead of being silently dropped (the #10413
+wrong-numbers defect on the ObjectQL analytics path). The
+`StrategyContext.executeAggregate` bridge (`@objectstack/spec/contracts`)
+gains the same optional `filter` on its aggregation entries so analytics
+strategies can lower measure filters into it (#10413 phase 2 consumes this
+seam next).
+
+Execution is the correct-first two-tier shape date bucketing and HAVING use:
+the engine lowers filtered aggregations in memory for every driver (unknown
+operators refuse loudly with `INVALID_FILTER`/400 naming the aggregation
+position; a group emptied by its filter answers the ruled empty-group values
+— count/sum 0, avg/min/max null). No driver compiles conditional aggregation
+natively today, so each native aggregate face (driver-sql — inherited by
+driver-sqlite-wasm and Turso local —, the Turso remote transport,
+driver-mongodb's pipeline builder, driver-memory's `performAggregation`)
+refuses a directly-delivered per-aggregation filter with
+`NOT_IMPLEMENTED`/501 instead of silently aggregating the unfiltered rows.
+Aggregations without a `filter` are byte-identically unchanged, including
+their native pushdown path.

content/docs/references/data/query.mdx

@@ -49,7 +49,7 @@ const result = AggregationFunction.parse(data);
 | **field** | `string` | optional | Field to aggregate (optional for COUNT(*)) |
 | **alias** | `string` | ✅ | Result column alias |
 | **distinct** | `never` | optional | [REMOVED] `query.aggregations[].distinct` was removed in @objectstack/spec 17 (#6815, ADR-0049) — exactly ONE of the six faces that read an aggregation honoured it. The objectql in-memory fallback deduplicated the values before applying the function, while `driver-sql`, `driver-turso`, `driver-mongodb`, `driver-memory` and the service-analytics SQL builder all ignored it — so `{ function: 'sum', field: 'amount', distinct: true }` answered a DEDUPLICATED sum when the engine fell back in memory and an ordinary sum on every SQL datasource: one query, two numbers, chosen by which backend happened to serve it. Both answers are plausible, so nothing surfaced the divergence. Delete the key. For a deduplicated COUNT the live spelling is the `count_distinct` aggregation function, which every SQL face compiles to `COUNT(DISTINCT field)` (#6409) and the in-memory fallback computes identically. `SUM(DISTINCT …)` / `AVG(DISTINCT …)` get no replacement: no backend ever computed them here, and a per-row measure that needs deduplicating is a modelling problem to fix in the data, not a flag on the read. |
-| **filter** | `any` | optional | [EXPERIMENTAL — not enforced] Per-aggregation filter (SQL FILTER (WHERE …)). Neither the SQL builders nor the in-memory fallback applies it (#4286); filter the whole query with `where` instead. |
+| **filter** | `any` | optional | Per-aggregation filter (SQL FILTER (WHERE …) semantics): narrows the source rows THIS aggregation reads, leaving sibling aggregations unfiltered. Enforced by engine.aggregate (#10576): lowered in memory for drivers without native conditional aggregation; a driver reached directly refuses rather than silently dropping it. |
 
 
 ---

packages/drivers/driver-memory/src/filter-refusal.ts

@@ -54,6 +54,36 @@ export function unsupportedFilterError(message: string): Error {
  return err;
 }
 
+/**
+ * [#10576] An aggregation entry carries a per-aggregation `filter`
+ * (`AggregationNodeSchema.filter`, the contract half of #10413) — the twin of
+ * `driver-sql`'s `unsupportedAggregationFilterError`, first sentence for first
+ * sentence, and the same NOT_IMPLEMENTED/501 class for the same reason (#5907,
+ * ADR-0112): the spec declares the key, this driver's `performAggregation`
+ * evaluates no per-aggregation predicate, so it is a capability gap in the
+ * backend rather than a mistake in the query. Building the evaluation here is
+ * a capability investment this refusal deliberately is not (#5499 freeze).
+ * Refused rather than silently aggregating the UNFILTERED rows — the #10413
+ * defect. Unreachable through `engine.aggregate` (the engine lowers filtered
+ * aggregations in memory for every driver); this fires only for a caller that
+ * reaches the driver's own aggregation faces directly (`find()` with
+ * aggregations, or `aggregate(AST)`).
+ */
+export function refusePerAggregationFilter(alias: string): never {
+ const err = new Error(
+ `Per-aggregation \`filter\` on "${alias}" is not supported by this backend (driver-memory). ` +
+ `The query is spelled correctly and @objectstack/spec AggregationNodeSchema declares the key — ` +
+ `this backend compiles no conditional-aggregate (SQL FILTER (WHERE …) / CASE WHEN) expression ` +
+ `for it, so it is refused rather than silently aggregating the UNFILTERED rows (#10413), which ` +
+ `is why it answers NOT_IMPLEMENTED/501 rather than a 400. \`engine.aggregate\` lowers filtered ` +
+ `aggregations in memory for every driver without native support — route the query through the ` +
+ `engine, or drop the \`filter\` key.`,
+ ) as Error & { code?: string; status?: number };
+ err.code = StandardErrorCode.enum.NOT_IMPLEMENTED;
+ err.status = 501;
+ throw err;
+}
+
 /**
  * [#5158] A `FilterArray` reached the driver unlowered.
  *

packages/drivers/driver-memory/src/memory-aggregation-filter-refusal.test.ts

@@ -0,0 +1,86 @@
+// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * [#10576] A per-aggregation `filter` (`AggregationNodeSchema.filter`, the
+ * contract half of #10413) reaching this driver's own aggregation faces is
+ * refused with the ADR-0112 envelope — never silently answered with the
+ * UNFILTERED aggregate, which is the #10413 defect at the driver seam.
+ *
+ * This driver has TWO doors into `performAggregation` — `aggregate(AST)` (the
+ * one objectql's engine uses) and `find()` with aggregations — so both are
+ * pinned: a guard on one door alone re-opens the drop through the other. The
+ * engine itself never pushes a filtered aggregation down (it lowers in
+ * memory); the refusal exists for direct callers. Evaluating the predicate
+ * here instead would be a capability build-out, which the #5499 family freeze
+ * rules out — the refusal path is the sanctioned scope.
+ *
+ * Every case asserts `code` AND `status`, never merely "it threw" (#6144).
+ */
+
+import { describe, it, expect, beforeEach } from 'vitest';
+import type { DriverQuery } from '@objectstack/spec/contracts';
+import { InMemoryDriver } from './memory-driver.js';
+
+interface WireBearingError extends Error {
+ code?: string;
+ status?: number;
+}
+
+const TABLE = 'deal';
+
+describe('[#10576] InMemoryDriver refuses a per-aggregation filter it does not evaluate', () => {
+ let driver: InMemoryDriver;
+
+ beforeEach(async () => {
+ driver = new InMemoryDriver();
+ await driver.connect();
+ await driver.create(TABLE, { id: '1', stage: 'closed_won', amount: 500 });
+ await driver.create(TABLE, { id: '2', stage: 'open', amount: 900 });
+ });
+
+ const filteredQuery = (): DriverQuery => ({
+ aggregations: [{ function: 'count', alias: 'won_count', filter: { stage: 'closed_won' } }],
+ }) as unknown as DriverQuery;
+
+ it('the aggregate(AST) door answers NOT_IMPLEMENTED / 501, naming the aggregation and the engine lowering', async () => {
+ let thrown: WireBearingError | undefined;
+ try {
+ await driver.aggregate(TABLE, filteredQuery());
+ } catch (e) {
+ thrown = e as WireBearingError;
+ }
+ expect(thrown, 'a filter this face does not evaluate must not be silently dropped').toBeDefined();
+ expect(thrown!.code).toBe('NOT_IMPLEMENTED');
+ expect(thrown!.status).toBe(501);
+ expect(thrown!.message).toContain(
+ 'Per-aggregation `filter` on "won_count" is not supported by this backend (driver-memory).',
+ );
+ expect(thrown!.message).toContain('`engine.aggregate` lowers filtered aggregations in memory');
+ });
+
+ it('the find()-with-aggregations door refuses identically — one guard covers both doors', async () => {
+ let thrown: WireBearingError | undefined;
+ try {
+ await driver.find(TABLE, filteredQuery());
+ } catch (e) {
+ thrown = e as WireBearingError;
+ }
+ expect(thrown).toBeDefined();
+ expect(thrown!.code).toBe('NOT_IMPLEMENTED');
+ expect(thrown!.status).toBe(501);
+ expect(thrown!.message).toContain('Per-aggregation `filter` on "won_count"');
+ });
+
+ it('control: the same aggregation WITHOUT a filter still computes through both doors', async () => {
+ const bare = { aggregations: [{ function: 'count', alias: 'n' }] } as unknown as DriverQuery;
+ await expect(driver.aggregate(TABLE, bare)).resolves.toEqual([{ n: 2 }]);
+ await expect(driver.find(TABLE, bare)).resolves.toEqual([{ n: 2 }]);
+ });
+
+ it('control: an EMPTY filter object is vacuous (the where/having convention) and computes', async () => {
+ const vacuous = {
+ aggregations: [{ function: 'sum', field: 'amount', alias: 'total', filter: {} }],
+ } as unknown as DriverQuery;
+ await expect(driver.aggregate(TABLE, vacuous)).resolves.toEqual([{ total: 1400 }]);
+ });
+});

packages/drivers/driver-memory/src/memory-driver.ts

@@ -28,6 +28,8 @@ import {
  // [#7536] The `$like`/`$ilike` comparand refusals, beside their siblings.
  likePatternComparandError,
  danglingLikeEscapeError,
+ // [#10576] The per-aggregation `filter` refusal — this driver evaluates none.
+ refusePerAggregationFilter,
 } from './filter-refusal.js';
 import {
  coerceTemporalValue,
@@ -1159,6 +1161,19 @@ export class InMemoryDriver implements IDataDriver {
 
  private performAggregation(records: any[], query: DriverQuery): any[] {
  const { groupBy, aggregations } = query;
+ // [#10576] A per-aggregation `filter` this face does not evaluate is
+ // refused before any group is built — silently answering the UNFILTERED
+ // aggregate is the #10413 defect. Guarded HERE because both of this
+ // driver's aggregation doors (`find()` with aggregations and
+ // `aggregate(AST)`) funnel through this method. `{}` is the vacuous
+ // filter, same convention as `where` / `having`. See
+ // {@link refusePerAggregationFilter}.
+ for (const agg of aggregations ?? []) {
+ const f = (agg as { filter?: unknown }).filter;
+ if (f && typeof f === 'object' && Object.keys(f).length > 0) {
+ refusePerAggregationFilter((agg as any).alias ?? (agg as any).field ?? '(unaliased)');
+ }
+ }
  const groups: Map<string, any[]> = new Map();
 
  const normalizeGroupBy = (node: any): { field: string; alias: string } => {

packages/drivers/driver-mongodb/src/mongodb-aggregation-translation.test.ts

@@ -330,3 +330,40 @@ describe('the in-process pipeline evaluator discriminates', () => {
  expect(counted).toBe(1);
  });
 });
+
+/**
+ * [#10576] A per-aggregation `filter` (`AggregationNodeSchema.filter`, the
+ * contract half of #10413) reaching this builder is refused with the ADR-0112
+ * envelope — never silently accumulated over the UNFILTERED rows, which is the
+ * #10413 defect at the driver seam. `AggregationInput.filter` has declared the
+ * key since #6850 and nothing ever read it; a declared key a builder ignores
+ * is exactly the silent drop this card closes. The engine never pushes a
+ * filtered aggregation down (it lowers in memory) — the refusal exists for the
+ * direct caller of this exported builder.
+ */
+describe('[#10576] per-aggregation filter refuses instead of silently dropping', () => {
+ it('answers NOT_IMPLEMENTED / 501, naming the aggregation and the engine lowering', () => {
+ let thrown: (Error & { code?: string; status?: number }) | undefined;
+ try {
+ buildAggregationPipeline({
+ aggregations: [{ function: 'count', alias: 'won_count', filter: { stage: 'closed_won' } }],
+ });
+ } catch (err) {
+ thrown = err as Error & { code?: string; status?: number };
+ }
+ expect(thrown, 'a filter this builder cannot lower must not be silently dropped').toBeDefined();
+ expect(thrown!.code).toBe('NOT_IMPLEMENTED');
+ expect(thrown!.status).toBe(501);
+ expect(thrown!.message).toContain(
+ 'Per-aggregation `filter` on "won_count" is not supported by this backend (driver-mongodb).',
+ );
+ expect(thrown!.message).toContain('`engine.aggregate` lowers filtered aggregations in memory');
+ });
+
+ it('control: an EMPTY filter object is vacuous (the where/having convention) and still lowers', () => {
+ const pipeline = buildAggregationPipeline({
+ aggregations: [{ function: 'count', alias: 'n', filter: {} }],
+ });
+ expect(pipeline[0]).toHaveProperty('$group');
+ });
+});

packages/drivers/driver-mongodb/src/mongodb-aggregation.ts

@@ -303,6 +303,33 @@ function refuseDateBucketedGroupBy(granularity: string): never {
  throw err;
 }
 
+/**
+ * [#10576] An aggregation entry carries a per-aggregation `filter`
+ * (`AggregationNodeSchema.filter`, the contract half of #10413) — the twin of
+ * `driver-sql`'s `unsupportedAggregationFilterError`, first sentence for first
+ * sentence, and the same NOT_IMPLEMENTED/501 class for the same reason (#5907,
+ * ADR-0112): the spec declares the key, this builder emits no conditional
+ * accumulator for it, so it is a capability gap in the backend rather than a
+ * mistake in the query. Refused rather than silently accumulating the
+ * UNFILTERED rows — the #10413 defect. Unreachable through `engine.aggregate`
+ * (the engine lowers filtered aggregations in memory for every driver); this
+ * fires only for a caller that drives the builder or driver directly.
+ */
+function refusePerAggregationFilter(alias: string): never {
+ const err = new Error(
+ `Per-aggregation \`filter\` on "${alias}" is not supported by this backend (driver-mongodb). ` +
+ `The query is spelled correctly and @objectstack/spec AggregationNodeSchema declares the key — ` +
+ `this backend compiles no conditional-aggregate (SQL FILTER (WHERE …) / CASE WHEN) expression ` +
+ `for it, so it is refused rather than silently aggregating the UNFILTERED rows (#10413), which ` +
+ `is why it answers NOT_IMPLEMENTED/501 rather than a 400. \`engine.aggregate\` lowers filtered ` +
+ `aggregations in memory for every driver without native support — route the query through the ` +
+ `engine, or drop the \`filter\` key.`,
+ ) as Error & { code?: string; status?: number };
+ err.code = StandardErrorCode.enum.NOT_IMPLEMENTED;
+ err.status = 501;
+ throw err;
+}
+
 /**
  * [#6850] A `groupBy` entry that is neither half of the declared union.
  *
@@ -386,6 +413,17 @@ export function buildAggregationPipeline(opts: {
 
  // Build accumulators from aggregation descriptors
  for (const agg of opts.aggregations) {
+ // [#10576] A per-aggregation `filter` (`AggregationNodeSchema.filter`,
+ // the contract half of #10413) has no lowering in this builder — a
+ // `$cond`-wrapped accumulator would be one, but building it is a
+ // capability investment this refusal deliberately is not. Refused before
+ // a pipeline exists rather than silently accumulating the UNFILTERED
+ // rows (the #10413 defect). Unreachable through `engine.aggregate`,
+ // which lowers filtered aggregations in memory for every driver; `{}` is
+ // the vacuous filter, same convention as `where` / `having`.
+ if (agg.filter && typeof agg.filter === 'object' && Object.keys(agg.filter).length > 0) {
+ refusePerAggregationFilter(agg.alias ?? agg.field ?? '(unaliased)');
+ }
  groupAccumulators[agg.alias] = buildAccumulator(agg);
  }
 

packages/drivers/driver-sql/src/sql-driver-aggregation-filter-refusal.test.ts

@@ -0,0 +1,93 @@
+// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * [#10576] A per-aggregation `filter` (`AggregationNodeSchema.filter`, the
+ * contract half of #10413) reaching this compiler DIRECTLY is refused with a
+ * wire identity — never silently dropped.
+ *
+ * Before #10576 this driver's `aggregate()` never read `agg.filter`: the
+ * statement it built aggregated EVERY row and reported success, which is the
+ * #10413 defect ("won deals" counting every opportunity) at the driver seam.
+ * The engine now lowers filtered aggregations in memory and never pushes one
+ * down here — so the only caller that can arrive with the key is a direct
+ * one, and the honest answers are exactly two: compile a conditional
+ * aggregate, or refuse loudly. This backend refuses (NOT_IMPLEMENTED/501, the
+ * #5907 class for "declared by the spec, not compiled by this face").
+ *
+ * Every case asserts `code` AND `status`, never merely "it threw" (#6144): a
+ * bare-throw assertion is green before and after the envelope exists.
+ */
+
+import { describe, it, expect, beforeEach } from 'vitest';
+import { SqlDriver } from './index.js';
+import type { QueryAST } from '@objectstack/spec/data';
+
+interface WireBearingError extends Error {
+ code?: string;
+ status?: number;
+}
+
+describe('[#10576] SqlDriver refuses a per-aggregation filter it cannot compile', () => {
+ let driver: SqlDriver;
+
+ beforeEach(async () => {
+ driver = new SqlDriver({
+ client: 'better-sqlite3',
+ connection: { filename: ':memory:' },
+ useNullAsDefault: true,
+ });
+ await driver.initObjects([
+ {
+ name: 'deal',
+ fields: {
+ id: { type: 'text', name: 'id' },
+ stage: { type: 'text', name: 'stage' },
+ amount: { type: 'number', name: 'amount' },
+ },
+ } as any,
+ ]);
+ await driver.create('deal', { id: '1', stage: 'closed_won', amount: 500 });
+ await driver.create('deal', { id: '2', stage: 'open', amount: 900 });
+ });
+
+ it('refuses NOT_IMPLEMENTED/501, naming the aggregation and the engine lowering', async () => {
+ const ast = {
+ object: 'deal',
+ aggregations: [
+ { function: 'count', alias: 'won_count', filter: { stage: 'closed_won' } },
+ ],
+ } as unknown as QueryAST;
+
+ let thrown: WireBearingError | undefined;
+ try {
+ await driver.aggregate('deal', ast);
+ } catch (e) {
+ thrown = e as WireBearingError;
+ }
+
+ expect(thrown).toBeDefined();
+ expect(thrown!.code).toBe('NOT_IMPLEMENTED');
+ expect(thrown!.status).toBe(501);
+ expect(thrown!.message).toContain(
+ 'Per-aggregation `filter` on "won_count" is not supported by this backend (driver-sql).',
+ );
+ // The remedy is named: the engine's in-memory lowering serves this query.
+ expect(thrown!.message).toContain('`engine.aggregate` lowers filtered aggregations in memory');
+ });
+
+ it('control: the same aggregation WITHOUT a filter still computes (only the refusal was added)', async () => {
+ const rows = await driver.aggregate('deal', {
+ object: 'deal',
+ aggregations: [{ function: 'count', alias: 'n' }],
+ } as unknown as QueryAST);
+ expect(rows).toEqual([{ n: 2 }]);
+ });
+
+ it('control: an EMPTY filter object is vacuous (the where/having convention) and computes', async () => {
+ const rows = await driver.aggregate('deal', {
+ object: 'deal',
+ aggregations: [{ function: 'sum', field: 'amount', alias: 'total', filter: {} }],
+ } as unknown as QueryAST);
+ expect(rows).toEqual([{ total: 1400 }]);
+ });
+});