You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "leekeiabstraction (via GitHub)" <gi...@apache.org> on 2026/08/22 21:29:50 UTC
[PR] [rust] Wire batch statistics into the log batch write and read paths [fluss]
leekeiabstraction opened a new pull request, #4074:
URL: https://github.com/apache/fluss/pull/4074
## Summary
- Wires the already-merged V1 batch statistics collection (#4011) into the write and read paths, closing #4009 and #4010.
- **Write** (mirroring Java's `MemoryLogRecordsArrowBuilder` / `RecordAccumulator`): for a table with `table.statistics.columns` set, the accumulator resolves the statistics column mapping and hands it to the Arrow builder, which upgrades the batch magic to V1, serializes the statistics between the header and the records data, writes the `StatisticsLength` header field, and reserves an estimated statistics size in its size accounting. A serialization failure degrades to an empty statistics section rather than failing the batch, as in Java.
- **Read** (mirroring Java's `DefaultLogRecordBatch.recordsDataOffset`): the records data is located per the batch's magic instead of a hardcoded V0 offset, and unknown magic versions are rejected with a clear error instead of being misparsed.
- The read side also fixes a live incompatibility: Java clients already produce V1 batches for statistics-enabled tables, and the server's projection path keeps the V1 magic (with `StatisticsLength = 0`) even after stripping the statistics, so the Rust reader was handing statistics bytes to the Arrow IPC parser.
## Test Plan
- New unit tests in `record/arrow.rs`: V1 header sizes; decoding V1 batches with and without a statistics section (append-only and changelog); rejection of unsupported magic, negative statistics length, and a statistics length past the batch end; V1 batches produced by both builder modes carry the exact serialized statistics and round-trip through the reader; V0 is kept when statistics are disabled; size estimate reserves room for statistics; out-of-range mapping index is rejected at construction.
- `cargo test -p fluss-rs --lib`: 699 passed. `cargo clippy --all-targets` and `cargo fmt --check` clean.
- An end-to-end filter pushdown integration test is deliberately left to #3845.
🤖 AI-assisted changes - reviewed by human developer
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_018NzSynxezK31J84gGAs5wN
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Wire batch statistics into the log batch write and read paths [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #4074:
URL: https://github.com/apache/fluss/pull/4074#discussion_r3845889545
##########
fluss-rust/crates/fluss/src/record/arrow.rs:
##########
@@ -508,14 +459,35 @@ impl MemoryLogRecordsArrowBuilder {
.update_estimation(actual_ratio);
}
- // now, write batch header and arrow batch
- let mut batch_bytes = vec![0u8; RECORD_BATCH_HEADER_SIZE + real_arrow_batch_bytes.len()];
+ let statistics_bytes = match &self.statistics {
+ Some((row_type, mapping)) => {
+ match serialize_statistics(record_batch.as_ref(), row_type, mapping) {
+ Ok(Some(bytes)) => bytes,
+ // Unlike Java's 27-byte empty block, an empty mapping
+ // writes length 0, which both parsers read as no statistics.
+ Ok(None) => Vec::new(),
+ // A failure degrades to an empty section rather than
+ // failing the batch, matching Java's builder.
+ Err(error) => {
+ log::error!("Failed to serialize statistics for record batch: {error}");
+ Vec::new()
+ }
+ }
+ }
+ None => Vec::new(),
+ };
+
+ // now, write batch header, statistics and arrow batch
+ let header_size = self.header_size();
+ let mut batch_bytes =
+ vec![0u8; header_size + statistics_bytes.len() + real_arrow_batch_bytes.len()];
// write batch header
- self.write_batch_header(&mut batch_bytes[..])?;
+ self.write_batch_header(&mut batch_bytes[..], statistics_bytes.len())?;
- // write arrow batch bytes
+ // write statistics and arrow batch bytes
let mut cursor = Cursor::new(&mut batch_bytes[..]);
- cursor.set_position(RECORD_BATCH_HEADER_SIZE as u64);
+ cursor.set_position(header_size as u64);
+ cursor.write_all(&statistics_bytes)?;
cursor.write_all(real_arrow_batch_bytes)?;
let calcute_crc_bytes = &cursor.get_ref()[SCHEMA_ID_OFFSET..];
Review Comment:
Added debug_assert!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Wire batch statistics into the log batch write and read paths [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #4074:
URL: https://github.com/apache/fluss/pull/4074#issuecomment-5387416794
@fresh-borzoni @platinumhamburg @charlesdong1991 Appreciate a review here 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Wire batch statistics into the log batch write and read paths [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #4074:
URL: https://github.com/apache/fluss/pull/4074#discussion_r3845886919
##########
fluss-rust/crates/fluss/src/record/log_record_batch.rs:
##########
@@ -0,0 +1,1296 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Fluss' log record batch wire format, mirroring Java's
+//! `LogRecordBatchFormat` / `DefaultLogRecordBatch`.
+//!
+//! A batch on the wire is `[header][statistics (V1 only)][change types][records
+//! data]`, where the records data is an Arrow IPC payload that is encoded and
+//! decoded in [`super::arrow`]. The versions here (`LOG_MAGIC_VALUE_*`) are
+//! versions of this Fluss framing and are unrelated to the Arrow IPC format
+//! version.
+
+use crate::error::{Error, Result};
+use crate::record::arrow::{ArrowLogRecordIterator, ArrowReader, ReadContext};
+use crate::record::{ChangeType, ScanRecord};
+use arrow::array::RecordBatch;
+use byteorder::{ByteOrder, LittleEndian};
+use bytes::Bytes;
+use crc32c::crc32c;
+use std::fs::File;
+use std::io::{Read, Seek, SeekFrom};
+use std::path::PathBuf;
+use std::sync::Arc;
+
+/// const for record batch
+pub const BASE_OFFSET_LENGTH: usize = 8;
+pub const LENGTH_LENGTH: usize = 4;
+pub const MAGIC_LENGTH: usize = 1;
+pub const COMMIT_TIMESTAMP_LENGTH: usize = 8;
+pub const CRC_LENGTH: usize = 4;
+pub const SCHEMA_ID_LENGTH: usize = 2;
+pub const ATTRIBUTE_LENGTH: usize = 1;
+pub const LAST_OFFSET_DELTA_LENGTH: usize = 4;
+pub const WRITE_CLIENT_ID_LENGTH: usize = 8;
+pub const BATCH_SEQUENCE_LENGTH: usize = 4;
+pub const RECORDS_COUNT_LENGTH: usize = 4;
+
+pub const BASE_OFFSET_OFFSET: usize = 0;
+pub const LENGTH_OFFSET: usize = BASE_OFFSET_OFFSET + BASE_OFFSET_LENGTH;
+pub const MAGIC_OFFSET: usize = LENGTH_OFFSET + LENGTH_LENGTH;
+pub const COMMIT_TIMESTAMP_OFFSET: usize = MAGIC_OFFSET + MAGIC_LENGTH;
+pub const CRC_OFFSET: usize = COMMIT_TIMESTAMP_OFFSET + COMMIT_TIMESTAMP_LENGTH;
+pub const SCHEMA_ID_OFFSET: usize = CRC_OFFSET + CRC_LENGTH;
+pub const ATTRIBUTES_OFFSET: usize = SCHEMA_ID_OFFSET + SCHEMA_ID_LENGTH;
+pub const LAST_OFFSET_DELTA_OFFSET: usize = ATTRIBUTES_OFFSET + ATTRIBUTE_LENGTH;
+pub const WRITE_CLIENT_ID_OFFSET: usize = LAST_OFFSET_DELTA_OFFSET + LAST_OFFSET_DELTA_LENGTH;
+pub const BATCH_SEQUENCE_OFFSET: usize = WRITE_CLIENT_ID_OFFSET + WRITE_CLIENT_ID_LENGTH;
+pub const RECORDS_COUNT_OFFSET: usize = BATCH_SEQUENCE_OFFSET + BATCH_SEQUENCE_LENGTH;
+pub const RECORDS_OFFSET: usize = RECORDS_COUNT_OFFSET + RECORDS_COUNT_LENGTH;
+
+pub const RECORD_BATCH_HEADER_SIZE: usize = RECORDS_OFFSET;
+pub const ARROW_CHANGETYPE_OFFSET: usize = RECORD_BATCH_HEADER_SIZE;
+pub const LOG_OVERHEAD: usize = LENGTH_OFFSET + LENGTH_LENGTH;
+
+pub const STATISTICS_LENGTH_LENGTH: usize = 4;
+/// V1 keeps the whole V0 header layout and appends a statistics length field,
+/// so every offset up to the records count is shared between the two versions.
+pub const V1_STATISTICS_LENGTH_OFFSET: usize = RECORDS_COUNT_OFFSET + RECORDS_COUNT_LENGTH;
+pub const V1_STATISTICS_DATA_OFFSET: usize = V1_STATISTICS_LENGTH_OFFSET + STATISTICS_LENGTH_LENGTH;
+pub const V1_RECORD_BATCH_HEADER_SIZE: usize = V1_STATISTICS_DATA_OFFSET;
+
+pub const LEADER_EPOCH_LENGTH: usize = 4;
+/// Leader epoch reported for batches whose magic predates V2, mirroring Java's
+/// `LogRecordBatchFormat.NO_LEADER_EPOCH`.
+pub const NO_LEADER_EPOCH: i32 = -1;
+
+/// V2 inserts a leader epoch between the commit timestamp and the CRC, so every
+/// field from the CRC onward sits [`LEADER_EPOCH_LENGTH`] bytes after its V1
+/// position; the statistics section and records data layout match V1.
+pub const V2_LEADER_EPOCH_OFFSET: usize = COMMIT_TIMESTAMP_OFFSET + COMMIT_TIMESTAMP_LENGTH;
+pub const V2_STATISTICS_LENGTH_OFFSET: usize = V1_STATISTICS_LENGTH_OFFSET + LEADER_EPOCH_LENGTH;
+pub const V2_STATISTICS_DATA_OFFSET: usize = V2_STATISTICS_LENGTH_OFFSET + STATISTICS_LENGTH_LENGTH;
+pub const V2_RECORD_BATCH_HEADER_SIZE: usize = V2_STATISTICS_DATA_OFFSET;
+
+/// Bit 0 of the attributes byte. When set, the batch is append-only and carries
+/// no change-type vector; when clear, a `record_count`-byte change-type vector
+/// precedes the Arrow IPC payload (the changelog of a primary-key table). Shares
+/// the wire layout of the Java client's `DefaultLogRecordBatch`.
+pub const APPEND_ONLY_FLAG_MASK: u8 = 0x01;
+
+/// Maximum batch size matches Java's Integer.MAX_VALUE limit.
+/// Java uses int type for batch size, so max value is 2^31 - 1 = 2,147,483,647 bytes (~2GB).
+/// This is the implicit limit in FileLogRecords.java and other Java components.
+pub const MAX_BATCH_SIZE: usize = i32::MAX as usize; // 2,147,483,647 bytes (~2GB)
+
+/// const for record
+/// The "magic" values.
+#[derive(Debug, Clone, Copy)]
+pub enum LogMagicValue {
+ V0 = 0,
+ /// V1 places a statistics section between the fixed header and the records
+ /// so the server can prune whole batches against a pushed-down filter.
+ V1 = 1,
+ /// V2 adds a leader epoch to the header so followers can build a
+ /// consistent leader epoch cache for log truncation on failover.
+ V2 = 2,
+}
+
+pub const LOG_MAGIC_VALUE_V0: u8 = LogMagicValue::V0 as u8;
+pub const LOG_MAGIC_VALUE_V1: u8 = LogMagicValue::V1 as u8;
+pub const LOG_MAGIC_VALUE_V2: u8 = LogMagicValue::V2 as u8;
+
+/// Fixed header size for the given magic, mirroring Java's
+/// `LogRecordBatchFormat.recordBatchHeaderSize`.
+pub fn record_batch_header_size(magic: u8) -> Result<usize> {
+ match magic {
+ LOG_MAGIC_VALUE_V0 => Ok(RECORD_BATCH_HEADER_SIZE),
+ LOG_MAGIC_VALUE_V1 => Ok(V1_RECORD_BATCH_HEADER_SIZE),
+ LOG_MAGIC_VALUE_V2 => Ok(V2_RECORD_BATCH_HEADER_SIZE),
+ _ => Err(Error::UnexpectedError {
+ message: format!("Unsupported magic value {magic}"),
+ source: None,
+ }),
+ }
+}
+
+/// Safely convert batch size from i32 to usize with validation.
+///
+/// Validates that:
+/// - batch_size_bytes is non-negative
+/// - batch_size_bytes + LOG_OVERHEAD doesn't overflow
+/// - Result is within reasonable bounds
+fn validate_batch_size(batch_size_bytes: i32) -> Result<usize> {
+ // Check for negative size (corrupted data)
+ if batch_size_bytes < 0 {
+ return Err(Error::UnexpectedError {
+ message: format!("Invalid negative batch size: {batch_size_bytes}"),
+ source: None,
+ });
+ }
+
+ let batch_size_u = batch_size_bytes as usize;
+
+ // Check for overflow when adding LOG_OVERHEAD
+ let total_size =
+ batch_size_u
+ .checked_add(LOG_OVERHEAD)
+ .ok_or_else(|| Error::UnexpectedError {
+ message: format!(
+ "Batch size {batch_size_u} + LOG_OVERHEAD {LOG_OVERHEAD} would overflow"
+ ),
+ source: None,
+ })?;
+
+ // Sanity check: reject unreasonably large batches
+ if total_size > MAX_BATCH_SIZE {
+ return Err(Error::UnexpectedError {
+ message: format!(
+ "Batch size {total_size} exceeds maximum allowed size {MAX_BATCH_SIZE}"
+ ),
+ source: None,
+ });
+ }
+
+ Ok(total_size)
+}
+
+pub const CURRENT_LOG_MAGIC_VALUE: u8 = LOG_MAGIC_VALUE_V0;
+
+/// Value used if writer ID is not available or non-idempotent.
+pub const NO_WRITER_ID: i64 = -1;
+
+/// Value used if batch sequence is not available.
+pub const NO_BATCH_SEQUENCE: i32 = -1;
+
+/// In-memory log record source.
+/// Used for local tablet server fetches (existing path).
+struct MemorySource {
+ data: Bytes,
+}
+
+impl MemorySource {
+ fn new(data: Vec<u8>) -> Self {
+ Self {
+ data: Bytes::from(data),
+ }
+ }
+
+ fn read_batch_header(&mut self, pos: usize) -> Result<(i64, usize)> {
+ if pos + LOG_OVERHEAD > self.data.len() {
+ return Err(Error::UnexpectedError {
+ message: format!(
+ "Position {} + LOG_OVERHEAD {} exceeds data size {}",
+ pos,
+ LOG_OVERHEAD,
+ self.data.len()
+ ),
+ source: None,
+ });
+ }
+
+ let base_offset = LittleEndian::read_i64(&self.data[pos + BASE_OFFSET_OFFSET..]);
+ let batch_size_bytes = LittleEndian::read_i32(&self.data[pos + LENGTH_OFFSET..]);
+
+ // Validate batch size to prevent integer overflow and corruption
+ let batch_size = validate_batch_size(batch_size_bytes)?;
+
+ Ok((base_offset, batch_size))
+ }
+
+ fn read_batch_data(&mut self, pos: usize, size: usize) -> Result<Bytes> {
+ if pos + size > self.data.len() {
+ return Err(Error::UnexpectedError {
+ message: format!(
+ "Read beyond data size: {} + {} > {}",
+ pos,
+ size,
+ self.data.len()
+ ),
+ source: None,
+ });
+ }
+ // Zero-copy slice (Bytes is Arc-based)
+ Ok(self.data.slice(pos..pos + size))
+ }
+
+ fn total_size(&self) -> usize {
+ self.data.len()
+ }
+}
+
+/// RAII guard that deletes a file when dropped.
+/// Used to ensure file deletion happens AFTER the file handle is closed.
+struct FileCleanupGuard {
+ file_path: PathBuf,
+}
+
+impl Drop for FileCleanupGuard {
+ fn drop(&mut self) {
+ // File handle is already closed (this guard drops after the file field)
+ if let Err(e) = std::fs::remove_file(&self.file_path) {
+ log::warn!(
+ "Failed to delete remote log file {}: {}",
+ self.file_path.display(),
+ e
+ );
+ } else {
+ log::debug!("Deleted remote log file: {}", self.file_path.display());
+ }
+ }
+}
+
+/// File-backed log record source.
+/// Used for remote log segments downloaded to local disk.
+/// Streams data on-demand instead of loading entire file into memory.
+///
+/// Uses seek + read_exact for cross-platform compatibility.
+/// Access pattern is sequential iteration (single consumer).
+struct FileSource {
+ file: File,
+ file_size: usize,
+ base_offset: usize,
+ _cleanup: Option<FileCleanupGuard>, // Drops AFTER file (field order matters!)
+}
+
+impl FileSource {
+ /// Create a new FileSource.
+ ///
+ /// The file at `file_path` will be deleted when this FileSource is dropped.
+ fn new(file: File, base_offset: usize, file_path: PathBuf) -> Result<Self> {
+ let file_size = file.metadata()?.len() as usize;
+
+ // Validate base_offset to prevent underflow in total_size()
+ if base_offset > file_size {
+ return Err(Error::UnexpectedError {
+ message: format!("base_offset ({base_offset}) exceeds file_size ({file_size})"),
+ source: None,
+ });
+ }
+
+ Ok(Self {
+ file,
+ file_size,
+ base_offset,
+ _cleanup: Some(FileCleanupGuard { file_path }),
+ })
+ }
+
+ /// Read data at a specific position using seek + read_exact.
+ /// This is cross-platform and adequate for sequential access patterns.
+ fn read_at(&mut self, pos: u64, buf: &mut [u8]) -> Result<()> {
+ self.file.seek(SeekFrom::Start(pos))?;
+ self.file.read_exact(buf)?;
+ Ok(())
+ }
+
+ fn read_batch_header(&mut self, pos: usize) -> Result<(i64, usize)> {
+ let actual_pos = self.base_offset + pos;
+ if actual_pos + LOG_OVERHEAD > self.file_size {
+ return Err(Error::UnexpectedError {
+ message: format!(
+ "Position {} exceeds file size {}",
+ actual_pos, self.file_size
+ ),
+ source: None,
+ });
+ }
+
+ // Read only the header to extract base_offset and batch_size
+ let mut header_buf = vec![0u8; LOG_OVERHEAD];
+ self.read_at(actual_pos as u64, &mut header_buf)?;
+
+ let base_offset = LittleEndian::read_i64(&header_buf[BASE_OFFSET_OFFSET..]);
+ let batch_size_bytes = LittleEndian::read_i32(&header_buf[LENGTH_OFFSET..]);
+
+ // Validate batch size to prevent integer overflow and corruption
+ let batch_size = validate_batch_size(batch_size_bytes)?;
+
+ Ok((base_offset, batch_size))
+ }
+
+ fn read_batch_data(&mut self, pos: usize, size: usize) -> Result<Bytes> {
+ let actual_pos = self.base_offset + pos;
+ if actual_pos + size > self.file_size {
+ return Err(Error::UnexpectedError {
+ message: format!(
+ "Read beyond file size: {} + {} > {}",
+ actual_pos, size, self.file_size
+ ),
+ source: None,
+ });
+ }
+
+ // Read the full batch data
+ let mut batch_buf = vec![0u8; size];
+ self.read_at(actual_pos as u64, &mut batch_buf)?;
+
+ Ok(Bytes::from(batch_buf))
+ }
+
+ fn total_size(&self) -> usize {
+ self.file_size - self.base_offset
+ }
+}
+
+/// Enum for different log record sources.
+enum LogRecordsSource {
+ Memory(MemorySource),
+ File(FileSource),
+}
+
+impl LogRecordsSource {
+ fn read_batch_header(&mut self, pos: usize) -> Result<(i64, usize)> {
+ match self {
+ Self::Memory(s) => s.read_batch_header(pos),
+ Self::File(s) => s.read_batch_header(pos),
+ }
+ }
+
+ fn read_batch_data(&mut self, pos: usize, size: usize) -> Result<Bytes> {
+ match self {
+ Self::Memory(s) => s.read_batch_data(pos, size),
+ Self::File(s) => s.read_batch_data(pos, size),
+ }
+ }
+
+ fn total_size(&self) -> usize {
+ match self {
+ Self::Memory(s) => s.total_size(),
+ Self::File(s) => s.total_size(),
+ }
+ }
+}
+
+pub struct LogRecordsBatches {
+ source: LogRecordsSource,
+ current_pos: usize,
+ remaining_bytes: usize,
+}
+
+impl LogRecordsBatches {
+ /// Create from in-memory Vec (existing path - backward compatible).
+ pub fn new(data: Vec<u8>) -> Self {
+ let source = LogRecordsSource::Memory(MemorySource::new(data));
+ let remaining_bytes = source.total_size();
+ Self {
+ source,
+ current_pos: 0,
+ remaining_bytes,
+ }
+ }
+
+ /// Create from file.
+ /// Enables streaming without loading entire file into memory.
+ ///
+ /// The file at `file_path` will be deleted when dropped.
+ /// This ensures the file is closed before deletion.
+ pub fn from_file(file: File, base_offset: usize, file_path: PathBuf) -> Result<Self> {
+ let source = FileSource::new(file, base_offset, file_path)?;
+ let remaining_bytes = source.total_size();
+ Ok(Self {
+ source: LogRecordsSource::File(source),
+ current_pos: 0,
+ remaining_bytes,
+ })
+ }
+
+ /// Try to get the size of the next batch.
+ fn next_batch_size(&mut self) -> Result<Option<usize>> {
+ if self.remaining_bytes < LOG_OVERHEAD {
+ return Ok(None);
+ }
+
+ // Read only header to get size
+ match self.source.read_batch_header(self.current_pos) {
+ Ok((_base_offset, batch_size)) => {
+ if batch_size > self.remaining_bytes {
+ Ok(None)
+ } else {
+ Ok(Some(batch_size))
+ }
+ }
+ Err(e) => Err(e),
+ }
+ }
+}
+
+impl Iterator for LogRecordsBatches {
+ type Item = Result<LogRecordBatch>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ match self.next_batch_size() {
+ Ok(Some(batch_size)) => {
+ // Read full batch data on-demand
+ match self.source.read_batch_data(self.current_pos, batch_size) {
+ Ok(data) => {
+ let record_batch = LogRecordBatch::new(data);
+ self.current_pos += batch_size;
+ self.remaining_bytes -= batch_size;
+ Some(Ok(record_batch))
+ }
+ Err(e) => Some(Err(e)),
+ }
+ }
+ Ok(None) => None,
+ Err(e) => Some(Err(e)),
+ }
+ }
+}
+
+pub struct LogRecordBatch {
+ data: Bytes,
+}
+
+#[allow(dead_code)]
+impl LogRecordBatch {
+ pub fn new(data: Bytes) -> Self {
+ LogRecordBatch { data }
+ }
+
+ pub fn magic(&self) -> u8 {
+ self.data[MAGIC_OFFSET]
+ }
+
+ /// Byte shift of every header field at or after the CRC, relative to the
+ /// V0/V1 layout: V2 inserts the leader epoch before the CRC. Decode entry
+ /// points gate on [`record_batch_header_size`] first, so unknown magics
+ /// never reach the accessors that use this.
+ fn header_field_shift(&self) -> usize {
+ if self.magic() >= LOG_MAGIC_VALUE_V2 {
+ LEADER_EPOCH_LENGTH
+ } else {
+ 0
+ }
+ }
+
+ pub fn commit_timestamp(&self) -> i64 {
+ let offset = COMMIT_TIMESTAMP_OFFSET;
+ LittleEndian::read_i64(&self.data[offset..offset + COMMIT_TIMESTAMP_LENGTH])
+ }
+
+ /// The epoch of the leader that wrote this batch, or [`NO_LEADER_EPOCH`]
+ /// for magics before V2.
+ pub fn leader_epoch(&self) -> i32 {
Review Comment:
Well caught, description is wrong. The code that allows setting of leader epoch is there but nothing calls it yet. Updated descriptions / comments
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Wire batch statistics into the log batch write and read paths [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #4074:
URL: https://github.com/apache/fluss/pull/4074#issuecomment-5398961138
TY for the thorough review as usual @fresh-borzoni , addressed your comments. PTAL 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Wire batch statistics into the log batch write and read paths [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4074:
URL: https://github.com/apache/fluss/pull/4074#discussion_r3850402687
##########
fluss-rust/crates/fluss/src/record/arrow.rs:
##########
@@ -188,6 +89,22 @@ pub struct MemoryLogRecordsArrowBuilder {
/// Matching Java's `ArrowWriter.estimatedCompressionRatio` which is
/// cached per batch and only refreshed on `reset()`.
estimated_compression_ratio: f32,
+ /// Statistics columns with their row type; `Some` upgrades the batch to V1.
+ statistics: Option<(RowType, Vec<usize>)>,
+ /// Per-schema estimate of the serialized statistics size.
+ estimated_statistics_size: usize,
+}
+
+/// Table-level configuration for building Arrow log batches, shared by
+/// `ArrowLogWriteBatch` and `MemoryLogRecordsArrowBuilder`.
+pub struct ArrowBatchConfig {
Review Comment:
Agree 👍 .
Two things: MemoryLogRecordsArrowBuilder::new takes it, so the builder has to go pub(crate) too, and the rest of pub use arrow::* can't follow as the python binding uses fluss::record::to_arrow_schema.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Wire batch statistics into the log batch write and read paths [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction merged PR #4074:
URL: https://github.com/apache/fluss/pull/4074
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Wire batch statistics into the log batch write and read paths [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4074:
URL: https://github.com/apache/fluss/pull/4074#discussion_r3844346675
##########
fluss-rust/crates/fluss/src/row/aligned/mod.rs:
##########
Review Comment:
This module grows another export while staying pub. Make it pub(crate) now?
##########
fluss-rust/crates/fluss/src/record/arrow.rs:
##########
@@ -508,14 +459,35 @@ impl MemoryLogRecordsArrowBuilder {
.update_estimation(actual_ratio);
}
- // now, write batch header and arrow batch
- let mut batch_bytes = vec![0u8; RECORD_BATCH_HEADER_SIZE + real_arrow_batch_bytes.len()];
+ let statistics_bytes = match &self.statistics {
+ Some((row_type, mapping)) => {
+ match serialize_statistics(record_batch.as_ref(), row_type, mapping) {
+ Ok(Some(bytes)) => bytes,
+ // Unlike Java's 27-byte empty block, an empty mapping
+ // writes length 0, which both parsers read as no statistics.
+ Ok(None) => Vec::new(),
+ // A failure degrades to an empty section rather than
+ // failing the batch, matching Java's builder.
+ Err(error) => {
+ log::error!("Failed to serialize statistics for record batch: {error}");
+ Vec::new()
+ }
+ }
+ }
+ None => Vec::new(),
+ };
+
+ // now, write batch header, statistics and arrow batch
+ let header_size = self.header_size();
+ let mut batch_bytes =
+ vec![0u8; header_size + statistics_bytes.len() + real_arrow_batch_bytes.len()];
// write batch header
- self.write_batch_header(&mut batch_bytes[..])?;
+ self.write_batch_header(&mut batch_bytes[..], statistics_bytes.len())?;
- // write arrow batch bytes
+ // write statistics and arrow batch bytes
let mut cursor = Cursor::new(&mut batch_bytes[..]);
- cursor.set_position(RECORD_BATCH_HEADER_SIZE as u64);
+ cursor.set_position(header_size as u64);
+ cursor.write_all(&statistics_bytes)?;
cursor.write_all(real_arrow_batch_bytes)?;
let calcute_crc_bytes = &cursor.get_ref()[SCHEMA_ID_OFFSET..];
Review 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?
##########
fluss-rust/crates/fluss/src/record/log_record_batch.rs:
##########
@@ -0,0 +1,1296 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Fluss' log record batch wire format, mirroring Java's
+//! `LogRecordBatchFormat` / `DefaultLogRecordBatch`.
+//!
+//! A batch on the wire is `[header][statistics (V1 only)][change types][records
+//! data]`, where the records data is an Arrow IPC payload that is encoded and
+//! decoded in [`super::arrow`]. The versions here (`LOG_MAGIC_VALUE_*`) are
+//! versions of this Fluss framing and are unrelated to the Arrow IPC format
+//! version.
+
+use crate::error::{Error, Result};
+use crate::record::arrow::{ArrowLogRecordIterator, ArrowReader, ReadContext};
+use crate::record::{ChangeType, ScanRecord};
+use arrow::array::RecordBatch;
+use byteorder::{ByteOrder, LittleEndian};
+use bytes::Bytes;
+use crc32c::crc32c;
+use std::fs::File;
+use std::io::{Read, Seek, SeekFrom};
+use std::path::PathBuf;
+use std::sync::Arc;
+
+/// const for record batch
+pub const BASE_OFFSET_LENGTH: usize = 8;
+pub const LENGTH_LENGTH: usize = 4;
+pub const MAGIC_LENGTH: usize = 1;
+pub const COMMIT_TIMESTAMP_LENGTH: usize = 8;
+pub const CRC_LENGTH: usize = 4;
+pub const SCHEMA_ID_LENGTH: usize = 2;
+pub const ATTRIBUTE_LENGTH: usize = 1;
+pub const LAST_OFFSET_DELTA_LENGTH: usize = 4;
+pub const WRITE_CLIENT_ID_LENGTH: usize = 8;
+pub const BATCH_SEQUENCE_LENGTH: usize = 4;
+pub const RECORDS_COUNT_LENGTH: usize = 4;
+
+pub const BASE_OFFSET_OFFSET: usize = 0;
+pub const LENGTH_OFFSET: usize = BASE_OFFSET_OFFSET + BASE_OFFSET_LENGTH;
+pub const MAGIC_OFFSET: usize = LENGTH_OFFSET + LENGTH_LENGTH;
+pub const COMMIT_TIMESTAMP_OFFSET: usize = MAGIC_OFFSET + MAGIC_LENGTH;
+pub const CRC_OFFSET: usize = COMMIT_TIMESTAMP_OFFSET + COMMIT_TIMESTAMP_LENGTH;
+pub const SCHEMA_ID_OFFSET: usize = CRC_OFFSET + CRC_LENGTH;
+pub const ATTRIBUTES_OFFSET: usize = SCHEMA_ID_OFFSET + SCHEMA_ID_LENGTH;
+pub const LAST_OFFSET_DELTA_OFFSET: usize = ATTRIBUTES_OFFSET + ATTRIBUTE_LENGTH;
+pub const WRITE_CLIENT_ID_OFFSET: usize = LAST_OFFSET_DELTA_OFFSET + LAST_OFFSET_DELTA_LENGTH;
+pub const BATCH_SEQUENCE_OFFSET: usize = WRITE_CLIENT_ID_OFFSET + WRITE_CLIENT_ID_LENGTH;
+pub const RECORDS_COUNT_OFFSET: usize = BATCH_SEQUENCE_OFFSET + BATCH_SEQUENCE_LENGTH;
+pub const RECORDS_OFFSET: usize = RECORDS_COUNT_OFFSET + RECORDS_COUNT_LENGTH;
+
+pub const RECORD_BATCH_HEADER_SIZE: usize = RECORDS_OFFSET;
+pub const ARROW_CHANGETYPE_OFFSET: usize = RECORD_BATCH_HEADER_SIZE;
+pub const LOG_OVERHEAD: usize = LENGTH_OFFSET + LENGTH_LENGTH;
+
+pub const STATISTICS_LENGTH_LENGTH: usize = 4;
+/// V1 keeps the whole V0 header layout and appends a statistics length field,
+/// so every offset up to the records count is shared between the two versions.
+pub const V1_STATISTICS_LENGTH_OFFSET: usize = RECORDS_COUNT_OFFSET + RECORDS_COUNT_LENGTH;
+pub const V1_STATISTICS_DATA_OFFSET: usize = V1_STATISTICS_LENGTH_OFFSET + STATISTICS_LENGTH_LENGTH;
+pub const V1_RECORD_BATCH_HEADER_SIZE: usize = V1_STATISTICS_DATA_OFFSET;
+
+pub const LEADER_EPOCH_LENGTH: usize = 4;
+/// Leader epoch reported for batches whose magic predates V2, mirroring Java's
+/// `LogRecordBatchFormat.NO_LEADER_EPOCH`.
+pub const NO_LEADER_EPOCH: i32 = -1;
+
+/// V2 inserts a leader epoch between the commit timestamp and the CRC, so every
+/// field from the CRC onward sits [`LEADER_EPOCH_LENGTH`] bytes after its V1
+/// position; the statistics section and records data layout match V1.
+pub const V2_LEADER_EPOCH_OFFSET: usize = COMMIT_TIMESTAMP_OFFSET + COMMIT_TIMESTAMP_LENGTH;
+pub const V2_STATISTICS_LENGTH_OFFSET: usize = V1_STATISTICS_LENGTH_OFFSET + LEADER_EPOCH_LENGTH;
+pub const V2_STATISTICS_DATA_OFFSET: usize = V2_STATISTICS_LENGTH_OFFSET + STATISTICS_LENGTH_LENGTH;
+pub const V2_RECORD_BATCH_HEADER_SIZE: usize = V2_STATISTICS_DATA_OFFSET;
+
+/// Bit 0 of the attributes byte. When set, the batch is append-only and carries
+/// no change-type vector; when clear, a `record_count`-byte change-type vector
+/// precedes the Arrow IPC payload (the changelog of a primary-key table). Shares
+/// the wire layout of the Java client's `DefaultLogRecordBatch`.
+pub const APPEND_ONLY_FLAG_MASK: u8 = 0x01;
+
+/// Maximum batch size matches Java's Integer.MAX_VALUE limit.
+/// Java uses int type for batch size, so max value is 2^31 - 1 = 2,147,483,647 bytes (~2GB).
+/// This is the implicit limit in FileLogRecords.java and other Java components.
+pub const MAX_BATCH_SIZE: usize = i32::MAX as usize; // 2,147,483,647 bytes (~2GB)
+
+/// const for record
+/// The "magic" values.
+#[derive(Debug, Clone, Copy)]
+pub enum LogMagicValue {
+ V0 = 0,
+ /// V1 places a statistics section between the fixed header and the records
+ /// so the server can prune whole batches against a pushed-down filter.
+ V1 = 1,
+ /// V2 adds a leader epoch to the header so followers can build a
+ /// consistent leader epoch cache for log truncation on failover.
+ V2 = 2,
+}
+
+pub const LOG_MAGIC_VALUE_V0: u8 = LogMagicValue::V0 as u8;
+pub const LOG_MAGIC_VALUE_V1: u8 = LogMagicValue::V1 as u8;
+pub const LOG_MAGIC_VALUE_V2: u8 = LogMagicValue::V2 as u8;
+
+/// Fixed header size for the given magic, mirroring Java's
+/// `LogRecordBatchFormat.recordBatchHeaderSize`.
+pub fn record_batch_header_size(magic: u8) -> Result<usize> {
+ match magic {
+ LOG_MAGIC_VALUE_V0 => Ok(RECORD_BATCH_HEADER_SIZE),
+ LOG_MAGIC_VALUE_V1 => Ok(V1_RECORD_BATCH_HEADER_SIZE),
+ LOG_MAGIC_VALUE_V2 => Ok(V2_RECORD_BATCH_HEADER_SIZE),
+ _ => Err(Error::UnexpectedError {
+ message: format!("Unsupported magic value {magic}"),
+ source: None,
+ }),
+ }
+}
+
+/// Safely convert batch size from i32 to usize with validation.
+///
+/// Validates that:
+/// - batch_size_bytes is non-negative
+/// - batch_size_bytes + LOG_OVERHEAD doesn't overflow
+/// - Result is within reasonable bounds
+fn validate_batch_size(batch_size_bytes: i32) -> Result<usize> {
+ // Check for negative size (corrupted data)
+ if batch_size_bytes < 0 {
+ return Err(Error::UnexpectedError {
+ message: format!("Invalid negative batch size: {batch_size_bytes}"),
+ source: None,
+ });
+ }
+
+ let batch_size_u = batch_size_bytes as usize;
+
+ // Check for overflow when adding LOG_OVERHEAD
+ let total_size =
+ batch_size_u
+ .checked_add(LOG_OVERHEAD)
+ .ok_or_else(|| Error::UnexpectedError {
+ message: format!(
+ "Batch size {batch_size_u} + LOG_OVERHEAD {LOG_OVERHEAD} would overflow"
+ ),
+ source: None,
+ })?;
+
+ // Sanity check: reject unreasonably large batches
+ if total_size > MAX_BATCH_SIZE {
+ return Err(Error::UnexpectedError {
+ message: format!(
+ "Batch size {total_size} exceeds maximum allowed size {MAX_BATCH_SIZE}"
+ ),
+ source: None,
+ });
+ }
+
+ Ok(total_size)
+}
+
+pub const CURRENT_LOG_MAGIC_VALUE: u8 = LOG_MAGIC_VALUE_V0;
+
+/// Value used if writer ID is not available or non-idempotent.
+pub const NO_WRITER_ID: i64 = -1;
+
+/// Value used if batch sequence is not available.
+pub const NO_BATCH_SEQUENCE: i32 = -1;
+
+/// In-memory log record source.
+/// Used for local tablet server fetches (existing path).
+struct MemorySource {
+ data: Bytes,
+}
+
+impl MemorySource {
+ fn new(data: Vec<u8>) -> Self {
+ Self {
+ data: Bytes::from(data),
+ }
+ }
+
+ fn read_batch_header(&mut self, pos: usize) -> Result<(i64, usize)> {
+ if pos + LOG_OVERHEAD > self.data.len() {
+ return Err(Error::UnexpectedError {
+ message: format!(
+ "Position {} + LOG_OVERHEAD {} exceeds data size {}",
+ pos,
+ LOG_OVERHEAD,
+ self.data.len()
+ ),
+ source: None,
+ });
+ }
+
+ let base_offset = LittleEndian::read_i64(&self.data[pos + BASE_OFFSET_OFFSET..]);
+ let batch_size_bytes = LittleEndian::read_i32(&self.data[pos + LENGTH_OFFSET..]);
+
+ // Validate batch size to prevent integer overflow and corruption
+ let batch_size = validate_batch_size(batch_size_bytes)?;
+
+ Ok((base_offset, batch_size))
+ }
+
+ fn read_batch_data(&mut self, pos: usize, size: usize) -> Result<Bytes> {
+ if pos + size > self.data.len() {
+ return Err(Error::UnexpectedError {
+ message: format!(
+ "Read beyond data size: {} + {} > {}",
+ pos,
+ size,
+ self.data.len()
+ ),
+ source: None,
+ });
+ }
+ // Zero-copy slice (Bytes is Arc-based)
+ Ok(self.data.slice(pos..pos + size))
+ }
+
+ fn total_size(&self) -> usize {
+ self.data.len()
+ }
+}
+
+/// RAII guard that deletes a file when dropped.
+/// Used to ensure file deletion happens AFTER the file handle is closed.
+struct FileCleanupGuard {
+ file_path: PathBuf,
+}
+
+impl Drop for FileCleanupGuard {
+ fn drop(&mut self) {
+ // File handle is already closed (this guard drops after the file field)
+ if let Err(e) = std::fs::remove_file(&self.file_path) {
+ log::warn!(
+ "Failed to delete remote log file {}: {}",
+ self.file_path.display(),
+ e
+ );
+ } else {
+ log::debug!("Deleted remote log file: {}", self.file_path.display());
+ }
+ }
+}
+
+/// File-backed log record source.
+/// Used for remote log segments downloaded to local disk.
+/// Streams data on-demand instead of loading entire file into memory.
+///
+/// Uses seek + read_exact for cross-platform compatibility.
+/// Access pattern is sequential iteration (single consumer).
+struct FileSource {
+ file: File,
+ file_size: usize,
+ base_offset: usize,
+ _cleanup: Option<FileCleanupGuard>, // Drops AFTER file (field order matters!)
+}
+
+impl FileSource {
+ /// Create a new FileSource.
+ ///
+ /// The file at `file_path` will be deleted when this FileSource is dropped.
+ fn new(file: File, base_offset: usize, file_path: PathBuf) -> Result<Self> {
+ let file_size = file.metadata()?.len() as usize;
+
+ // Validate base_offset to prevent underflow in total_size()
+ if base_offset > file_size {
+ return Err(Error::UnexpectedError {
+ message: format!("base_offset ({base_offset}) exceeds file_size ({file_size})"),
+ source: None,
+ });
+ }
+
+ Ok(Self {
+ file,
+ file_size,
+ base_offset,
+ _cleanup: Some(FileCleanupGuard { file_path }),
+ })
+ }
+
+ /// Read data at a specific position using seek + read_exact.
+ /// This is cross-platform and adequate for sequential access patterns.
+ fn read_at(&mut self, pos: u64, buf: &mut [u8]) -> Result<()> {
+ self.file.seek(SeekFrom::Start(pos))?;
+ self.file.read_exact(buf)?;
+ Ok(())
+ }
+
+ fn read_batch_header(&mut self, pos: usize) -> Result<(i64, usize)> {
+ let actual_pos = self.base_offset + pos;
+ if actual_pos + LOG_OVERHEAD > self.file_size {
+ return Err(Error::UnexpectedError {
+ message: format!(
+ "Position {} exceeds file size {}",
+ actual_pos, self.file_size
+ ),
+ source: None,
+ });
+ }
+
+ // Read only the header to extract base_offset and batch_size
+ let mut header_buf = vec![0u8; LOG_OVERHEAD];
+ self.read_at(actual_pos as u64, &mut header_buf)?;
+
+ let base_offset = LittleEndian::read_i64(&header_buf[BASE_OFFSET_OFFSET..]);
+ let batch_size_bytes = LittleEndian::read_i32(&header_buf[LENGTH_OFFSET..]);
+
+ // Validate batch size to prevent integer overflow and corruption
+ let batch_size = validate_batch_size(batch_size_bytes)?;
+
+ Ok((base_offset, batch_size))
+ }
+
+ fn read_batch_data(&mut self, pos: usize, size: usize) -> Result<Bytes> {
+ let actual_pos = self.base_offset + pos;
+ if actual_pos + size > self.file_size {
+ return Err(Error::UnexpectedError {
+ message: format!(
+ "Read beyond file size: {} + {} > {}",
+ actual_pos, size, self.file_size
+ ),
+ source: None,
+ });
+ }
+
+ // Read the full batch data
+ let mut batch_buf = vec![0u8; size];
+ self.read_at(actual_pos as u64, &mut batch_buf)?;
+
+ Ok(Bytes::from(batch_buf))
+ }
+
+ fn total_size(&self) -> usize {
+ self.file_size - self.base_offset
+ }
+}
+
+/// Enum for different log record sources.
+enum LogRecordsSource {
+ Memory(MemorySource),
+ File(FileSource),
+}
+
+impl LogRecordsSource {
+ fn read_batch_header(&mut self, pos: usize) -> Result<(i64, usize)> {
+ match self {
+ Self::Memory(s) => s.read_batch_header(pos),
+ Self::File(s) => s.read_batch_header(pos),
+ }
+ }
+
+ fn read_batch_data(&mut self, pos: usize, size: usize) -> Result<Bytes> {
+ match self {
+ Self::Memory(s) => s.read_batch_data(pos, size),
+ Self::File(s) => s.read_batch_data(pos, size),
+ }
+ }
+
+ fn total_size(&self) -> usize {
+ match self {
+ Self::Memory(s) => s.total_size(),
+ Self::File(s) => s.total_size(),
+ }
+ }
+}
+
+pub struct LogRecordsBatches {
+ source: LogRecordsSource,
+ current_pos: usize,
+ remaining_bytes: usize,
+}
+
+impl LogRecordsBatches {
+ /// Create from in-memory Vec (existing path - backward compatible).
+ pub fn new(data: Vec<u8>) -> Self {
+ let source = LogRecordsSource::Memory(MemorySource::new(data));
+ let remaining_bytes = source.total_size();
+ Self {
+ source,
+ current_pos: 0,
+ remaining_bytes,
+ }
+ }
+
+ /// Create from file.
+ /// Enables streaming without loading entire file into memory.
+ ///
+ /// The file at `file_path` will be deleted when dropped.
+ /// This ensures the file is closed before deletion.
+ pub fn from_file(file: File, base_offset: usize, file_path: PathBuf) -> Result<Self> {
+ let source = FileSource::new(file, base_offset, file_path)?;
+ let remaining_bytes = source.total_size();
+ Ok(Self {
+ source: LogRecordsSource::File(source),
+ current_pos: 0,
+ remaining_bytes,
+ })
+ }
+
+ /// Try to get the size of the next batch.
+ fn next_batch_size(&mut self) -> Result<Option<usize>> {
+ if self.remaining_bytes < LOG_OVERHEAD {
+ return Ok(None);
+ }
+
+ // Read only header to get size
+ match self.source.read_batch_header(self.current_pos) {
+ Ok((_base_offset, batch_size)) => {
+ if batch_size > self.remaining_bytes {
+ Ok(None)
+ } else {
+ Ok(Some(batch_size))
+ }
+ }
+ Err(e) => Err(e),
+ }
+ }
+}
+
+impl Iterator for LogRecordsBatches {
+ type Item = Result<LogRecordBatch>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ match self.next_batch_size() {
+ Ok(Some(batch_size)) => {
+ // Read full batch data on-demand
+ match self.source.read_batch_data(self.current_pos, batch_size) {
+ Ok(data) => {
+ let record_batch = LogRecordBatch::new(data);
+ self.current_pos += batch_size;
+ self.remaining_bytes -= batch_size;
+ Some(Ok(record_batch))
+ }
+ Err(e) => Some(Err(e)),
+ }
+ }
+ Ok(None) => None,
+ Err(e) => Some(Err(e)),
+ }
+ }
+}
+
+pub struct LogRecordBatch {
+ data: Bytes,
+}
+
+#[allow(dead_code)]
+impl LogRecordBatch {
+ pub fn new(data: Bytes) -> Self {
+ LogRecordBatch { data }
+ }
+
+ pub fn magic(&self) -> u8 {
+ self.data[MAGIC_OFFSET]
+ }
+
+ /// Byte shift of every header field at or after the CRC, relative to the
+ /// V0/V1 layout: V2 inserts the leader epoch before the CRC. Decode entry
+ /// points gate on [`record_batch_header_size`] first, so unknown magics
+ /// never reach the accessors that use this.
+ fn header_field_shift(&self) -> usize {
+ if self.magic() >= LOG_MAGIC_VALUE_V2 {
+ LEADER_EPOCH_LENGTH
+ } else {
+ 0
+ }
+ }
+
+ pub fn commit_timestamp(&self) -> i64 {
+ let offset = COMMIT_TIMESTAMP_OFFSET;
+ LittleEndian::read_i64(&self.data[offset..offset + COMMIT_TIMESTAMP_LENGTH])
+ }
+
+ /// The epoch of the leader that wrote this batch, or [`NO_LEADER_EPOCH`]
+ /// for magics before V2.
+ pub fn leader_epoch(&self) -> i32 {
Review 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.
##########
fluss-rust/crates/fluss/src/record/statistics.rs:
##########
@@ -921,4 +1085,137 @@ mod tests {
assert_eq!(i64::from_le_bytes(min[8..16].try_into().unwrap()), 1_000);
assert_eq!(i64::from_le_bytes(max[8..16].try_into().unwrap()), 2_000);
}
+
+ /// 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 {
Review 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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Wire batch statistics into the log batch write and read paths [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #4074:
URL: https://github.com/apache/fluss/pull/4074#discussion_r3845905151
##########
fluss-rust/crates/fluss/src/record/statistics.rs:
##########
@@ -921,4 +1085,137 @@ mod tests {
assert_eq!(i64::from_le_bytes(min[8..16].try_into().unwrap()), 1_000);
assert_eq!(i64::from_le_bytes(max[8..16].try_into().unwrap()), 2_000);
}
+
+ /// 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 {
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Wire batch statistics into the log batch write and read paths [fluss]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on code in PR #4074:
URL: https://github.com/apache/fluss/pull/4074#discussion_r3846567740
##########
fluss-rust/crates/fluss/src/record/arrow.rs:
##########
@@ -188,6 +89,22 @@ pub struct MemoryLogRecordsArrowBuilder {
/// Matching Java's `ArrowWriter.estimatedCompressionRatio` which is
/// cached per batch and only refreshed on `reset()`.
estimated_compression_ratio: f32,
+ /// Statistics columns with their row type; `Some` upgrades the batch to V1.
+ statistics: Option<(RowType, Vec<usize>)>,
+ /// Per-schema estimate of the serialized statistics size.
+ estimated_statistics_size: usize,
+}
+
+/// Table-level configuration for building Arrow log batches, shared by
+/// `ArrowLogWriteBatch` and `MemoryLogRecordsArrowBuilder`.
+pub struct ArrowBatchConfig {
Review Comment:
i think this is only builder plumbing, should be pub(crate)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org