From 653b88e4c5e0837efe071a61470ad469f300cd8c Mon Sep 17 00:00:00 2001 From: Eugene Gu Date: Wed, 26 Aug 2026 04:45:25 +0000 Subject: [PATCH] fix(workflow-operator): MAX aggregation checks the wrong empty-group sentinel (#7532) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### What changes were proposed in this PR? One-line fix in `AggregationOperation.scala`: the `max` aggregation starts its running maximum from the type's minimum value, so its final "was this group empty?" check must compare against that same minimum value, but it compared against the type's maximum value instead (the line was copied from `min`, where that comparison is correct because `min` starts from the maximum). Because the check looked at the wrong sentinel, two results were silently wrong while the workflow completed with no error: - A group with only null values returned the sentinel itself instead of `null`, e.g. `-2147483648` for INTEGER or `1970-01-01 00:00:00` for TIMESTAMP. - A true maximum equal to the type's maximum value was mistaken for "no value seen" and discarded, so `max` over `{1, 5, 2147483647}` reported `5` (the largest value from the other local aggregation worker) instead of `2147483647`. #### Before-and-after Test data: group `g1` has only nulls, `g2` contains `{1, 5, 2147483647}`, `g3` contains `{10, 42}`. Screenshot 2026-08-07 at 2 27 40 PM Before the fix, `max(v)` grouped by `k` returns `-2147483648` for `g1` (expected `null`) and `5` for `g2` (expected `2147483647`): Screenshot 2026-08-07 at 2 27 51 PM `min(v)` on the same data is correct (`null` for `g1`, `1` for `g2`, `10` for `g3`), confirming only `max`'s empty-group check is broken: Screenshot 2026-08-07 at 2 28 00 PM After the fix, `max(v)` returns `null` for `g1`, `2147483647` for `g2`, and `42` for `g3`: Screenshot 2026-08-07 at 5 54 17 PM ### Any related issues, documentation, discussions? Closes #7531 The regression was introduced by #1840, which generalised the hard-coded `Double` sentinels to per-type `minValue`/`maxValue` helpers and updated `maxAgg`'s initialiser but not its finaliser. ### How was this PR tested? Added 5 regression tests, all of which fail without the one-line fix and pass with it (verified in both directions): - `AggregateOpSpec`: `max` over empty input and over all-null input returns `null`; `max` keeps a true maximum equal to the type's maximum value, covering INTEGER, LONG, DOUBLE, and TIMESTAMP. - `AggregationOperationSpec`: a worker-to-final pipeline via `getFinal` keeps `Int.MaxValue` when partial results are re-aggregated, reproducing the two-worker scenario shown above; `max`'s merge stays neutral when one side saw no values. Full aggregate suite: `sbt "WorkflowOperator/testOnly org.apache.texera.amber.operator.aggregate.*"` — 68 tests, all passing. Also verified end to end in the UI with the workflow shown above (screenshots are from before and after rebuilding the backend with this fix). ### Was this PR authored or co-authored using generative AI tooling? Co-authored by: Claude Code (Claude Fable 5) (backported from commit b0dd3ffc5a2356b9d2c62626d5e6e8a407c60f74) --- .../aggregate/AggregationOperation.scala | 2 +- .../operator/aggregate/AggregateOpSpec.scala | 63 +++++ .../aggregate/AggregationOperationSpec.scala | 252 ++++++++++++++++++ 3 files changed, 316 insertions(+), 1 deletion(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregationOperation.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregationOperation.scala index 70105de9ef4..bec4b46c10d 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregationOperation.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregationOperation.scala @@ -222,7 +222,7 @@ class AggregationOperation { (partial1, partial2) => if (AttributeTypeUtils.compare(partial1, partial2, attributeType) > 0) partial1 else partial2, - partial => if (partial == AttributeTypeUtils.maxValue(attributeType)) null else partial + partial => if (partial == AttributeTypeUtils.minValue(attributeType)) null else partial ) } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregateOpSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregateOpSpec.scala index cb7925ec41d..f1e32c8bba5 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregateOpSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregateOpSpec.scala @@ -23,6 +23,8 @@ import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tup import org.apache.texera.amber.util.JSONUtils.objectMapper import org.scalatest.funsuite.AnyFunSuite +import java.sql.Timestamp + class AggregateOpSpec extends AnyFunSuite { /** Helpers */ @@ -408,6 +410,67 @@ class AggregateOpSpec extends AnyFunSuite { assert(result == maxValue) } + test("MAX aggregation finds largest INTEGER and returns null when given no values") { + val schema = makeSchema("temperature" -> AttributeType.INTEGER) + val tuple1 = makeTuple(schema, 10) + val tuple2 = makeTuple(schema, -2) + val tuple3 = makeTuple(schema, 5) + + val operation = makeAggregationOp(AggregationFunction.MAX, "temperature", "max_temp") + val agg = operation.getAggFunc(AttributeType.INTEGER) + + // Empty case: never iterate, just finalize init + val emptyPartial = agg.init() + val emptyResult = agg.finalAgg(emptyPartial) + assert(emptyResult == null) + + // Non-empty case + var partial = agg.init() + partial = agg.iterate(partial, tuple1) + partial = agg.iterate(partial, tuple2) + partial = agg.iterate(partial, tuple3) + + val result = agg.finalAgg(partial).asInstanceOf[Number].intValue() + assert(result == 10) + } + + test("MAX aggregation returns null when all values are null") { + val schema = makeSchema("temperature" -> AttributeType.INTEGER) + + val operation = makeAggregationOp(AggregationFunction.MAX, "temperature", "max_temp") + val agg = operation.getAggFunc(AttributeType.INTEGER) + + var partial = agg.init() + Seq.fill(3)(makeTuple(schema, null)).foreach(tp => partial = agg.iterate(partial, tp)) + assert(agg.finalAgg(partial) == null) + } + + test("MAX aggregation keeps the type's maximum value when it is the true maximum") { + // Regression: the finalizer used to mistake a partial equal to + // maxValue(attributeType) for the "no value seen" sentinel and emit null. + val cases: Seq[(AttributeType, Seq[Any], Any)] = Seq( + (AttributeType.INTEGER, Seq(1, 5, Int.MaxValue), Int.MaxValue), + (AttributeType.LONG, Seq(1L, 5L, Long.MaxValue), Long.MaxValue), + (AttributeType.DOUBLE, Seq(1.0, 5.0, Double.MaxValue), Double.MaxValue), + ( + AttributeType.TIMESTAMP, + Seq(new Timestamp(1L), new Timestamp(Long.MaxValue)), + new Timestamp(Long.MaxValue) + ) + ) + + for ((attrType, values, expected) <- cases) { + val schema = makeSchema("v" -> attrType) + val operation = makeAggregationOp(AggregationFunction.MAX, "v", "max_v") + val agg = operation.getAggFunc(attrType) + + var partial = agg.init() + values.foreach(v => partial = agg.iterate(partial, makeTuple(schema, v))) + + assert(agg.finalAgg(partial) == expected, s"MAX over $attrType must keep $expected") + } + } + test("AVERAGE aggregation ignores nulls and returns null when all values are null") { val schema = makeSchema("price" -> AttributeType.DOUBLE) val tuple1 = makeTuple(schema, 10.0) diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregationOperationSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregationOperationSpec.scala index 4ebba05fbc2..b11bc3e9596 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregationOperationSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/aggregate/AggregationOperationSpec.scala @@ -173,6 +173,35 @@ class AggregationOperationSpec extends AnyFlatSpec { assert(finalSum == 36, "single-pass SUM(1+2+3+10+20) == 36") } + it should + "keep the type's maximum value when partial MAX results are re-aggregated via getFinal" in { + // Regression: the local stage used to finalize a partial equal to + // maxValue(attributeType) to null, so the final stage reported the largest + // value from the other worker instead of the true maximum. + val workerOp = op(AggregationFunction.MAX, attribute = "v", resultAttribute = "max_v") + val workerAgg = workerOp.getAggFunc(AttributeType.INTEGER) + + val w1State = Seq[AnyRef](Int.box(1), Int.box(Int.MaxValue)) + .map(v => tupleOf("v", AttributeType.INTEGER, v)) + .foldLeft(workerAgg.init())(workerAgg.iterate) + val w1Out = workerAgg.finalAgg(w1State) + assert(w1Out == Int.box(Int.MaxValue), "worker 1's true maximum is Int.MaxValue") + + val w2State = Seq[AnyRef](Int.box(5), Int.box(2)) + .map(v => tupleOf("v", AttributeType.INTEGER, v)) + .foldLeft(workerAgg.init())(workerAgg.iterate) + val w2Out = workerAgg.finalAgg(w2State) + assert(w2Out == Int.box(5)) + + val finalOp = workerOp.getFinal + assert(finalOp.aggFunction == AggregationFunction.MAX) + val finalAgg = finalOp.getAggFunc(AttributeType.INTEGER) + val finalState = Seq(w1Out, w2Out) + .map(p => tupleOf("max_v", AttributeType.INTEGER, p)) + .foldLeft(finalAgg.init())(finalAgg.iterate) + assert(finalAgg.finalAgg(finalState) == Int.box(Int.MaxValue)) + } + // --- AveragePartialObj ----------------------------------------------------- "AveragePartialObj" should "expose its sum and count fields and support value equality" in { @@ -183,4 +212,227 @@ class AggregationOperationSpec extends AnyFlatSpec { assert(a == b) assert(a.hashCode == b.hashCode) } +<<<<<<< HEAD +======= + + // --- getAggFunc: the accepting side of the four-clause type guard ---------- + + // SUM/MIN/MAX each guard on + // `!= INTEGER && != DOUBLE && != LONG && != TIMESTAMP`. The existing tests + // only drive the rejecting side plus INTEGER/DOUBLE, so the chain is never + // walked to its later clauses; TIMESTAMP in particular has to pass all four. + private val supportedAggTypes = Seq( + AttributeType.INTEGER, + AttributeType.DOUBLE, + AttributeType.LONG, + AttributeType.TIMESTAMP + ) + + private val guardedAggregations = Seq( + AggregationFunction.SUM -> "sum", + AggregationFunction.MIN -> "min", + AggregationFunction.MAX -> "max" + ) + + it should "accept every supported attribute type on SUM, MIN and MAX" in { + for ((func, name) <- guardedAggregations; t <- supportedAggTypes) + assert(op(func).getAggFunc(t) != null, s"$name should accept $t") + } + + it should "reject unsupported attribute types on SUM, MIN and MAX, naming the aggregation and the type" in { + val unsupported = + Seq(AttributeType.STRING, AttributeType.BOOLEAN, AttributeType.BINARY) + for ((func, name) <- guardedAggregations; t <- unsupported) { + val ex = intercept[UnsupportedOperationException](op(func).getAggFunc(t)) + assert(ex.getMessage == s"Unsupported attribute type for $name aggregation: $t") + } + } + + // --- AVERAGE over TIMESTAMP: the timestamp branch of getNumericalValue ----- + + // Everywhere else AVERAGE is driven over DOUBLE, which takes the + // `value.toString.toDouble` path. A TIMESTAMP column is the only route into + // the `parseTimestamp(...).getTime` branch. + private val earlier = Timestamp.valueOf("2020-03-05 10:00:00") + private val later = Timestamp.valueOf("2020-03-05 11:00:00") + private val midpointMillis = (earlier.getTime + later.getTime) / 2.0 + + "AVERAGE over a TIMESTAMP column" should "average the values' epoch milliseconds" in { + val agg = op(AggregationFunction.AVERAGE).getAggFunc(AttributeType.TIMESTAMP) + val state = Seq(earlier, later) + .map(ts => tupleOf("v", AttributeType.TIMESTAMP, ts)) + .foldLeft(agg.init())(agg.iterate) + + assert(agg.finalAgg(state).asInstanceOf[java.lang.Double] == midpointMillis) + } + + it should "combine per-worker partials through merge" in { + val agg = op(AggregationFunction.AVERAGE).getAggFunc(AttributeType.TIMESTAMP) + val p1 = agg.iterate(agg.init(), tupleOf("v", AttributeType.TIMESTAMP, earlier)) + val p2 = agg.iterate(agg.init(), tupleOf("v", AttributeType.TIMESTAMP, later)) + + val merged = agg.merge(p1, p2) + + assert(agg.finalAgg(merged).asInstanceOf[java.lang.Double] == midpointMillis) + } + + it should "ignore null timestamps and return null when every value is null" in { + val agg = op(AggregationFunction.AVERAGE).getAggFunc(AttributeType.TIMESTAMP) + + val mixed = Seq[AnyRef](earlier, null, later) + .map(ts => tupleOf("v", AttributeType.TIMESTAMP, ts)) + .foldLeft(agg.init())(agg.iterate) + assert(agg.finalAgg(mixed).asInstanceOf[java.lang.Double] == midpointMillis) + + val allNull = + agg.iterate(agg.init(), tupleOf("v", AttributeType.TIMESTAMP, null)) + assert(agg.finalAgg(allNull) == null) + } + + // --- merge: the partial-combination lambda of each aggregation -------------- + + // `AggregateOpSpec` drives init/iterate/finalAgg for every aggregation but only + // ever merges AVERAGE/CONCAT partials. The `merge` lambdas of SUM, COUNT, MIN + // and MAX are what the global stage calls when several workers report in, so + // each one is exercised here directly and cross-checked against the equivalent + // single-pass aggregation. + + private def aggregateAll( + agg: DistributedAggregation[Object], + t: AttributeType, + values: Seq[AnyRef] + ): Object = + agg.finalAgg(values.map(v => tupleOf("v", t, v)).foldLeft(agg.init())(agg.iterate)) + + "SUM aggregation merge" should "add two partials, matching a single-pass SUM" in { + val agg = op(AggregationFunction.SUM).getAggFunc(AttributeType.LONG) + val left = Seq[AnyRef](Long.box(1L), Long.box(2L)) + val right = Seq[AnyRef](Long.box(10L), Long.box(20L)) + + val p1 = left.map(v => tupleOf("v", AttributeType.LONG, v)).foldLeft(agg.init())(agg.iterate) + val p2 = right.map(v => tupleOf("v", AttributeType.LONG, v)).foldLeft(agg.init())(agg.iterate) + + assert(agg.finalAgg(agg.merge(p1, p2)) == Long.box(33L)) + assert(agg.finalAgg(agg.merge(p1, p2)) == aggregateAll(agg, AttributeType.LONG, left ++ right)) + // merging with an untouched (zero) partial must be a no-op + assert(agg.finalAgg(agg.merge(p1, agg.init())) == Long.box(3L)) + } + + "COUNT aggregation merge" should "add the per-worker counts" in { + val agg = op(AggregationFunction.COUNT).getAggFunc(AttributeType.INTEGER) + val p1 = Seq[AnyRef](Int.box(1), null, Int.box(3)) + .map(v => tupleOf("v", AttributeType.INTEGER, v)) + .foldLeft(agg.init())(agg.iterate) + val p2 = Seq[AnyRef](Int.box(4)) + .map(v => tupleOf("v", AttributeType.INTEGER, v)) + .foldLeft(agg.init())(agg.iterate) + + // COUNT(v) skips the null, so 2 + 1 == 3 + assert(agg.finalAgg(agg.merge(p1, p2)) == Int.box(3)) + assert(agg.finalAgg(agg.merge(agg.init(), agg.init())) == Int.box(0)) + } + + "MIN and MAX aggregation merge" should "pick the smaller and larger partial respectively" in { + val minAgg = op(AggregationFunction.MIN).getAggFunc(AttributeType.DOUBLE) + val maxAgg = op(AggregationFunction.MAX).getAggFunc(AttributeType.DOUBLE) + val left = Seq[AnyRef](Double.box(4.0), Double.box(9.0)) + val right = Seq[AnyRef](Double.box(-1.5), Double.box(2.0)) + + def partialOf(agg: DistributedAggregation[Object], values: Seq[AnyRef]): Object = + values.map(v => tupleOf("v", AttributeType.DOUBLE, v)).foldLeft(agg.init())(agg.iterate) + + assert( + minAgg.finalAgg(minAgg.merge(partialOf(minAgg, left), partialOf(minAgg, right))) + == Double.box(-1.5) + ) + // merge must be symmetric + assert( + minAgg.finalAgg(minAgg.merge(partialOf(minAgg, right), partialOf(minAgg, left))) + == Double.box(-1.5) + ) + assert( + maxAgg.finalAgg(maxAgg.merge(partialOf(maxAgg, left), partialOf(maxAgg, right))) + == Double.box(9.0) + ) + assert( + maxAgg.finalAgg(maxAgg.merge(partialOf(maxAgg, right), partialOf(maxAgg, left))) + == Double.box(9.0) + ) + } + + "MIN aggregation merge" should "stay neutral when one side saw no values" in { + val agg = op(AggregationFunction.MIN).getAggFunc(AttributeType.INTEGER) + val seen = agg.iterate(agg.init(), tupleOf("v", AttributeType.INTEGER, Int.box(7))) + + // An empty partial is the sentinel maxValue, so it must lose the comparison. + assert(agg.finalAgg(agg.merge(seen, agg.init())) == Int.box(7)) + assert(agg.finalAgg(agg.merge(agg.init(), seen)) == Int.box(7)) + // Two empty partials still finalize to null (no rows anywhere). + assert(agg.finalAgg(agg.merge(agg.init(), agg.init())) == null) + } + + "MAX aggregation merge" should "stay neutral when one side saw no values" in { + val agg = op(AggregationFunction.MAX).getAggFunc(AttributeType.INTEGER) + val seen = agg.iterate(agg.init(), tupleOf("v", AttributeType.INTEGER, Int.box(7))) + + // An empty partial is the sentinel minValue, so it must lose the comparison. + assert(agg.finalAgg(agg.merge(seen, agg.init())) == Int.box(7)) + assert(agg.finalAgg(agg.merge(agg.init(), seen)) == Int.box(7)) + // Two empty partials still finalize to null (no rows anywhere). + assert(agg.finalAgg(agg.merge(agg.init(), agg.init())) == null) + } + + // --- CONCAT: a null first value seeds the partial with an empty string ------ + + "CONCAT aggregation" should "swallow leading nulls but keep interior ones as empty slots" in { + // AggregateOpSpec only ever concatenates a null in the middle of the stream. + // A null on the very first tuple takes the `partial == ""` side of the branch + // and leaves the partial empty, so — unlike an interior null — it does not + // occupy a slot in the comma-joined output. + val agg = op(AggregationFunction.CONCAT).getAggFunc(AttributeType.STRING) + val result = Seq[AnyRef](null, "red", null, "blue") + .map(v => tupleOf("v", AttributeType.STRING, v)) + .foldLeft(agg.init())(agg.iterate) + + assert(agg.finalAgg(result) == "red,,blue") + } + + it should "return an empty string when every value is null" in { + val agg = op(AggregationFunction.CONCAT).getAggFunc(AttributeType.STRING) + val onlyNull = agg.iterate(agg.init(), tupleOf("v", AttributeType.STRING, null)) + + assert(agg.finalAgg(onlyNull) == "") + } + + it should "stringify non-string values it is pointed at" in { + // CONCAT is schema-restricted to STRING columns in the UI, but the executor + // only ever calls `.toString`, so an INTEGER column still concatenates. + val agg = op(AggregationFunction.CONCAT).getAggFunc(AttributeType.STRING) + val result = Seq[AnyRef](Int.box(1), Int.box(2)) + .map(v => tupleOf("v", AttributeType.INTEGER, v)) + .foldLeft(agg.init())(agg.iterate) + + assert(agg.finalAgg(result) == "1,2") + } + + // --- getAggregationAttribute / getFinal: message and identity guarantees ---- + + "getAggregationAttribute" should "name the unknown aggregation function in its error" in { + val ex = intercept[RuntimeException](op(null).getAggregationAttribute(AttributeType.INTEGER)) + assert(ex.getMessage == "Unknown aggregation function: null") + } + + "getFinal" should "produce a detached copy that re-reads the result column" in { + val original = op(AggregationFunction.MAX, attribute = "src", resultAttribute = "dst") + val copy = original.getFinal + + assert(copy ne original) + assert(copy.aggFunction == AggregationFunction.MAX) + // the final stage reads and writes the same (result) column + assert(copy.attribute == "dst") + assert(copy.resultAttribute == "dst") + // the original is left untouched + assert(original.attribute == "src") + } +>>>>>>> b0dd3ffc5 (fix(workflow-operator): MAX aggregation checks the wrong empty-group sentinel (#7532)) }