Skip to content

feat(workflow-operator): export the CSV sources as Python - #8341

Merged
aglinxinyuan merged 61 commits into
apache:mainfrom
kz930:feat/standalone-sources
Oct 26, 2026
Merged

aglinxinyuan merged 61 commits into
apache:mainfrom
kz930:feat/standalone-sources

Conversation

@kz930

@kz930 kz930 commented Sep 2, 2026 •

Copy link
Copy Markdown
Contributor

What changes were proposed in this PR?

The three CSV source operators (CSV, CSV Old and Parallel CSV) implement StandaloneCodeGenerator, so an exported script starts from the same data the workflow did rather than from a variable the reader has to fill in. They also carry what every source after them uses: the placeholder a source writes where its file is named, and the parse that reads a timestamp column from text.

A source is the one place where the script cannot simply repeat what the operator does. The engine resolves a dataset through Texera's storage and hands the operator a URI; a script has no such resolver, so it reads the file from its own directory. Each source writes a placeholder where its file should be named and declares the path it reads, and the translator assigns the names once it can see the whole plan: the last segment of the path where that is free, and a numbered form of it where two sources read different files whose paths end the same way. The last segment rather than a parse of the whole string, because the resolver percent-encodes the file-relative parts but leaves the repository and version names as the user typed them, and a dataset version with a space in its name makes new URI throw before any code is generated. That is what makes an exported script portable, and it is equally its one precondition: the data has to sit beside the script, under the names the script states.

A timestamp is read as the engine reads one. The engine hands each field to DateParserUtils, so a row states its own format, any year reads, an offset is moved to the local zone, and the reading is cut to the millisecond. pandas' date parsing does none of that. So each CSV reader reads the column as text and converts it with the helper #8597 adds.

The end of an exported scan window is counted in Long: two Ints the operator accepts can add up past what an Int holds, and the slice came out negative.

Two of the changes made in review are deliberately not the literal instruction, because the three CSV readers do not agree with each other about a blank field. Old CSV sets keep_default_na=False but names no missing value at all: scala-csv hands a blank back as the empty string rather than as a null, so the na_values the main CSV source needs would null a value this reader keeps. Parallel CSV asks for such a column as text rather than as a nullable integer: its schema is read with scala-csv, where one blank types the whole column as text, while its executor nulls the blank and parses the rest as that text. Each fix follows the reader it belongs to rather than the one next to it.

Any related issues, documentation, discussions?

Part of #8325, step 12a; #8702 and #8703 follow. That issue lists the set in order. It needs #8327 for the trait, so it does not compile until that lands, and the rows these operators add to the verification runner follow with the harness rather than as whole new files here. It also needs #8597, whose helpers the timestamp reading calls, so #8597 merges first.

Closes #8415, the task this change is the whole of.

One engine defect was found while matching these readers, and it is fixed here rather than worked around: the export alone could not close it. #8602, where a scan window of no rows left the type inference nothing to read, so CSV declared a schema of no columns and CSV Old threw. Its JSONL half follows in #8702, which closes it.

How was this PR tested?

Each operator asserts the block it emits in its own spec. Where a difference is one of behaviour rather than of text, the spec runs the engine's own executor and the generated script over the same file and compares what each produced: the row window the parallel reader ignores, and the nullable long each reader has to keep exact. Once the verification lands, every operator is also run both ways on every configuration its schema offers; this branch is cut from main and does not carry that machinery, so those runs are not on this diff's CI.

Was this PR authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Opus 5)

kz930 and others added 5 commits September 1, 2026 16:23
…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>
…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>
Ten sources say how they read outside the engine, so an exported script
starts from the same data the workflow did rather than from a variable
the reader has to fill in: the CSV family, JSON Lines, Arrow, plain
text, and the two that read a file named at runtime.

A source is the one place where the script cannot simply repeat what the
operator does. The engine resolves a dataset through Texera's storage
and hands the operator a URI; a script has no such resolver, so it reads
the file from its own directory by the name the URI ended with. That is
what makes an exported script portable, and it is also its one
precondition: the data has to sit beside the script.

Two are reported as unverifiable rather than exported blind. File Scan
takes its filenames from an input port at run time, which a source
harness has nothing to feed; URL Fetcher reads a live URL, so no two
runs are required to agree.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
@github-actions github-actions Bot added feature dependencies Pull requests that update a dependency file common platform Non-amber Scala service paths labels Sep 2, 2026
@github-actions

github-actions Bot commented Sep 2, 2026 •

Copy link
Copy Markdown
Contributor

Automated Reviewer Suggestions

Based on the git blame history of the changed files, we recommend the following reviewers:

  • Contributors with relevant context: @sshiv012, @eugenegujing, @roshiiiz
    You can notify them by mentioning @sshiv012, @eugenegujing, @roshiiiz in a comment.

