Yicong-Huang commented on code in PR #6143:
URL: https://github.com/apache/texera/pull/6143#discussion_r3606737513
##########
common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalPlan.scala:
##########
@@ -81,6 +80,11 @@ case class LogicalPlan(
def getOperator(opId: OperatorIdentity): LogicalOp = operatorMap(opId)
+ def getTerminalOperatorIds: List[OperatorIdentity] =
+ operatorMap.keys
+ .filter(op => jgraphtDag.outDegreeOf(op) == 0)
+ .toList
+
def addOperator(op: LogicalOp): LogicalPlan = {
Review Comment:
Dropped both (and the stale TODO) in
bbbf9f5841677f4708c4c17c3b77f98889fcae80 — confirmed no callers anywhere in
amber / compiling-service / common.
##########
common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/WorkflowCompilerSpec.scala:
##########
@@ -314,4 +316,168 @@ class WorkflowCompilerSpec extends AnyFlatSpec {
s"expected both csvs in errors, got ${result.operatorIdToError.keySet}"
)
}
+
+ // -------------------- physical-plan shape --------------------
+
+ private def pojo(
+ operators: List[org.apache.texera.amber.operator.LogicalOp],
+ links: List[LogicalLink],
+ opsToViewResult: List[String] = List.empty
+ ): LogicalPlanPojo =
+ LogicalPlanPojo(operators, links, opsToViewResult, List.empty)
+
+ // Re-anchor the subject after the sub-section.
+ "WorkflowCompiler" should "produce a physical plan that contains at least
one physical op per logical op" in {
+ val csv = TestOperators.smallCsvScanOpDesc()
+ val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia")
+
+ val result = new WorkflowCompiler(newContext()).compile(
+ pojo(
+ List(csv, keyword),
+ List(
+ LogicalLink(
+ csv.operatorIdentifier,
+ PortIdentity(),
+ keyword.operatorIdentifier,
+ PortIdentity()
+ )
+ )
+ )
+ )
+
+ assert(result.logicalPlan.operators.size == 2)
+ val physicalPlan = result.physicalPlan.get
+
assert(physicalPlan.getPhysicalOpsOfLogicalOp(csv.operatorIdentifier).nonEmpty)
+
assert(physicalPlan.getPhysicalOpsOfLogicalOp(keyword.operatorIdentifier).nonEmpty)
+ }
+
+ it should "translate a logical link into a physical link between the two
logical ops' physical ops" in {
+ val csv = TestOperators.smallCsvScanOpDesc()
+ val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia")
+
+ val result = new WorkflowCompiler(newContext()).compile(
+ pojo(
+ List(csv, keyword),
+ List(
+ LogicalLink(
+ csv.operatorIdentifier,
+ PortIdentity(),
+ keyword.operatorIdentifier,
+ PortIdentity()
+ )
+ )
+ )
+ )
+
+ val physicalPlan = result.physicalPlan.get
+ val csvPhysIds =
+
physicalPlan.getPhysicalOpsOfLogicalOp(csv.operatorIdentifier).map(_.id).toSet
+ val keywordPhysIds =
+
physicalPlan.getPhysicalOpsOfLogicalOp(keyword.operatorIdentifier).map(_.id).toSet
+
+ val bridging = physicalPlan.links.filter(l =>
+ csvPhysIds.contains(l.fromOpId) && keywordPhysIds.contains(l.toOpId)
+ )
+ assert(bridging.nonEmpty, "expected at least one physical link from csv to
keyword")
+ }
+
+ // -------------------- storage-port collection --------------------
+
+ // The compiler walks `logicalPlan.getTerminalOperatorIds` (logical ops with
+ // out-degree 0) plus `opsToViewResult`, and for every physical op of those
+ // logical ops collects every non-internal output port into the result's
+ // `outputPortsNeedingStorage`. These tests pin both the terminal-default and
+ // the opsToViewResult-additive paths, and that internal ports are filtered.
+
+ "WorkflowCompiler" should "mark the terminal op's output port as needing
storage" in {
+ val csv = TestOperators.smallCsvScanOpDesc()
+ val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia")
+
+ val result = new WorkflowCompiler(newContext()).compile(
+ pojo(
+ List(csv, keyword),
+ List(
+ LogicalLink(
+ csv.operatorIdentifier,
+ PortIdentity(),
+ keyword.operatorIdentifier,
+ PortIdentity()
+ )
+ )
+ )
+ )
+
+ val storage = result.outputPortsNeedingStorage
+ assert(
+ storage.exists(_.opId.logicalOpId == keyword.operatorIdentifier),
+ s"expected keyword to be marked for storage, got
${storage.map(_.opId.logicalOpId)}"
+ )
+ assert(
+ !storage.exists(_.opId.logicalOpId == csv.operatorIdentifier),
+ "csv is not terminal and was not requested via opsToViewResult; it
should not be in storage"
+ )
+ }
+
+ it should "also mark a non-terminal op for storage when it is named in
opsToViewResult" in {
+ val csv = TestOperators.smallCsvScanOpDesc()
+ val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia")
+
+ val result = new WorkflowCompiler(newContext()).compile(
+ pojo(
+ List(csv, keyword),
+ List(
+ LogicalLink(
+ csv.operatorIdentifier,
+ PortIdentity(),
+ keyword.operatorIdentifier,
+ PortIdentity()
+ )
+ ),
+ opsToViewResult = List(csv.operatorIdentifier.id)
+ )
+ )
+
+ val logicalOpsInStorage =
result.outputPortsNeedingStorage.map(_.opId.logicalOpId)
+ assert(
+ logicalOpsInStorage.contains(csv.operatorIdentifier),
+ s"opsToViewResult should add csv to storage, got $logicalOpsInStorage"
+ )
+ assert(
+ logicalOpsInStorage.contains(keyword.operatorIdentifier),
+ s"terminal keyword should remain in storage, got $logicalOpsInStorage"
+ )
+ }
+
+ it should "treat a single source op as terminal and mark its output port for
storage" in {
+ val csv = TestOperators.smallCsvScanOpDesc()
+
+ val result = new WorkflowCompiler(newContext()).compile(pojo(List(csv),
List.empty))
+
+ val storage = result.outputPortsNeedingStorage
+ assert(
+ storage.exists(_.opId.logicalOpId == csv.operatorIdentifier),
+ "single op has out-degree 0, so its output port should land in storage"
+ )
+ assert(
+ storage.forall(!_.portId.internal),
+ "compiler must filter out internal ports; storage should expose only
user-visible outputs"
+ )
+ }
+
+ // -------------------- strict-mode error semantics --------------------
+
+ // Re-anchor the subject after the sub-section.
+ "WorkflowCompiler in strict mode" should "throw when a scan source has no
fileName set" in {
Review Comment:
Added in bbbf9f5841677f4708c4c17c3b77f98889fcae80: a Strict success case
(asserts `physicalPlan` is defined, no errors, storage ports still collected)
and a Strict schema-propagation-error case pinning the current drop-behavior
(plan stays defined, port schema is `None`) with a comment pointing at the
follow-up. Left the Python code-gen branch untested for now — exercising it
needs a Python operator descriptor whose code-gen fails, which drags Python
tooling into this module's test scope; happy to cover it in the follow-up.
##########
amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala:
##########
@@ -110,8 +110,16 @@ class WorkflowExecutionService(
def executeWorkflow(): Unit = {
try {
- workflow = new WorkflowCompiler(workflowContext)
- .compile(request.logicalPlan)
+ val compilationResult = new WorkflowCompiler(workflowContext)
+ .compile(request.logicalPlan, CompilationErrorHandling.Strict)
+ workflowContext.workflowSettings = workflowContext.workflowSettings.copy(
+ outputPortsNeedingStorage = compilationResult.outputPortsNeedingStorage
+ )
+ workflow = Workflow(
+ workflowContext,
+ compilationResult.logicalPlan,
+ compilationResult.physicalPlan.get
+ )
} catch {
case err: Throwable =>
errorHandler(err)
Review Comment:
Done in bbbf9f5841677f4708c4c17c3b77f98889fcae80 — early `return` added
right after `errorHandler(err)`.
--
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]
