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 78ee70579c6..67146e81af2 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 @@ -237,7 +237,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 3730181d9dd..1cf05a0dc25 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 */ @@ -438,6 +440,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 44f0f1b3dae..444d8247c8e 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 @@ -175,6 +175,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 { @@ -342,6 +371,17 @@ class AggregationOperationSpec extends AnyFlatSpec { 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 {