test(verify): run a Python operator the way the engine runs it by kz930 · Pull Request #8357 · apache/texera

kz930 · 2026-09-02T23:31:11Z

What changes were proposed in this PR?

A Python operator is not called; it is handed to an interpreter that imports
the generated module and drives it through the same open, process and close
the engine uses. PyOpExecHarness writes that module, starts the driver, and
reads back what the operator emitted.

The driver is the engine's side of the contract written out plainly: it is
what makes the answer this side produces the engine's answer rather than an
approximation of it.

A timestamp reaches the operator the way the engine hands one to a Python worker: a naive datetime at microsecond resolution. It is read without pandas, so a year past 2262 arrives intact. That is the zoneless microsecond Arrow type #8704 makes the engine send.

A port that carried no rows reaches the operator with its declared columns, as #8766 makes the engine hand them. The driver declares each input port's schema right before that port finishes, where the worker does.

Eight lines of build.sbt are unrelated to the export and fixed while this change is in the file: File Service's test grouping read its fork options inside the lambda, which sbt hoists anyway, so the warning it printed on every build said the code meant something it did not.

Any related issues, documentation, discussions?

Part of #8325, 4 of 27; that issue lists the set in order. The timestamp handling matches the engine once #8704 lands. It needs #8766, which adds the input_schemas the driver writes, so it merges after it. It also needs #8506, which adds the Scala OperatorExecutor.inputSchemas the harness fills before it finishes a port.

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

How was this PR tested?

The tests in this change cover it. PyOpExecHarnessSpec also checks that a row the output schema does not describe is refused. It also checks that 2500-01-01 00:00:00.123456789 reaches the operator as 2500-01-01 00:00:00.123456. It also checks that a port that carried no rows hands the operator its declared columns. Without the driver's change the operator sees a table with no columns. The whole set is exercised together once the last piece lands: every operator run through the engine and through its generated script, and the two answers compared.

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

Generated-by: Claude Code (Claude Opus 5, Claude Opus 5.5)

A Python operator is not called; it is handed to an interpreter that imports the generated module and drives it through the same open, process and close the engine uses. `PyOpExecHarness` writes that module, starts the driver, and reads back what the operator emitted. The driver is the engine's side of the contract written out plainly: it is what makes the answer this side produces the engine's answer rather than an approximation of it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

github-actions · 2026-09-02T23:31:29Z

Automated Reviewer Suggestions

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

Contributors with relevant context: @tanishqgandhi1908
You can notify them by mentioning @tanishqgandhi1908 in a comment.

codecov-commenter · 2026-09-02T23:34:52Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 92.18%. Comparing base (7b8422e) to head (c3ace93).

Additional details and impacted files

@@            Coverage Diff            @@
##               main    #8357   +/-   ##
=========================================
  Coverage     92.18%   92.18%           
+ Complexity     5051     5050    -1     
=========================================
  Files          1286     1286           
  Lines         55937    55937           
  Branches       7140     7140           
=========================================
+ Hits          51563    51564    +1     
+ Misses         2612     2610    -2     
- Partials       1762     1763    +1

Flag	Coverage Δ		*Carryforward flag
access-control-service	`77.44% <ø> (ø)`
agent-service	`99.16% <ø> (ø)`		Carriedforward from f225ca5
amber	`86.93% <ø> (-0.01%)`	⬇️	Carriedforward from f225ca5
computing-unit-managing-service	`64.02% <ø> (ø)`
config-service	`87.61% <ø> (ø)`
file-service	`82.29% <ø> (ø)`
frontend	`96.48% <ø> (-0.01%)`	⬇️	Carriedforward from f225ca5
notebook-migration-service	`83.77% <ø> (ø)`
pyamber	`98.54% <ø> (+0.06%)`	⬆️	Carriedforward from f225ca5
workflow-compiling-service	`74.25% <ø> (ø)`		Carriedforward from f225ca5

*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-02T23:37:45Z

⚠️ Benchmark changes need a look

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

Compared against main 7b8422e 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	390	0.238	24,919/35,257/35,257 us	🔴 +20.1% / 🔴 +150.3%
🔴	bs=100 sw=10 sl=64	827	0.505	120,924/143,757/143,757 us	🔴 +6.5% / 🔴 +49.1%
🔴	bs=1000 sw=10 sl=64	948	0.579	1,040,005/1,376,479/1,376,479 us	🔴 +20.1% / 🔴 +48.2%

