Repository navigation
feat(workflow-operator): export the samplers, Split and the control-flow operators as Python - #8710
Conversation
…low operators as Python Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Automated Reviewer SuggestionsBased on the
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #8710 +/- ##
============================================
- Coverage 92.78% 92.72% -0.06%
- Complexity 4899 4953 +54
============================================
Files 1236 1238 +2
Lines 52130 52462 +332
Branches 6407 6457 +50
============================================
+ Hits 48369 48647 +278
- Misses 2185 2206 +21
- Partials 1576 1609 +33
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 346 | 0.211 | 27,374/39,939/39,939 us | 🔴 +8.6% / 🔴 +134.3% |
| 🔴 | bs=100 sw=10 sl=64 | 750 | 0.458 | 127,376/184,085/184,085 us | 🔴 +14.2% / 🔴 +59.9% |
| ⚪ | bs=1000 sw=10 sl=64 | 889 | 0.542 | 1,111,322/1,278,038/1,278,038 us | ⚪ within ±5% / 🔴 +15.4% |
Baseline details
Latest main ba2667a from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 346 tuples/sec | 372 tuples/sec | 721.84 tuples/sec | -7.0% | -52.1% |
| bs=10 sw=10 sl=64 | MB/s | 0.211 MB/s | 0.227 MB/s | 0.441 MB/s | -7.0% | -52.1% |
| bs=10 sw=10 sl=64 | p50 | 27,374 us | 26,750 us | 13,416 us | +2.3% | +104.0% |
| bs=10 sw=10 sl=64 | p95 | 39,939 us | 36,774 us | 17,046 us | +8.6% | +134.3% |
| bs=10 sw=10 sl=64 | p99 | 39,939 us | 36,774 us | 20,349 us | +8.6% | +96.3% |
| bs=100 sw=10 sl=64 | throughput | 750 tuples/sec | 807 tuples/sec | 914.39 tuples/sec | -7.1% | -18.0% |
| bs=100 sw=10 sl=64 | MB/s | 0.458 MB/s | 0.493 MB/s | 0.558 MB/s | -7.1% | -17.9% |
| bs=100 sw=10 sl=64 | p50 | 127,376 us | 121,115 us | 108,507 us | +5.2% | +17.4% |
| bs=100 sw=10 sl=64 | p95 | 184,085 us | 161,224 us | 115,131 us | +14.2% | +59.9% |
| bs=100 sw=10 sl=64 | p99 | 184,085 us | 161,224 us | 127,181 us | +14.2% | +44.7% |
| bs=1000 sw=10 sl=64 | throughput | 889 tuples/sec | 893 tuples/sec | 937.6 tuples/sec | -0.4% | -5.2% |
| bs=1000 sw=10 sl=64 | MB/s | 0.542 MB/s | 0.545 MB/s | 0.572 MB/s | -0.6% | -5.3% |
| bs=1000 sw=10 sl=64 | p50 | 1,111,322 us | 1,113,787 us | 1,065,461 us | -0.2% | +4.3% |
| bs=1000 sw=10 sl=64 | p95 | 1,278,038 us | 1,244,549 us | 1,107,736 us | +2.7% | +15.4% |
| bs=1000 sw=10 sl=64 | p99 | 1,278,038 us | 1,244,549 us | 1,133,556 us | +2.7% | +12.7% |
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,578.04,200,128000,346,0.211,27374.06,39939.07,39939.07
1,100,10,64,20,2665.19,2000,1280000,750,0.458,127376.44,184084.93,184084.93
2,1000,10,64,20,22509.65,20000,12800000,889,0.542,1111322.01,1278037.77,1278037.77
carloea2
left a comment
There was a problem hiding this comment.
Checked the seeded Split output and the Java random sequence locally. Both match Java 17. Looks good.
…pache#8506) ### 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. apache#8340 keeps the operators that work on text. Aggregate and the sort family follow in apache#8709, and the samplers and control flow in apache#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 apache#8766 does for Python operators. ### Any related issues, documentation, discussions? Part of apache#8325, step 11a; apache#8709 and apache#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 apache#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 apache#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 apache#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](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
What changes were proposed in this PR?
The two samplers, Split, and If, Dummy and Sleep implement
StandaloneCodeGenerator.The samplers need more than a pandas expression. A sampler decides per row whether to keep it, so which rows survive is fixed by the exact sequence
java.util.Randomproduces; seeding Python's own generator with the same number selects a different set, and the script would then report a different sample than the workflow it came from.SamplingHelperstranscribes the generator, and the three operators that draw from one, the two samplers and Split, emit it once per script.If cannot receive the condition the workflow sets upstream, since that only exists at run time. The script takes the True branch as before, and now says so on stderr. The message names the variable to set before the step to take the False branch.
Review then found two places where the transcribed generator answered a question the engine answers differently. It carried its rejection sum as a Python integer where Java lets an int overflow, and never refused the bound of zero a reservoir of nothing hands it.
The edge values the shared table now carries found one more. The reservoir sampler rebuilt its output from the rows' values, so pandas inferred each column again and a timestamp past 2262 came back as an object column. It now takes the rows from the input by position.
Any related issues, documentation, discussions?
Part of #8325, step 11c; that issue lists the set in order.
Closes #8708, the task this change is the whole of.
How was this PR tested?
Each operator asserts the block it emits in its own spec. What the verify tables cannot reach stays here as executed tests: the samplers' draws against
java.util.Random, a reservoir of nothing, and the warning If prints when nothing sets its condition.Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)
🤖 Generated with Claude Code