fix(db-ivm): treat 0 and empty string as valid min/max extremes (#1809)

* fix(db-ivm): treat 0 and empty string as valid min/max extremes

min() and max() used truthiness to detect an unset accumulator, so 0,
0n, and "" were skipped or overwritten. Compare against undefined.

Fixes #1775

* fix(db-ivm): narrow min and max inputs

* test(db-ivm): cover falsy extrema in oracle

---------

Co-authored-by: Kyle Mathews <mathews.kyle@gmail.com>

Author: aakbarpour

.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.

packages/db-ivm/src/operators/groupBy.ts

@@ -237,7 +237,10 @@ export function min<T, V extends CanMinMax>(
     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<T, V extends CanMinMax>(
     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
         }
       }

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<Keyed>): Weighted<GroupedOutput> {
-  const groups = new Map<number, number>()
+/**
+ * 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<Keyed>): Weighted<GroupedOutput> {
+  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/)
   })
 })

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<number>()
+      const maxNum = max<number>()
+      const minStr = min<string>()
+      const minBig = min<bigint>()
+      const maxBig = max<bigint>()
+
+      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<{