feat(workflow-operator): export the join and set operators as Python

[kz930]

What changes were proposed in this PR?

Six operators implement StandaloneCodeGenerator: the three joins and the three set operations. What they have in common is that they combine or compare tables rather than read what is inside a row.

#8340 keeps the operators that work on text. Aggregate and the sort family follow in #8709, and the samplers and control flow in #8710.

Review then found two places where the block answered a question the engine answers differently. The hash join asked pandas for a suffix, which appends once and then refuses the duplicate it just made, and let an outer join's holes cost an integer column its type.

Running the engine on those cases found the platform out of step with itself. HashJoin declared a right column's new name by checking only the left names, while the executor also avoids the other right names. So a right x beside a right x#@1 was declared as a second x#@1, and one value was lost. An unmatched right row in an outer join kept its raw names and landed in the left column. Both now go through one renaming rule in JoinUtils.

The shared table then found one more. In a right or full outer join on keys both sides name alike, the engine writes an unmatched right row's key into the left key column, and the export left it empty. The engine had its own half: an unmatched right key named like a left payload column landed in that payload. The export now fills the key, and the engine leaves it out of the payload.

The set operations matched rows the way pandas does. The engine's HashSet compares rows with Scala's ==, under which NaN equals nothing, and hashes them by each double's bits, which tell 0.0 from -0.0. The export now selects rows by a key that follows both rules.

A probe worker that received no build rows could not name an unmatched right row's columns, because it read the left names off the first build row. This happens when the left input is empty or no left key hashes to that worker. The runtime now records each input port's declared schema on the executor before the port finishes (OperatorExecutor.inputSchemas, filled in EndChannelHandler), and the probe reads the left names from it. This is the Scala counterpart of what #8766 does for Python operators.

Any related issues, documentation, discussions?

Part of #8325, step 11a; #8709 and #8710 follow. That issue lists the set in order.

The set operations and joins keyed on a null or a NaN need the nullable dtypes #8704 reads an Arrow file into. Without it a null and a NaN share a slot and both behave as they did before, so this PR is no worse on its own and only complete alongside that one.

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

How was this PR tested?

Each operator asserts the block it emits in its own spec. The interval join reads its keys off the merged frame rather than writing over an input column that shares a temporary's name.

The join cases the review named run in the verify tables of #8363, which put the same input through the engine and through the script and compare the two. What those tables cannot reach stays here as executed tests: a null kept apart from a NaN in the set operations and the join key. HashJoinProbeOpExecSpec pins the platform fixes.

Each set operation's spec runs its native executor and the export over NaN, signed-zero, finite and null rows. HashJoinProbeOpExecSpec covers an empty build side, and EndChannelHandlerSpec checks that the schema is in place before the operator finishes the port.

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

Generated-by: Claude Code (Claude Opus 5)

🤖 Generated with Claude Code

[github-actions]

Automated Reviewer Suggestions

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

• Contributors with relevant context: @aglinxinyuan, @Yicong-Huang, @eugenegujing
  You can notify them by mentioning @aglinxinyuan, @Yicong-Huang, @eugenegujing in a comment.

[github-actions]

⚠️ Benchmark changes need a look

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

Compared against main 6a29c59 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	612	0.373	15,841/23,042/23,042 us	🔴 +22.3% / 🔴 +57.7%
🔴	bs=100 sw=10 sl=64	1,541	0.94	63,273/79,398/79,398 us	🔴 +10.7% / 🟢 -33.6%
⚪	bs=1000 sw=10 sl=64	1,906	1.163	521,247/615,530/615,530 us	⚪ within ±5% / 🟢 +55.5%

Baseline details

