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