[rust] Wire batch statistics into the log batch write and read paths - #4074
[rust] Wire batch statistics into the log batch write and read paths#4074leekeiabstraction wants to merge 3 commits into
Conversation
e33bb99 to
c09cb91
Compare
Tables with 'table.statistics.columns' set now produce V1 batches carrying serialized statistics, matching Java's MemoryLogRecordsArrowBuilder. The reader now locates the records data per the batch's magic, fixing the misparsing of V1 batches that Java clients and the server's projection path already produce.
c09cb91 to
48718e9
Compare
|
@fresh-borzoni @platinumhamburg @charlesdong1991 Appreciate a review here 🙏 |
fresh-borzoni
left a comment
fresh-borzoni
left a comment
There was a problem hiding this comment.
@leekeiabstraction Thank you for the great PR, I verified the roundtrip as well 👍
Really great work, LGTM overall, left minor comments, PTAL
| @@ -1 +1 @@ | |||
| // Licensed to the Apache Software Foundation (ASF) under one | |||
There was a problem hiding this comment.
This module grows another export while staying pub. Make it pub(crate) now?
|
|
||
| /// The epoch of the leader that wrote this batch, or [`NO_LEADER_EPOCH`] | ||
| /// for magics before V2. | ||
| pub fn leader_epoch(&self) -> i32 { |
There was a problem hiding this comment.
Which code produces V2 today?
ArrowWalBuilder passes a null collector so CDC stays V0, CURRENT_LOG_MAGIC_VALUE is still V0, and the only V2 branches I find are dormant ones in the two builders.
Did I miss a producer? Asking because the description says the server already produces them.
There was a problem hiding this comment.
Well caught, description is wrong. The code that allows setting of leader epoch is there but nothing calls it yet. Updated descriptions / comments
| cursor.write_all(&statistics_bytes)?; | ||
| cursor.write_all(real_arrow_batch_bytes)?; | ||
|
|
||
| let calcute_crc_bytes = &cursor.get_ref()[SCHEMA_ID_OFFSET..]; |
There was a problem hiding this comment.
nit: the CRC is read and written at V0/V1 offsets while the reader shifts them for V2.
Add a debug_assert!(magic < LOG_MAGIC_VALUE_V2) so a future V2 writer fails loudly?
There was a problem hiding this comment.
Added debug_assert!
| /// The Java reference block, generated and asserted by | ||
| /// `LogRecordBatchStatisticsCompatibilityTest` against the same checked-in | ||
| /// fixture so both languages pin to one set of bytes. | ||
| fn java_statistics_block_hex() -> String { |
There was a problem hiding this comment.
nit: this resolves only inside the monorepo, so the test panics in a vendored crate.
Is the published crate expected to pass its own tests?
There was a problem hiding this comment.
Well caught, added .hex file within the crate. Added assertion to check that Rust side .hex file's content is equivalent to Java side when Java side .hex file is accessible.
The log batch framing and the aligned row module are now crate-private, dropping two dead items the tightening surfaced. The golden statistics fixture is embedded in the crate so the test passes outside the monorepo, and a debug assert guards the builder's V0/V1 header layout against a future V2 writer.
|
TY for the thorough review as usual @fresh-borzoni , addressed your comments. PTAL 🙏 |
2 participants
Summary
table.statistics.columnsset emit V1 batches carrying the serialized statistics between the header and the records data, mirroring Java'sMemoryLogRecordsArrowBuilder. A serialization failure degrades to an empty statistics section rather than failing the batch.DefaultLogRecordBatch, which also reads V2 while no builder writes it yet. The writer stays on V0/V1 like Java's client-side builder.arrow.rsinto a newlog_record_batch.rs, separating the Fluss wire format from Arrow IPC concerns. The Arrow builder's long argument lists collapsed into anArrowBatchConfigstruct.Test Plan
fluss-common/src/test/resources/encoding/statistics_block.hex) pins the statistics block byte-for-byte across all supported types, nulls, both decimal widths, both timestamp precisions, and float NaN/signed-zero ordering (bounds follow Java'sFloat.comparesemantics, not Arrow's IEEE totalOrder). A new Java test generates it, asserts it and parses it back verifying the decoded bounds; the Rust test reads the same file.cargo test -p fluss-rs --libpasses 707 tests; clippy and fmt are clean; the end-to-end integration test is left to [rust] Filter pushdown integration test #3845.🤖 AI-assisted changes - reviewed by human developer