Latest main 6a29c59 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	612 tuples/sec	695 tuples/sec	912.36 tuples/sec	-11.9%	-32.9%
bs=10 sw=10 sl=64	MB/s	0.373 MB/s	0.424 MB/s	0.557 MB/s	-12.0%	-33.0%
bs=10 sw=10 sl=64	p50	15,841 us	12,955 us	11,391 us	+22.3%	+39.1%
bs=10 sw=10 sl=64	p95	23,042 us	23,545 us	14,608 us	-2.1%	+57.7%
bs=10 sw=10 sl=64	p99	23,042 us	23,545 us	17,250 us	-2.1%	+33.6%
bs=100 sw=10 sl=64	throughput	1,541 tuples/sec	1,609 tuples/sec	1,161 tuples/sec	-4.2%	+32.7%
bs=100 sw=10 sl=64	MB/s	0.94 MB/s	0.982 MB/s	0.709 MB/s	-4.3%	+32.6%
bs=100 sw=10 sl=64	p50	63,273 us	60,696 us	91,471 us	+4.2%	-30.8%
bs=100 sw=10 sl=64	p95	79,398 us	71,746 us	100,192 us	+10.7%	-20.8%
bs=100 sw=10 sl=64	p99	79,398 us	71,746 us	119,565 us	+10.7%	-33.6%
bs=1000 sw=10 sl=64	throughput	1,906 tuples/sec	1,924 tuples/sec	1,226 tuples/sec	-0.9%	+55.5%
bs=1000 sw=10 sl=64	MB/s	1.163 MB/s	1.174 MB/s	0.748 MB/s	-0.9%	+55.4%
bs=1000 sw=10 sl=64	p50	521,247 us	510,864 us	887,282 us	+2.0%	-41.3%
bs=1000 sw=10 sl=64	p95	615,530 us	593,373 us	924,352 us	+3.7%	-33.4%
bs=1000 sw=10 sl=64	p99	615,530 us	593,373 us	951,858 us	+3.7%	-35.3%

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,327.02,200,128000,612,0.373,15841.33,23041.69,23041.69
1,100,10,64,20,1298.09,2000,1280000,1541,0.940,63272.58,79397.85,79397.85
2,1000,10,64,20,10495.81,20000,12800000,1906,1.163,521247.42,615530.31,615530.31

[codecov-commenter]

Codecov Report

❌ Patch coverage is 60.78431% with 40 lines in your changes missing coverage. Please review.
✅ Project coverage is 92.27%. Comparing base (6a29c59) to head (2036f15).
⚠️ Report is 18 commits behind head on main.

Files with missing lines	Patch %	Lines
...exera/amber/operator/hashJoin/HashJoinOpDesc.scala	53.65%	9 Missing and 10 partials ⚠️
...ber/operator/intervalJoin/IntervalJoinOpDesc.scala	66.66%	2 Missing and 8 partials ⚠️
...ache/texera/amber/operator/StandaloneHelpers.scala	0.00%	2 Missing ⚠️
...a/amber/operator/difference/DifferenceOpDesc.scala	33.33%	2 Missing ⚠️
...era/amber/operator/intersect/IntersectOpDesc.scala	33.33%	2 Missing ⚠️
...ymmetricDifference/SymmetricDifferenceOpDesc.scala	33.33%	2 Missing ⚠️
.../texera/amber/core/executor/OperatorExecutor.scala	66.66%	1 Missing ⚠️
...ator/cartesianProduct/CartesianProductOpDesc.scala	50.00%	1 Missing ⚠️
.../amber/operator/hashJoin/HashJoinProbeOpExec.scala	92.85%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #8506      +/-   ##
============================================
- Coverage     92.61%   92.27%   -0.35%     
+ Complexity     5059     5055       -4     
============================================
  Files          1258     1258              
  Lines         53780    53984     +204     
  Branches       6707     6742      +35     
============================================
+ Hits          49811    49816       +5     
- Misses         2313     2495     +182     
- Partials       1656     1673      +17     

Flag	Coverage Δ		*Carryforward flag
access-control-service	77.38% <ø> (ø)
agent-service	99.16% <ø> (ø)		Carriedforward from f4004e8
amber	87.18% <60.78%> (-0.93%)	⬇️
computing-unit-managing-service	60.48% <ø> (ø)
config-service	87.37% <ø> (ø)
file-service	82.00% <ø> (ø)
frontend	96.59% <ø> (ø)		Carriedforward from f4004e8
notebook-migration-service	83.73% <ø> (ø)
pyamber	98.58% <ø> (ø)		Carriedforward from f4004e8
workflow-compiling-service	72.43% <ø> (ø)

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

[carloea2]

The Java random copy does not match Java integer overflow in next_int. Java rejects some values after the addition overflows, but this Python code always accepts them. This can select different rows for Split and Reservoir Sampling. Please copy the overflow check and add a seed that reaches the rejection case.

[carloea2]

There are more behavior gaps. If always defaults to the true branch because the exported script never receives the state. Sampling only matches one worker, while native runs with one random generator per worker. Aggregate CONCAT also writes Python boolean names instead of lowercase Java names. Reservoir size zero fails natively but silently returns empty in standalone. These cases need matching behavior or a clear unsupported error.

[carloea2]

More operators still differ. Sleep removes the delay. Timestamp sum fails and timestamp average returns the wrong type. Outer joins can turn integers into floats and fail chained name collisions. Dummy loses extra port routing. Sort order differs for NaN and some Unicode strings. Please add executed parity tests for these cases.

[carloea2]

Integer sum can return different data in the two paths. With 2147483647 and 1, native integer aggregation wraps to negative 2147483648, while pandas returns positive 2147483648 as a long. Please match the native result and add a boundary test.

[carloea2]