Baseline details

Latest main 7b8422e from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	390 tuples/sec	457 tuples/sec	888.52 tuples/sec	-14.7%	-56.1%
bs=10 sw=10 sl=64	MB/s	0.238 MB/s	0.279 MB/s	0.542 MB/s	-14.7%	-56.1%
bs=10 sw=10 sl=64	p50	24,919 us	20,747 us	11,467 us	+20.1%	+117.3%
bs=10 sw=10 sl=64	p95	35,257 us	32,558 us	14,087 us	+8.3%	+150.3%
bs=10 sw=10 sl=64	p99	35,257 us	32,558 us	17,519 us	+8.3%	+101.3%
bs=100 sw=10 sl=64	throughput	827 tuples/sec	867 tuples/sec	1,174 tuples/sec	-4.6%	-29.5%
bs=100 sw=10 sl=64	MB/s	0.505 MB/s	0.529 MB/s	0.716 MB/s	-4.5%	-29.5%
bs=100 sw=10 sl=64	p50	120,924 us	113,499 us	89,952 us	+6.5%	+34.4%
bs=100 sw=10 sl=64	p95	143,757 us	145,654 us	96,426 us	-1.3%	+49.1%
bs=100 sw=10 sl=64	p99	143,757 us	145,654 us	110,188 us	-1.3%	+30.5%
bs=1000 sw=10 sl=64	throughput	948 tuples/sec	964 tuples/sec	1,204 tuples/sec	-1.7%	-21.3%
bs=1000 sw=10 sl=64	MB/s	0.579 MB/s	0.588 MB/s	0.735 MB/s	-1.5%	-21.2%
bs=1000 sw=10 sl=64	p50	1,040,005 us	1,038,669 us	886,711 us	+0.1%	+17.3%
bs=1000 sw=10 sl=64	p95	1,376,479 us	1,145,640 us	928,644 us	+20.1%	+48.2%
bs=1000 sw=10 sl=64	p99	1,376,479 us	1,145,640 us	972,356 us	+20.1%	+41.6%

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,512.82,200,128000,390,0.238,24918.74,35257.32,35257.32
1,100,10,64,20,2418.86,2000,1280000,827,0.505,120923.84,143756.74,143756.74
2,1000,10,64,20,21096.02,20000,12800000,948,0.579,1040005.28,1376478.98,1376478.98

Eight lines this change is already in the file for. sbt lifts a `.value` written inside a lambda to the top of the task, so the options were read once rather than per suite either way; written where they were, they said the other thing, and sbt warned on it in every build. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Matches apache#8357, which carries this upstream. sbt lifts a `.value` written inside a lambda to the top of the task, so the options were read once either way; written where they were, they said the other thing. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

The two helpers here were a second copy of OpExecHarness's, carrying a note that they were kept inline so each harness read on its own and would be consolidated if a third arrived. Two copies of the same twenty lines is already the cost that note was deferring: they have to be changed together, and nothing says so at either site. Preparing the plan does not vary with the executor, so it happens once now. What differs between the harnesses stays here: how the prepared op is run. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Three kinds of comment came out. A drawing of the string the code below assembles. A restatement of a branch the reader can see. And the word MVP, which dated the scope to a moment rather than stating it. What replaces them says the same thing shorter, or says what the code cannot: which cases the harness does not drive and why none of them has an operator asking for it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

carloea2

The Python operator harness looks good.

`PhysicalPlan` is named only by `getPhysicalPlan` and by a comment, so scalafix reports the import and the check fails before any test runs. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

carloea2

The unused import was removed. Looks good.

carloea2

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.

The driver projected the operator's output onto the schema's attribute names and coerced each value with int/float/str. That is looser than the engine, which hands every yielded value to all_output_to_tuple and finalizes the result: an operator putting 1.5 in an INTEGER field, or a row with a field the schema does not have, or one missing a field, all raise there and none of the three raised here. A reference path that accepts output the engine rejects cannot say what the engine would do, so call the engine's own two functions instead of re-deriving them. finalize pickles a BINARY field's object itself and tags it, where the standalone side pickles the same object untagged, so drop the tag on the way out and both sides carry the same payload. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

