Copilot commented on code in PR #6157:
URL: https://github.com/apache/texera/pull/6157#discussion_r3524611499
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGeneratorSpec.scala:
##########
@@ -552,4 +552,132 @@ 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 defaults to 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()
+
+ // 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 only enqueues one neighbor per state, so it must
explore fewer states than a global search
+ // that enqueues every unvisited neighbor.
+ val globalResult = new CostBasedScheduleGenerator(
+ workflow.context,
+ workflow.physicalPlan,
+ COORDINATOR
+ ).bottomUpSearch(globalSearch = true)
+ assert(greedyResult.numStatesExplored < globalResult.numStatesExplored)
Review Comment:
The assertion that greedy search must explore fewer states than global
search is not guaranteed. With early-stop pruning and FIFO queue ordering, a
global search can find a schedulable state earlier and prune more aggressively,
resulting in fewer explored states than the greedy path. This makes the test
brittle across implementation changes and even iteration-order/tie-break
differences. Prefer asserting a property that is guaranteed for greedy search
(linear exploration bound) rather than comparing against globalSearch=true.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGeneratorSpec.scala:
##########
@@ -552,4 +552,132 @@ 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 defaults to 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()
+
+ // 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 only enqueues one neighbor per state, so it must
explore fewer states than a global search
+ // that enqueues every unvisited neighbor.
+ val globalResult = new CostBasedScheduleGenerator(
+ workflow.context,
+ workflow.physicalPlan,
+ COORDINATOR
+ ).bottomUpSearch(globalSearch = true)
+ assert(greedyResult.numStatesExplored < globalResult.numStatesExplored)
+
+ // 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 defaults to 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()
Review Comment:
Same as the bottom-up case: this test is intended to cover the greedy
topDownSearch path (globalSearch = false) but currently relies on the default
argument. Pass globalSearch = false explicitly so the test can’t silently stop
covering the greedy branch if defaults change.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGeneratorSpec.scala:
##########
@@ -552,4 +552,132 @@ 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 defaults to 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()
Review Comment:
This test’s stated purpose is to exercise the greedy path (globalSearch =
false), but it currently relies on the default argument of bottomUpSearch(). If
the default ever changes, the test would silently stop covering the greedy
branch while still passing. Make the intent explicit by passing globalSearch =
false.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
