feat(workflow-compiling-service): translate a workflow into a standalone Python script - #8327
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% 93.89% -0.14%
- Complexity 4821 4844 +23
============================================
Files 1207 1205 -2
Lines 49109 49139 +30
Branches 5963 5979 +16
============================================
- Hits 46180 46140 -40
- Misses 1461 1515 +54
- Partials 1468 1484 +16
*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 | 361 | 0.22 | 26,395/40,951/40,951 us | ⚪ within ±5% / 🔴 +153.8% |
| 🔴 | bs=100 sw=10 sl=64 | 749 | 0.457 | 126,205/184,729/184,729 us | 🔴 +25.4% / 🔴 +69.9% |
| 🔴 | bs=1000 sw=10 sl=64 | 905 | 0.552 | 1,105,662/1,202,135/1,202,135 us | 🔴 +6.3% / 🔴 +15.8% |
Baseline details
Latest main ec3a9dd from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 361 tuples/sec | 364 tuples/sec | 755.28 tuples/sec | -0.8% | -52.2% |
| bs=10 sw=10 sl=64 | MB/s | 0.22 MB/s | 0.222 MB/s | 0.461 MB/s | -0.9% | -52.3% |
| bs=10 sw=10 sl=64 | p50 | 26,395 us | 26,015 us | 12,957 us | +1.5% | +103.7% |
| bs=10 sw=10 sl=64 | p95 | 40,951 us | 39,206 us | 16,134 us | +4.5% | +153.8% |
| bs=10 sw=10 sl=64 | p99 | 40,951 us | 39,206 us | 20,333 us | +4.5% | +101.4% |
| bs=100 sw=10 sl=64 | throughput | 749 tuples/sec | 797 tuples/sec | 980.1 tuples/sec | -6.0% | -23.6% |
| bs=100 sw=10 sl=64 | MB/s | 0.457 MB/s | 0.486 MB/s | 0.598 MB/s | -6.0% | -23.6% |
| bs=100 sw=10 sl=64 | p50 | 126,205 us | 121,838 us | 101,894 us | +3.6% | +23.9% |
| bs=100 sw=10 sl=64 | p95 | 184,729 us | 147,321 us | 108,718 us | +25.4% | +69.9% |
| bs=100 sw=10 sl=64 | p99 | 184,729 us | 147,321 us | 122,482 us | +25.4% | +50.8% |
| bs=1000 sw=10 sl=64 | throughput | 905 tuples/sec | 921 tuples/sec | 1,011 tuples/sec | -1.7% | -10.5% |
| bs=1000 sw=10 sl=64 | MB/s | 0.552 MB/s | 0.562 MB/s | 0.617 MB/s | -1.8% | -10.6% |
| bs=1000 sw=10 sl=64 | p50 | 1,105,662 us | 1,087,553 us | 996,422 us | +1.7% | +11.0% |
| bs=1000 sw=10 sl=64 | p95 | 1,202,135 us | 1,131,207 us | 1,037,670 us | +6.3% | +15.8% |
| bs=1000 sw=10 sl=64 | p99 | 1,202,135 us | 1,131,207 us | 1,072,152 us | +6.3% | +12.1% |
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,554.35,200,128000,361,0.220,26394.81,40950.97,40950.97
1,100,10,64,20,2671.30,2000,1280000,749,0.457,126204.50,184729.23,184729.23
2,1000,10,64,20,22094.24,20000,12800000,905,0.552,1105662.21,1202135.20,1202135.20…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>
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.
carloea2
left a comment
There was a problem hiding this comment.
Reviewed alongside the related export and verification PRs. These findings are based on code inspection and focused Python checks, not a full Scala suite run.
… name each operator's files The substitution rewrote every occurrence of a placeholder in an operator's block, string literals included, so a Projection selecting a column named in1df asked the frame for a column named after the variable instead and raised KeyError. Only the code parts are rewritten now: a scan splits the block into code, string literals and comments, and the leftover-placeholder warning reads the code alone. The same substitution binds outputHtml and outputJson, the names a chart writes its files under. The whole plan runs as one program in one directory, so each operator is handed a stem of its own and a workflow ending in two charts keeps two pictures. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
696a192 to
298bbc7
Compare
A scan source generates its reader from the schema it reads off the file, and only a resolved URI can be opened. Compilation resolves the user-given name before it expands the plan; the export skipped that step, so the name stayed as typed, the schema could not be read, and the CSV reader quietly dropped the timestamp columns it would otherwise have named to pandas. A name that does not resolve is collected rather than thrown, so a workflow whose file is not chosen yet still exports as it did before. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
A file carries no types and pandas infers them, so a column of whole numbers with one empty cell comes back as a float where the engine, which has the schema, still has an integer. No reading of the values can settle it: a genuine DOUBLE holding 6.0 looks the same. The generator gains an overload that takes the input ports' schemas, defaulting to the schema-free form so an operator that does not care is untouched, and `renderedAsText` for the operators that turn a column into text. The translator folds each operator's output schemas onto the ports its downstream reads. The endpoint now compiles the plan rather than resolving the scan sources by hand. Compilation resolves them on the way, so the reader still reads its file, and it is also where the schemas come from. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
|
/unrequest-review @mengw15 |
|
/request-review @aglinxinyuan |
aglinxinyuan
left a comment
There was a problem hiding this comment.
This PR is still too large. For example, the change of limit and union doesn't have to be in this PR.
The trait, the translator, and nothing that uses them. The five operators that implemented the trait here, the endpoint that serves the script, the literal-escaping helper and the test-only column annotation each go to a change of their own, since none of them is what a reader of this one needs to follow. The translator's spec now drives stubs. What the translator does is place a block and bind the variables around it, so a stub that names its own block keeps these assertions off any real operator's emitted text, which would otherwise turn a change to that operator into a failure here. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
|
Split. The five operators and the endpoint are now in #8502, and this change is the trait and the translator alone. |
aglinxinyuan
left a comment
There was a problem hiding this comment.
LGTM! We can make change based on the caller later if needed.
…he script (apache#8502) ### What changes were proposed in this PR? Five operators implement `StandaloneCodeGenerator`, which is what turns the translator from a mechanism into something that produces a script that runs: Distinct, Filter, Limit, Projection and Union. Four of them read a single input; Union reads a variadic port, so it names the whole list of upstreams rather than a fixed count, which a fixed count gets wrong in both directions. The endpoint the editor calls comes with them, because what it is worth testing on is a script that runs rather than one made of placeholders. It takes a plan, compiles it first so a scan source can read the schema off the file it points at, hands the translator the output schemas that gives, and returns the script. A failed compile is logged and translation goes on without them, so a workflow whose file is not chosen yet still exports. `pyStringLiteral` comes with them too. A generator has to write a column name into the source it emits, and writing the quotes by hand lets any quote, backslash or newline in the name close the literal early and change, or break, the emitted program. apache#8327 is the trait and the translator, and nothing that uses them. ### Any related issues, documentation, discussions? Part of apache#8325, 2 of 27; that issue lists the set in order. Closes apache#8501, the task this change is the whole of. ### How was this PR tested? Each operator asserts the block it emits in its own spec. `PythonTemplateBuilderApiSpec` covers what `pyStringLiteral` escapes, including the NUL that Python refuses to compile anywhere in a source file. `WorkflowToPythonResourceSpec` drives the endpoint over a plan built from these operators and reads the script back. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
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 is the mechanism that produces one: a trait an operator implements to say how it reads outside the engine, and a translator that stitches what they return into a single script.
StandaloneCodeGeneratorreturns 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. A variadic port 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.
Nothing here implements it. The first five operators, and the endpoint that serves the script, are in #8502.
Any related issues, documentation, discussions?
Part of #8325, 1 of 27; 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, the file each operator writing one is given a name for, and the operator that has no generator.It drives stub operators rather than real ones. What the translator does is place a block and bind the variables around it, so a stub that names its own block keeps these assertions off any operator's emitted text, which would otherwise turn a change to that operator into a failure here.
Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)