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/06/13 17:17:39 UTC
(texera) 01/01: test(amber): add unit test coverage for LogicalPlan (#5441)
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-5441-7ae9b35f12748616daf7bcc925fdde2e5def5187
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 6032413f6f1e65b6a32e53542c4668f00698ec26
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Sat Jun 13 10:17:17 2026 -0700
test(amber): add unit test coverage for LogicalPlan (#5441)
### What changes were proposed in this PR?
Adds `LogicalPlanSpec` covering
`amber/src/main/scala/org/apache/texera/workflow/LogicalPlan.scala` —
the user-facing logical workflow graph case class plus its companion
factory.
| Surface | Pinned |
| --- | --- |
| Construction | `LogicalPlan(operators, links)` exposes both fields
verbatim. |
| `LogicalPlan.apply(LogicalPlanPojo)` | Lifts the POJO's operators +
links into a `LogicalPlan`, ignoring the POJO-only `opsToViewResult` /
`opsToReuseResult` fields. |
| `getTopologicalOpIds` | Topological order on a linear chain; respects
edge directionality across a fan-out (`a → b, a → c`) — source first,
two sinks unordered in the tail. |
| `getOperator` | Returns the operator with the requested id; throws
`NoSuchElementException` for an unknown id. |
| `getTerminalOperatorIds` | Single sink in a linear chain; every
out-degree-0 operator in a fan-out plan; every operator when there are
no links; empty list for an empty plan. |
| `getUpstreamLinks` | Returns every link whose `toOpId` matches the
argument; preserves construction order when multiple links flow into the
same target; returns an empty list when nothing flows in. |
| `resolveScanSourceOpFileName` | Failures with `Some(errorList)` are
appended per-operator instead of throwing; with `None` the first failure
rethrows; non-`ScanSourceOpDesc` operators are left untouched (no
errors, no resolution). Failures are forced deterministically by
pointing a `ScanSourceOpDesc` fixture at a non-existent file path. |
A happy-path `resolveScanSourceOpFileName` test is intentionally
omitted: `FileResolver` reaches the LakeFS / dataset service in
production and is environment-dependent, so a deterministic unit test
would have to mock that surface — out of scope for this spec.
No production code changed; this is test-only.
### Any related issues, documentation, discussions?
Closes #5438
### How was this PR tested?
```
sbt "WorkflowExecutionService/Test/testOnly org.apache.texera.workflow.LogicalPlanSpec"
# → 16 tests, all pass
sbt "WorkflowExecutionService/Test/scalafmtCheck"
# → clean
```
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.7)
---
.../apache/texera/workflow/LogicalPlanSpec.scala | 344 +++++++++++++++++++++
1 file changed, 344 insertions(+)
diff --git a/amber/src/test/scala/org/apache/texera/workflow/LogicalPlanSpec.scala b/amber/src/test/scala/org/apache/texera/workflow/LogicalPlanSpec.scala
new file mode 100644
index 0000000000..cbc28388dd
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/workflow/LogicalPlanSpec.scala
@@ -0,0 +1,344 @@
+/*
+ * 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.workflow
+
+import org.apache.commons.vfs2.FileNotFoundException
+import org.apache.texera.amber.core.virtualidentity.OperatorIdentity
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.operator.aggregate.AggregateOpDesc
+import org.apache.texera.amber.operator.hashJoin.HashJoinOpDesc
+import org.apache.texera.amber.operator.keywordSearch.KeywordSearchOpDesc
+import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc
+import org.apache.texera.amber.operator.source.scan.json.JSONLScanSourceOpDesc
+import org.apache.texera.amber.operator.split.SplitOpDesc
+import org.apache.texera.amber.operator.udf.python.DualInputPortsPythonUDFOpDescV2
+import org.apache.texera.amber.operator.udf.python.source.PythonUDFSourceOpDescV2
+import org.apache.texera.amber.operator.union.UnionOpDesc
+import org.apache.texera.amber.operator.{LogicalOp, TestOperators}
+import org.apache.texera.web.model.websocket.request.LogicalPlanPojo
+import org.scalatest.flatspec.AnyFlatSpec
+
+import scala.collection.mutable.ArrayBuffer
+import scala.jdk.CollectionConverters.IteratorHasAsScala
+
+class LogicalPlanSpec extends AnyFlatSpec {
+
+ // ---------------------------------------------------------------------------
+ // Fixtures
+ // ---------------------------------------------------------------------------
+
+ private def csv(): CSVScanSourceOpDesc = TestOperators.headerlessSmallCsvScanOpDesc()
+ private def keyword(): KeywordSearchOpDesc = TestOperators.keywordSearchOpDesc("column-1", "Asia")
+
+ /** Give an operator a readable, stable id so failure messages and graph
+ * assertions read in terms of the operator's role rather than a UUID.
+ */
+ private def named[T <: LogicalOp](op: T, id: String): T = {
+ op.setOperatorId(id)
+ op
+ }
+
+ private def link(
+ from: OperatorIdentity,
+ to: OperatorIdentity,
+ fromPort: Int = 0,
+ toPort: Int = 0
+ ): LogicalLink =
+ LogicalLink(from, PortIdentity(fromPort), to, PortIdentity(toPort))
+
+ // ---------------------------------------------------------------------------
+ // A realistic, non-trivial workflow used by the graph-query tests below.
+ //
+ // Sources (three different operator types):
+ // csv : CSV scan json : JSONL scan pySrc : Python UDF source
+ //
+ // Edges (target port in parentheses):
+ // csv ─▶ join (0, "left") join ─▶ udf2 (0, "model")
+ // json ─▶ join (1, "right") pySrc ─▶ udf2 (1, "tuples")
+ // udf2 ─▶ union (0) json ─▶ union (0) ← fan-out: 2nd link into the SAME port
+ // union ─▶ split (0)
+ // split ─▶ aggA (from out 0) split ─▶ aggB (from out 1) ← two distinct OUTPUT ports
+ //
+ // Shapes exercised: a join (two distinct input ports), a dual-input UDF
+ // (two distinct input ports), a union (two links merged into ONE input
+ // port), a split (two distinct output ports), and a fan-out source
+ // (json feeds both the join and the union).
+ // Sources (in-degree 0): csv, json, pySrc. Terminals (out-degree 0): aggA, aggB.
+ // ---------------------------------------------------------------------------
+
+ private case class Rich(
+ plan: LogicalPlan,
+ csv: CSVScanSourceOpDesc,
+ json: JSONLScanSourceOpDesc,
+ pySrc: PythonUDFSourceOpDescV2,
+ join: HashJoinOpDesc[String],
+ udf2: DualInputPortsPythonUDFOpDescV2,
+ union: UnionOpDesc,
+ split: SplitOpDesc,
+ aggA: AggregateOpDesc,
+ aggB: AggregateOpDesc
+ )
+
+ private def richPlan(): Rich = {
+ val csv = named(TestOperators.headerlessSmallCsvScanOpDesc(), "csv-source")
+ val json = named(TestOperators.smallJSONLScanOpDesc(), "jsonl-source")
+ val pySrc = named(TestOperators.pythonSourceOpDesc(5), "python-source")
+ val join = named(TestOperators.joinOpDesc("country", "country"), "hash-join")
+ val udf2 = named(new DualInputPortsPythonUDFOpDescV2(), "dual-input-udf")
+ val union = named(new UnionOpDesc(), "union")
+ val split = named(new SplitOpDesc(), "split")
+ val aggA = named(new AggregateOpDesc(), "aggregate-a")
+ val aggB = named(new AggregateOpDesc(), "aggregate-b")
+
+ // Order matters for getUpstreamLinks: links are returned in construction order.
+ val links = List(
+ link(csv.operatorIdentifier, join.operatorIdentifier, toPort = 0), // join: left
+ link(json.operatorIdentifier, join.operatorIdentifier, toPort = 1), // join: right
+ link(join.operatorIdentifier, udf2.operatorIdentifier, toPort = 0), // udf2: model
+ link(pySrc.operatorIdentifier, udf2.operatorIdentifier, toPort = 1), // udf2: tuples
+ link(udf2.operatorIdentifier, union.operatorIdentifier, toPort = 0), // union <- udf2
+ link(
+ json.operatorIdentifier,
+ union.operatorIdentifier,
+ toPort = 0
+ ), // union <- json (2nd into port 0)
+ link(union.operatorIdentifier, split.operatorIdentifier, toPort = 0),
+ link(split.operatorIdentifier, aggA.operatorIdentifier, fromPort = 0), // split: output 0
+ link(split.operatorIdentifier, aggB.operatorIdentifier, fromPort = 1) // split: output 1
+ )
+
+ // Deliberately pass the operators in a NON-topological order so the
+ // topological-sort test proves real reordering, not input identity.
+ val operators = List[LogicalOp](aggB, split, union, udf2, join, csv, pySrc, json, aggA)
+
+ Rich(LogicalPlan(operators, links), csv, json, pySrc, join, udf2, union, split, aggA, aggB)
+ }
+
+ // ---------------------------------------------------------------------------
+ // Construction
+ // ---------------------------------------------------------------------------
+
+ "LogicalPlan" should "expose the operators and links it was constructed with" in {
+ val a = csv()
+ val b = keyword()
+ val l = link(a.operatorIdentifier, b.operatorIdentifier)
+ val plan = LogicalPlan(List(a, b), List(l))
+ assert(plan.operators == List(a, b))
+ assert(plan.links == List(l))
+ }
+
+ "LogicalPlan.apply(LogicalPlanPojo)" should
+ "lift the POJO's operators and links, ignoring opsToViewResult / opsToReuseResult" in {
+ val a = csv()
+ val b = keyword()
+ val l = link(a.operatorIdentifier, b.operatorIdentifier)
+ val pojo = LogicalPlanPojo(
+ operators = List(a, b),
+ links = List(l),
+ // `operatorId` is private — use the public OperatorIdentity wrapper.
+ opsToViewResult = List(b.operatorIdentifier.id), // intentionally non-empty
+ opsToReuseResult = List(a.operatorIdentifier.id)
+ )
+ val plan = LogicalPlan(pojo)
+ assert(plan.operators == List(a, b))
+ assert(plan.links == List(l))
+ }
+
+ // ---------------------------------------------------------------------------
+ // getTopologicalOpIds
+ // ---------------------------------------------------------------------------
+
+ "LogicalPlan.getTopologicalOpIds" should "yield a topological order on a linear chain" in {
+ val a = csv()
+ val b = keyword()
+ val plan = LogicalPlan(List(a, b), List(link(a.operatorIdentifier, b.operatorIdentifier)))
+ val order = plan.getTopologicalOpIds.asScala.toList
+ assert(order == List(a.operatorIdentifier, b.operatorIdentifier))
+ }
+
+ it should "honor every edge on a realistic plan (join, union, dual-input UDF, split fan-out)" in {
+ val r = richPlan()
+ val order = r.plan.getTopologicalOpIds.asScala.toList
+
+ // Every operator appears exactly once.
+ assert(order.size == r.plan.operators.size)
+ assert(order.toSet == r.plan.operators.map(_.operatorIdentifier).toSet)
+
+ // The defining property of a topological order: each edge points "forward".
+ val positionOf = order.zipWithIndex.toMap
+ r.plan.links.foreach { l =>
+ assert(
+ positionOf(l.fromOpId) < positionOf(l.toOpId),
+ s"topological order violates edge ${l.fromOpId.id} -> ${l.toOpId.id}; " +
+ s"order = ${order.map(_.id)}"
+ )
+ }
+ }
+
+ // ---------------------------------------------------------------------------
+ // getOperator
+ // ---------------------------------------------------------------------------
+
+ "LogicalPlan.getOperator" should "return the operator with the requested identifier" in {
+ val r = richPlan()
+ assert(r.plan.getOperator(r.join.operatorIdentifier) eq r.join)
+ assert(r.plan.getOperator(r.split.operatorIdentifier) eq r.split)
+ assert(r.plan.getOperator(r.pySrc.operatorIdentifier) eq r.pySrc)
+ }
+
+ it should "throw NoSuchElementException for an unknown operator id" in {
+ val a = csv()
+ val plan = LogicalPlan(List(a), List.empty)
+ intercept[NoSuchElementException] {
+ plan.getOperator(OperatorIdentity("not-in-plan"))
+ }
+ }
+
+ // ---------------------------------------------------------------------------
+ // getTerminalOperatorIds
+ // ---------------------------------------------------------------------------
+
+ "LogicalPlan.getTerminalOperatorIds" should
+ "return every out-degree-0 operator on a realistic plan (the two split sinks)" in {
+ val r = richPlan()
+ assert(
+ r.plan.getTerminalOperatorIds.toSet ==
+ Set(r.aggA.operatorIdentifier, r.aggB.operatorIdentifier)
+ )
+ }
+
+ it should "return the single sink in a linear chain" in {
+ val a = csv()
+ val b = keyword()
+ val plan = LogicalPlan(List(a, b), List(link(a.operatorIdentifier, b.operatorIdentifier)))
+ assert(plan.getTerminalOperatorIds == List(b.operatorIdentifier))
+ }
+
+ it should "return every operator when there are no links" in {
+ // An isolated set of operators with no edges — every operator has
+ // out-degree 0 and is therefore terminal.
+ val a = csv()
+ val b = keyword()
+ val plan = LogicalPlan(List(a, b), List.empty)
+ assert(plan.getTerminalOperatorIds.toSet == Set(a.operatorIdentifier, b.operatorIdentifier))
+ }
+
+ it should "return an empty list for an empty plan" in {
+ val plan = LogicalPlan(List.empty, List.empty)
+ assert(plan.getTerminalOperatorIds.isEmpty)
+ }
+
+ // ---------------------------------------------------------------------------
+ // getUpstreamLinks
+ // ---------------------------------------------------------------------------
+
+ "LogicalPlan.getUpstreamLinks" should
+ "return both inbound links on their distinct ports for a join" in {
+ val r = richPlan()
+ val up = r.plan.getUpstreamLinks(r.join.operatorIdentifier)
+ // Construction order: csv (left, port 0) then json (right, port 1).
+ assert(up.map(_.fromOpId) == List(r.csv.operatorIdentifier, r.json.operatorIdentifier))
+ assert(up.map(_.toPortId) == List(PortIdentity(0), PortIdentity(1)))
+ }
+
+ it should "return both inbound links on their distinct ports for a dual-input UDF" in {
+ val r = richPlan()
+ val up = r.plan.getUpstreamLinks(r.udf2.operatorIdentifier)
+ assert(up.map(_.fromOpId) == List(r.join.operatorIdentifier, r.pySrc.operatorIdentifier))
+ assert(up.map(_.toPortId) == List(PortIdentity(0), PortIdentity(1)))
+ }
+
+ it should "return every inbound link merged into a union's single input port, in order" in {
+ val r = richPlan()
+ val up = r.plan.getUpstreamLinks(r.union.operatorIdentifier)
+ // udf2 and json both feed union's port 0; the result preserves construction order.
+ assert(up.map(_.fromOpId) == List(r.udf2.operatorIdentifier, r.json.operatorIdentifier))
+ assert(up.map(_.toPortId).toSet == Set(PortIdentity(0)))
+ }
+
+ it should "return an empty list for a source operator with no inbound links" in {
+ val r = richPlan()
+ assert(r.plan.getUpstreamLinks(r.csv.operatorIdentifier).isEmpty)
+ }
+
+ // ---------------------------------------------------------------------------
+ // resolveScanSourceOpFileName
+ // ---------------------------------------------------------------------------
+ //
+ // A successful resolution of a real file path is environment-dependent
+ // (resolved through FileResolver, which can reach LakeFS / dataset
+ // service), so we pin the FAILURE behavior here. The method has two modes,
+ // both used in production:
+ // 1. Some(errorList) — every failing operator is appended to errorList by
+ // opId and resolution CONTINUES (collect-all). Used by the editing-path
+ // compiler in workflow-compiling-service so the UI can surface every
+ // bad operator at once.
+ // 2. None — the first failure rethrows (fail-fast). Used by the amber
+ // WorkflowCompiler on the execution path, right before a run.
+ // To force a failure deterministically, we point a ScanSource fixture at a
+ // non-existent file and assert the error surfaces both ways.
+
+ private def csvWithMissingFile(): CSVScanSourceOpDesc = {
+ val op = TestOperators.headerlessSmallCsvScanOpDesc()
+ // Use a relative single-segment file name so `FileResolver` short-
+ // circuits BOTH resolvers without touching the DB: localResolveFunc
+ // fails (file doesn't exist) and datasetResolveFunc's parser bails
+ // immediately (path segments < 4) before any dataset DB query.
+ op.fileName = Some("nonexistent-test-file.csv")
+ op
+ }
+
+ "LogicalPlan.resolveScanSourceOpFileName" should
+ "append a per-operator error for every failing scan source instead of throwing" in {
+ // Two independent broken scans: Some(errorList) must collect BOTH (it does
+ // not stop at the first failure), keyed by their distinct operator ids.
+ val brokenA = csvWithMissingFile()
+ val brokenB = csvWithMissingFile()
+ val plan = LogicalPlan(List(brokenA, brokenB), List.empty)
+ val errors = ArrayBuffer.empty[(OperatorIdentity, Throwable)]
+ plan.resolveScanSourceOpFileName(Some(errors))
+ assert(errors.size == 2, s"expected both failures captured, got: $errors")
+ assert(
+ errors.map(_._1).toSet ==
+ Set(brokenA.operatorIdentifier, brokenB.operatorIdentifier)
+ )
+ }
+
+ it should "rethrow FileNotFoundException when no errorList is provided" in {
+ // `FileResolver.resolve` raises `org.apache.commons.vfs2.FileNotFoundException`
+ // when all resolvers fail. Pin the specific type so an unrelated failure
+ // mode (e.g. NPE) doesn't silently satisfy this test.
+ val brokenOp = csvWithMissingFile()
+ val plan = LogicalPlan(List(brokenOp), List.empty)
+ intercept[FileNotFoundException] {
+ plan.resolveScanSourceOpFileName(None)
+ }
+ }
+
+ it should "leave non-ScanSourceOpDesc operators untouched (no errors, no resolution)" in {
+ // KeywordSearch is not a ScanSourceOpDesc — the `case _` branch
+ // skips it entirely. An errorList provided here must remain empty.
+ val k = keyword()
+ val plan = LogicalPlan(List(k), List.empty)
+ val errors = ArrayBuffer.empty[(OperatorIdentity, Throwable)]
+ plan.resolveScanSourceOpFileName(Some(errors))
+ assert(errors.isEmpty)
+ }
+}