Repository navigation
feat(arrow): store and read a timestamp as the wall clock it is - #7672
Conversation
Every other source hands the engine a wall clock. CSV and JSONL carry no zone at all, so what the file states is what the operator gets, and a Texera TIMESTAMP has no zone to record one either. Arrow was the exception: its fields carry a zone, and both directions reconciled that through the value's own epoch, which reads the wall clock in the JVM's zone and puts the machine's setting where the file cannot record it. Reading was the visible half. A zoned vector hands back epoch milliseconds, and `new Timestamp(millis)` renders those in the local zone, so one file read out 00:00 in Los Angeles, 08:00 in UTC and 17:00 in Tokyo. Nothing in the file accounted for the difference and nothing reported it. Writing was the same mismatch from the other end: the number stored was the local instant of the wall clock while the label beside it said UTC, so every reader other than a Texera in that same zone saw the value moved. Both now go through the label the field already carries, which puts Arrow on the same footing as the rest: what the file states is what the engine gets, wherever it runs. Zoneless columns are untouched, those handing back a LocalDateTime that is already the wall clock itself. The spec asserted the old identity, that the stored long equals the value's own epoch. It states the wall clock explicitly now, so what it expects no longer depends on where it runs. Note for anyone with existing files: bytes written before this are read by the new rule, so a timestamp in them shifts by the offset of the zone that wrote it. Closes apache#7666 Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Automated Reviewer SuggestionsBased on the
|
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #7672 +/- ##
============================================
- Coverage 93.52% 93.51% -0.02%
Complexity 4727 4727
============================================
Files 1195 1195
Lines 48492 48509 +17
Branches 5379 5391 +12
============================================
+ Hits 45352 45362 +10
- Misses 1693 1694 +1
- Partials 1447 1453 +6
*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 | 424 | 0.259 | 23,256/30,625/30,625 us | 🔴 +12.3% / 🔴 +115.4% |
| 🟢 | bs=100 sw=10 sl=64 | 947 | 0.578 | 103,019/138,419/138,419 us | 🟢 -9.1% / 🔴 +45.6% |
| ⚪ | bs=1000 sw=10 sl=64 | 1,121 | 0.684 | 895,633/970,639/970,639 us | ⚪ within ±5% / 🔴 +5.9% |
Baseline details
Latest main 604f109 from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 424 tuples/sec | 455 tuples/sec | 894.27 tuples/sec | -6.8% | -52.6% |
| bs=10 sw=10 sl=64 | MB/s | 0.259 MB/s | 0.278 MB/s | 0.546 MB/s | -6.8% | -52.5% |
| bs=10 sw=10 sl=64 | p50 | 23,256 us | 20,713 us | 11,164 us | +12.3% | +108.3% |
| bs=10 sw=10 sl=64 | p95 | 30,625 us | 32,010 us | 14,219 us | -4.3% | +115.4% |
| bs=10 sw=10 sl=64 | p99 | 30,625 us | 32,010 us | 16,863 us | -4.3% | +81.6% |
| bs=100 sw=10 sl=64 | throughput | 947 tuples/sec | 965 tuples/sec | 1,158 tuples/sec | -1.9% | -18.2% |
| bs=100 sw=10 sl=64 | MB/s | 0.578 MB/s | 0.589 MB/s | 0.707 MB/s | -1.9% | -18.2% |
| bs=100 sw=10 sl=64 | p50 | 103,019 us | 99,201 us | 88,711 us | +3.8% | +16.1% |
| bs=100 sw=10 sl=64 | p95 | 138,419 us | 152,194 us | 95,062 us | -9.1% | +45.6% |
| bs=100 sw=10 sl=64 | p99 | 138,419 us | 152,194 us | 102,275 us | -9.1% | +35.3% |
| bs=1000 sw=10 sl=64 | throughput | 1,121 tuples/sec | 1,116 tuples/sec | 1,188 tuples/sec | +0.4% | -5.6% |
| bs=1000 sw=10 sl=64 | MB/s | 0.684 MB/s | 0.681 MB/s | 0.725 MB/s | +0.4% | -5.6% |
| bs=1000 sw=10 sl=64 | p50 | 895,633 us | 888,882 us | 874,204 us | +0.8% | +2.5% |
| bs=1000 sw=10 sl=64 | p95 | 970,639 us | 969,719 us | 916,845 us | +0.1% | +5.9% |
| bs=1000 sw=10 sl=64 | p99 | 970,639 us | 969,719 us | 947,186 us | +0.1% | +2.5% |
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,471.19,200,128000,424,0.259,23256.44,30625.39,30625.39
1,100,10,64,20,2111.18,2000,1280000,947,0.578,103018.90,138418.97,138418.97
2,1000,10,64,20,17846.36,20000,12800000,1121,0.684,895633.36,970638.98,970638.98
carloea2
left a comment
There was a problem hiding this comment.
The conversion always uses UTC, but an Arrow timestamp field can name another zone. getTexeraTuple maps every Arrow timestamp to Texera TIMESTAMP, so a field labeled America New York will return the UTC clock instead of the clock in that field. setTexeraTuple has the same issue with a non UTC timestamp vector. Please use the timezone from the Arrow field and add a non UTC test.
Reading a zoned vector as UTC was right only for the fields this writes, which are the UTC millisecond ones. An .arrow file handed to the source operator can label a timestamp field with any zone, and Arrow hands a zoned vector its number unscaled, so the field's unit is what says how far from the epoch that number reaches. A pandas tz-aware column arrives as nanoseconds in its own zone: read as UTC milliseconds it landed in 1970. Both directions now go through the field. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
|
Solved. Both directions now take the zone from the Arrow field and there are tests on non-UTC ones. |
|
@aglinxinyuan May you take a look at it? |
|
/request-review @aglinxinyuan |
carloea2
left a comment
There was a problem hiding this comment.
The timestamp result looks correct now, but the benchmark shows a large slowdown, including the string batches. getTexeraTuple now reads and matches the Arrow field type for every value, even when the Texera type is not TIMESTAMP. Please keep the new field work inside the TIMESTAMP case and rerun the benchmark.
Selecting on the Arrow field type put that read on every value's path, including the columns whose type the schema already settles. The schema's attribute type is what decides now, so only a TIMESTAMP consults its field and every other column goes straight to the parse it did before. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
|
Moved the field read into the TIMESTAMP case. The schema's attribute type selects now, so a non-timestamp column never touches the field. Also merged main so the benchmark has a current baseline, and pushed a rerun. |
|
Benchmark is green: all three configs within ±5% of main on the same runner. The empty commit above was just to trigger the rerun, since the first run's small-batch delta did not repeat. |
carloea2
left a comment
There was a problem hiding this comment.
The field read now happens only for timestamps, and the rerun shows no benchmark slowdown. The timezone fix looks good.
|
@aglinxinyuan May you take a look at this PR please? |
Both sides carried this change in different shapes; the PR's is kept. It dispatches on the Texera attribute type rather than on the Arrow field, so an Arrow timestamp read into a column declared something else is parsed as that something else instead of becoming a Timestamp, and `wallClockOf` widens to accept a field that is not a timestamp at all. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
…ython (apache#8704) ### 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 apache#8597 adds. This PR also changes the engine. `ArrowUtils` handed 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 apache#8325, step 12b; that issue lists the set in order. It uses apache#8341's shared source code and apache#8597's timestamp helpers. Its JSON Lines export reads a JSON `null` as the text `"null"`, as the engine does today (apache#8674). If apache#8740 merges first, that part changes to read a null before this one merges. Closes apache#8702, the task this change is the whole of. It also finishes apache#8602, where a scan window of no rows left the type inference nothing to read and JSONL declared a schema of no columns. apache#8341 fixes the CSV half. Closes apache#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 apache#7672 is in: before that the engine read the file's numbers through the JVM's zone while `pd.read_feather` read 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) --------- Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
What changes were proposed in this PR?
Every other source hands the engine a wall clock. CSV and JSONL carry no zone at all, so what the file states is what the operator gets, and
TypeCastingand the time-series plot both parse with a plainpd.to_datetime. A Texera TIMESTAMP has no zone to record one either. Arrow was the exception: its timestamp fields state how their own numbers are to be read, andArrowUtilsignored that and went through the value's own epoch, which reads the wall clock in the JVM's zone and puts the machine's setting where the file cannot record it.So one file read out different values on servers in different places, with nothing to account for the difference:
Writing was the same mismatch from the other end: the number stored was the local instant of the wall clock while the label beside it said UTC, so every reader other than a Texera in that same zone saw the value moved. Reading that file with
pandas.read_featherreported 08:00 before and reports 00:00 now, which is what Texera itself shows.Both directions now go through what the field itself states, which puts Arrow on the same footing as the rest: what the file states is what the engine gets, wherever it runs. A field states two things and both matter. One is the zone. Every field this writes is labelled UTC, but an
.arrowfile handed to the source operator may name any zone, and the wall clock to hand over is the one that zone gives. The other is the unit. Arrow hands a zoned vector its number unscaled, so the field's unit is what says how far from the epoch that number reaches. A tz-aware pandas column arrives as nanoseconds, and taken for milliseconds it landed in 1970. Unlabelled columns, which is what an ordinary.arrowfile written by pandas holds, still hand back aLocalDateTime, already the wall clock itself.One thing for anyone holding existing files: bytes written before this are read by the new rule, so a timestamp in them shifts by the offset of the zone that wrote it.
Any related issues, documentation, discussions?
Closes #7666
How was this PR tested?
Two
ArrowUtilsSpecs cover it. Inworkflow-operator, one assertion moved with the change: it stated the identity between the stored long and the value's own epoch, which is machine-dependent, and now states the wall clock explicitly. The round-trip case needed no change, preserving the wall clock being the property both before and after.In
workflow-core, six cases cover the timestamp fieldsfromTexeraSchemanever produces, since every root built from it is UTC milliseconds. Three read: a zoned millisecond field, whose number is one reading in the zone it names and another five hours off in UTC; a zoned nanosecond field, which taken for milliseconds would land in 1970; and an unlabelled field, which hands back the wall clock already. Two write, one per axis, pinning the number that reaches the vector. The last sends a wall clock out and back through five combinations of unit and zone, an unlabelled one among them.Both specs were also run in four JVMs started with different
-Duser.timezone(America/Los_Angeles, UTC, Asia/Tokyo, Europe/Berlin) and pass identically in each, which is the property the table above states.workflow-core(765 tests) andworkflow-operator(2214) pass, as do the Python-channel specs that share this conversion,PythonProxyServerSpecandPythonProxyClientSpec.Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)