[replay #7532] fix(workflow-operator): MAX aggregation checks the wrong empty-group sentinel

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
     )
   }
 

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)

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 {