Conversation
Seventeen operators implement the trait: the three joins, the three set operations, Aggregate, the sort family, Split, the two samplers, and If, Dummy and Sleep. What they have in common is that they rearrange rows rather than read what is inside one. A sampler decides per row whether to keep it, so which rows survive is fixed by the exact sequence java.util.Random produces. Seeding Python's own generator selects a different set, and the script would then report a different sample than the workflow it came from, so SamplingHelpers transcribes the generator rather than approximating it. Split out of apache#8340 on review, which had grown past what one reading can hold. That change keeps the operators that work on text, and the cast transcription they share. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Automated Reviewer SuggestionsBased on the
|
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 462 | 0.282 | 20,781/30,902/30,902 us | 🔴 +13.0% / 🔴 +98.5% |
| 🔴 | bs=100 sw=10 sl=64 | 971 | 0.592 | 100,600/139,580/139,580 us | 🔴 +11.7% / 🔴 +28.9% |
| 🟢 | bs=1000 sw=10 sl=64 | 1,167 | 0.713 | 855,529/902,049/902,049 us | 🟢 -6.6% / 🟢 -15.3% |
Baseline details
Latest main 5264df2 from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 462 tuples/sec | 481 tuples/sec | 775.4 tuples/sec | -4.0% | -40.4% |
| bs=10 sw=10 sl=64 | MB/s | 0.282 MB/s | 0.294 MB/s | 0.473 MB/s | -4.1% | -40.4% |
| bs=10 sw=10 sl=64 | p50 | 20,781 us | 18,392 us | 12,716 us | +13.0% | +63.4% |
| bs=10 sw=10 sl=64 | p95 | 30,902 us | 33,953 us | 15,572 us | -9.0% | +98.5% |
| bs=10 sw=10 sl=64 | p99 | 30,902 us | 33,953 us | 19,527 us | -9.0% | +58.3% |
| bs=100 sw=10 sl=64 | throughput | 971 tuples/sec | 997 tuples/sec | 992.44 tuples/sec | -2.6% | -2.2% |
| bs=100 sw=10 sl=64 | MB/s | 0.592 MB/s | 0.608 MB/s | 0.606 MB/s | -2.6% | -2.3% |
| bs=100 sw=10 sl=64 | p50 | 100,600 us | 97,468 us | 102,140 us | +3.2% | -1.5% |
| bs=100 sw=10 sl=64 | p95 | 139,580 us | 124,957 us | 108,282 us | +11.7% | +28.9% |
| bs=100 sw=10 sl=64 | p99 | 139,580 us | 124,957 us | 118,936 us | +11.7% | +17.4% |
| bs=1000 sw=10 sl=64 | throughput | 1,167 tuples/sec | 1,165 tuples/sec | 1,023 tuples/sec | +0.2% | +14.1% |
| bs=1000 sw=10 sl=64 | MB/s | 0.713 MB/s | 0.711 MB/s | 0.624 MB/s | +0.3% | +14.2% |
| bs=1000 sw=10 sl=64 | p50 | 855,529 us | 850,558 us | 998,263 us | +0.6% | -14.3% |
| bs=1000 sw=10 sl=64 | p95 | 902,049 us | 965,575 us | 1,036,365 us | -6.6% | -13.0% |
| bs=1000 sw=10 sl=64 | p99 | 902,049 us | 965,575 us | 1,064,941 us | -6.6% | -15.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,432.69,200,128000,462,0.282,20780.59,30902.26,30902.26
1,100,10,64,20,2060.74,2000,1280000,971,0.592,100600.34,139579.75,139579.75
2,1000,10,64,20,17132.15,20000,12800000,1167,0.713,855528.59,902049.04,902049.04
Codecov Report❌ Patch coverage is ❌ Your patch status has failed because the patch coverage (47.80%) is below the target coverage (60.00%). You can increase the patch coverage or adjust the target coverage. Additional details and impacted files@@ Coverage Diff @@
## main #8506 +/- ##
============================================
- Coverage 92.78% 92.59% -0.20%
- Complexity 4899 4943 +44
============================================
Files 1236 1238 +2
Lines 52130 52420 +290
Branches 6407 6464 +57
============================================
+ Hits 48369 48537 +168
- Misses 2185 2273 +88
- Partials 1576 1610 +34
*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:
|
…he#8340) ### What changes were proposed in this PR? Six operators that read or rewrite a text column implement `StandaloneCodeGenerator`: Type Casting, Keyword Search, Substring Search, Unnest String, Regex and Dictionary Matcher. The sixteen that rearrange rows rather than read inside one are in apache#8506. Each is written against what its executor does rather than against what its name suggests, and the verification that came with the export runs both and compares them. Some of what that turned up is visible in the code: a cast reads the column's declared type rather than the value in front of it, because a hole widens an integer column to float and "6" would otherwise come back "6.0"; the 32-bit narrowing wraps for an integral source and saturates for a double, as `Long.toInt` and `Double.toInt` do; Dictionary Matcher's substring branch asks whether the cell sits inside a dictionary entry, which is the direction the executor tests and not the one the name suggests. `StandaloneHelpers` holds what these six share: a transcription of `AttributeTypeUtils`, emitted once near the top of a script rather than inlined per operator. A cast goes through `parseField(force = true)`, whose numeric branch is `java.text.NumberFormat`, so the engine reads "12abc" as 12 and "false" as false where Python's own conversions answer differently. Four of these operators change what they do, not only how they export. Regex, Substring Search and Unnest String answered a null cell by raising, which a generated script has no way to reproduce and no reason to: nothing in a column matches nothing, and unnests to no rows. They answer it that way now, and their specs say so. Keyword Search states which keywords its query parser will take, which a user had no way to know from the field alone. ### Any related issues, documentation, discussions? Part of apache#8325, 10 of 27; that issue lists the set in order. apache#8327 has since merged, so the trait is in place and this branch compiles on its own. The rows these operators add to the verification runner follow with the harness rather than as whole new files here. Closes apache#8074, closes apache#7548. Closes apache#8414, the task this change is the whole of. ### How was this PR tested? Each operator asserts the block it emits in its own spec. Type Casting's spec goes further: it runs the block it emits through a real Python and compares every answer against `AttributeTypeUtils`, so the two findings the review raised are held there, and that part does run on this diff's CI. Once the verification lands each operator is also run through the engine and through its generated script, on every configuration its schema offers, and the two answers compared; 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) --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-authored-by: Xinyuan Lin <xinyual3@uci.edu>
carloea2
left a comment
carloea2
left a comment
There was a problem hiding this comment.
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
left a comment
carloea2
left a comment
There was a problem hiding this comment.
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
left a comment
carloea2
left a comment
There was a problem hiding this comment.
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
left a comment
carloea2
left a comment
There was a problem hiding this comment.
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
left a comment
carloea2
left a comment
There was a problem hiding this comment.
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
left a comment
carloea2
left a comment
There was a problem hiding this comment.
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.
…ational-and-sampling
The engine picks If's route from a State message on the Condition port, and the verification harness writes rows per port with no State channel, so a fixture can only ever reach the default route, True. The exported block reads the same decision from a global instead, which the spec can set, so both routes are run there: the rows leave by one output and the other comes back empty. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The engine compares with String.compareTo, which reads UTF-16 code units, while pandas compares by code point. They agree across the basic plane and differ only where a character above U+FFFF meets one in U+E000..U+FFFF, because the surrogate pair encoding the first holds units below that range. U+1F600 against U+FFFD is the shape of it: pandas reads the first as greater, the engine as smaller. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The transcribed nextInt carried its rejection sum as a Python integer, where Java lets an int overflow: a draw within bound of 2**31 wraps negative there and is taken again. Carrying it keeps the value Java throws away and shifts every later draw, which seed 12345 at bound 1000 reaches on draw 2126624. It also never refused a bound of zero, which Reservoir Sampling hands it when the reservoir holds nothing. The engine ends the run there; the script answered with an empty table instead. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ers them Four places where pandas is asked the same question and answers differently. SUM keeps the column's declared type, so an INTEGER sum is Java int arithmetic and wraps at 2**31 where pandas widens, and a TIMESTAMP sum is a timestamp built from the added epoch milliseconds where pandas refuses to reduce datetime64 at all. AVERAGE is declared DOUBLE whatever it reads, so a timestamp average is the mean of those milliseconds and not a timestamp. CONCAT folds with Java's toString, which spells a boolean in lower case. The first three need the declared type rather than the dtype pandas inferred: a holed INTEGER column arrives as a float, and an INTEGER and a LONG arrive alike. The schema-free form keeps pandas' own answers rather than guessing. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ger integral The engine appends "#@1" to a colliding right column until the name is free, so one that meets an already-suffixed column lands on "#@1#@1". The export asked pandas for `suffixes` instead, which appends once and then refuses the duplicate it just made, ending the run. Compute the rename the way HashJoinProbeOpExec computes it, the way the Cartesian product export already did, and hand the merge columns that no longer collide. An outer join also leaves holes, and pandas pays for a hole in an integer column by reading the whole column as float. The engine writes a null and the column stays INTEGER, so the declared integer columns are read back as the nullable integer that says the same thing. Only the declared type can say which those are, and without a schema the widening stands. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The reasoning behind each fix was written twice, once in the commit that made it and again beside the line. Keep the half a reader needs to follow the code and drop the worked examples and the derivations. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The engine puts a null last whichever way a key points, and compares a NaN above every number, so a NaN goes last ascending and first descending. The export asked for na_position="last", which is right for the null and wrong for the NaN on the way down. Sort each key in three tiers instead: whether the value is missing, always ascending; whether it is a NaN, following the key; then the value. A column read into a numpy dtype holds a null and a NaN in the same slot, and there both land in the missing tier, which is where they were before. Telling them apart needs the nullable dtypes the Arrow source now reads into. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…d paths The engine reads a missing value and a stored NaN as two different values, so a difference keeps the NaN row a null row does not cancel, a join matches the missing key and not the NaN one, and a group keyed on each stays two groups. Each is run on the nullable dtype an Arrow file is read into, which is the only place the two are distinct. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
|
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 Your two newest both reproduce here. Fixing them now. |
carloea2
left a comment
carloea2
left a comment
There was a problem hiding this comment.
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.
carloea2
left a comment
carloea2
left a comment
There was a problem hiding this comment.
The outer join type fix mixes up columns with the same name on different inputs. Give the left input an integer x with value 10 and the right input a double x with value 1.5, then use a full outer join with different keys. The right x is renamed, but the fix treats it as the left integer and tries to cast 1.5 to Int64. I reproduced the crash with pandas 2.2.3. Please track the type of each output column through the rename.
…he column that declared an integer The integers behind a timestamp column count nanoseconds or microseconds depending on the resolution it was read at, so reading them and dividing by a million answers a microsecond column with a 1970 date. A nanosecond total also leaves the range of a 64-bit integer after six modern dates. Cast to milliseconds first, sum as Python integers, and wrap the total the way the engine's Java longs wrap. The outer-join integer cast pooled the column names of both inputs, so a double on the right that collided with an integer on the left was cast back to an integer and raised. Name the right columns as the merge will name them, and cast those names alone. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
|
Both fixed in c18b1e5: the timestamp helpers cast to milliseconds before reading the integers out and carry the sum in Python integers, so your six copies of 2024-01-01 give 2293-12-31 at either resolution, and the outer-join cast now names the right columns the way the merge names them, so your double |
The everyday join, the same column name on both inputs, carried a second key column the operator never promises: `k`, `name`, `k#@1`, `score` where the engine emits `k`, `name`, `score`. Moving the right frame's columns aside before the merge, so a chained name collision resolves, also moves the key aside when the two sides name it alike. The drop was still decided by comparing the operator's own two names, which read that case as one shared column and kept the copy. It now asks the name the rename settled on, which is the only place that knows, and covers the three shapes alike: the same name, a free name, and one that met a left column. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
carloea2
left a comment
carloea2
left a comment
There was a problem hiding this comment.
The outer join still rounds large integer values before restoring their type. A Parquet input with key 1 and id 9007199254740993 is read exactly. Full outer join it with an unmatched key 2, and the output id becomes 9007199254740992. The merge creates a float column before the Int64 cast. Please preserve nullable integer types before merging. I reproduced this with pandas 2.2.3.
carloea2
left a comment
carloea2
left a comment
There was a problem hiding this comment.
Joining two inputs on a key named k produces an extra key column in the export. Left columns k and x joined with right columns k and y should return k, x, y, as native does. The export renames the right key to k#@1 and keeps it. Please always remove the probe key after the merge and correct the test that expects the extra column.
An unmatched row leaves a hole, and a hole costs a pandas integer column its type: int64 becomes float64. float64 cannot hold an integer past 2^53, so the cast that put INTEGER back was reading a value already rounded. A column holding 9007199254740993 came out of a full outer join as 9007199254740992. Each input's declared integer columns are now cast to the nullable integer dtype on that input's own frame, before the merge digs the hole. The cast runs before the rename, so a right column is typed by the right schema rather than by the left column it collided with, and the merged frame is never cast back. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
|
That case is fixed in b8e5dbf, pushed shortly before this review: joining |
carloea2
left a comment
carloea2
left a comment
There was a problem hiding this comment.
Three new findings reproduced from the generated Python blocks at 7ccb215. The timestamp case was checked against Java 17 Timestamp arithmetic in UTC and America/Mexico_City. Passing controls covered ordinary column names and UTC. I did not run a frontend workflow or the full Scala suite.
| |for _texera_col, _texera_a in zip($cols, $ascending): | ||
| | _texera_null = "_texera_null_" + _texera_col | ||
| | _texera_nan = "_texera_nan_" + _texera_col | ||
| | _texera_sorted[_texera_null] = _texera_sorted[_texera_col].isna() |
There was a problem hiding this comment.
Sorting by x deletes an existing _texera_null_x or _texera_nan_x column. I tested x=[2,1] with payload values in each name: the helper overwrites the payload, then the final drop removes it. The ordinary-name control passes. Please allocate collision-free helpers or sort by separate temporary keys so all input columns survive.
| |_rename = {} | ||
| |for _col in _right_cols: | ||
| | _new = _col | ||
| | _others = _right_set - {_col} |
There was a problem hiding this comment.
The probe key affects payload renaming even though it is dropped later. Joining left {k:1, x:'left'} with right {'x#@1':1, x:'right'} on k and x#@1 emits x#@1#@1. Native schema propagation and JoinUtils exclude the probe key first and name the payload x#@1, so a downstream projection fails. Please exclude the discarded key from payload collision checks and add this chained-join case.
| | # resolution the column carries, and reading the integers out of a | ||
| | # microsecond column asks for a different number than a nanosecond | ||
| | # one, so cast to milliseconds before reading them. | ||
| | return series.dropna().astype("datetime64[ms]").astype("int64") |
There was a problem hiding this comment.
This treats a naive wall-clock timestamp as UTC. With the JVM in America/Mexico_City, summing midnight on 2024-01-01 and 2024-01-02 using the engine's Timestamp arithmetic gives 2078-01-01 06:00:00; the generated helper returns midnight. The average also differs by 21600000 milliseconds. UTC passes. Please match the engine's timezone when converting to epoch milliseconds and back, and test a non-UTC zone.
The sort built its null and NaN tiers as columns on the input frame, named "_texera_null_" and "_texera_nan_" plus the key. An input already carrying one of those names lost it twice over: the tier overwrote the payload, and the drop that cleared the tiers afterwards took the column with it. Sorting a frame of "x", "_texera_null_x" and "_texera_nan_x" by "x" returned "x" alone. The tiers now go in a frame of their own, named by position, and the input is reordered by the index that frame sorts to. Nothing is added to the input and nothing is dropped from it, so no name it carries can collide. The ordering is unchanged: a null still goes last whichever way the key points, and a NaN still follows the key in a dtype that tells the two apart. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The probe key is dropped after the merge, but it was still counted when the payload columns looked for a free name, so it pushed one of them a suffix further along than the engine does. Joining a left "k", "x" with a right "x#@1", "x" on "x#@1" emitted "x#@1#@1" where JoinUtils and the operator's own schema propagation both emit "x#@1", and a downstream projection asking for the promised name failed. A right input that has already been through a join is where those names come from, so a chain of two joins is enough to reach it. JoinUtils filters the probe key out before it renames anything, and the rename now does the same. The key still rides along for the merge to join on, and is given a name nothing else claims before it is dropped again. Checked against the engine's own loop over every left and right column set that can be built from "k", "x", "x#@1", "x#@1#@1" and "pk". Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
A timestamp column holds a wall clock, and the engine reads one through `Timestamp.getTime`, which answers for the instant that wall clock names in the JVM's default zone. The generated helper read the column as if the zone were UTC, so every epoch millisecond it added was a whole offset out. With the JVM in America/Mexico_City, summing midnight on 2024-01-01 and 2024-01-02 gives 2078-01-01 06:00:00 where the script said midnight, and the average differed by 21600000. UTC agreed, which is why this went unseen. The column is now localized before its integers are read and the total is carried back through the same zone, the way `new Timestamp(long)` renders one. The two daylight-saving flags are java.time's own readings: the later of a repeated hour, and a skipped one shifted past the gap. Both were checked against Java 17 in America/New_York. The two timestamp tests that were already here read as UTC without saying so, so they now name the zone to the child process, and a third runs the arithmetic in a zone that is not UTC. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
3 participants
What changes were proposed in this PR?
Sixteen operators implement
StandaloneCodeGenerator: the three joins, the three set operations, Aggregate, the sort family, Split, the two samplers, and If, Dummy and Sleep. What they have in common is that they rearrange rows rather than read what is inside one.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 emit it once per script.#8340 keeps the operators that work on text, and the cast transcription they share.
Review then found eight places where the block answered a question the engine answers differently. The transcribed generator 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. Aggregate summed an INTEGER column without the wrap Java ints have, refused a TIMESTAMP column outright, answered a timestamp average with a timestamp where AVERAGE is declared DOUBLE, and spelled a boolean in Python's capitals rather than Java's lower case. 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. The sort put a NaN last on the way down, where the engine puts it first.
Any related issues, documentation, discussions?
Part of #8325, 11 of 27; that issue lists the set in order.
Sorting a NaN apart from a null, and the set operations and joins keyed on one, need the nullable dtypes #8341 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 joins carry the cases a review found: a hash join whose left frame already holds the probe key's name drops the renamed right key rather than the left payload, and an interval join reads its keys off the merged frame rather than writing over an input column that happens to share a temporary's name.
The cases the review named are run rather than asserted as text, since the arithmetic is the point: the sampler's draws against what
java.util.Randomanswers, the four aggregations against the engine's own results, both join repairs against the columns and dtypes that come back, and the sort against where a null and a NaN land in each direction.Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)
🤖 Generated with Claude Code