This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6907-38d94ead0ec267fb205857d16b61a1eed36172df in repository https://gitbox.apache.org/repos/asf/texera.git
commit 03a0e3a3b12424b99fa829d56fdb0a7512c6f173 Author: Meng Wang <[email protected]> AuthorDate: Sat Jul 25 23:44:15 2026 -0700 fix(workflow-operator): fall back to day semantics for an unset interval type; extend aggregate and interval-join tests (#6907) ### What changes were proposed in this PR? Writing the coverage tests asked for in #6901 turned up a live defect in `IntervalJoinOpExec`, so this fixes that and adds the tests around it. **An unset `timeIntervalType` crashed the timestamp branch instead of falling back to days.** `IntervalJoinOpDesc` declares `var timeIntervalType: Option[TimeIntervalType] = _`, and an absent JSON field deserializes to `Some(null)` — not `None`, and not `null`. The match in `intervalCompare` only handled `Some(<each unit>)` and `None`, so any TIMESTAMP interval join whose descriptor omits the field died with `scala.MatchError: Some(null)` rather than taking the documented day fallback. Replaced `case None` with a catch-all, which also covers `None` and any unit added later. This makes the fallback arm reachable for the first time; it was dead code before. On the coverage side, `AggregationOperationSpec` gains the accepting half of the SUM/MIN/MAX type guard — the four-clause `!= INTEGER && != DOUBLE && != LONG && != TIMESTAMP` chain was only ever driven with INTEGER/DOUBLE and the rejecting types, so its later clauses were never walked — plus AVERAGE over a TIMESTAMP column, which is the only route into the `parseTimestamp(...).getTime` branch of `getNumericalValue` (AVERAGE is otherwise only exercised over DOUBLE). Two items from the issue are deliberately **not** included, because they are already covered on `main`: the `TimeIntervalType` ladder and the `processNumValue` bound-combination arms were covered by #6098 (`"join timestamps across every interval unit"` plus the four `[] [) (] ()` tests), and `newAggFunc`'s non-COUNT arm is covered by the existing SUM `getFinal` pipeline test. `AveragePartialObj` is likewise already constructed by an existing test. The `maxAgg` finaliser asymmetry the issue mentions is left alone — it is a behavior change in live code and deserves its own PR. ### Any related issues, documentation, discussions? Closes #6901. ### How was this PR tested? Both touched specs pass (27 tests), and the full `WorkflowOperator/test` suite passes with 1957 tests and no regressions; `scalafmtCheck` and `scalafix --check` are clean. The new interval-join test is the regression test for the fix: reverting the catch-all back to `case None` reddens it with the original `scala.MatchError: Some(null)`, which is how the defect was found in the first place. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Fable 5) --- .../operator/intervalJoin/IntervalJoinOpExec.scala | 6 +- .../aggregate/AggregationOperationSpec.scala | 76 ++++++++++++++++++++++ .../operator/intervalJoin/IntervalOpExecSpec.scala | 29 +++++++++ 3 files changed, 110 insertions(+), 1 deletion(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpExec.scala index 4a4c987540..4dd6832d4d 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpExec.scala @@ -210,7 +210,11 @@ class IntervalJoinOpExec(descString: String) extends OperatorExecutor { Timestamp.valueOf(leftBoundValue.toLocalDateTime.plusMinutes(desc.constant)) case Some(TimeIntervalType.SECOND) => Timestamp.valueOf(leftBoundValue.toLocalDateTime.plusSeconds(desc.constant)) - case None => + case _ => + // Unset interval type falls back to day semantics. This has to be a + // catch-all rather than `case None`: an absent JSON field + // deserializes to Some(null), not None, so matching only on None + // threw a MatchError instead of taking the fallback. Timestamp.valueOf(leftBoundValue.toLocalDateTime.plusDays(desc.constant)) } result = processNumValue( 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 4ebba05fbc..0931e3bab7 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 @@ -22,6 +22,8 @@ package org.apache.texera.amber.operator.aggregate import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} import org.scalatest.flatspec.AnyFlatSpec +import java.sql.Timestamp + /** * Coverage notes: * `AggregateOpSpec` (in this same package) already exercises the happy paths for @@ -183,4 +185,78 @@ class AggregationOperationSpec extends AnyFlatSpec { assert(a == b) assert(a.hashCode == b.hashCode) } + + // --- 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) + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/intervalJoin/IntervalOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/intervalJoin/IntervalOpExecSpec.scala index aceab33f9d..cfc3f360e6 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/intervalJoin/IntervalOpExecSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/intervalJoin/IntervalOpExecSpec.scala @@ -519,6 +519,35 @@ class IntervalOpExecSpec extends AnyFlatSpec with BeforeAndAfter { } } + it should "fall back to day semantics when timeIntervalType is left unset" in { + // Every other test builds the desc through the 6-argument constructor, + // which always supplies Some(...). Leaving `timeIntervalType` unset omits + // the property, and it deserializes back as Some(null) — neither a + // recognised unit nor None. That is what used to throw a MatchError, and + // it is why the fallback has to be a catch-all instead of `case None`. + val desc = new IntervalJoinOpDesc() + desc.leftAttributeName = "point" + desc.rightAttributeName = "range" + desc.constant = 3L + desc.includeLeftBound = true + desc.includeRightBound = true + + val exec = new IntervalJoinOpExec(objectMapper.writeValueAsString(desc)) + exec.open() + try { + val base = Timestamp.valueOf("2020-03-05 00:00:00") + assert(exec.processTuple(timeStampTuple("range", 1, base), right).isEmpty) + + // The fallback adds `constant` DAYS, so +2 days is inside [base, base+3d] + // and +4 days is outside. Dates avoid Feb 29 and DST boundaries. + val inside = Timestamp.valueOf("2020-03-07 00:00:00") + assert(exec.processTuple(timeStampTuple("point", 1, inside), left).toList.size == 1) + + val outside = Timestamp.valueOf("2020-03-09 00:00:00") + assert(exec.processTuple(timeStampTuple("point", 1, outside), left).toList.isEmpty) + } finally exec.close() + } + it should "reject a join key whose type does not support interval comparison" in { val desc = new IntervalJoinOpDesc("point", "range", 3L, true, true, TimeIntervalType.DAY) val exec = new IntervalJoinOpExec(objectMapper.writeValueAsString(desc))
