Skip to content

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

Draft
kz930 wants to merge 59 commits into
apache:mainfrom
kz930:feat/standalone-sources
Draft

kz930 wants to merge 59 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 61.58940% with 58 lines in your changes missing coverage. Please review.
✅ Project coverage is 91.66%. Comparing base (2082021) to head (915fae5).
⚠️ Report is 103 commits behind head on main.
✅ All tests successful. No failed tests found.

Files with missing lines Patch % Lines
...or/source/scan/csvOld/CSVOldScanSourceOpDesc.scala 23.07% 29 Missing and 1 partial ⚠️
...exera/amber/operator/StandaloneCodeGenerator.scala 69.44% 10 Missing and 1 partial ⚠️
...operator/source/scan/csv/CSVScanSourceOpDesc.scala 75.75% 5 Missing and 3 partials ⚠️
.../source/scan/csv/ParallelCSVScanSourceOpDesc.scala 77.41% 3 Missing and 4 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.66%   -0.92%     
- Complexity     5046     5075      +29     
============================================
  Files          1251     1253       +2     
  Lines         53368    54240     +872     
  Branches       6618     6747     +129     
============================================
+ Hits          49405    49718     +313     
- Misses         2319     2820     +501     
- Partials       1644     1702      +58     
Flag Coverage Δ *Carryforward flag
access-control-service 77.44% <ø> (+0.06%) ⬆️
agent-service 99.16% <ø> (ø) Carriedforward from 1c80ab6
amber 85.77% <60.28%> (-2.32%) ⬇️
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 1c80ab6
notebook-migration-service 83.77% <ø> (+0.03%) ⬆️
pyamber 98.56% <ø> (ø) Carriedforward from 1c80ab6
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 354 0.216 28,452/32,984/32,984 us 🟢 -15.8% / 🔴 +216.4%
⚪ bs=100 sw=10 sl=64 811 0.495 122,175/142,148/142,148 us ⚪ within ±5% / 🔴 +89.9%
🔴 bs=1000 sw=10 sl=64 957 0.584 1,030,798/1,194,804/1,194,804 us 🔴 +10.9% / 🔴 +69.1%
Baseline details

Latest main 2082021 from same runner

config metric PR latest main 7d avg Δ latest Δ 7d
bs=10 sw=10 sl=64 throughput 354 tuples/sec 383 tuples/sec 1,119 tuples/sec -7.6% -68.4%
bs=10 sw=10 sl=64 MB/s 0.216 MB/s 0.234 MB/s 0.683 MB/s -7.7% -68.4%
bs=10 sw=10 sl=64 p50 28,452 us 26,382 us 8,992 us +7.8% +216.4%
bs=10 sw=10 sl=64 p95 32,984 us 39,189 us 11,255 us -15.8% +193.1%
bs=10 sw=10 sl=64 p99 32,984 us 39,189 us 13,808 us -15.8% +138.9%
bs=100 sw=10 sl=64 throughput 811 tuples/sec 847 tuples/sec 1,518 tuples/sec -4.3% -46.6%
bs=100 sw=10 sl=64 MB/s 0.495 MB/s 0.517 MB/s 0.927 MB/s -4.3% -46.6%
bs=100 sw=10 sl=64 p50 122,175 us 117,399 us 68,504 us +4.1% +78.3%
bs=100 sw=10 sl=64 p95 142,148 us 137,646 us 74,866 us +3.3% +89.9%
bs=100 sw=10 sl=64 p99 142,148 us 137,646 us 89,939 us +3.3% +58.1%
bs=1000 sw=10 sl=64 throughput 957 tuples/sec 966 tuples/sec 1,569 tuples/sec -0.9% -39.0%
bs=1000 sw=10 sl=64 MB/s 0.584 MB/s 0.59 MB/s 0.957 MB/s -1.0% -39.0%
bs=1000 sw=10 sl=64 p50 1,030,798 us 1,032,742 us 669,501 us -0.2% +54.0%
bs=1000 sw=10 sl=64 p95 1,194,804 us 1,077,784 us 706,487 us +10.9% +69.1%
bs=1000 sw=10 sl=64 p99 1,194,804 us 1,077,784 us 737,560 us +10.9% +62.0%
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,565.62,200,128000,354,0.216,28452.03,32984.13,32984.13
1,100,10,64,20,2467.12,2000,1280000,811,0.495,122174.81,142148.12,142148.12
2,1000,10,64,20,20894.91,20000,12800000,957,0.584,1030798.13,1194804.20,1194804.20

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
@kz930 kz930 changed the title feat(workflow-operator): export the source operators as Python feat(workflow-operator): export the CSV sources as Python Sep 27, 2026
@github-actions github-actions Bot removed the pyamber label Sep 27, 2026
@kz930

kz930 commented Sep 27, 2026

Copy link
Copy Markdown
Contributor Author

@carloea2 I split this into #8341 (the CSV family), #8704 (JSON Lines, Arrow and the engine's timestamp type) and #8705 (the line sources and the URL fetcher), which merge in that order. Each source now reads a timestamp from text as DateParserUtils does.

kz930 and others added 2 commits October 1, 2026 01:06
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
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.

One remaining empty-input mismatch, reproduced from the emitted Python read with pandas 2.2.3. This was a focused check, not a full workflow run.

…ses the end

Without a header, an offset at or past the last row left pandas no line to
count the columns from, so the exported read raised EmptyDataError where the
engine emits no rows. Both CSV readers now name the columns by position when
the schema resolves, and a parity test in each spec checks the engine and the
script agree on zero rows with the schema's columns.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
kz930 added a commit to Nicoleee1108/texera_workflow_to_py that referenced this pull request Oct 8, 2026
The scan sources take apache#8341's resolution: the Limit 0 sample now lives in the
shared inferSampleSize, and the CSV delimiters use the picker constant. JSONL
reads its sample through inferSampleSize too.

CSVOld's optionals variant is red until the delimiter examples land in apache#8388.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@kz930
kz930 requested a review from carloea2 October 8, 2026 07:51

@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 10 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>

This branch has not been deployed

No deployments
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