Repository navigation
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 | 647 | 0.395 | 14,905/22,646/22,646 us | 🔴 +8.5% / 🔴 +101.2% |
| 🔴 | bs=100 sw=10 sl=64 | 1,649 | 1.007 | 59,500/71,472/71,472 us | 🟢 -7.4% / 🟢 -20.5% |
| 🔴 | bs=1000 sw=10 sl=64 | 2,021 | 1.233 | 486,583/582,941/582,941 us | 🔴 +9.1% / 🟢 +28.8% |
Baseline details
Latest main 6a29c59 from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 647 tuples/sec | 699 tuples/sec | 1,119 tuples/sec | -7.4% | -42.2% |
| bs=10 sw=10 sl=64 | MB/s | 0.395 MB/s | 0.427 MB/s | 0.683 MB/s | -7.5% | -42.2% |
| bs=10 sw=10 sl=64 | p50 | 14,905 us | 13,735 us | 8,992 us | +8.5% | +65.7% |
| bs=10 sw=10 sl=64 | p95 | 22,646 us | 24,657 us | 11,255 us | -8.2% | +101.2% |
| bs=10 sw=10 sl=64 | p99 | 22,646 us | 24,657 us | 13,808 us | -8.2% | +64.0% |
| bs=100 sw=10 sl=64 | throughput | 1,649 tuples/sec | 1,731 tuples/sec | 1,518 tuples/sec | -4.7% | +8.6% |
| bs=100 sw=10 sl=64 | MB/s | 1.007 MB/s | 1.057 MB/s | 0.927 MB/s | -4.7% | +8.7% |
| bs=100 sw=10 sl=64 | p50 | 59,500 us | 55,941 us | 68,504 us | +6.4% | -13.1% |
| bs=100 sw=10 sl=64 | p95 | 71,472 us | 77,182 us | 74,866 us | -7.4% | -4.5% |
| bs=100 sw=10 sl=64 | p99 | 71,472 us | 77,182 us | 89,939 us | -7.4% | -20.5% |
| bs=1000 sw=10 sl=64 | throughput | 2,021 tuples/sec | 2,009 tuples/sec | 1,569 tuples/sec | +0.6% | +28.8% |
| bs=1000 sw=10 sl=64 | MB/s | 1.233 MB/s | 1.226 MB/s | 0.957 MB/s | +0.6% | +28.8% |
| bs=1000 sw=10 sl=64 | p50 | 486,583 us | 491,355 us | 669,501 us | -1.0% | -27.3% |
| bs=1000 sw=10 sl=64 | p95 | 582,941 us | 534,074 us | 706,487 us | +9.1% | -17.5% |
| bs=1000 sw=10 sl=64 | p99 | 582,941 us | 534,074 us | 737,560 us | +9.1% | -21.0% |
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,309.14,200,128000,647,0.395,14904.52,22645.69,22645.69
1,100,10,64,20,1212.66,2000,1280000,1649,1.007,59499.53,71472.29,71472.29
2,1000,10,64,20,9898.28,20000,12800000,2021,1.233,486583.11,582941.21,582941.21
❌ 13 Tests Failed:
View the top 3 failed test(s) by shortest run time
To view more test analytics, go to the Test Analytics Dashboard |
…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
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
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
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
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
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
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
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.
The engine's HashSet[Tuple] 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. pandas' duplicate checks count both pairs as equal, so Intersect, Difference and Symmetric Difference now select rows by a key that follows the engine. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
carloea2
left a comment
There was a problem hiding this comment.
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'
…e declared build schema A probe worker read the left column names off its first build row, so one that received no build rows put an unmatched right row's columns into the left ones. The runtime now records each input port's declared schema on the executor before the port finishes, and the probe reads the left names from it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
OperatorExecutor: upstream now keeps executor state beside the executors rather than in trait fields, which a Java class implementing the trait would have to declare. inputSchemas follows suit and becomes a final def over a weak-keyed map in the companion object. HashJoinOpDesc: keep both imports, NoLoopVariable from upstream and AttributeType for the export. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
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>
…the engine does The probe looks each key up in a HashMap under Scala's ==, where a NaN finds nothing, while merge paired NaN with NaN. Merge on codes that follow the engine's equality instead, each NaN taking one of its own. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
carloea2
left a comment
There was a problem hiding this comment.
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.
… the engine does IntervalJoinOpExec compares a timestamp by getTime, so two moments in the same millisecond are one point to it, and a left key just past the upper bound but inside its millisecond still joins. The export compared the full resolution and dropped those rows. Floor the two keys and the bound to the millisecond for the comparison, adding the bound before it is cut. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…e, as the engine does An INTEGER bound is an Int sum: the engine cuts the constant to 32 bits and the sum wraps past Int.MaxValue, where pandas held the column as int64 and did neither. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…l clock, as the engine does The executor adds the constant to toLocalDateTime and reads the sum back with Timestamp.valueOf, so a day across a daylight-saving change is 23 or 25 hours and a month counts from the local date. The export added it to the UTC wall clock. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
carloea2
left a comment
There was a problem hiding this comment.
The 76 operator tests, 14 EndChannelHandlerSpec tests and formatting pass with #8597 and #9003 supplied. A cross-operator timestamp regression still fails: two controls pass and both clock-change cases fail.
sbt -batch 'WorkflowOperator/testOnly *ReviewIntervalWallClockSpec -- -oDF'
This compares the native IntervalJoinOpExec with emitted code fed through the shared source timestamp reader, not a full connected workflow.
…ipt is handed A source already hands the script a timestamp as the machine's wall clock, so labelling it UTC and converting it again moved the bound by the local offset. The interval is now added to that wall clock as it stands, and the keys and the bound are read back into instants as Timestamp.valueOf reads them. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…the exported script _texera_set_row_keys, _texera_set_select and _texera_join_codes explained themselves in Python comments, which every exported script with a set operator or a Hash Join printed. The notes now sit in Scala, one line per function. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…exported script Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
carloea2
left a comment
There was a problem hiding this comment.
The daylight-saving finding is fixed. The 64 focused join and set tests pass, including the original four source-reader wall-clock probes. All 14 EndChannelHandlerSpec tests and scalafmtCheckAll pass too. Tested with the shared source and timestamp prerequisites supplied. CI remains red and must clear before merge.
…ared helper The shared helper also reads a wall clock in the hours around 1900-01-01 at the offset the engine does, where the old one was a whole offset off. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…odes Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… comments Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…ed key codes Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
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
xbeside a rightx#@1was declared as a secondx#@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 inJoinUtils.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 inEndChannelHandler), 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.
HashJoinProbeOpExecSpecpins the platform fixes.Each set operation's spec runs its native executor and the export over NaN, signed-zero, finite and null rows.
HashJoinProbeOpExecSpeccovers an empty build side, andEndChannelHandlerSpecchecks 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