refactor(workflow-compiler): unify duplicate compilers into common/workflow-compiler#6143
refactor(workflow-compiler): unify duplicate compilers into common/workflow-compiler#6143Yicong-Huang wants to merge 1 commit into
Conversation
github-actions
Bot
added
engine
dependencies
Automated Reviewer SuggestionsBased on the
|
aglinxinyuan
left a comment
aglinxinyuan
left a comment
There was a problem hiding this comment.
Let's get input from @bobbai00 and @Xiao-zhen-Liu as well.
There was a problem hiding this comment.
Pull request overview
This PR consolidates the previously duplicated workflow compiler implementations into a single shared common/workflow-compiler module, repointing amber and workflow-compiling-service to depend on and use the unified API with explicit Lenient (editing-time) vs Strict (execution) error-handling semantics.
Changes:
- Introduced a shared
common/workflow-compilermodule exposingWorkflowCompiler.compile(..., CompilationErrorHandling)returningWorkflowCompilationResult(logical plan, optional physical plan, schemas, per-op errors, storage ports). - Updated amber execution/adapters and workflow-compiling-service to import and use the shared compiler/models; removed amber’s legacy compiler/model copies.
- Migrated/expanded compiler unit coverage into the shared module and updated affected amber tests/imports accordingly.
Reviewed changes
Copilot reviewed 35 out of 36 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowCompilationResourceSpec.scala | Updates model/spec references to the shared compiler package. |
| workflow-compiling-service/src/main/scala/org/apache/texera/service/resource/WorkflowCompilationResource.scala | Switches resource to use shared WorkflowCompiler + shared LogicalPlanPojo. |
| common/workflow-compiler/src/test/scala/org/apache/texera/compiler/WorkflowCompilerSpec.scala | Migrates and expands compiler behavior tests (physical plan shape, storage ports, strict mode). |
| common/workflow-compiler/src/test/scala/org/apache/texera/compiler/model/LogicalLinkSpec.scala | Moves test into the unified org.apache.texera.compiler.model package. |
| common/workflow-compiler/src/test/scala/org/apache/texera/compiler/LogicalPlanSpec.scala | Repoints plan tests to unified LogicalPlan/Pojo/Link model types. |
| common/workflow-compiler/src/main/scala/org/apache/texera/compiler/WorkflowCompiler.scala | Implements unified compiler API/result, adds storage-port computation and strict/lenient handling. |
| common/workflow-compiler/src/main/scala/org/apache/texera/compiler/model/LogicalPlanPojo.scala | Moves LogicalPlanPojo into shared compiler model package. |
| common/workflow-compiler/src/main/scala/org/apache/texera/compiler/model/LogicalPlan.scala | Moves LogicalPlan and adds getTerminalOperatorIds used for storage-port selection. |
| common/workflow-compiler/src/main/scala/org/apache/texera/compiler/model/LogicalLink.scala | Moves LogicalLink into shared compiler model package. |
| common/workflow-compiler/src/main/scala/org/apache/texera/compiler/CompilationErrorHandling.scala | Introduces explicit strict vs lenient compilation error-handling mode. |
| common/workflow-compiler/DESIGN.md | Documents rationale, API contract, and module architecture for the unification. |
| common/workflow-compiler/build.sbt | Adds module-local sbt settings/deps for the new shared compiler module. |
| build.sbt | Registers WorkflowCompiler project and rewires amber/workflow-compiling-service dependencies. |
| amber/src/test/scala/org/apache/texera/workflow/WorkflowCompilerSpec.scala | Removes legacy amber compiler spec now covered in the shared module. |
| amber/src/test/scala/org/apache/texera/web/service/WorkflowExecutionServiceSpec.scala | Updates imports to shared LogicalPlanPojo. |
| amber/src/test/scala/org/apache/texera/amber/engine/faulttolerance/CheckpointSpec.scala | Repoints LogicalLink import to shared compiler model. |
| amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala | Executes compilation in strict mode and adapts to WorkflowCompilationResult for workflow construction. |
| amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala | Repoints LogicalLink import to shared compiler model. |
| amber/src/test/scala/org/apache/texera/amber/engine/e2e/PauseSpec.scala | Repoints LogicalLink import to shared compiler model. |
| amber/src/test/scala/org/apache/texera/amber/engine/e2e/DataProcessingSpec.scala | Repoints LogicalLink import to shared compiler model. |
| amber/src/test/scala/org/apache/texera/amber/engine/e2e/BatchSizePropagationSpec.scala | Repoints LogicalLink import to shared compiler model. |
| amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/resourcePolicies/ResourcePoliciesSpec.scala | Repoints LogicalLink import to shared compiler model. |
| amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/ExpansionGreedyScheduleGeneratorSpec.scala | Repoints LogicalLink import to shared compiler model. |
| amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/DefaultCostEstimatorSpec.scala | Repoints LogicalLink import to shared compiler model. |
| amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGeneratorSpec.scala | Repoints LogicalLink import to shared compiler model. |
| amber/src/test/scala/org/apache/texera/amber/engine/architecture/controller/WorkflowSchedulerSpec.scala | Repoints LogicalLink import to shared compiler model. |
| amber/src/test/integration/org/apache/texera/amber/engine/e2e/ReconfigurationIntegrationSpec.scala | Repoints LogicalLink import to shared compiler model. |
| amber/src/main/scala/org/apache/texera/workflow/WorkflowCompiler.scala | Removes legacy amber compiler implementation (replaced by shared compiler + thin adapter). |
| amber/src/main/scala/org/apache/texera/workflow/LogicalPlan.scala | Removes legacy amber logical-plan model (replaced by shared compiler model). |
| amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala | Repoints LogicalPlan import to shared compiler model. |
| amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala | Adapts execution path to strict compilation + uses shared compilation result and storage ports. |
| amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala | Uses strict compilation for validation paths and repoints model/compiler imports. |
| amber/src/main/scala/org/apache/texera/web/model/websocket/request/WorkflowExecuteRequest.scala | Switches request to shared LogicalPlanPojo type (removes inline duplicate model). |
| amber/src/main/scala/org/apache/texera/web/model/websocket/request/EditingTimeCompilationRequest.scala | Repoints LogicalLink/LogicalPlanPojo imports to shared compiler model. |
| amber/src/main/scala/org/apache/texera/amber/engine/architecture/controller/Workflow.scala | Repoints LogicalPlan to shared compiler model in engine workflow wrapper. |
| .github/workflows/build.yml | Adds jacoco aggregation for the new WorkflowCompiler module. |
Comments suppressed due to low confidence (1)
common/workflow-compiler/src/main/scala/org/apache/texera/compiler/WorkflowCompiler.scala:262
- In Strict mode, schema-propagation failures from collectOutputSchemaFromPhysicalPlan are currently written into a throwaway buffer and then ignored (no throw, and they don’t land in operatorIdToError). This violates the Strict "fail-fast" contract and can let execution proceed with schema errors that would only surface later.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Yicong-Huang
force-pushed
the
refactor/unify-workflow-compiler
branch
2 times, most recently
from
fb89a9b to
526c9bb
Compare
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #6143 +/- ##
============================================
- Coverage 59.80% 59.76% -0.05%
+ Complexity 3295 3283 -12
============================================
Files 1133 1131 -2
Lines 43801 43777 -24
Branches 4748 4749 +1
============================================
- Hits 26197 26163 -34
- Misses 16174 16186 +12
+ Partials 1430 1428 -2
☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 362 | 0.221 | 26,790/37,081/37,081 us | 🔴 +23.7% / 🔴 +144.9% |
| 🔴 | bs=100 sw=10 sl=64 | 777 | 0.474 | 126,929/158,025/158,025 us | 🔴 +20.6% / 🔴 +46.4% |
| ⚪ | bs=1000 sw=10 sl=64 | 916 | 0.559 | 1,092,689/1,148,923/1,148,923 us | ⚪ within ±5% / 🔴 +11.7% |
Baseline details
Latest main 44ec60d from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 362 tuples/sec | 438 tuples/sec | 776.36 tuples/sec | -17.4% | -53.4% |
| bs=10 sw=10 sl=64 | MB/s | 0.221 MB/s | 0.267 MB/s | 0.474 MB/s | -17.2% | -53.4% |
| bs=10 sw=10 sl=64 | p50 | 26,790 us | 21,657 us | 12,636 us | +23.7% | +112.0% |
| bs=10 sw=10 sl=64 | p95 | 37,081 us | 33,173 us | 15,143 us | +11.8% | +144.9% |
| bs=10 sw=10 sl=64 | p99 | 37,081 us | 33,173 us | 18,954 us | +11.8% | +95.6% |
| bs=100 sw=10 sl=64 | throughput | 777 tuples/sec | 834 tuples/sec | 985.33 tuples/sec | -6.8% | -21.1% |
| bs=100 sw=10 sl=64 | MB/s | 0.474 MB/s | 0.509 MB/s | 0.601 MB/s | -6.9% | -21.2% |
| bs=100 sw=10 sl=64 | p50 | 126,929 us | 120,681 us | 101,671 us | +5.2% | +24.8% |
| bs=100 sw=10 sl=64 | p95 | 158,025 us | 131,040 us | 107,939 us | +20.6% | +46.4% |
| bs=100 sw=10 sl=64 | p99 | 158,025 us | 131,040 us | 113,798 us | +20.6% | +38.9% |
| bs=1000 sw=10 sl=64 | throughput | 916 tuples/sec | 929 tuples/sec | 1,016 tuples/sec | -1.4% | -9.9% |
| bs=1000 sw=10 sl=64 | MB/s | 0.559 MB/s | 0.567 MB/s | 0.62 MB/s | -1.4% | -9.9% |
| bs=1000 sw=10 sl=64 | p50 | 1,092,689 us | 1,082,805 us | 989,693 us | +0.9% | +10.4% |
| bs=1000 sw=10 sl=64 | p95 | 1,148,923 us | 1,095,793 us | 1,028,327 us | +4.8% | +11.7% |
| bs=1000 sw=10 sl=64 | p99 | 1,148,923 us | 1,095,793 us | 1,059,969 us | +4.8% | +8.4% |
Raw CSV
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,552.10,200,128000,362,0.221,26790.24,37080.74,37080.74
1,100,10,64,20,2575.27,2000,1280000,777,0.474,126928.97,158025.16,158025.16
2,1000,10,64,20,21829.46,20000,12800000,916,0.559,1092689.32,1148922.74,1148922.74
Yicong-Huang
force-pushed
the
refactor/unify-workflow-compiler
branch
from
526c9bb to
dd92287
Compare
…rkflow-compiler Merge amber's and workflow-compiling-service's separate WorkflowCompiler implementations into a shared common/workflow-compiler module exposing one compile API with Lenient (editing-time, accumulate per-op errors) and Strict (execution, fail-fast) modes. amber's controller Workflow becomes a thin wrapper over the compilation result. Migrate the compiler/model unit specs into the module and add WorkflowCompiler/jacoco to the amber Codecov job so the new module reports coverage.
Yicong-Huang
force-pushed
the
refactor/unify-workflow-compiler
branch
from
dd92287 to
be99b36
Compare
bobbai00
left a comment
bobbai00
left a comment
There was a problem hiding this comment.
LGTM! Left few minor comments
| @@ -0,0 +1,61 @@ | |||
| # Design: Unified `WorkflowCompiler` | |||
There was a problem hiding this comment.
Should this one be removed ?
| ) | ||
| } catch { | ||
| case err: Throwable => | ||
| errorHandler(err) |
There was a problem hiding this comment.
can we directly return to end this function here ?
| { | ||
| val externalLinks = logicalPlan | ||
| .getUpstreamLinks(logicalOp.operatorIdentifier) | ||
| .filter(link => physicalOp.inputPorts.contains(link.toPortId)) |
There was a problem hiding this comment.
can we save the upstream links as a variable outside of this loop ?
|
@Xiao-zhen-Liu Please chime in. @Yicong-Huang A general question: should we create a discussion first? Or in what cases should we create a discussion before raising a PR? |
We create discussions when there could have different opinions, for instance when we are unsure if support a feature or not, when we are not sure whether support something or not, or whether we want to remove/disable something. The discussion usually is on the direction. For this particular case, I feel having two copies of the compiler source code is problematic and adds maintenance cost. Thus, I did not ask for input to unifying them by creating a shared lib. There is no change on the design, or experience: we still do two compilations, one at editing time, through compilation service, one at execution time, through execution service. It's only a code level refactoring. |
Xiao-zhen-Liu
left a comment
Xiao-zhen-Liu
left a comment
There was a problem hiding this comment.
Nice refactor — merging the two hand-synced compiler copies into one module is a clear win, and the boundary is clean (common/workflow-compiler depends only on WorkflowOperator; the WorkflowCompilationResult / CompilationErrorHandling split reads well).
One ask: could you update the PR description to make clear this is source-level de-duplication, not a change to how many times a workflow compiles? We still compile twice at runtime (editing + execution). As written it reads like it might be collapsing them, which threw me off — a line like "two compilations stays; this only shares the code" would clear it up.
I checked both callers against their old versions — faithful almost everywhere. A few non-blocking things inline: Strict doesn't actually fail fast on schema errors (drops them) though the docs say otherwise; one real behavior change on the editing path (Try vs the old catch Throwable); the failed-compile path falls through to a null workflow (bobbai00 spotted it); and Strict is thinly tested. One more small one: the compiling-service caller relies on the default Lenient while amber passes Strict explicitly — passing it explicitly there too would read symmetrically.
Also +1 to bobbai00 on hoisting getUpstreamLinks and the DESIGN.md question (no other module ships one). Good to merge once the Strict/schema wording is sorted or a follow-up is filed.
| // even if error is encountered when logical => physical, we still want to get the input schemas | ||
| // for the rest of the no-error operators. In Strict mode there is no buffer (errors already thrown), | ||
| // so schema errors go to a throwaway buffer and do not affect the result. | ||
| val schemaErrorList = errorList.getOrElse(new ArrayBuffer[(OperatorIdentity, Throwable)]()) |
There was a problem hiding this comment.
Strict sets errorList = None, so this buffer is never read and hasErrors (line 263) is always false. Schema-propagation errors (e.g. a Projection on a missing column) get dropped, physicalPlan stays Some, and the run proceeds. Matches old amber (not a regression) but contradicts the fail-fast claim above — and such a plan can still be launched from the UI. Fix the wording, or throw when the buffer is non-empty (a behavior change → probably a follow-up).
| val logicalOp = logicalPlan.getOperator(logicalOpId) | ||
| val allUpstreamLinks = logicalPlan.getUpstreamLinks(logicalOp.operatorIdentifier) | ||
| logicalPlan.getTopologicalOpIds.asScala.foreach(logicalOpId => | ||
| Try { |
There was a problem hiding this comment.
Behavior change from the merge: the old compiling-service used catch { case e: Throwable }; scala.util.Try catches only non-fatal errors. So a fatal error during expansion now fails the /compile request instead of being recorded per-operator. Probably fine — flagging since the PR says nothing changes.
| * | ||
| * - [[CompilationErrorHandling.Lenient]] collects every error and returns them in the result, | ||
| * with `physicalPlan = None` when any error occurred. Used for editing-time validation. | ||
| * - [[CompilationErrorHandling.Strict]] throws on the first error. Used before execution. |
There was a problem hiding this comment.
Small accuracy fix: Strict doesn't throw on schema-propagation errors — they become Left and are never thrown (see WorkflowCompiler.scala:260).
| .filter(op => jgraphtDag.outDegreeOf(op) == 0) | ||
| .toList | ||
|
|
||
| def addOperator(op: LogicalOp): LogicalPlan = { |
There was a problem hiding this comment.
Stale TODO, and addOperator/addLink came over from the old code with no callers I could find — drop them or note why they stay.
| // -------------------- 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 { |
There was a problem hiding this comment.
Only one Strict test. Worth adding a Strict success case (the execution path relies on physicalPlan.get) and a schema-error case to pin the behavior. The Python code-gen error branch (~line 179) is untested in both modes too.
| ) | ||
| } catch { | ||
| case err: Throwable => | ||
| errorHandler(err) |
There was a problem hiding this comment.
+1 to bobbai00's point here: if compile throws, this reports the error but execution then falls through to workflow.context below with workflow still null → a second NPE that masks the real message. A return after errorHandler(err) fixes it. Pre-existing, but a good spot to fix now that Strict is the fail-fast path.
What changes were proposed in this PR?
amber and workflow-compiling-service each maintained a separate copy of the workflow compiler and its models (
WorkflowCompiler,LogicalPlan,LogicalLink,LogicalPlanPojo), which had to be kept in sync by hand. This PR consolidates them into a single sharedcommon/workflow-compilermodule that both services and amber depend on; the old per-service copies are removed.The unified
WorkflowCompiler.compilereturns aWorkflowCompilationResult(logical plan, optional physical plan, per-port output schemas, per-operator errors, and the output ports needing storage) and takes aCompilationErrorHandlingmode:Lenientfor the editing-time path (accumulate per-operator errors so the UI can render them) andStrictfor the execution path (fail-fast before a run). amber's controllerWorkflowbecomes a thin wrapper over the compilation result.Any related issues, documentation, discussions?
Closes #6142. Design notes live in
common/workflow-compiler/DESIGN.md.How was this PR tested?
Migrated the compiler and model unit specs into the new module and added strict-mode coverage; amber and workflow-compiling-service compile and test against the shared library.
Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8 (Claude Code)