[rust] Add V1 batch statistics collection - #4011
Conversation
7595c4d to
4692a22
Compare
Adds the pieces a writer needs to produce V1 log record batch statistics: `AlignedRowWriter` for the min and max rows, `record::statistics` to reduce an Arrow batch to per-column bounds and null counts, and the `table.statistics.columns` plumbing that selects the columns. Nothing emits them yet, so batches stay in the V0 format.
4692a22 to
4bbfbd0
Compare
|
@fresh-borzoni @charlesdong1991 Appreciate review here 🙏 |
fresh-borzoni
left a comment
fresh-borzoni
left a comment
There was a problem hiding this comment.
@leekeiabstraction Thank you for the PR, left a couple of comments, PTAL
| let mut null_counts = Vec::with_capacity(mapping.len()); | ||
| let mut bounds = Vec::with_capacity(mapping.len()); | ||
| for &column_index in mapping { | ||
| let column = batch.column(column_index); |
There was a problem hiding this comment.
mapping is checked against the table schema but not against the batch, and batch.column(i) panics out of range.
There was a problem hiding this comment.
Well spotted, added mapped index range checks. Note that check against table schema is done in column_bounds
| /// | ||
| /// # Errors | ||
| /// Returns an error if a named column is absent from the table schema. | ||
| pub fn get_stats_index_mapping(&self) -> Result<Vec<usize>> { |
There was a problem hiding this comment.
Java rejects unsupported statistics columns at table creation, not at write time. Why do we prefer to differ here?
There was a problem hiding this comment.
Well spotted, I've made an error in the documentation. The implementation is already mirroring Java's TableInfo.getStatsIndexMapping() which is called at write time.
| fn write_decimal(&mut self, value: &Decimal, precision: u32) { | ||
| let pos = self.current_pos; | ||
| if Decimal::is_compact_precision(precision) { | ||
| let unscaled = value.to_unscaled_long().unwrap_or(0); |
There was a problem hiding this comment.
unwrap_or(0) turns a decimal that doesn't fit in i64 into a silent 0, which then ships as a min or max. Java throws here. Shall we propagate it or there is a reason for clamping?
There was a problem hiding this comment.
Well spotted, I've updated to mirror Java side client. Using panic instead of Result::Error due to existing signature of writers. Changing the signature or the reducing writer visibility to pub(crate) probably requires some discussion and is out of scope of this PR.
Bounds-check the statistics mapping against the batch rather than panicking on a mismatch, and reject a decimal built at a precision other than the column's instead of writing a silent zero bound.
The Arrow array holds seconds for TIME(0) while the statistics format is always millis of day, so the bounds were a thousand times too small.
Adds tests for the Time64 micro and nano, timestamp second and nano, and non compact decimal arms, which were the conversions still going unchecked.
9962e3a to
0d7b031
Compare
|
@fresh-borzoni TY for your review, I've addressed your comments and also push a separate fix to a bug I found. PTAL. |
2 participants
Summary
AlignedRowWriterfor the min and max rows,record::statisticsto reduce an Arrow batch to per-column bounds and null counts, and thetable.statistics.columnsplumbing that selects the columns.append_arrow_batchpaths are covered by one implementation.Test Plan
cargo test -p fluss-rs --lib: 646 passed, including 16 tests assertingAlignedRowWriteroutput byte by byte against Java's layout and 5 covering the statistics block.cargo fmt --all --checkandcargo clippy -p fluss-rs --all-targetsclean.🤖 AI-assisted changes - reviewed by human developer