[rust] Add V1 batch statistics collection by leekeiabstraction · Pull Request #4011 · apache/fluss

leekeiabstraction · 2026-08-16T17:59:34Z

Summary

Closes [rust] Batch statistics implementation #4008.
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.
Statistics are derived from the finished Arrow batch rather than row by row as Java does, so the row-append and append_arrow_batch paths are covered by one implementation.
Nothing emits them yet, so batches stay in the V0 format and behaviour is unchanged.

Test Plan

cargo test -p fluss-rs --lib: 646 passed, including 16 tests asserting AlignedRowWriter output byte by byte against Java's layout and 5 covering the statistics block.
cargo fmt --all --check and cargo clippy -p fluss-rs --all-targets clean.

🤖 AI-assisted changes - reviewed by human developer

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.

leekeiabstraction · 2026-08-16T18:41:05Z

@fresh-borzoni @charlesdong1991 Appreciate review here 🙏

fresh-borzoni

@leekeiabstraction Thank you for the PR, left a couple of comments, PTAL

fresh-borzoni · 2026-08-17T00:27:05Z

+    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);


mapping is checked against the table schema but not against the batch, and batch.column(i) panics out of range.

Well spotted, added mapped index range checks. Note that check against table schema is done in column_bounds

fresh-borzoni · 2026-08-17T00:29:24Z

+    ///
+    /// # Errors
+    /// Returns an error if a named column is absent from the table schema.
+    pub fn get_stats_index_mapping(&self) -> Result<Vec<usize>> {


Java rejects unsupported statistics columns at table creation, not at write time. Why do we prefer to differ here?

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.

fresh-borzoni · 2026-08-17T00:31:00Z

+    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);


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?

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.

The module row::aligned is introduced in this PR, so pub(crate) mod aligned would not change any writer signature and would keep the ARRAY and MAP panics out of the public API.

ValueWriter::write_value routes ARRAY and MAP into any BinaryWriter, and Java's AlignedRow supports both through AlignedArray and AlignedMap, so a caller using this writer outside statistics would hit an abort rather than an error. With the module made internal, the only new warning is that size_in_bytes has no caller, which is already true today.

Would it make sense to keep the module internal until the writers can return Result? The visibility can be widened later without a compatibility concern, while the reverse is not true.

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.

leekeiabstraction · 2026-08-17T07:32:53Z

@fresh-borzoni TY for your review, I've addressed your comments and also push a separate fix to a bug I found. PTAL.

naivedogger

Thanks @leekeiabstraction for working on this! Overall, LGTM. I left one minor comment, PTAL.

naivedogger · 2026-08-19T07:02:14Z

+
+    /// Inlines a payload of at most 7 bytes into the field slot, with
+    /// `len | 0x80` in the slot's high byte as Java's `writeBytesToFixLenPart`.
+    fn write_bytes_to_fix_len_part(&mut self, pos: usize, bytes: &[u8]) {


The low-level encoding logic here substantially overlaps with FlussArrayWriter in binary_array.rs. Both implement the same wire-format rules for inline bytes, aligned variable-length storage, offset/size packing, and non-compact Decimal/Timestamp encoding, but through separate code paths.

On the Java side, the common encoding logic is centralized in AbstractBinaryWriter and reused by both AlignedRowWriter and BinaryArrayWriter, while writer-specific field addressing remains separate.

Would it make sense to extract a small internal encoding/buffer helper in Rust so that AlignedRowWriter and FlussArrayWriter can share these rules without requiring the same API or trait? This would reduce duplication and help prevent format divergence.

+1, and there is a third copy: PaimonBinaryRowWriter implements the same rules again

fresh-borzoni

@leekeiabstraction Thank you, left some additional comments and question, PTAL

fresh-borzoni · 2026-08-20T12:35:11Z

+                .as_any()
+                .downcast_ref::<PrimitiveArray<Decimal128Type>>()
+                .ok_or_else(|| unexpected_array(column, data_type))?;
+            let (precision, scale) = (decimal_type.precision(), decimal_type.scale());


DECIMAL takes the scale from the type, not the array, so a Decimal128(10,2) batch on a DECIMAL(10,3) column gives 1.234 instead of 12.34. column_vector.rs uses Decimal::from_arrow_decimal128 for this. Should the source scale come from column.data_type() here too?

fresh-borzoni · 2026-08-20T12:35:52Z

+            ),
+            _ => Err(unexpected_array(column, data_type)),
+        },
+        DataType::String(_) | DataType::Char(_) => {


Java's collector has no CHAR branch so it ships a null bound, while we ship a real one. A CHAR predicate would prune our batches and never Java's.
Should we file an issue on the Java side?

fresh-borzoni · 2026-08-20T12:38:13Z

+    ///
+    /// # Errors
+    /// Returns an error if a named column is absent from the table schema.
+    pub fn get_stats_index_mapping(&self) -> Result<Vec<usize>> {


Java caches this because it's on the per-batch path, here every call re-splits the property string. Should we cache it on TableInfo?

fresh-borzoni · 2026-08-20T12:54:53Z

+    for &column_index in mapping {
+        let column = batch.column(column_index);
+        null_counts.push(column.null_count() as i32);
+        bounds.push(column_bounds(


Java never fails a write over statistics, MemoryLogRecordsArrowBuilder catches and writes length 0, while here any column_bounds error reaches the caller.
Is it intended?

leekeiabstraction force-pushed the rust-v1-statistics-standalone branch 2 times, most recently from 7595c4d to 4692a22 Compare August 16, 2026 18:12

leekeiabstraction force-pushed the rust-v1-statistics-standalone branch from 4692a22 to 4bbfbd0 Compare August 16, 2026 18:36

fresh-borzoni reviewed Aug 17, 2026

View reviewed changes

leekeiabstraction added 3 commits August 17, 2026 07:56

[rust] Address review on batch statistics

9172270

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.

[rust] Scale second precision time bounds to millis

a270717

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.

[rust] Cover the remaining unit conversions with tests

0d7b031

Adds tests for the Time64 micro and nano, timestamp second and nano, and non compact decimal arms, which were the conversions still going unchecked.

leekeiabstraction force-pushed the rust-v1-statistics-standalone branch from 9962e3a to 0d7b031 Compare August 17, 2026 07:31

naivedogger approved these changes Aug 19, 2026

View reviewed changes

fresh-borzoni reviewed Aug 20, 2026

View reviewed changes

Uh oh!

Conversation

leekeiabstraction commented Aug 16, 2026

Summary

Test Plan

Uh oh!

leekeiabstraction commented Aug 16, 2026

Uh oh!

fresh-borzoni left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leekeiabstraction Aug 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leekeiabstraction commented Aug 17, 2026

Uh oh!

naivedogger left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fresh-borzoni left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

leekeiabstraction Aug 17, 2026 •

edited

Loading