carloea2

Further review with local regression tests and connected execution checks.

The driver collected everything the operator yielded and converted the list afterwards. `DataProcessor` reads a yield where it happens, turning it into tuples and finalizing them on the spot, so a UDF that yields one dict twice and changes it in between is recorded as the two values it produced. The driver recorded the last one twice. The operator now takes a callback and hands each value over at the yield, before the generator is advanced. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

DataProcessor writes each input port's schema into the executor before calling on_finish, so a TableOperator handed no rows can still say what its columns were. The driver stands in for DataProcessor here, so it has to do the same or the reference path answers on a frame of no columns where the engine answers on the declared ones. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

The handlers are written nine steps later in the set, and this borrowed a fixture writer from them for a single one-column row. Writing it with TupleIO, which the spec already reads with, leaves the order free of the one edge that pointed backwards. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

carloea2

Each emitted tuple is now finalized when it is yielded. This fixes reused mutable output values. Looks good.

carloea2

The harness declares large binary as supported, but both input coercion and output conversion reject it. An identity Python operator with a valid large binary field cannot be checked at all. Please handle large binary and add an input and output case.

The schema sidecar names large_binary and the driver's type map reads it, but neither the input coercion nor the output conversion knew the type, so an operator handed such a column could not be run at all. A large binary is a reference: the bytes live in S3 and the field is the s3:// URI that points at them. Both harnesses now carry that URI, which is what the worker sends and what LargeBinary holds on the other side, and building one touches no storage. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

kz930 · 2026-09-18T20:17:17Z

Fixed in f4cdf6a: a large binary is a reference, so both harnesses now carry the s3:// URI the field holds, which is what the worker sends and what LargeBinary holds on the other side. Two cases cover it, one handing the column to the operator as a reference it can read, one writing back a large binary the operator made.

carloea2

The large binary handling is now covered on both the Python and Scala sides, including input and output. The earlier blocker is fixed. Looks good.

apache#8488 is closed: a table with no rows carries no schema by design, so there is nothing for the driver to hand the operator and no engine behavior here to match. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

carloea2

One input-decoding mismatch reproduced against the actual Arrow tuple provider. Ordinary bytes passed as a control.

The worker's ArrowTableTupleProvider unpickles a BINARY cell that starts with the cast's pickle marker, so a model column reaches the next operator as a model. The driver handed the operator the bytes instead, and a predict on them raised AttributeError. Decoding the cell alone was not enough. The driver then finalized each input tuple, and the cast in finalize pickles any non-bytes value back into bytes. The worker's InputManager attaches the schema without finalizing, so the driver now builds input tuples the same way. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

… spec The pickled-cell test used a pickled list in place of a model. It now pickles a fitted DecisionTreeClassifier with the interpreter the harness runs and has the operator call predict on it, which is the call that raised AttributeError when the driver handed over bytes. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

…and refuse a row the schema does not describe Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

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

@carloea2 A Python operator now receives a timestamp as the engine sends it after #8704: a naive datetime at microsecond resolution.

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

carloea2

The current driver passes 12 direct Python probes with the declared #8766 SDK dependency. Seven Scala harness cases pass too, but the new model fixture fails on Windows before reaching the driver. scalafmtCheckAll passes.

Windows drops the quotes from a multiline `-c` argument, so the bytes literal reached Python as `bpickle` and the fixture failed before the harness ran. Write the script to a temporary .py file and run that. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

kz930 · 2026-10-13T02:28:27Z

@carloea2 The model helper now runs its script from a temporary .py file (b581ae0). Could you take another look when you have time?

carloea2

The Windows model-fixture fix passes. With the declared #8766 SDK selected through TEXERA_AMBER_PYTHON_HOME, all 8 Scala tests and 12 direct Python 3.12 driver probes pass. Formatting passes. No new finding in this update; the SDK prerequisite is still needed.

sbt -batch 'WorkflowCompilingService/testOnly *PyOpExecHarnessSpec -- -oDF' 'scalafmtCheckAll'

As the worker's EndChannelHandler now does (apache#8506). Without it the native side of the HashJoin right-outer variant put the right columns into the left ones. Moves to apache#8357 once apache#8506 is merged upstream. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

