This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 03a0e3a3b1 fix(workflow-operator): fall back to day semantics for an
unset interval type; extend aggregate and interval-join tests (#6907)
03a0e3a3b1 is described below
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))
