feat(workflow-operator): export the CSV sources as Python by kz930 · Pull Request #8341 · apache/texera

kz930 · 2026-09-02T18:18:44Z

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)

…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>

…rces

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 · 2026-09-02T18:20:13Z

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 · 2026-09-02T18:23:33Z

Codecov Report

❌ Patch coverage is 65.77181% with 51 lines in your changes missing coverage. Please review.
✅ Project coverage is 92.50%. Comparing base (2082021) to head (1b126e0).
⚠️ Report is 43 commits behind head on main.
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
...or/source/scan/csvOld/CSVOldScanSourceOpDesc.scala	20.45%	34 Missing and 1 partial ⚠️
.../source/scan/csv/ParallelCSVScanSourceOpDesc.scala	77.41%	3 Missing and 4 partials ⚠️
...operator/source/scan/csv/CSVScanSourceOpDesc.scala	84.21%	3 Missing and 3 partials ⚠️
.../amber/translator/WorkflowToPythonTranslator.scala	80.00%	0 Missing and 2 partials ⚠️
...exera/amber/operator/StandaloneCodeGenerator.scala	95.83%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #8341      +/-   ##
============================================
- Coverage     92.57%   92.50%   -0.08%     
- Complexity     5046     5069      +23     
============================================
  Files          1251     1249       -2     
  Lines         53368    53495     +127     
  Branches       6618     6631      +13     
============================================
+ Hits          49405    49484      +79     
- Misses         2319     2351      +32     
- Partials       1644     1660      +16

Flag	Coverage Δ		*Carryforward flag
access-control-service	`77.38% <ø> (ø)`
agent-service	`99.16% <ø> (ø)`		Carriedforward from 5acfad6
amber	`87.93% <64.74%> (-0.16%)`	⬇️
computing-unit-managing-service	`60.48% <ø> (+0.07%)`	⬆️
config-service	`87.37% <ø> (ø)`
file-service	`82.00% <ø> (+0.47%)`	⬆️
frontend	`96.56% <ø> (-0.01%)`	⬇️	Carriedforward from 5acfad6
notebook-migration-service	`83.73% <ø> (ø)`
pyamber	`98.56% <ø> (ø)`		Carriedforward from 5acfad6
workflow-compiling-service	`74.09% <80.00%> (ø)`

*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 · 2026-09-02T18:27:07Z

⚠️ 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	379	0.232	25,655/36,107/36,107 us	🟢 -8.9% / 🔴 +143.3%
🔴	bs=100 sw=10 sl=64	776	0.473	124,768/158,548/158,548 us	🔴 +13.8% / 🔴 +57.5%
⚪	bs=1000 sw=10 sl=64	903	0.551	1,096,114/1,187,707/1,187,707 us	⚪ within ±5% / 🔴 +24.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	379 tuples/sec	389 tuples/sec	840.06 tuples/sec	-2.6%	-54.9%
bs=10 sw=10 sl=64	MB/s	0.232 MB/s	0.237 MB/s	0.513 MB/s	-2.1%	-54.8%
bs=10 sw=10 sl=64	p50	25,655 us	24,068 us	11,813 us	+6.6%	+117.2%
bs=10 sw=10 sl=64	p95	36,107 us	39,614 us	14,838 us	-8.9%	+143.3%
bs=10 sw=10 sl=64	p99	36,107 us	39,614 us	17,837 us	-8.9%	+102.4%
bs=100 sw=10 sl=64	throughput	776 tuples/sec	818 tuples/sec	1,087 tuples/sec	-5.1%	-28.6%
bs=100 sw=10 sl=64	MB/s	0.473 MB/s	0.499 MB/s	0.663 MB/s	-5.2%	-28.7%
bs=100 sw=10 sl=64	p50	124,768 us	119,797 us	94,074 us	+4.1%	+32.6%
bs=100 sw=10 sl=64	p95	158,548 us	139,366 us	100,688 us	+13.8%	+57.5%
bs=100 sw=10 sl=64	p99	158,548 us	139,366 us	116,015 us	+13.8%	+36.7%
bs=1000 sw=10 sl=64	throughput	903 tuples/sec	912 tuples/sec	1,128 tuples/sec	-1.0%	-20.0%
bs=1000 sw=10 sl=64	MB/s	0.551 MB/s	0.557 MB/s	0.689 MB/s	-1.1%	-20.0%
bs=1000 sw=10 sl=64	p50	1,096,114 us	1,092,702 us	916,643 us	+0.3%	+19.6%
bs=1000 sw=10 sl=64	p95	1,187,707 us	1,156,862 us	957,169 us	+2.7%	+24.1%
bs=1000 sw=10 sl=64	p99	1,187,707 us	1,156,862 us	978,578 us	+2.7%	+21.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,527.16,200,128000,379,0.232,25654.71,36107.15,36107.15
1,100,10,64,20,2578.87,2000,1280000,776,0.473,124768.43,158548.01,158548.01
2,1000,10,64,20,22143.37,20000,12800000,903,0.551,1096113.83,1187707.36,1187707.36

