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 ddfe28469b test(amber): cover CostBasedScheduleGenerator greedy search
paths (#6157)
ddfe28469b is described below
commit ddfe28469bc164937bbf24488b4f84c9c91b6871
Author: Xinyuan Lin <[email protected]>
AuthorDate: Mon Jul 6 22:43:30 2026 -0700
test(amber): cover CostBasedScheduleGenerator greedy search paths (#6157)
### What changes were proposed in this PR?
Extends `CostBasedScheduleGeneratorSpec` to cover the greedy
(`globalSearch = false`) search paths, which the existing tests never
exercised (they only ever call `bottomUpSearch`/`topDownSearch` with
`globalSearch = true`). Reusing the spec's existing pure-JVM harness
(the 4-op `csv → filter → join → filter` workflow), the new tests call
`bottomUpSearch()` and `topDownSearch()` with default args (greedy) and
assert a schedulable result (non-empty region DAG, finite cost, chosen
state ⊆ physical links, and fewer states explored than global search).
No production code changed.
### Any related issues, documentation, discussions?
Coverage gaps identified from the Codecov report for `apache/texera`
(amber module coverage push toward 80%).
### How was this PR tested?
```
sbt "WorkflowExecutionService/testOnly *CostBasedScheduleGeneratorSpec"
```
All 11 pass. `scalafmt` and `scalafixAll --check` are clean.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8 [1M context])
---------
Co-authored-by: Claude Opus 4.8 <[email protected]>
---
.../CostBasedScheduleGeneratorSpec.scala | 125 +++++++++++++++++++++
1 file changed, 125 insertions(+)
diff --git
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGeneratorSpec.scala
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGeneratorSpec.scala
index c9c2843acd..73f80c1b3c 100644
---
a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGeneratorSpec.scala
+++
b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGeneratorSpec.scala
@@ -552,4 +552,129 @@ class CostBasedScheduleGeneratorSpec extends AnyFlatSpec
with MockFactory {
)
}
+ "CostBasedRegionPlanGenerator" should "finish bottom-up greedy search
(globalSearch=false) in csv->->filter->join->filter2 workflow" in {
+ val headerlessCsvOpDesc1 = TestOperators.headerlessSmallCsvScanOpDesc()
+ val keywordOpDesc = TestOperators.keywordSearchOpDesc("column-1", "Asia")
+ val joinOpDesc = TestOperators.joinOpDesc("column-1", "column-1")
+ val keywordOpDesc2 = TestOperators.keywordSearchOpDesc("column-1", "Asia")
+ val workflow = buildWorkflow(
+ List(
+ headerlessCsvOpDesc1,
+ keywordOpDesc,
+ joinOpDesc,
+ keywordOpDesc2
+ ),
+ List(
+ LogicalLink(
+ headerlessCsvOpDesc1.operatorIdentifier,
+ PortIdentity(),
+ joinOpDesc.operatorIdentifier,
+ PortIdentity()
+ ),
+ LogicalLink(
+ headerlessCsvOpDesc1.operatorIdentifier,
+ PortIdentity(),
+ keywordOpDesc.operatorIdentifier,
+ PortIdentity()
+ ),
+ LogicalLink(
+ keywordOpDesc.operatorIdentifier,
+ PortIdentity(),
+ joinOpDesc.operatorIdentifier,
+ PortIdentity(1)
+ ),
+ LogicalLink(
+ joinOpDesc.operatorIdentifier,
+ PortIdentity(),
+ keywordOpDesc2.operatorIdentifier,
+ PortIdentity()
+ )
+ ),
+ new WorkflowContext()
+ )
+
+ val scheduleGenerator = new CostBasedScheduleGenerator(
+ workflow.context,
+ workflow.physicalPlan,
+ COORDINATOR
+ )
+
+ // Greedy search (globalSearch = false): at each schedulable/unschedulable
state the frontier keeps only
+ // the single lowest-cost neighbor, driving the greedy branch
(filteredNeighborStates.nonEmpty + minBy).
+ val greedyResult = scheduleGenerator.bottomUpSearch(globalSearch = false)
+
+ // A schedulable plan should have been found: the region DAG is non-empty
and the cost is finite.
+ assert(greedyResult.regionDAG.vertexSet().asScala.nonEmpty)
+ assert(greedyResult.cost < Double.PositiveInfinity)
+
+ // The greedy search enqueues at most one neighbor per explored state, and
each bottom-up transition materializes
+ // one more edge, so the number of states it explores is bounded linearly
by the number of physical links. This is
+ // a guaranteed property of greedy search, unlike a comparison against
global search whose explored count depends on
+ // early-stop pruning and queue ordering.
+ assert(greedyResult.numStatesExplored <=
scheduleGenerator.physicalPlan.links.size + 1)
+
+ // The chosen state is a set of materialized non-blocking edges, all of
which must be links of the physical plan.
+ assert(greedyResult.state.subsetOf(scheduleGenerator.physicalPlan.links))
+ }
+
+ "CostBasedRegionPlanGenerator" should "finish top-down greedy search
(globalSearch=false) in csv->->filter->join->filter2 workflow" in {
+ val headerlessCsvOpDesc1 = TestOperators.headerlessSmallCsvScanOpDesc()
+ val keywordOpDesc = TestOperators.keywordSearchOpDesc("column-1", "Asia")
+ val joinOpDesc = TestOperators.joinOpDesc("column-1", "column-1")
+ val keywordOpDesc2 = TestOperators.keywordSearchOpDesc("column-1", "Asia")
+ val workflow = buildWorkflow(
+ List(
+ headerlessCsvOpDesc1,
+ keywordOpDesc,
+ joinOpDesc,
+ keywordOpDesc2
+ ),
+ List(
+ LogicalLink(
+ headerlessCsvOpDesc1.operatorIdentifier,
+ PortIdentity(),
+ joinOpDesc.operatorIdentifier,
+ PortIdentity()
+ ),
+ LogicalLink(
+ headerlessCsvOpDesc1.operatorIdentifier,
+ PortIdentity(),
+ keywordOpDesc.operatorIdentifier,
+ PortIdentity()
+ ),
+ LogicalLink(
+ keywordOpDesc.operatorIdentifier,
+ PortIdentity(),
+ joinOpDesc.operatorIdentifier,
+ PortIdentity(1)
+ ),
+ LogicalLink(
+ joinOpDesc.operatorIdentifier,
+ PortIdentity(),
+ keywordOpDesc2.operatorIdentifier,
+ PortIdentity()
+ )
+ ),
+ new WorkflowContext()
+ )
+
+ val scheduleGenerator = new CostBasedScheduleGenerator(
+ workflow.context,
+ workflow.physicalPlan,
+ COORDINATOR
+ )
+
+ // Greedy search (globalSearch = false): starting from the fully
materialized seed state, each transition
+ // keeps only the single lowest-cost neighbor, driving the greedy branch
(unvisitedNeighborStates.nonEmpty + minBy)
+ // over both the schedulable (Left) and unschedulable-intermediate (Right)
legs.
+ val greedyResult = scheduleGenerator.topDownSearch(globalSearch = false)
+
+ // A schedulable plan should have been found: the region DAG is non-empty
and the cost is finite.
+ assert(greedyResult.regionDAG.vertexSet().asScala.nonEmpty)
+ assert(greedyResult.cost < Double.PositiveInfinity)
+
+ // The chosen state is a set of materialized non-blocking edges, all of
which must be links of the physical plan.
+ assert(greedyResult.state.subsetOf(scheduleGenerator.physicalPlan.links))
+ }
+
}
