(texera) 01/01: test(workflow-operator): add unit test coverage for interval-join, and sleep (#6098)

[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-6098-3084881cd96a6896664ab1bed0e97aa11a85039d
in repository https://gitbox.apache.org/repos/asf/texera.git

commit 3a12e6214c5f45477042374ae3a4a569835d1fda
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Sat Jul 4 18:32:04 2026 -0700

    test(workflow-operator): add unit test coverage for interval-join, and sleep (#6098)
    
    ### What changes were proposed in this PR?
    
    Add unit test coverage for two `workflow-operator` executors, selected
    from the Codecov report. No production-code changes.
    
    | File | Codecov before | New coverage |
    | --- | --- | --- |
    | `IntervalJoinOpExec.scala` | 81% | extends the existing spec to cover
    the YEAR/MONTH/HOUR/MINUTE/SECOND timestamp interval-unit arms and the
    unsupported-key-type `WorkflowRuntimeException` |
    | `SleepOpExec.scala` | 0% | new spec: construction from a serialized
    desc + the pass-through `processTuple` (sleepTime 0) |
    
    The `IntervalJoin` `case None` interval arm requires a JSON payload with
    the field absent (Jackson maps an explicit `None` to `Some(null)`), so
    it is left to integration.
    
    > Note: an earlier revision of this PR also added a
    `CacheSourceOpExecSpec`. `CacheSourceOpExec` turned out to be dead code
    — cache source operators were replaced by input-port materialization
    reader threads in #3425, and its only factory
    (`SpecialPhysicalOpFactory.newSourcePhysicalOp`) has no production
    caller — so that spec was dropped and removal of the operator is tracked
    in #6125.
    
    ### Any related issues, documentation, discussions?
    
    Follow-up to the review feedback on #6043: prioritize tests that fill
    uncovered code paths.
    
    ### How was this PR tested?
    
    - `sbt "WorkflowOperator/testOnly *IntervalOpExecSpec *SleepOpExecSpec"`
    — all green
    - `sbt "WorkflowOperator/Test/scalafmtCheck"` and `sbt
    "WorkflowOperator/scalafixAll --check"` — clean
    
    ### Was this PR authored or co-authored using generative AI tooling?
    
    Generated-by: Claude Code (Opus 4.8 [1M context])
---
 .../operator/intervalJoin/IntervalOpExecSpec.scala | 35 +++++++++++++
 .../amber/operator/sleep/SleepOpExecSpec.scala     | 57 ++++++++++++++++++++++
 2 files changed, 92 insertions(+)

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 97c6012983..aceab33f9d 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
@@ -498,4 +498,39 @@ class IntervalOpExecSpec extends AnyFlatSpec with BeforeAndAfter {
     )
   }
 
+  it should "join timestamps across every interval unit" in {
+    val base = Timestamp.valueOf("2020-01-01 00:00:00")
+    Seq(
+      TimeIntervalType.YEAR,
+      TimeIntervalType.MONTH,
+      TimeIntervalType.HOUR,
+      TimeIntervalType.MINUTE,
+      TimeIntervalType.SECOND
+    ).foreach { unit =>
+      val desc = new IntervalJoinOpDesc("point", "range", 3L, true, true, unit)
+      val exec = new IntervalJoinOpExec(objectMapper.writeValueAsString(desc))
+      exec.open()
+      try {
+        assert(exec.processTuple(timeStampTuple("range", 1, base), right).isEmpty)
+        val out = exec.processTuple(timeStampTuple("point", 1, base), left).toList
+        assert(out.size == 1, s"expected a match for unit $unit")
+        assert(out.head.getFields.length == 4)
+      } 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))
+    exec.open()
+    assert(
+      exec.processTuple(newTuple[String]("range", 1, "a", AttributeType.STRING), right).isEmpty
+    )
+    val ex = intercept[org.apache.texera.amber.core.WorkflowRuntimeException] {
+      exec.processTuple(newTuple[String]("point", 1, "a", AttributeType.STRING), left).toList
+    }
+    assert(ex.getMessage == "The data type can not support comparison: string")
+    exec.close()
+  }
+
 }
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sleep/SleepOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sleep/SleepOpExecSpec.scala
new file mode 100644
index 0000000000..2aa143f294
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sleep/SleepOpExecSpec.scala
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.operator.sleep
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.scalatest.flatspec.AnyFlatSpec
+
+class SleepOpExecSpec extends AnyFlatSpec {
+
+  private val schema: Schema = Schema().add(new Attribute("v", AttributeType.INTEGER))
+
+  private def tuple(v: Int): Tuple =
+    Tuple.builder(schema).add(new Attribute("v", AttributeType.INTEGER), Integer.valueOf(v)).build()
+
+  // SleepOpDesc is a LogicalOp: serialize a real instance so the polymorphic operatorType
+  // discriminator is present (a hand-written JSON string would fail to deserialize).
+  private def descString(sleepTime: Int): String = {
+    val desc = new SleepOpDesc()
+    desc.sleepTime = sleepTime
+    objectMapper.writeValueAsString(desc)
+  }
+
+  "SleepOpExec" should "construct from a serialized SleepOpDesc" in {
+    val exec = new SleepOpExec(descString(0))
+    assert(exec != null)
+  }
+
+  "SleepOpExec.processTuple" should "return the input tuple unchanged" in {
+    // sleepTime = 0 -> Thread.sleep(0), instant
+    val exec = new SleepOpExec(descString(0))
+    assert(exec.processTuple(tuple(7), 0).toList == List(tuple(7)))
+  }
+
+  it should "emit exactly one tuple per input" in {
+    val exec = new SleepOpExec(descString(0))
+    val emitted = (0 until 5).flatMap(i => exec.processTuple(tuple(i), 0).toList)
+    assert(emitted == (0 until 5).map(tuple).toList)
+  }
+}