diff --git a/.changeset/minmax-falsy-extremes.md b/.changeset/minmax-falsy-extremes.md new file mode 100644 index 0000000000..327ad64030 --- /dev/null +++ b/.changeset/minmax-falsy-extremes.md @@ -0,0 +1,5 @@ +--- +'@tanstack/db-ivm': patch +--- + +Compare min and max aggregates against `undefined` instead of truthiness so `0`, `0n`, and `""` can be the extreme of a group. diff --git a/packages/db-ivm/src/operators/groupBy.ts b/packages/db-ivm/src/operators/groupBy.ts index 435156d25d..808b4f224a 100644 --- a/packages/db-ivm/src/operators/groupBy.ts +++ b/packages/db-ivm/src/operators/groupBy.ts @@ -237,7 +237,10 @@ export function min( reduce: (values) => { let minValue: V | undefined for (const [value, _multiplicity] of values) { - if (!minValue || (value && value < minValue)) { + if ( + value !== undefined && + (minValue === undefined || value < minValue) + ) { minValue = value } } @@ -267,7 +270,10 @@ export function max( reduce: (values) => { let maxValue: V | undefined for (const [value, _multiplicity] of values) { - if (!maxValue || (value && value > maxValue)) { + if ( + value !== undefined && + (maxValue === undefined || value > maxValue) + ) { maxValue = value } } diff --git a/packages/db-ivm/tests/incrementalization-law.property.test.ts b/packages/db-ivm/tests/incrementalization-law.property.test.ts index 3c58913b35..73f1de7ac0 100644 --- a/packages/db-ivm/tests/incrementalization-law.property.test.ts +++ b/packages/db-ivm/tests/incrementalization-law.property.test.ts @@ -42,14 +42,28 @@ import type { Weighted } from './incrementalization-law.js' * * The generated domain uses small JSON tuples with integer weights. Named * cases force empty batches, duplicate weights, replacements, cancellation, - * presence changes, boundary ties, and a zero-width window. Fault controls - * prove that the checker rejects missing, sign-flipped, and wrong-member output. + * presence changes, falsey group extrema, boundary ties, and a zero-width + * window. Fault controls prove that the checker rejects missing, sign-flipped, + * wrong-member, and truthiness-filtered aggregate output. + * + * Before this repair the groupBy branch observed only sums. Truthiness defects + * in `min` and `max` were therefore outside both its model and its assertions. + * The small direct reducer cases remain readable replay witnesses; this suite + * owns the generated incremental-versus-recompute law. */ type Keyed = [number, number] type JoinOutput = [number, [number, number]] type OuterJoinOutput = [number, [number | null, number | null]] -type GroupedOutput = [string, { bucket: number; total: number }] +type GroupedOutput = [ + string, + { + bucket: number + total: number + minimum: number | undefined + maximum: number | undefined + }, +] const FIXED_SEED = 1741 type GeneratedCampaign = { @@ -219,14 +233,32 @@ function joinedSums( return [...sums].map(([key, value]) => [[key, value], 1]) } -function groupedSums(input: Weighted): Weighted { - const groups = new Map() +/** + * Group extrema come from every retained defined value. Zero is a value, not + * absence. The model recomputes from plain weighted rows and shares no + * aggregate reducer with production. + */ +function groupedAggregates(input: Weighted): Weighted { + const groups = new Map< + number, + { total: number; minimum: number; maximum: number } + >() for (const [[key, value], weight] of keyedIdentity(input)) { const bucket = key % 2 - groups.set(bucket, (groups.get(bucket) ?? 0) + value * weight) + const current = groups.get(bucket) + groups.set( + bucket, + current === undefined + ? { total: value * weight, minimum: value, maximum: value } + : { + total: current.total + value * weight, + minimum: Math.min(current.minimum, value), + maximum: Math.max(current.maximum, value), + }, + ) } - return [...groups].map(([bucket, total]) => [ - [JSON.stringify({ bucket }), { bucket, total }], + return [...groups].map(([bucket, aggregates]) => [ + [JSON.stringify({ bucket }), { bucket, ...aggregates }], 1, ]) } @@ -332,9 +364,11 @@ describe(`DBSP incrementalization laws`, () => { input.pipe( groupBy(([key]) => ({ bucket: key % 2 }), { total: groupByOperators.sum(([, value]) => value), + minimum: groupByOperators.min(([, value]) => value), + maximum: groupByOperators.max(([, value]) => value), }), ), - evaluate: groupedSums, + evaluate: groupedAggregates, }) assertUnaryIncrementalization({ name: `top-K`, @@ -555,6 +589,35 @@ describe(`DBSP incrementalization laws`, () => { splitDeliveries: 4, }) + const extremaReach = assertUnaryIncrementalization({ + name: `grouped falsey extrema`, + initial: [ + [[0, 5], 1], + [[2, 0], 1], + [[1, -2], 1], + [[3, 0], 1], + ], + batches: [], + inputPolicy: keyedPolicy, + outputPolicy: groupedPolicy, + splitDomain: uniqueRowSplitDomain, + build: (input) => + input.pipe( + groupBy(([key]) => ({ bucket: key % 2 }), { + total: groupByOperators.sum(([, value]) => value), + minimum: groupByOperators.min(([, value]) => value), + maximum: groupByOperators.max(([, value]) => value), + }), + ), + evaluate: groupedAggregates, + }) + expect(extremaReach).toEqual({ + atomicCheckpoints: 1, + atomicDeliveries: 1, + splitCheckpoints: 1, + splitDeliveries: 4, + }) + assertUnaryIncrementalization({ name: `grouped-order boundary tie`, initial: [ @@ -617,7 +680,7 @@ describe(`DBSP incrementalization laws`, () => { }) }) - it(`rejects omitted, sign-flipped, wrong-member, and wrong-window output`, () => { + it(`rejects omitted, sign-flipped, wrong-member, wrong-window, and truthiness-filtered output`, () => { expect(() => assertUnaryIncrementalization({ name: `omitted output fault`, @@ -674,5 +737,56 @@ describe(`DBSP incrementalization laws`, () => { evaluate: firstThree, }), ).toThrow(/output delta diverged/) + + const truthinessMinimum = { + preMap: ([, value]: Keyed): number | undefined => value, + reduce: (values: Array<[number | undefined, number]>) => { + let minimum: number | undefined + for (const [value] of values) { + if (!minimum || (value !== undefined && value && value < minimum)) { + minimum = value + } + } + return minimum + }, + postMap: (result: number | undefined) => result, + } + const truthinessMaximum = { + preMap: ([, value]: Keyed): number | undefined => value, + reduce: (values: Array<[number | undefined, number]>) => { + let maximum: number | undefined + for (const [value] of values) { + if (!maximum || (value !== undefined && value && value > maximum)) { + maximum = value + } + } + return maximum + }, + postMap: (result: number | undefined) => result, + } + expect(() => + assertUnaryIncrementalization({ + name: `truthiness-filtered extrema fault`, + initial: [ + [[0, 5], 1], + [[2, 0], 1], + [[1, -2], 1], + [[3, 0], 1], + ], + batches: [], + inputPolicy: keyedPolicy, + outputPolicy: groupedPolicy, + splitDomain: uniqueRowSplitDomain, + build: (input) => + input.pipe( + groupBy(([key]) => ({ bucket: key % 2 }), { + total: groupByOperators.sum(([, value]) => value), + minimum: truthinessMinimum, + maximum: truthinessMaximum, + }), + ), + evaluate: groupedAggregates, + }), + ).toThrow(/output delta diverged/) }) }) diff --git a/packages/db-ivm/tests/operators/groupBy.test.ts b/packages/db-ivm/tests/operators/groupBy.test.ts index 52b0fac653..6f700b2ad5 100644 --- a/packages/db-ivm/tests/operators/groupBy.test.ts +++ b/packages/db-ivm/tests/operators/groupBy.test.ts @@ -624,6 +624,122 @@ describe(`Operators`, () => { expect(latestMessage.getInner()).toEqual(expectedResult) }) + // These are readable replay witnesses. The generated groupBy law lives in + // incrementalization-law.property.test.ts. + test(`min and max reduce keep 0, 0n, and empty string as extremes`, () => { + const minNum = min() + const maxNum = max() + const minStr = min() + const minBig = min() + const maxBig = max() + + if ( + !(`reduce` in minNum) || + !(`reduce` in maxNum) || + !(`reduce` in minStr) || + !(`reduce` in minBig) || + !(`reduce` in maxBig) + ) { + throw new Error(`Expected direct min/max aggregates`) + } + + expect( + minNum.reduce([ + [undefined, 1], + [5, 1], + [0, 1], + ]), + ).toBe(0) + expect( + minNum.reduce([ + [0, 1], + [3, 1], + ]), + ).toBe(0) + expect( + maxNum.reduce([ + [undefined, 1], + [-2, 1], + [0, 1], + [-1, 1], + ]), + ).toBe(0) + expect( + maxNum.reduce([ + [0, 1], + [-1, 1], + ]), + ).toBe(0) + expect( + minStr.reduce([ + [`b`, 1], + [``, 1], + ]), + ).toBe(``) + expect( + minStr.reduce([ + [``, 1], + [`a`, 1], + ]), + ).toBe(``) + expect( + minBig.reduce([ + [5n, 1], + [0n, 1], + ]), + ).toBe(0n) + expect( + maxBig.reduce([ + [-2n, 1], + [0n, 1], + ]), + ).toBe(0n) + }) + + test(`with min and max aggregates including a zero amount`, () => { + const graph = new D2() + const input = graph.newInput<{ + category: string + amount: number + }>() + let latestMessage: any = null + + input.pipe( + groupBy((data) => ({ category: data.category }), { + minimum: min((data) => data.amount), + maximum: max((data) => data.amount), + }), + output((message) => { + latestMessage = message + }), + ) + + graph.finalize() + + input.sendData( + new MultiSet([ + [{ category: `A`, amount: 10 }, 1], + [{ category: `A`, amount: 0 }, 1], + [{ category: `A`, amount: 7 }, 1], + ]), + ) + graph.run() + + expect(latestMessage.getInner()).toEqual([ + [ + [ + serializeValue({ category: `A` }), + { + category: `A`, + minimum: 0, + maximum: 10, + }, + ], + 1, + ], + ]) + }) + test(`with median and mode aggregates`, () => { const graph = new D2() const input = graph.newInput<{