You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/07/26 06:32:38 UTC
(texera) branch gh-readonly-queue/main/pr-6907-38d94ead0ec267fb205857d16b61a1eed36172df created (now 00b80151e6)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6907-38d94ead0ec267fb205857d16b61a1eed36172df
in repository https://gitbox.apache.org/repos/asf/texera.git
at 00b80151e6 fix(workflow-operator): fall back to day semantics for an unset interval type; extend aggregate and interval-join tests (#6907)
This branch includes the following new commits:
new 00b80151e6 fix(workflow-operator): fall back to day semantics for an unset interval type; extend aggregate and interval-join tests (#6907)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: fix(workflow-operator): fall back to day semantics for an unset interval type; extend aggregate and interval-join tests (#6907)
Posted by gi...@apache.org.
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 00b80151e6a53f3cf0f0dff04cfffb88cf33524d
Author: Meng Wang <me...@uci.edu>
AuthorDate: Sat Jul 25 23:32:17 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))