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/07 05:43:51 UTC
(texera) branch gh-readonly-queue/main/pr-6157-db00f941646739688aad8251c6ab37527b9f1092 created (now ddfe28469b)
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-6157-db00f941646739688aad8251c6ab37527b9f1092
in repository https://gitbox.apache.org/repos/asf/texera.git
at ddfe28469b test(amber): cover CostBasedScheduleGenerator greedy search paths (#6157)
This branch includes the following new commits:
new ddfe28469b test(amber): cover CostBasedScheduleGenerator greedy search paths (#6157)
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: test(amber): cover CostBasedScheduleGenerator greedy search paths (#6157)
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-6157-db00f941646739688aad8251c6ab37527b9f1092
in repository https://gitbox.apache.org/repos/asf/texera.git
commit ddfe28469bc164937bbf24488b4f84c9c91b6871
Author: Xinyuan Lin <xi...@uci.edu>
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 <no...@anthropic.com>
---
.../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))
+ }
+
}