Repository navigation
feat(workflow-operator): export the JSON Lines and Arrow sources as Python - #8704
Conversation
… zoneless microsecond timestamp Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Automated Reviewer SuggestionsBased on the
|
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #8704 +/- ##
============================================
- Coverage 92.17% 92.17% -0.01%
+ Complexity 5043 5041 -2
============================================
Files 1287 1287
Lines 55799 55813 +14
Branches 7144 7146 +2
============================================
+ Hits 51434 51444 +10
- Misses 2598 2604 +6
+ Partials 1767 1765 -2
*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 | 360 | 0.22 | 25,891/39,543/39,543 us | 🔴 +19.2% / 🔴 +178.5% |
| 🔴 | bs=100 sw=10 sl=64 | 790 | 0.482 | 123,256/179,143/179,143 us | 🔴 +22.9% / 🔴 +85.7% |
| ⚪ | bs=1000 sw=10 sl=64 | 944 | 0.576 | 1,052,508/1,135,462/1,135,462 us | ⚪ within ±5% / 🔴 +21.7% |
Baseline details
Latest main 05c19a4 from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 360 tuples/sec | 416 tuples/sec | 880.32 tuples/sec | -13.5% | -59.1% |
| bs=10 sw=10 sl=64 | MB/s | 0.22 MB/s | 0.254 MB/s | 0.537 MB/s | -13.4% | -59.1% |
| bs=10 sw=10 sl=64 | p50 | 25,891 us | 21,717 us | 11,404 us | +19.2% | +127.0% |
| bs=10 sw=10 sl=64 | p95 | 39,543 us | 38,372 us | 14,198 us | +3.1% | +178.5% |
| bs=10 sw=10 sl=64 | p99 | 39,543 us | 38,372 us | 17,818 us | +3.1% | +121.9% |
| bs=100 sw=10 sl=64 | throughput | 790 tuples/sec | 848 tuples/sec | 1,146 tuples/sec | -6.8% | -31.1% |
| bs=100 sw=10 sl=64 | MB/s | 0.482 MB/s | 0.518 MB/s | 0.7 MB/s | -6.9% | -31.1% |
| bs=100 sw=10 sl=64 | p50 | 123,256 us | 116,046 us | 90,069 us | +6.2% | +36.8% |
| bs=100 sw=10 sl=64 | p95 | 179,143 us | 145,714 us | 96,453 us | +22.9% | +85.7% |
| bs=100 sw=10 sl=64 | p99 | 179,143 us | 145,714 us | 109,096 us | +22.9% | +64.2% |
| bs=1000 sw=10 sl=64 | throughput | 944 tuples/sec | 958 tuples/sec | 1,174 tuples/sec | -1.5% | -19.6% |
| bs=1000 sw=10 sl=64 | MB/s | 0.576 MB/s | 0.585 MB/s | 0.717 MB/s | -1.5% | -19.6% |
| bs=1000 sw=10 sl=64 | p50 | 1,052,508 us | 1,038,365 us | 886,710 us | +1.4% | +18.7% |
| bs=1000 sw=10 sl=64 | p95 | 1,135,462 us | 1,153,050 us | 933,314 us | -1.5% | +21.7% |
| bs=1000 sw=10 sl=64 | p99 | 1,135,462 us | 1,153,050 us | 979,175 us | -1.5% | +16.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,556.05,200,128000,360,0.220,25891.22,39542.92,39542.92
1,100,10,64,20,2530.26,2000,1280000,790,0.482,123255.57,179142.50,179142.50
2,1000,10,64,20,21191.17,20000,12800000,944,0.576,1052508.31,1135461.80,1135461.80…ned integer can arrive in Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
read_json's default float parser rounds 1.0000000000000002 to 1.0 where Jackson keeps it, so a filter on v > 1 kept different rows in the two paths. precise_float=True reads the value exactly. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… last line The schema sampled only the rows inside the window, so an offset past the end left nothing to infer from and the operator declared no columns. An empty window now samples from the top of the file instead. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
carloea2
left a comment
There was a problem hiding this comment.
Retested the current head. The offset past EOF regression is fixed. All 75 focused tests and scalafmtCheckAll pass with the shared source API from #8341 supplied locally.
Command: sbt -batch "WorkflowCore/testOnly *ArrowUtilsSpec -- -oDF" "WorkflowOperator/testOnly *ArrowSourceOpDescSpec *JSONLScanSourceOpDescSpec *ArrowUtilsSpec -- -oDF" "scalafmtCheckAll".
Current CI still fails compilation on that missing API. This verifies the fix, not standalone merge readiness. #8341 and #8597 still need to land first, as the description states.
carloea2
left a comment
There was a problem hiding this comment.
The 75 focused Scala tests and scalafmtCheckAll pass with #8341 and #8597 supplied. The Python suites have 70 passes and three tests blocked by their unconditional time.tzset calls on Windows. A separate seven-case production-parser probe finds three pre-1970 offset inputs now raising OSError; ordinary, future, no-offset and invalid-input controls pass. The upstream parser accepts the same pre-1970 inputs.
sbt -batch 'WorkflowCore/testOnly *ArrowUtilsSpec -- -oDF' 'WorkflowOperator/testOnly *ArrowSourceOpDescSpec *JSONLScanSourceOpDescSpec *ArrowUtilsSpec -- -oDF' 'scalafmtCheckAll'
python -m pytest amber/src/test/python/core/models/schema/test_attribute_type.py amber/src/test/python/pytexera/udf/test_udf_operator.py -q
…ut localtime before 1970 Windows' localtime refuses an instant before 1970, so a bare astimezone() raised OSError there. The machine's zone is now asked by name, and where it has none such an instant takes the standard offset. The tests that change the zone skip where time.tzset is missing. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
carloea2
left a comment
There was a problem hiding this comment.
The original Windows crash is fixed. The original timestamp probes now pass, along with 75 focused Scala checks and 81 Python checks, with seven platform skips. Formatting passes with the declared source prerequisites supplied. A separate value check still finds a mismatch before 1900, confirmed against JDK 17 using the engine's DateParserUtils. Its two historical cases fail while the 1960 and 2024 controls pass.
python -m pytest amber/src/test/python/core/models/schema/test_attribute_type.py amber/src/test/python/pytexera/udf/test_udf_operator.py -q
sbt -batch 'WorkflowCore/testOnly *ArrowUtilsSpec -- -oDF' 'WorkflowOperator/testOnly *ArrowSourceOpDescSpec *JSONLScanSourceOpDescSpec *ArrowUtilsSpec -- -oDF' 'scalafmtCheckAll'
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…port lookup Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…, as the engine does Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
@carloea2 Before 1900 an offset is now read at the zone's standard offset, the same rule as |
…es (apache#8597) ### What changes were proposed in this PR? Type Casting to a timestamp read text with `pd.to_datetime(col, errors="coerce")` in the exported script. That is wrong in two ways, and both empty a cell the run itself filled. pandas parses into nanoseconds, which reach only 1677-09-21 to 2262-04-11. The engine holds a `java.sql.Timestamp`, where `2500-01-01 00:00:00` is an ordinary moment: it reads it, and the exported script answered with an empty cell. The same for `1500-06-15 08:30:00` and `9999-12-31 23:59:59`. pandas also infers one format for the whole column and coerces every row that does not match it. The engine hands `DateParserUtils` one field at a time, so a row states its own format. A column holding `2024-03-05 14:09:07` and `March 5, 2024` kept the first and emptied the second. The text branch now reads the column cell by cell and holds the result at microsecond resolution, which covers the years the engine covers. Text no format reads now ends the run, as the engine's `parseField` ends its own. The script raises the engine's message, `Failed to parse type str to Timestamp`, where it used to leave the cell empty. dateutil reads every format the verification hands the engine, so the two refuse the same text. Review and the verification's edge values found more places where the cast parted from the engine: - A reading that states an offset, such as `Z` or `+05:30`, stopped the cast. The engine moves it to the machine's zone and keeps that wall clock, and the script now does the same. - A column that already held timestamps went through the text parse and lost its nanoseconds. `parseField` returns a `java.sql.Timestamp` untouched, so the script now hands such a column back at the resolution it arrived in. - Text with digits past the millisecond kept them. `DateParserUtils` reads into a `java.util.Date`, which counts milliseconds, so the script now cuts there too. - Casting to text wrote a value the way Python's `str()` does. Java writes a double past 1e7 or under 1e-3 in E notation, a boolean in lower case, and a timestamp with its nanoseconds and at least `.0`. A shared `_texera_java_text` now writes each one as Java's `toString` does. - A double column holding a NaN as a value, as an Arrow or Parquet source hands one over, lost it. The casts asked each cell whether it was empty, and pandas hands an empty cell over as NaN too, so a stored NaN came out null where the engine writes `NaN` as text, 0 as an integer and `true` as a boolean. The casts now read empty cells from the column's own mask, and the double cast builds a nullable `Float64`, so a NaN read from text stays one through the next cast. A double past an integer's bounds now saturates as `toInt` and `toLong` do, where `int()` raised on an infinity. - A moment was read with dateutil's `tzlocal()`, which knows only today's two offsets, so an instant from before a zone changed its rules came out an hour or more off on every platform. The script now reads the machine's zone by its IANA name and, before 1900, keeps today's standard offset as `java.util.Date` does. - On Windows, `localtime` refuses an instant before 1970, which stopped the cast. The same helper now falls back to the zone's standard offset there. That `Float64` column changed what a Python operator after this cast is handed. pyamber hands a Python operator a DOUBLE column as `float64`, where a NaN and a null are one value, and writes a NaN out as null. Given `Float64` instead, Sort ordered a NaN apart from the nulls, a chart drew points the engine dropped, and a passed-through NaN stayed NaN. `StandaloneCodeGenerator` gains `standaloneInputAdapter()`, and for a Python operator the translator feeds the fragment a `float64` copy of each input, leaving the upstream frame as it was. The number branch is left alone. A LONG is read as epoch milliseconds and has the same window, but it also disagrees about the calendar before 1582-10-15, where `new Timestamp(long)` is Julian and pandas is proleptic Gregorian. Widening the window there would answer with a moment eight days off instead of an empty cell, so what that branch should do is a separate decision. The issue records it. ### Any related issues, documentation, discussions? Closes apache#8595. apache#8480 closed the same nanosecond gap for Extract Date/Time Fields, which parses strictly and could take the one-line form this operator cannot. apache#8341, apache#8704, apache#8705 and apache#8709 call the timestamp and text helpers this adds, so this PR merges before them. ### How was this PR tested? Eight tests in `TypeCastingOpDescSpec` run the generated Python and compare it cell by cell against `AttributeTypeUtils.parseField`. They are tagged to run in the integration job, which has pandas. - The text-to-timestamp test reads moments in three formats, the rows either side of the nanosecond edge, `2500-01-01`, `1500-06-15`, `9999-12-31`, readings with `Z` and `+05:30`, and readings past the millisecond. It fails on the old generator. - One test gives a column that already holds timestamps and checks its nanoseconds are kept. - The cast-to-string test covers a double at 1e7 and 1e-4. - One test checks that unreadable text ends the script with the engine's message, as the engine refuses it. - Two tests cast a column holding a stored NaN, both infinities and an empty cell to each type and compare against `AttributeTypeUtils`, and check that a NaN read from text survives a second cast. - Two tests read moments from `America/Mexico_City` in 1899, 1921, 1931 and 2021, at the offset the zone had then, and the same moments with `localtime` refusing those before 1970. - Three tests in `WorkflowToPythonTranslatorSpec` check that a Python operator reads a converted copy of its input and an operator the engine runs on the JVM reads its upstream as it is. Two of them fail without the change. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
carloea2
left a comment
There was a problem hiding this comment.
Rechecked the source and timestamp fixes. The focused Arrow and parser checks pass, and CI is green.
…ython (apache#8709) ### What changes were proposed in this PR? Aggregate and the sort family (Sort, Stable Merge Sort and Sort Partitions) implement `StandaloneCodeGenerator`. The sort family's row order is its contract, so `LogicalOp` gains `orderSensitive`, which those three override. Anything comparing two runs of an operator reads it to decide whether the rows have to arrive in the same order or only be the same rows. Review then found five places where the block answered a question the engine answers differently. 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 sort put a NaN last on the way down, where the engine puts it first. The edge values the shared table now carries found one more. Aggregate's concatenation wrote a double past 1e7 or under 1e-3, and a timestamp, as Python prints them, where the engine writes what Java's toString gives. It now writes every value, the lower-case boolean the review asked for included, through the helper apache#8597 adds for casting to text. One change reaches code the engine runs. `getPhysicalPlan` swapped Aggregate's aggregations for their global-stage form and left them on the operator. The export compiles a plan and then translates the same operators, so the script grouped on the result column names and failed. The operator now keeps what the user set. The desc the global stage receives is unchanged. ### Any related issues, documentation, discussions? Part of apache#8325, step 11b; that issue lists the set in order. Aggregate's concatenation calls a helper apache#8597 adds. Sorting a NaN apart from a null, and grouping on one, 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#8707, 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 Aggregate 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 timestamp at two resolutions and in a zone other than UTC, and a null kept apart from a NaN in the sort and the group key. `AggregateOpDescSpec` builds the physical plan and then checks that the operator still holds the user's aggregations and that its generated code still reads the input column. ### 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.5 (1M context) <noreply@anthropic.com>
What changes were proposed in this PR?
The JSON Lines and Arrow sources implement
StandaloneCodeGenerator, so an exported script starts from the same data the workflow did.Arrow is read into pandas' nullable dtypes rather than its numpy ones. Arrow says of every value whether it is there, and a numpy column has nowhere to put that: a missing double and a stored NaN both land on NaN, and a missing integer costs the column its integer type. The engine reads the file's own answer, so a set operation keeps a NaN row a null row does not match, a holed integer column is still made of integers, and a long past 2^53 is not rounded on its way through a float. The other sources state the types they need one at a time, their formats having no way to say it once.
A file also states the width and the sign of each of its numbers, and Texera has no column narrower than a double or a 32-bit integer. A single-precision or 16-bit column was arriving null, the parse having no case for the Float and Short those vectors hand back, and an unsigned one arrived as the -1 its storage counts down to. Each now reads as the Texera type that holds it, an unsigned 32-bit column as a long and an unsigned 64-bit one refused for want of a wider one, and the script normalizes the same widths so a later sum comes to the same number on both sides. This is the rule the Parquet source already follows. The end of an exported Arrow window is counted in Long, as the CSV readers count theirs.
A timestamp is read as the engine reads one. The engine hands each field to DateParserUtils, so a row states its own format, any year reads, an offset is moved to the local zone, and the reading is cut to the millisecond. pandas' date parsing does none of that. So JSONL reads the column as text and converts it with the helper #8597 adds.
This PR also changes the engine.
ArrowUtilshanded a TIMESTAMP column to a Python worker as an Arrow millisecond timestamp labelled UTC. A Python operator saw an aware datetime cut to the millisecond, and a year past 2262 overflowed. It is now a zoneless microsecond timestamp. That is what Texera's own Python side declares for a TIMESTAMP (pa.timestamp("us")), and what an exported script holds. Two things a user can see change. A Python UDF now receives a naive datetime instead of one in UTC. A result downloaded as an Arrow file now carries a zoneless microsecond timestamp column instead of a UTC millisecond one. A TIMESTAMP parameter a Python UDF declares is now parsed the same way, as a naive wall clock with any offset moved to the machine's zone, so it compares with a row's instead of raising.Any related issues, documentation, discussions?
Part of #8325, step 12b; that issue lists the set in order. It uses #8341's shared source code and #8597's timestamp helpers.
Its JSON Lines export reads a JSON
nullas the text"null", as the engine does today (#8674). If #8740 merges first, that part changes to read a null before this one merges.Closes #8702, the task this change is the whole of.
It also finishes #8602, where a scan window of no rows left the type inference nothing to read and JSONL declared a schema of no columns. #8341 fixes the CSV half.
Closes #8602.
How was this PR tested?
Each operator asserts the block it emits in its own spec, and the Arrow spec runs the engine's own executor and the generated script over the same file. The engine change is covered in both ArrowUtilsSpec files, which include a round trip on either side of the nanosecond range, and in the Python tests of the timestamp parser. Once the verification lands, every operator is also run both ways on every configuration its schema offers; this branch is cut from main and does not carry that machinery, so those runs are not on this diff's CI.
Arrow is the one worth calling out. Its column of timestamps only matches once #7672 is in: before that the engine read the file's numbers through the JVM's zone while
pd.read_featherread the wall clock the file states, and the two differed by whatever the machine was set to. That change has since landed.Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)