@codecov-commenter

codecov-commenter commented Sep 2, 2026 •

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 62.68657% with 50 lines in your changes missing coverage. Please review.
✅ Project coverage is 91.67%. Comparing base (2082021) to head (253241c).
⚠️ Report is 106 commits behind head on main.

Files with missing lines Patch % Lines
...or/source/scan/csvOld/CSVOldScanSourceOpDesc.scala 26.47% 24 Missing and 1 partial ⚠️
...exera/amber/operator/StandaloneCodeGenerator.scala 69.44% 10 Missing and 1 partial ⚠️
...operator/source/scan/csv/CSVScanSourceOpDesc.scala 75.00% 4 Missing and 3 partials ⚠️
.../source/scan/csv/ParallelCSVScanSourceOpDesc.scala 79.16% 2 Missing and 3 partials ⚠️
.../amber/translator/WorkflowToPythonTranslator.scala 80.00% 0 Missing and 2 partials ⚠️
Additional details and impacted files
@@             Coverage Diff              @@
##               main    #8341      +/-   ##
============================================
- Coverage     92.57%   91.67%   -0.90%     
- Complexity     5046     5079      +33     
============================================
  Files          1251     1253       +2     
  Lines         53368    54229     +861     
  Branches       6618     6743     +125     
============================================
+ Hits          49405    49715     +310     
- Misses         2319     2813     +494     
- Partials       1644     1701      +57     
Flag Coverage Δ *Carryforward flag
access-control-service 77.44% <ø> (+0.06%) ⬆️
agent-service 99.16% <ø> (ø) Carriedforward from 377b4bb
amber 85.80% <61.29%> (-2.29%) ⬇️ Carriedforward from 377b4bb
computing-unit-managing-service 61.84% <ø> (+1.42%) ⬆️
config-service 87.50% <ø> (+0.12%) ⬆️
file-service 81.92% <ø> (+0.38%) ⬆️
frontend 96.56% <ø> (-0.01%) ⬇️ Carriedforward from 377b4bb
notebook-migration-service 83.77% <ø> (+0.03%) ⬆️
pyamber 98.56% <ø> (ø) Carriedforward from 377b4bb
workflow-compiling-service 74.25% <80.00%> (+0.15%) ⬆️

*This pull request uses carry forward flags. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

@github-actions

github-actions Bot commented Sep 2, 2026 •

Copy link
Copy Markdown
Contributor

⚠️ Benchmark changes need a look

🟢 2 better · 🔴 5 worse · ⚪ 8 noise (<±5%) · 0 without baseline

Compared against main 2082021 benchmarked on this same runner, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.

Dashboard · Run

config throughput MB/s latency max Δ latest / 7d
🔴 bs=10 sw=10 sl=64 504 0.308 19,436/23,619/23,619 us 🟢 -42.5% / 🔴 +93.2%
🔴 bs=100 sw=10 sl=64 1,202 0.733 80,783/108,842/108,842 us 🔴 +6.3% / 🔴 +24.9%
🔴 bs=1000 sw=10 sl=64 1,443 0.881 690,958/788,086/788,086 us 🔴 +5.4% / 🟢 -11.5%
Baseline details

Latest main 2082021 from same runner

config metric PR latest main 7d avg Δ latest Δ 7d
bs=10 sw=10 sl=64 throughput 504 tuples/sec 498 tuples/sec 1,019 tuples/sec +1.2% -50.6%
bs=10 sw=10 sl=64 MB/s 0.308 MB/s 0.304 MB/s 0.622 MB/s +1.3% -50.5%
bs=10 sw=10 sl=64 p50 19,436 us 17,584 us 10,144 us +10.5% +91.6%
bs=10 sw=10 sl=64 p95 23,619 us 41,067 us 12,225 us -42.5% +93.2%
bs=10 sw=10 sl=64 p99 23,619 us 41,067 us 15,527 us -42.5% +52.1%
bs=100 sw=10 sl=64 throughput 1,202 tuples/sec 1,254 tuples/sec 1,371 tuples/sec -4.1% -12.3%
bs=100 sw=10 sl=64 MB/s 0.733 MB/s 0.765 MB/s 0.837 MB/s -4.2% -12.4%
bs=100 sw=10 sl=64 p50 80,783 us 79,013 us 79,910 us +2.2% +1.1%
bs=100 sw=10 sl=64 p95 108,842 us 102,370 us 87,155 us +6.3% +24.9%
bs=100 sw=10 sl=64 p99 108,842 us 102,370 us 106,643 us +6.3% +2.1%
bs=1000 sw=10 sl=64 throughput 1,443 tuples/sec 1,451 tuples/sec 1,415 tuples/sec -0.6% +2.0%
bs=1000 sw=10 sl=64 MB/s 0.881 MB/s 0.886 MB/s 0.864 MB/s -0.6% +2.0%
bs=1000 sw=10 sl=64 p50 690,958 us 691,820 us 780,407 us -0.1% -11.5%
bs=1000 sw=10 sl=64 p95 788,086 us 747,360 us 817,814 us +5.4% -3.6%
bs=1000 sw=10 sl=64 p99 788,086 us 747,360 us 848,683 us +5.4% -7.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,396.75,200,128000,504,0.308,19435.53,23619.01,23619.01
1,100,10,64,20,1664.27,2000,1280000,1202,0.733,80783.30,108841.70,108841.70
2,1000,10,64,20,13855.82,20000,12800000,1443,0.881,690957.79,788085.95,788085.95