As the worker's EndChannelHandler does since apache#8506: a HashJoin probe that received no build rows names its right columns from the build port's declared schema. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

…iter does json.dumps writes a bare Infinity, which is not JSON; the engine's writer spells it as text. A NaN still leaves as null, as the worker casts it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

kz930 · 2026-10-22T08:02:38Z

@carloea2 Two commits since your review: the executor gets each finished port's declared schema, and a Python operator writes infinities as the engine does. Could you take another look?

carloea2

Retested the updated input-schema and infinity handling on fresh upstream main. All 8 PyOpExecHarnessSpec tests and scalafmtCheckAll pass with the declared SDK, executor-schema and Arrow-type APIs supplied locally. No new harness finding. Code looks good; the prerequisite PRs and CI still need clearing. Keeping the draft unchanged.

…rator-python # Conflicts: # build.sbt

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

build: read the fork options once, outside the grouping loop #8369

Closed

kz930 and others added 2 commits September 3, 2026 15:11

carloea2 approved these changes Sep 4, 2026

View reviewed changes

kz930 marked this pull request as draft September 4, 2026 17:44

fix(verify): drop an import nothing uses

9f51cd5

`PhysicalPlan` is named only by `getPhysicalPlan` and by a comment, so scalafix reports the import and the check fails before any test runs. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

carloea2 approved these changes Sep 4, 2026

View reviewed changes

carloea2 reviewed Sep 9, 2026

View reviewed changes

Comment thread workflow-compiling-service/src/test/resources/python/py_op_driver.py Resolved

carloea2 reviewed Sep 10, 2026

View reviewed changes

Comment thread workflow-compiling-service/src/test/resources/python/py_op_driver.py Resolved

kz930 and others added 2 commits September 10, 2026 13:02

kz930 mentioned this pull request Sep 14, 2026

test(verify): compare the charts the two paths drew #8509

Draft

Merge branch 'main' into feat/verify-run-operator-python

72c3570

kz930 mentioned this pull request Sep 15, 2026

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

Draft

carloea2 approved these changes Sep 18, 2026

View reviewed changes

carloea2 suggested changes Sep 18, 2026

View reviewed changes

carloea2 approved these changes Sep 18, 2026

View reviewed changes

kz930 mentioned this pull request Sep 18, 2026

test(verify): compare what the two paths produced #8359

Draft

carloea2 reviewed Sep 23, 2026

View reviewed changes

Comment thread workflow-compiling-service/src/test/resources/python/py_op_driver.py Resolved

kz930 and others added 2 commits September 23, 2026 16:16

kz930 mentioned this pull request Sep 27, 2026

test(verify): run every operator both ways and report what it did #8362

Draft

test(verify): hand a Python operator the timestamp the engine sends, …

a2f9668

…and refuse a row the schema does not describe Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

test(verify): hand a port that carried no rows its declared columns

135b9be

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

carloea2 suggested changes Oct 13, 2026

View reviewed changes

Comment thread ...g-service/src/test/scala/org/apache/texera/amber/translator/verify/PyOpExecHarnessSpec.scala Resolved

carloea2 reviewed Oct 13, 2026

View reviewed changes

kz930 requested a review from carloea2 October 13, 2026 05:42

test(verify): write a Python operator's infinities as the engine's wr…

3f7a878

…iter does json.dumps writes a bare Infinity, which is not JSON; the engine's writer spells it as text. A NaN still leaves as null, as the worker casts it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

carloea2 mentioned this pull request Oct 23, 2026

test(verify): hand-write the configurations a schema cannot describe #8363

Draft

carloea2 approved these changes Oct 23, 2026

View reviewed changes

Merge remote-tracking branch 'upstream/main' into feat/verify-run-ope…

c3ace93

…rator-python # Conflicts: # build.sbt

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!

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!

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!

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!

kz930 commented Sep 18, 2026

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 Sep 27, 2026

Uh oh!

carloea2 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

kz930 commented Oct 13, 2026

Uh oh!

carloea2 left a comment

Choose a reason for hiding this comment

Uh oh!

kz930 commented Oct 22, 2026

Uh oh!

carloea2 left a comment

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 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