…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>

…into feat/standalone-sources

… 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>

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>

kz930 · 2026-09-25T21:48:28Z

The source tests that each read a file of their own, including the ones added for your findings, now run in #8364's shared source table, which reads every source through the same cases and checks dtypes as well as values; this PR keeps only what that table cannot reach, and is down to about 2,100 lines.

…zoneless microsecond timestamp Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

…o PRs that follow Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

kz930 · 2026-09-27T03:44:45Z

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

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

carloea2

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>

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>

carloea2

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.

… 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

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 · 2026-10-14T08:52:43Z

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

kz930 and others added 5 commits September 1, 2026 16:23

Merge remote-tracking branch 'upstream/main' into feat/standalone-sou…

97e05ec

…rces

github-actions Bot assigned kz930 Sep 2, 2026

github-actions Bot added feature dependencies Pull requests that update a dependency file common platform Non-amber Scala service paths labels Sep 2, 2026

kz930 and others added 2 commits September 2, 2026 11:32

Merge remote-tracking branch 'myfork/feat/standalone-verify-harness' …

9be0dbd

…into feat/standalone-sources

github-actions Bot added the ci changes related to CI label Sep 2, 2026

kz930 and others added 3 commits September 2, 2026 16:10

Merge upstream/main

3f07020

chore: leave the two verify files to apache#8327 as well

df10bd8

This branch held an older copy of both and changes neither. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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 mentioned this pull request Sep 2, 2026

A workflow cannot be taken out of Texera as a script that runs on its own #8325

Open

13 of 31 tasks

fix(operator): read timestamps as the engine does, and hand Python a …

f874af5

…zoneless microsecond timestamp Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

github-actions Bot added the pyamber label Sep 27, 2026

refactor(operator): leave the JSONL, Arrow and line sources to the tw…

7ed4f40

…o PRs that follow Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

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 Bot removed the pyamber label Sep 27, 2026

kz930 and others added 2 commits October 1, 2026 01:06

Merge upstream/main into feat/standalone-sources

6b0a3b5

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

Merge upstream/main into feat/standalone-sources

37f8cc2

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

carloea2 suggested changes Oct 7, 2026

View reviewed changes

Comment thread .../main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala

kz930 requested a review from carloea2 October 8, 2026 07:51

carloea2 approved these changes Oct 8, 2026

View reviewed changes

This was referenced Oct 13, 2026

test(verify): run the script the export produces #8358

Draft

feat(operator): read a Parquet file as a source #8512

Draft

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

carloea2 suggested changes Oct 14, 2026

View reviewed changes

Comment thread ...on/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneHelpers.scala Outdated

fix(workflow-operator): read an overflowing hex double as an infinity

1b126e0

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

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

Navigation Menu

Repository navigation

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

feat(workflow-operator): export the CSV sources as Python#8341
kz930 wants to merge 49 commits into
apache:mainfrom
kz930:feat/standalone-sources

kz930 commented Sep 2, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Sep 2, 2026 •

edited

Loading

Uh oh!

codecov-commenter commented Sep 2, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Sep 2, 2026 •

edited

Loading

Uh oh!

kz930 commented Sep 25, 2026

Uh oh!

kz930 commented Sep 27, 2026

Uh oh!

carloea2 left a comment

Uh oh!

Uh oh!

carloea2 left a comment

Uh oh!

carloea2 left a comment

Uh oh!

Uh oh!

kz930 commented Oct 14, 2026

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Conversation

kz930 commented Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this PR?

Any related issues, documentation, discussions?

How was this PR tested?

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

Uh oh!

github-actions Bot commented Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Automated Reviewer Suggestions

Uh oh!

codecov-commenter commented Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

github-actions Bot commented Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

⚠️ Benchmark changes need a look

Uh oh!

kz930 commented Sep 25, 2026

Uh oh!

kz930 commented Sep 27, 2026

Uh oh!

carloea2 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

carloea2 left a comment

Choose a reason for hiding this comment

Uh oh!

carloea2 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

kz930 commented Oct 14, 2026

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

kz930 commented Sep 2, 2026 •

edited

Loading

github-actions Bot commented Sep 2, 2026 •

edited

Loading

codecov-commenter commented Sep 2, 2026 •

edited

Loading

github-actions Bot commented Sep 2, 2026 •

edited

Loading