kz930 and others added 2 commits September 2, 2026 11:32
…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>
@github-actions github-actions Bot added the ci changes related to CI label Sep 2, 2026
… 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>
kz930 and others added 3 commits September 2, 2026 16:10
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>
This branch held an older copy of both and changes neither.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
@github-actions github-actions Bot removed dependencies Pull requests that update a dependency file ci changes related to CI platform Non-amber Scala service paths labels Sep 2, 2026

@carloea2 carloea2 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The empty-input finding is fixed. Ran the CSV source and translator specs locally: 53 tests passed, including both native versus exported empty-input cases. Looks good. This still needs #8597 to merge first.

kz930 and others added 2 commits October 13, 2026 15:07
… toDouble does

The engine reads a field with toDouble, so "NaN", a trailing d or f, and a
hex literal are numbers and the column is inferred DOUBLE. pandas' float
parser takes none of them and kept the whole column as strings, and an
empty cell beside a "NaN" read as NaN too. Read such a column cell by cell
after the read, and keep it nullable only where it holds both.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…no empty cell

float64 reports a NaN as missing to every operator that reads the mask, so
a column holding "NaN" and no empty cell read its NaN as null downstream.
Put any column holding a NaN into the nullable Float64.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

@carloea2 carloea2 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The 42 existing CSV tests pass. Six finite-value controls pass too, but six overflow regressions fail across the three readers. Formatting passes and the PR files are unchanged locally. The new hex-double helper still differs from the native readers on valid input.

sbt -batch 'WorkflowOperator/testOnly *CSVScanSourceOpDescSpec *CSVOldScanSourceOpDescSpec *CSVHexOverflowReviewSpec -- -oDF' 'scalafmtCheckAll'

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@kz930

kz930 commented Oct 14, 2026

Copy link
Copy Markdown
Contributor Author

@carloea2 The hex overflow case is fixed in 1b126e0. Could you take another look?

@kz930
kz930 marked this pull request as draft October 15, 2026 22:57

@carloea2 carloea2 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The hex-overflow finding is fixed. All 42 CSV source tests and 16 translator tests pass, including positive and negative overflow in all three readers. Formatting passes after mechanically combining the #8597 prerequisites. Please rebase after #8597: the shared helper conflicts, and the two companion-object additions must be combined. No new CSV finding. Keeping the draft unchanged.

kz930 and others added 12 commits October 23, 2026 15:11
…ported script

_texera_java_double and _texera_read_doubles explained themselves in
Python comments, which every exported CSV script printed. The notes now
live in the Scaladoc, one line per function.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
A CSV source wrote one line per DOUBLE or TIMESTAMP column, 19 lines for
the shared table. Several columns of one type now go through one loop
over their names; a single column keeps its one line.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… columns

The CSV sources already rename the frame to the schema's columns, so the
empty-read line's reindex listed every name a second time to change
nothing. They now set only the types; a JSONL read, which has no columns
when empty, keeps the reindex.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…cross the exports

Filter, the text sources and the CSV readers each carried their own copy
of String.trim, Long.parseLong and Double.parseDouble. AttributeCasts now
holds one of each, _texera_java_trim, _texera_parse_int and a strict
_texera_java_double, for every operator to call.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…types

A CSV source named every column three times: in the rename, in the
reader loops and in the empty-read types. It now writes one _schema dict
of name to type and takes the rename, the readers and the empty-read
types from it. The loop names only the readers the table's types need.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… key

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… and parses it

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…apart

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@kz930

kz930 commented Oct 26, 2026

Copy link
Copy Markdown
Contributor Author

@carloea2 Since your approval, the CSV sources and the shared cast helpers were tightened: one Java trim, int and double parser and one key-code helper now serve every operator, the cast boolean and double read text the way the engine does (they disagreed on several inputs before), and each source infers its schema once per export. Could you take another look?

@aglinxinyuan aglinxinyuan left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM!

@aglinxinyuan
aglinxinyuan added this pull request to the merge queue Oct 26, 2026
Merged via the queue into apache:main with commit 8dcdf05 Oct 26, 2026
33 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

common feature platform Non-amber Scala service paths

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Export the CSV sources as Python

4 participants