Set operations merge a null double and a real NaN. For example, if the left input has null and NaN and the right input has null, native Difference keeps the NaN row, but the exported code returns no rows. Please preserve this distinction and add a set operation test with both values.

[carloea2]

The same null and NaN problem also changes Join and Aggregate results. Left join keys null and NaN against a right null produce one native match but two exported matches. Grouping null and NaN with values 1 and 2 produces two native groups but one exported group totaling 3. Please cover keyed operators too.

[kz930]

Fixed in this PR: next_int's overflow rejection, a reservoir bound of zero, the INTEGER sum wrap, CONCAT's lower-case boolean, the join's rename and its integer column, and the NaN sort. If now has a spec that runs both routes. The null and NaN cases were the Arrow read rather than the operators, so #8341 reads the nullable dtypes instead, and Difference, the hash join and Aggregate each carry your case as a run test.

Four I would push back on. Dummy never overrides getPhysicalOp, so it cannot execute at all and has no routing to lose. Sleep's delay never reaches the output. Sampling is seeded with the worker count by the engine, and one process can only state one seed. A sorted string parts only where a character above U+FFFF meets one in U+E000..U+FFFF, since the engine reads UTF-16 code units and pandas reads code points.

Your two newest both reproduce here. Fixing them now.

[carloea2]

The new timestamp aggregation helpers assume the integer values are nanoseconds. With a microsecond column containing 2024-01-01, Sum returns 1970-01-20 17:21:07.200 instead. The nanosecond sum also overflows before conversion: six copies of 2024-01-01 return a date in 1709 instead of 2293. I reproduced both with pandas 2.2.3. Please convert each value to epoch milliseconds before aggregating and test both resolutions.

[kz930]

@carloea2 I split this into #8506 (joins and sets), #8709 (Aggregate and the sort family) and #8710 (the samplers and control flow). Since your last pass the outer join fills its key as the engine does, and the rest moved with its operators.

[carloea2]

The 53 existing focused tests and scalafmtCheckAll pass. Additional native versus export probes found two gaps: column placement with an empty join build side, and set equality for stored NaN and signed zero. Ordinary finite values, nulls and NaN versus null pass.

Commands: sbt -batch "WorkflowOperator/testOnly *CartesianProductOpDescSpec *HashJoinOpDescSpec *HashJoinProbeOpExecSpec *IntervalJoinOpDescSpec *DifferenceOpDescSpec *IntersectOpDescSpec *SymmetricDifferenceOpDescSpec -- -oDF" "scalafmtCheckAll", then sbt -batch "WorkflowOperator/testOnly *HashJoinProbeOpExecSpec *SetExportReviewSpec -- -oDF": 26 passed, eight regressions failed. Only diagnostic tests were added locally; production code was unchanged.

[carloea2]

The set equality fixes pass the original NaN and signed-zero reproductions. The empty full-outer-join finding remains: with output columns [k, x, x#@1], the native result is [2, right, null] instead of [2, null, right]. The differently named key case still puts the right key into a left payload column. No duplicate inline comment added; these are the existing join finding.

The expanded focused run gives 83 passed and 2 failed, with formatting passing. Please address that remaining case before merging.

sbt -batch 'WorkflowOperator/testOnly *HashJoin*Spec *JoinUtilsSpec *DifferenceOpDescSpec *IntersectOpDescSpec *SymmetricDifferenceOpDescSpec *Union*Spec *SetExportReviewSpec -- -oDF'

[kz930]

@carloea2 Both findings are fixed, the set operations in 77c64a3 and the empty build side in ce2b893. The second adds OperatorExecutor.inputSchemas, an engine change described in the PR body. Could you take another look?

[kz930]

@carloea2 One more case from your keyed-operator finding: two NaN keys were still treated as equal, while the engine never matches them. Hash Join now lets a NaN key match nothing (fa07950), and Aggregate in #8709 gives each NaN key its own group (14d7de2).

[carloea2]

The empty-build join fix passes, including colliding payload names and differently named keys. All 101 focused operator tests and 13 EndChannelHandler tests pass, with formatting clean. The original NaN and signed-zero set reproductions also pass. The remaining CI failure is LoopIntegrationSpec, covered by #8938. Code looks good. Leaving the draft status unchanged.

[kz930]

@carloea2 One more commit since your approval: Interval Join now compares timestamps by the millisecond, as the engine does (c59c02f).

[kz930]

@carloea2 One more: Interval Join now adds an INTEGER constant as an Int sum, cut to 32 bits and wrapping as the engine does (f4004e8).

[kz930]

@carloea2 One more: Interval Join now adds a time interval on the machine's wall clock and reads it back in the machine's zone, as the engine does, so a day across a daylight-saving change is 23 or 25 hours (2036f15). It calls the zone helpers from #8597, so its Python tests pass once #8597 lands.