feat(workflow-compiling-service): export a workflow as a standalone Python script - #8327
feat(workflow-compiling-service): export a workflow as a standalone Python script#8327kz930 wants to merge 17 commits into
Conversation
…ython script A workflow can be read in the editor but not taken away: there is no form of it that runs anywhere else, so a user who wants to keep a pipeline, hand it to someone without Texera, or step through it in a notebook has nothing to take. This adds the seam for one and the first few operators through it. An operator says how it reads outside the engine by implementing `StandaloneCodeGenerator`, returning a block of pandas that names its inputs and outputs as `in1df` / `out1df`. The translator walks the plan in topological order, gives every port a variable, substitutes those placeholders, and prints the leaves; `inAlldf` stands for the whole list of upstreams, which is what a variadic port like Union's needs, since any fixed count the code stated would be wrong for some workflow. An operator with no generator yet leaves a commented TODO rather than a line that looks like it works. `GET /workflow-to-python` on the compiling service returns the script for a plan it is given. Five operators implement it here — Distinct, Limit, Projection, Filter and Union — chosen to cover the shapes the translator has to handle: a single input, a config-driven one, one that renames columns, one that builds a predicate, and the variadic port. The rest of the operator set follows in later changes. `pyStringLiteral` renders a value as a Python literal with the escaping that keeps a quote or a newline in a column name from ending the literal early. The generators cannot use the runtime's decode expression, which needs an operator instance to decode through. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…re the files The standalone export claims that a generated script does what the operator does. Nothing checks it. This adds the two runners that make the claim checkable, and the file format they meet in. `OpExecHarness` runs a LogicalOp the way the engine does — compiling it to a physical plan and driving the executor — but outside a workflow, against JSONL files rather than a live upstream. `PyOpExecHarness` does the same for a Python operator, through the worker the engine uses. `StandaloneRunner` takes the other path: it asks the operator for its standalone code, wraps it in a script that binds `in1df` from the same files, and runs it. `TupleIO` is what the two meet in. A JSONL row carries values and no types, so the schema travels beside it in a sidecar; without one, a column written as INTEGER reads back as a number and the two paths disagree over a difference neither operator made. Both runners produce files, not assertions, so what to make of a difference is left to a later change. What is here is enough to run one operator both ways and see that the answers match, which is what the spec does with Distinct. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Automated Reviewer SuggestionsBased on the
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #8327 +/- ##
============================================
- Coverage 94.03% 94.02% -0.02%
- Complexity 4821 4839 +18
============================================
Files 1207 1205 -2
Lines 49109 48997 -112
Branches 5963 5951 -12
============================================
- Hits 46180 46069 -111
+ Misses 1461 1458 -3
- Partials 1468 1470 +2
*This pull request uses carry forward flags. Click here to find out more. ☔ 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 | 463 | 0.283 | 21,147/27,074/27,074 us | 🔴 +25.9% / 🔴 +65.8% |
| 🟢 | bs=100 sw=10 sl=64 | 1,011 | 0.617 | 96,127/129,877/129,877 us | 🟢 -20.1% / 🔴 +17.7% |
| ⚪ | bs=1000 sw=10 sl=64 | 1,198 | 0.731 | 826,716/922,288/922,288 us | ⚪ within ±5% / 🟢 +20.0% |
Baseline details
Latest main dcb25ef from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 463 tuples/sec | 544 tuples/sec | 749.39 tuples/sec | -14.9% | -38.2% |
| bs=10 sw=10 sl=64 | MB/s | 0.283 MB/s | 0.332 MB/s | 0.457 MB/s | -14.8% | -38.1% |
| bs=10 sw=10 sl=64 | p50 | 21,147 us | 19,038 us | 13,045 us | +11.1% | +62.1% |
| bs=10 sw=10 sl=64 | p95 | 27,074 us | 21,506 us | 16,328 us | +25.9% | +65.8% |
| bs=10 sw=10 sl=64 | p99 | 27,074 us | 21,506 us | 19,518 us | +25.9% | +38.7% |
| bs=100 sw=10 sl=64 | throughput | 1,011 tuples/sec | 937 tuples/sec | 968.07 tuples/sec | +7.9% | +4.4% |
| bs=100 sw=10 sl=64 | MB/s | 0.617 MB/s | 0.572 MB/s | 0.591 MB/s | +7.9% | +4.4% |
| bs=100 sw=10 sl=64 | p50 | 96,127 us | 99,644 us | 103,330 us | -3.5% | -7.0% |
| bs=100 sw=10 sl=64 | p95 | 129,877 us | 162,632 us | 110,305 us | -20.1% | +17.7% |
| bs=100 sw=10 sl=64 | p99 | 129,877 us | 162,632 us | 121,264 us | -20.1% | +7.1% |
| bs=1000 sw=10 sl=64 | throughput | 1,198 tuples/sec | 1,220 tuples/sec | 998 tuples/sec | -1.8% | +20.0% |
| bs=1000 sw=10 sl=64 | MB/s | 0.731 MB/s | 0.745 MB/s | 0.609 MB/s | -1.9% | +20.0% |
| bs=1000 sw=10 sl=64 | p50 | 826,716 us | 818,946 us | 1,009,354 us | +0.9% | -18.1% |
| bs=1000 sw=10 sl=64 | p95 | 922,288 us | 881,747 us | 1,052,748 us | +4.6% | -12.4% |
| bs=1000 sw=10 sl=64 | p99 | 922,288 us | 881,747 us | 1,081,982 us | +4.6% | -14.8% |
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,431.66,200,128000,463,0.283,21147.08,27073.84,27073.84
1,100,10,64,20,1977.34,2000,1280000,1011,0.617,96126.72,129876.70,129876.70
2,1000,10,64,20,16698.39,20000,12800000,1198,0.731,826715.69,922288.26,922288.26…he operator it came from The standalone export claims a generated script does what the operator does. This is what checks it, for every operator, on every configuration the operator offers. An operator is run twice. `OpExecHarness` drives it the way the engine does, compiled to a physical plan but outside a workflow, reading JSONL files rather than a live upstream; `PyOpExecHarness` does the same for a Python operator through the worker the engine uses. `StandaloneRunner` takes the other path, wrapping the operator's standalone code in a script that binds the same files. Both write files, and `Comparator` reads them back: order-insensitive by default, since the engine interleaves across workers and only the sort family promises an order. A visualization is compared as a figure rather than as a frame. What to run an operator ON is decided rather than written by hand for each. `ConfigGenerator` reads the operator's own schema — its enums, defaults, declared ranges and column pickers — and produces a base configuration plus one variant per branch the operator offers, so a switch nobody thought to try is still tried. `CanonicalFixture` is the table they run against, one column per shape an operator might ask for. `CuratedHandlers` is the escape hatch for an operator whose input cannot be derived, and `TransformVerificationRunner` decides which of the three tiers each operator takes and reports what it could not run and why. `LogicalOp.orderSensitive` and `@SampleColumn` are the two things the operators had to say for this to read them: whether row order is part of the contract, and which column a field should be pointed at when the first unused one would be a poor choice. Most of the operator set does not implement the generator yet — it arrives a family at a time — and the runner reports each of those rather than passing over it. The tier assertions for a family land with the change that gives that family its generator. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…f the one without The split this change relies on was declared but never wired. The specs carry `@IntegrationTest` and `build.sbt` reads `WCS_TEST_FILTER` to act on it, but nothing set that variable, so the filter was a no-op and the specs that fork Python ran in the job that provisions none — failing on `No module named 'pandas'` rather than on anything they were testing. The platform job now sets `skip-integration`, which excludes them. The platform-integration job sets `integration-only` and provisions what they need: Python 3.12, amber's requirements, protoc, and the generated proto bindings, which are gitignored and so have to be regenerated before a forked driver can import pyamber. Every step is guarded on the service, so the other entries in that matrix are untouched. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The comment claimed a knob carrying a `defaultValue` is left alone. It is not: a knob holding what a fresh instance holds counts as untouched, because a config where the user typed the default is indistinguishable from one where they typed nothing. Moving it off that value is the only way the branch for a knob that IS set gets exercised. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
… have one The example was an operator another batch gives a generator to, so the assertion held only until that batch landed. A Python UDF holds whatever order these land in: its body is written by whoever drops the operator, so there is nothing for a generator to emit. The word cloud assertion goes for the same reason. It says the operator is withheld, which a later batch stops being true once its placement is seeded, and the prediction op alone already covers what the test is for. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
One row is withheld because the fixture cannot carry what the operator consumes, the other because the operator has no physical execution to run at all. Only the first was asserted. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
sbt lifts a `.value` written inside a lambda to the top of the task, so the options were already read once rather than per suite. Written where it was, it read as the other thing, and sbt warned on every run. Hoisting it says what actually happens and quiets the warning. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The licence check compares the file against what pip resolved, and the two had drifted apart, which blocks every run on Python 3.12: the file claimed 4.14.1 while 4.16.0 was installed. The pin in requirements.txt still says 4.14.1 and does not hold, because operator-requirements.txt is installed in a second pass whose resolution the first file's constraint does not reach. That is worth fixing on its own; this only stops the check reporting a version nothing installs. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…talled" This reverts commit f7a0243.
Everything here that is not a source operator belongs to apache#8327 and was carried only so this branch could compile and run its own tests before that one landed. Reviewing it twice costs more than the red build does: what is left is the thirteen files this change is actually about. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
|
/request-review @mengw15 |
|
@carloea2 May you take a look at this? |
This configured a filter selecting on `IntegrationTest`, a tag the next change in the set adds. Named here it selects on nothing, and a reader looking for the tag does not find it. It moves to that change, where the tag it names arrives with it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
carloea2
left a comment
There was a problem hiding this comment.
The export flow and the first operator set look good.
There was a problem hiding this comment.
🟡 Changes recommended
Critical runtime and semantic divergences, plus moderate output inconsistencies, remain unresolved.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
Adds an endpoint and translator for exporting workflows as standalone pandas-based Python scripts.
Changes:
- Introduces the export API and code-generation contract.
- Supports Distinct, Filter, Limit, Projection, and Union.
- Adds literal rendering and translator/operator tests.
File summaries
| File | Review |
|---|---|
workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslatorSpec.scala |
Tests translation, ordering, ports, placeholders, and unsupported operators. |
workflow-compiling-service/src/main/scala/org/apache/texera/service/WorkflowCompilingService.scala |
Registers the export endpoint. |
workflow-compiling-service/src/main/scala/org/apache/texera/service/resource/WorkflowToPythonResource.scala |
Exposes the export API; nit: resource, malformed-plan, registration, and access-control coverage is missing. |
workflow-compiling-service/src/main/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslator.scala |
Generates scripts; critical: unnecessarily requires Plotly. Moderate: truncates leaf output to five rows. |
workflow-compiling-service/build.sbt |
Adds test concurrency configuration; nit: premature scaffolding for absent tests. |
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/union/UnionOpDescSpec.scala |
Tests Union generation. |
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpDescSpec.scala |
Tests Projection generation. |
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/limit/LimitOpDescSpec.scala |
Tests Limit generation. |
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala |
Tests Filter generation. |
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/distinct/DistinctOpDescSpec.scala |
Tests Distinct generation. |
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/union/UnionOpDesc.scala |
Generates pandas concatenation. |
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneCodeGenerator.scala |
Defines the standalone generation contract. |
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala |
Generates projection code; critical: empty projections become pass-through instead of failing like the engine. |
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/annotations/SampleColumn.java |
Adds metadata; nit: currently unused and lacks its claimed reader. |
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala |
Adds an ordering flag; nit: neither overridden nor consumed. |
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/limit/LimitOpDesc.scala |
Generates row limiting; moderate: negative limits diverge from engine behavior. |
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala |
Generates predicates; critical: empty predicate lists incorrectly pass through rows. |
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/distinct/DistinctOpDesc.scala |
Generates pandas deduplication. |
common/pybuilder/src/test/scala/org/apache/texera/amber/pybuilder/PythonTemplateBuilderApiSpec.scala |
Tests Python literal rendering. |
common/pybuilder/src/main/scala/org/apache/texera/amber/pybuilder/PythonTemplateBuilder.scala |
Renders Python literals; critical: NUL characters produce invalid Python source. |
Review details
Suppressed comments (5)
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala:80
- The generated filter does not preserve
FilterPredicatesemantics because it infers the RHS type only from its text. For example, a BOOLEAN column compared withTRUEis lowercased and compared as a boolean byFilterPredicate, but this emits a comparison between a pandas boolean Series and the string"TRUE", producing no matches; numeric-looking values against STRING columns can instead cause a string-versus-numberTypeError. The generated code needs dtype-aware coercion (including boolean, timestamp, and the runtime's numeric-string fallback) rather thancoerceValue(p.value).
s"""(in1df[$colLit].notna() & (in1df[$colLit] $pyOp ${coerceValue(p.value)}))"""
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/annotations/SampleColumn.java:39
- The comment claims a test-side
ConfigGeneratorreads this annotation, but there is no reference toSampleColumnanywhere else in the repository. As added, this is unused production-source metadata with no effect and does not support the export described by this PR; move it to the later parity-test change together with its reader, or add the missing reader and tests.
* <p>This has <em>no effect on production</em>: it is not a Jackson / JSON-schema
* annotation and is read only by the test-side ConfigGenerator.
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala:130
- A valid projection may select the same source column more than once under different aliases (for example,
a -> xanda -> y). This emits a Python dict with duplicate"a"keys, so the last alias wins and pandas renames both selected columns toy;ProjectionOpExecinstead produces distinctxandyfields. Build the result perAttributeUnit(for example by concatenating individually renamed Series) rather than using a source-name-keyed rename map.
val renameMap = renames.mkString("{", ", ", "}")
s"out1df = in1df[$originals].rename(columns=$renameMap)"
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala:117
- Drop-mode projection is case-insensitive in Texera (
ProjectionOpExeclowercases selected and actual names, andSchema.removedoes likewise), whileDataFrame.drop(columns=...)matches labels exactly. A configuredFIELD2drops an input column namedfield2in Texera but this script raisesKeyError. Resolve requested names againstin1df.columnscase-insensitively before dropping.
if (isDrop) {
// Drop mode ignores aliases (matches ProjectionOpExec).
val cols = units.map(u => pyStringLiteral(u.getOriginalAttribute)).mkString("[", ", ", "]")
s"out1df = in1df.drop(columns=$cols)"
workflow-compiling-service/build.sbt:53
- This concurrency setting is justified entirely by
OperatorBehaviorSpec,WCS_TEST_FILTER, and Python subprocesses, but none of those exist in this module in the current change. With no suite usingParallelTestExecution,-P4does not support the translator tests described by the PR and is premature scaffolding for a later parity-run change. Keep this PR scoped by removing it until that suite is introduced.
// -P4 bounds ScalaTest's ParallelTestExecution pool, and only this module wants
// it: OperatorBehaviorSpec forks a Python subprocess per operator, and at
// core-count concurrency (e.g. 12) resource contention caused rare flakes. A
// fixed 4 stays deterministic across machines (incl. CI runners) while still
// running ~3x faster than serial, and it matches PythonWorkerPool's own default
- Files reviewed: 20/20 changed files
- Comments generated: 8
- Review effort level: Balanced
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Six places where the export disagreed with the run it was exporting. An empty predicate list passed every row, where the executor's `predicates.exists` keeps none. An empty projection passed the frame through, where schema propagation refuses it outright. A negative limit dropped only the last row, where `count < limit` keeps nothing. A NUL in a column name reached the generated source verbatim, and Python refuses to compile a file holding one. Every script imported plotly whether or not anything in it drew, so a pandas-only environment could not start one that reshapes a table. And a leaf printed `head()`, which shows five rows and does not say how many there were. An operator now states what it needs beyond pandas through `standaloneImports()`, collected across the plan the way helpers already are. `orderSensitive` moves out: nothing here overrides it and nothing reads it, and the sort family that does both arrives with the base transforms. `WorkflowToPythonResourceSpec` covers the endpoint, which had no resource-level test: the success discriminator and payload, and a plan it cannot read coming back as a failure rather than a 500. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
carloea2
left a comment
There was a problem hiding this comment.
I checked the new error handling and import changes. They look good.
|
@aglinxinyuan May you take a look at this PR? This is the first PR of my workflow to Python project. After this first PR being merge, I can raise more PR, since this is the foundation for most of the PR. |
carloea2
left a comment
There was a problem hiding this comment.
I checked the main merge. The PR changes are unchanged and look good.
What changes were proposed in this PR?
A workflow can be built and read in the editor, but there is no form of it that runs anywhere else. This adds one: given a plan, the compiling service returns a single Python file that reads the same sources, applies the same operators in the same order, and prints its results.
An operator says how it reads outside the engine by implementing
StandaloneCodeGenerator, returning a block of pandas that names its inputs and outputs by position —in1df,in2df,out1df. The translator walks the plan in topological order, gives every output port a variable, substitutes those placeholders for the variables its upstreams were given, and prints the leaves. Union takes the whole list of upstreams rather than a fixed count, since any count an operator states would be wrong for some workflow.An operator that has no generator yet leaves a commented placeholder rather than a line that looks like it works, so the export is useful before every operator implements the trait.
Five operators implement it here to show the shape and to give the translator something real to walk: Distinct, Filter, Limit, Projection and Union. The rest of the operator set follows a family at a time.
Any related issues, documentation, discussions?
Part of #8325, 1 of 21; that issue lists the set in order.
Closes #8407, the task this change is the whole of.
How was this PR tested?
WorkflowToPythonTranslatorSpeccovers what the translator does with a plan: the topological order, the variable each port is given, the placeholder substitution, the variadic port, and the operator that has no generator. Each of the five operators asserts the block it emits in its own spec.Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)