diff --git a/fluss-common/src/test/java/org/apache/fluss/record/LogRecordBatchStatisticsCompatibilityTest.java b/fluss-common/src/test/java/org/apache/fluss/record/LogRecordBatchStatisticsCompatibilityTest.java new file mode 100644 index 00000000000..5c1635bf8a5 --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/record/LogRecordBatchStatisticsCompatibilityTest.java @@ -0,0 +1,239 @@ +/* + * 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. + */ + +package org.apache.fluss.record; + +import org.apache.fluss.memory.MemorySegment; +import org.apache.fluss.memory.MemorySegmentOutputView; +import org.apache.fluss.row.Decimal; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.row.TimestampLtz; +import org.apache.fluss.row.TimestampNtz; +import org.apache.fluss.testutils.DataTestUtils; +import org.apache.fluss.types.DataField; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.types.RowType; + +import org.junit.jupiter.api.Test; + +import java.io.InputStream; +import java.math.BigDecimal; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Locks the serialized statistics block against the checked-in reference so that other language + * clients (e.g. fluss-rust) can verify byte-for-byte compatibility. The reference lives in {@code + * src/test/resources/encoding/statistics_block.hex}; CHAR is excluded because the collector does + * not record CHAR bounds. + */ +public class LogRecordBatchStatisticsCompatibilityTest { + + private static final RowType ROW_TYPE = + DataTypes.ROW( + new DataField("bool", DataTypes.BOOLEAN()), + new DataField("i8", DataTypes.TINYINT()), + new DataField("i16", DataTypes.SMALLINT()), + new DataField("i32", DataTypes.INT()), + new DataField("i64", DataTypes.BIGINT()), + new DataField("f32", DataTypes.FLOAT()), + new DataField("f64", DataTypes.DOUBLE()), + new DataField("str", DataTypes.STRING()), + new DataField("dec5", DataTypes.DECIMAL(5, 2)), + new DataField("dec20", DataTypes.DECIMAL(20, 3)), + new DataField("date", DataTypes.DATE()), + new DataField("time", DataTypes.TIME()), + new DataField("ts3", DataTypes.TIMESTAMP(3)), + new DataField("ts6", DataTypes.TIMESTAMP(6)), + new DataField("ltz3", DataTypes.TIMESTAMP_LTZ(3)), + new DataField("ltz6", DataTypes.TIMESTAMP_LTZ(6)), + new DataField("strnull", DataTypes.STRING()), + new DataField("f32x", DataTypes.FLOAT()), + new DataField("f64x", DataTypes.DOUBLE())); + + private static List rows() { + return Arrays.asList( + new Object[] { + true, + (byte) 1, + (short) 100, + 10, + 1000L, + 1.5f, + 3.25, + "banana", + Decimal.fromBigDecimal(new BigDecimal("123.45"), 5, 2), + Decimal.fromBigDecimal(new BigDecimal("12345678.901"), 20, 3), + 19000, + 3600000, + TimestampNtz.fromMillis(1700000000123L), + TimestampNtz.fromMillis(1700000000123L, 456000), + TimestampLtz.fromEpochMillis(1700000000123L), + TimestampLtz.fromEpochMillis(1700000000123L, 456000), + null, + -0.0f, + 0.0 + }, + new Object[] { + false, + (byte) -3, + (short) 200, + null, + -2000L, + -2.5f, + 1.25, + "apple", + Decimal.fromBigDecimal(new BigDecimal("67.89"), 5, 2), + Decimal.fromBigDecimal(new BigDecimal("1.234"), 20, 3), + 18000, + 7200000, + TimestampNtz.fromMillis(1600000000000L), + TimestampNtz.fromMillis(1600000000000L, 1000), + TimestampLtz.fromEpochMillis(1600000000000L), + TimestampLtz.fromEpochMillis(1600000000000L, 1000), + null, + Float.intBitsToFloat(0xFFC00000), + -0.0 + }, + new Object[] { + true, + (byte) 7, + (short) -50, + 30, + 3000L, + 0.5f, + 9.75, + "cherry", + Decimal.fromBigDecimal(new BigDecimal("500.00"), 5, 2), + Decimal.fromBigDecimal(new BigDecimal("99999999999999.999"), 20, 3), + 20000, + 1800000, + TimestampNtz.fromMillis(1800000000999L), + TimestampNtz.fromMillis(1800000000999L, 999000), + TimestampLtz.fromEpochMillis(1800000000999L), + TimestampLtz.fromEpochMillis(1800000000999L, 999000), + null, + 0.0f, + Double.longBitsToDouble(0xFFF8000000000000L) + }); + } + + @Test + void testStatisticsBlockMatchesReference() throws Exception { + LogRecordBatchStatisticsCollector collector = + new LogRecordBatchStatisticsCollector( + ROW_TYPE, + LogRecordBatchStatisticsTestUtils.createAllColumnsStatsMapping(ROW_TYPE)); + for (Object[] data : rows()) { + collector.processRow(DataTestUtils.row(data)); + } + + MemorySegment segment = MemorySegment.allocateHeapMemory(4096); + int bytesWritten = collector.writeStatistics(new MemorySegmentOutputView(segment)); + byte[] bytes = new byte[bytesWritten]; + segment.get(0, bytes, 0, bytesWritten); + StringBuilder hex = new StringBuilder(); + for (byte b : bytes) { + hex.append(String.format("%02x", b)); + } + String expected = readReferenceHex(); + assertThat(hex.toString()).isEqualTo(expected); + } + + @Test + void testReferenceBlockParsesBack() throws Exception { + byte[] bytes = hexToBytes(readReferenceHex()); + DefaultLogRecordBatchStatistics stats = + LogRecordBatchStatisticsParser.parseStatistics( + MemorySegment.wrap(bytes), 0, ROW_TYPE, 1); + + assertThat(stats).isNotNull(); + assertThat(stats.getNullCounts()) + .containsExactly(0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0); + + InternalRow min = stats.getMinValues(); + InternalRow max = stats.getMaxValues(); + assertThat(min.getBoolean(0)).isFalse(); + assertThat(max.getBoolean(0)).isTrue(); + assertThat(min.getByte(1)).isEqualTo((byte) -3); + assertThat(max.getByte(1)).isEqualTo((byte) 7); + assertThat(min.getShort(2)).isEqualTo((short) -50); + assertThat(max.getShort(2)).isEqualTo((short) 200); + assertThat(min.getInt(3)).isEqualTo(10); + assertThat(max.getInt(3)).isEqualTo(30); + assertThat(min.getLong(4)).isEqualTo(-2000L); + assertThat(max.getLong(4)).isEqualTo(3000L); + assertThat(min.getFloat(5)).isEqualTo(-2.5f); + assertThat(max.getFloat(5)).isEqualTo(1.5f); + assertThat(min.getDouble(6)).isEqualTo(1.25); + assertThat(max.getDouble(6)).isEqualTo(9.75); + assertThat(min.getString(7).toString()).isEqualTo("apple"); + assertThat(max.getString(7).toString()).isEqualTo("cherry"); + assertThat(min.getDecimal(8, 5, 2).toBigDecimal()).isEqualTo(new BigDecimal("67.89")); + assertThat(max.getDecimal(8, 5, 2).toBigDecimal()).isEqualTo(new BigDecimal("500.00")); + assertThat(min.getDecimal(9, 20, 3).toBigDecimal()).isEqualTo(new BigDecimal("1.234")); + assertThat(max.getDecimal(9, 20, 3).toBigDecimal()) + .isEqualTo(new BigDecimal("99999999999999.999")); + assertThat(min.getInt(10)).isEqualTo(18000); + assertThat(max.getInt(10)).isEqualTo(20000); + assertThat(min.getInt(11)).isEqualTo(1800000); + assertThat(max.getInt(11)).isEqualTo(7200000); + assertThat(min.getTimestampNtz(12, 3).getMillisecond()).isEqualTo(1600000000000L); + assertThat(max.getTimestampNtz(12, 3).getMillisecond()).isEqualTo(1800000000999L); + assertThat(min.getTimestampNtz(13, 6)) + .isEqualTo(TimestampNtz.fromMillis(1600000000000L, 1000)); + assertThat(max.getTimestampNtz(13, 6)) + .isEqualTo(TimestampNtz.fromMillis(1800000000999L, 999000)); + assertThat(min.getTimestampLtz(14, 3).getEpochMillisecond()).isEqualTo(1600000000000L); + assertThat(max.getTimestampLtz(14, 3).getEpochMillisecond()).isEqualTo(1800000000999L); + assertThat(min.getTimestampLtz(15, 6)) + .isEqualTo(TimestampLtz.fromEpochMillis(1600000000000L, 1000)); + assertThat(max.getTimestampLtz(15, 6)) + .isEqualTo(TimestampLtz.fromEpochMillis(1800000000999L, 999000)); + // The all-null column carries a null count but no bounds. + assertThat(min.isNullAt(16)).isTrue(); + assertThat(max.isNullAt(16)).isTrue(); + // Float.compare semantics: -0.0 below 0.0, NaN above everything, and + // the retained NaN keeps its raw (here negative) bits. + assertThat(Float.floatToRawIntBits(min.getFloat(17))).isEqualTo(0x80000000); + assertThat(Float.floatToRawIntBits(max.getFloat(17))).isEqualTo(0xFFC00000); + assertThat(Double.doubleToRawLongBits(min.getDouble(18))).isEqualTo(0x8000000000000000L); + assertThat(Double.doubleToRawLongBits(max.getDouble(18))).isEqualTo(0xFFF8000000000000L); + } + + private static byte[] hexToBytes(String hex) { + byte[] bytes = new byte[hex.length() / 2]; + for (int i = 0; i < bytes.length; i++) { + bytes[i] = (byte) Integer.parseInt(hex.substring(2 * i, 2 * i + 2), 16); + } + return bytes; + } + + private static String readReferenceHex() throws Exception { + try (InputStream in = + LogRecordBatchStatisticsCompatibilityTest.class.getResourceAsStream( + "/encoding/statistics_block.hex")) { + assertThat(in).as("missing resource encoding/statistics_block.hex").isNotNull(); + byte[] buffer = new byte[8192]; + int length = in.read(buffer); + return new String(buffer, 0, length, StandardCharsets.UTF_8).trim(); + } + } +} diff --git a/fluss-common/src/test/resources/encoding/statistics_block.hex b/fluss-common/src/test/resources/encoding/statistics_block.hex new file mode 100644 index 00000000000..bce645214fe --- /dev/null +++ b/fluss-common/src/test/resources/encoding/statistics_block.hex @@ -0,0 +1 @@ +01130000000100020003000400050006000700080009000a000b000c000d000e000f0010001100120000000000000000000000000001000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000030000000000000000000000c000000000000001000000000000000000000000fd00000000000000ceff0000000000000a0000000000000030f8ffffffffffff000020c000000000000000000000f43f6170706c65000085851a00000000000002000000a0000000504600000000000040771b000000000000806e8774010000e8030000b000000000806e8774010000e8030000b800000000000000000000000000008000000000000000000000008004d2000000000000000000000000000000806e877401000000806e8774010000c0000000000000010000000001000000000000000700000000000000c8000000000000001e00000000000000b80b0000000000000000c03f000000000000000000802340636865727279008650c300000000000008000000a0000000204e00000000000000dd6d0000000000e7535c18a3010000583e0f00b0000000e7535c18a3010000583e0f00b800000000000000000000000000c0ff00000000000000000000f8ff016345785d89ffff0000000000000000e7535c18a3010000e7535c18a3010000 \ No newline at end of file diff --git a/fluss-rust/.licenserc.yaml b/fluss-rust/.licenserc.yaml index a3647d7f278..b61b4a3bbb2 100644 --- a/fluss-rust/.licenserc.yaml +++ b/fluss-rust/.licenserc.yaml @@ -36,4 +36,6 @@ header: - '**/*.md' - 'fluss-rust/**/DEPENDENCIES.*.tsv' - 'fluss-rust/**/*.env' + # Cross-language encoding fixtures. + - 'fluss-rust/**/testdata/*.hex' comment: on-failure diff --git a/fluss-rust/crates/fluss/src/client/table/batch_scanner.rs b/fluss-rust/crates/fluss/src/client/table/batch_scanner.rs index 1b9b5d6cf2a..1519ccc3e8a 100644 --- a/fluss-rust/crates/fluss/src/client/table/batch_scanner.rs +++ b/fluss-rust/crates/fluss/src/client/table/batch_scanner.rs @@ -459,10 +459,6 @@ fn project_batch( mod tests { use super::*; use crate::client::WriteRecord; - use crate::compression::{ - ArrowCompressionInfo, ArrowCompressionRatioEstimator, ArrowCompressionType, - DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }; use crate::metadata::{ Column, DataField, DataType, DataTypes, PhysicalTablePath, Schema, TableInfo, TablePath, }; @@ -470,7 +466,7 @@ mod tests { use crate::row::GenericRow; use crate::row::binary::BinaryWriter; use crate::row::compacted::CompactedRowWriter; - use crate::test_utils::build_table_info_with_columns; + use crate::test_utils::{build_table_info_with_columns, uncompressed_arrow_batch_config}; use arrow::array::{Array, Int32Array, Int64Array, StringArray}; fn build_two_col_table_info() -> TableInfo { @@ -499,15 +495,8 @@ mod tests { table_info.table_path.clone(), ))); let mut builder = MemoryLogRecordsArrowBuilder::new( - schema_id, - table_info.get_row_type(), + uncompressed_arrow_batch_config(schema_id, table_info.get_row_type(), usize::MAX), false, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - usize::MAX, - Arc::new(ArrowCompressionRatioEstimator::default()), ) .expect("builder"); for (i, row) in rows.iter().enumerate() { diff --git a/fluss-rust/crates/fluss/src/client/table/log_fetch_buffer.rs b/fluss-rust/crates/fluss/src/client/table/log_fetch_buffer.rs index 6d3a2ad4a62..1e90c8a6491 100644 --- a/fluss-rust/crates/fluss/src/client/table/log_fetch_buffer.rs +++ b/fluss-rust/crates/fluss/src/client/table/log_fetch_buffer.rs @@ -1016,10 +1016,6 @@ mod tests { use super::*; use crate::client::table::read_context_resolver::ReadContextResolver; use crate::client::{EARLIEST_OFFSET, WriteRecord}; - use crate::compression::{ - ArrowCompressionInfo, ArrowCompressionRatioEstimator, ArrowCompressionType, - DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }; use crate::metadata::{ Column, DataField, DataTypes, PhysicalTablePath, RowType, Schema, TableDescriptor, TableInfo, TablePath, @@ -1029,7 +1025,9 @@ mod tests { MemoryLogRecordsArrowBuilder, RECORDS_OFFSET, ReadContext, to_arrow_schema, }; use crate::row::GenericRow; - use crate::test_utils::{build_table_info, build_table_info_with_columns}; + use crate::test_utils::{ + build_table_info, build_table_info_with_columns, uncompressed_arrow_batch_config, + }; use arrow::array::{Array, StringArray}; use std::sync::Arc; @@ -1081,15 +1079,12 @@ mod tests { source_table_info.get_table_path().clone(), ))); let mut builder = MemoryLogRecordsArrowBuilder::new( - source_schema_id, - source_table_info.get_row_type(), + uncompressed_arrow_batch_config( + source_schema_id, + source_table_info.get_row_type(), + usize::MAX, + ), false, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - usize::MAX, - Arc::new(ArrowCompressionRatioEstimator::default()), )?; let record = WriteRecord::for_append( Arc::clone(&source_table_info), @@ -1225,15 +1220,8 @@ mod tests { let physical_table_path = Arc::new(PhysicalTablePath::of(Arc::new(table_path))); let mut builder = MemoryLogRecordsArrowBuilder::new( - 1, - &row_type, + uncompressed_arrow_batch_config(1, &row_type, usize::MAX), false, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - usize::MAX, - Arc::new(ArrowCompressionRatioEstimator::default()), )?; let mut row = GenericRow::new(2); @@ -1331,15 +1319,8 @@ mod tests { let physical_table_path = Arc::new(PhysicalTablePath::of(Arc::new(table_path))); let mut builder = MemoryLogRecordsArrowBuilder::new( - 1, - &row_type, + uncompressed_arrow_batch_config(1, &row_type, usize::MAX), false, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - usize::MAX, - Arc::new(ArrowCompressionRatioEstimator::default()), )?; let mut row = GenericRow::new(2); row.set_field(0, 1_i32); @@ -1438,15 +1419,8 @@ mod tests { let physical_table_path = Arc::new(PhysicalTablePath::of(Arc::new(table_path))); let mut builder = MemoryLogRecordsArrowBuilder::new( - 0, - &old_row_type, + uncompressed_arrow_batch_config(0, &old_row_type, usize::MAX), false, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - usize::MAX, - Arc::new(ArrowCompressionRatioEstimator::default()), )?; let mut row = GenericRow::new(1); @@ -1584,15 +1558,8 @@ mod tests { let physical_table_path = Arc::new(PhysicalTablePath::of(Arc::new(table_path))); let mut builder = MemoryLogRecordsArrowBuilder::new( - 1, - &row_type, + uncompressed_arrow_batch_config(1, &row_type, usize::MAX), false, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - usize::MAX, - Arc::new(ArrowCompressionRatioEstimator::default()), )?; for id in [10_i32, 20, 20] { let mut row = GenericRow::new(1); diff --git a/fluss-rust/crates/fluss/src/client/table/scanner.rs b/fluss-rust/crates/fluss/src/client/table/scanner.rs index 99331bf7268..c4fd89d9fea 100644 --- a/fluss-rust/crates/fluss/src/client/table/scanner.rs +++ b/fluss-rust/crates/fluss/src/client/table/scanner.rs @@ -2514,10 +2514,6 @@ mod tests { use crate::client::admin::FlussAdmin; use crate::client::metadata::Metadata; use crate::client::table::read_context_resolver::ReadContextResolver; - use crate::compression::{ - ArrowCompressionInfo, ArrowCompressionRatioEstimator, ArrowCompressionType, - DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }; use crate::metadata::{DataTypes, PhysicalTablePath, Schema, TableInfo, TablePath}; use crate::proto::{PbFetchLogRespForBucket, PbFetchLogRespForTable}; use crate::record::MemoryLogRecordsArrowBuilder; @@ -2525,6 +2521,7 @@ mod tests { use crate::rpc::FlussError; use crate::test_utils::{ assert_scanner_entries_labeled, build_cluster_arc, build_table_info, test_scanner_metrics, + uncompressed_arrow_batch_config, }; fn test_admin(metadata: &Arc) -> Arc { @@ -2568,15 +2565,8 @@ mod tests { fn build_records(table_info: &TableInfo, table_path: Arc) -> Result> { let mut builder = MemoryLogRecordsArrowBuilder::new( - 1, - table_info.get_row_type(), + uncompressed_arrow_batch_config(1, table_info.get_row_type(), usize::MAX), false, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - usize::MAX, - Arc::new(ArrowCompressionRatioEstimator::default()), )?; let physical_table_path = Arc::new(PhysicalTablePath::of(table_path)); let row = GenericRow { diff --git a/fluss-rust/crates/fluss/src/client/write/accumulator.rs b/fluss-rust/crates/fluss/src/client/write/accumulator.rs index 0d026d83c15..9334bb43de9 100644 --- a/fluss-rust/crates/fluss/src/client/write/accumulator.rs +++ b/fluss-rust/crates/fluss/src/client/write/accumulator.rs @@ -26,7 +26,7 @@ use crate::compression::ArrowCompressionRatioEstimator; use crate::config::Config; use crate::error::{Error, Result}; use crate::metadata::{PhysicalTablePath, TableBucket}; -use crate::record::{NO_BATCH_SEQUENCE, NO_WRITER_ID}; +use crate::record::{ArrowBatchConfig, NO_BATCH_SEQUENCE, NO_WRITER_ID}; use crate::util::current_time_ms; use crate::{BucketId, PartitionId, TableId}; use dashmap::DashMap; @@ -285,17 +285,30 @@ impl RecordAccumulator { let schema_id = table_info.schema_id; + let stats_index_mapping = if table_info + .get_table_config() + .get_statistics_columns() + .is_enabled() + { + Some(table_info.get_stats_index_mapping()?.to_vec()) + } else { + None + }; + let mut batch: WriteBatch = match record.record() { Record::Log(_) => ArrowLog(ArrowLogWriteBatch::new( self.batch_id.fetch_add(1, Ordering::Relaxed), Arc::clone(physical_table_path), - schema_id, - arrow_compression_info, - row_type, + ArrowBatchConfig { + schema_id, + row_type: row_type.clone(), + stats_index_mapping, + compression: arrow_compression_info, + write_limit: alloc_size, + compression_ratio_estimator, + }, current_time_ms(), matches!(&record.record, Record::Log(LogWriteRecord::RecordBatch(_))), - alloc_size, - compression_ratio_estimator, )?), Record::Kv(kv_record) => Kv(KvWriteBatch::new( self.batch_id.fetch_add(1, Ordering::Relaxed), diff --git a/fluss-rust/crates/fluss/src/client/write/batch.rs b/fluss-rust/crates/fluss/src/client/write/batch.rs index 50148835dce..fdfafbb1008 100644 --- a/fluss-rust/crates/fluss/src/client/write/batch.rs +++ b/fluss-rust/crates/fluss/src/client/write/batch.rs @@ -17,11 +17,10 @@ use crate::client::broadcast::{BatchWriteResult, BroadcastOnce}; use crate::client::{Record, ResultHandle, WriteRecord}; -use crate::compression::{ArrowCompressionInfo, ArrowCompressionRatioEstimator}; use crate::error::{Error, Result}; -use crate::metadata::{KvFormat, PhysicalTablePath, RowType}; -use crate::record::MemoryLogRecordsArrowBuilder; +use crate::metadata::{KvFormat, PhysicalTablePath}; use crate::record::kv::KvRecordBatchBuilder; +use crate::record::{ArrowBatchConfig, MemoryLogRecordsArrowBuilder}; use crate::record::{NO_BATCH_SEQUENCE, NO_WRITER_ID}; use bytes::Bytes; use std::cmp::max; @@ -238,34 +237,22 @@ impl WriteBatch { pub struct ArrowLogWriteBatch { pub write_batch: InnerWriteBatch, - pub arrow_builder: MemoryLogRecordsArrowBuilder, + pub(crate) arrow_builder: MemoryLogRecordsArrowBuilder, built_records: Option, } impl ArrowLogWriteBatch { - #[allow(clippy::too_many_arguments)] - pub fn new( + pub(crate) fn new( batch_id: i64, physical_table_path: Arc, - schema_id: i32, - arrow_compression_info: ArrowCompressionInfo, - row_type: &RowType, + config: ArrowBatchConfig, create_ms: i64, to_append_record_batch: bool, - write_limit: usize, - compression_ratio_estimator: Arc, ) -> Result { let base = InnerWriteBatch::new(batch_id, physical_table_path, create_ms); Ok(Self { write_batch: base, - arrow_builder: MemoryLogRecordsArrowBuilder::new( - schema_id, - row_type, - to_append_record_batch, - arrow_compression_info, - write_limit, - compression_ratio_estimator, - )?, + arrow_builder: MemoryLogRecordsArrowBuilder::new(config, to_append_record_batch)?, built_records: None, }) } @@ -441,9 +428,14 @@ impl KvWriteBatch { mod tests { use super::*; use crate::client::{RowBytes, WriteFormat}; - use crate::metadata::TablePath; + use crate::metadata::{RowType, TablePath}; use crate::test_utils::build_table_info; + /// An uncompressed [`ArrowBatchConfig`] for schema 1. + fn uncompressed_config(row_type: &RowType, write_limit: usize) -> ArrowBatchConfig { + crate::test_utils::uncompressed_arrow_batch_config(1, row_type, write_limit) + } + #[test] fn complete_only_once() { let table_path = TablePath::new("db".to_string(), "tbl".to_string()); @@ -480,9 +472,6 @@ mod tests { #[test] fn record_count_reflects_appended_rows() { use crate::client::WriteRecord; - use crate::compression::{ - ArrowCompressionInfo, ArrowCompressionType, DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }; use crate::metadata::{DataField, DataTypes, RowType}; use crate::row::GenericRow; @@ -498,16 +487,9 @@ mod tests { let arrow_batch = ArrowLogWriteBatch::new( 1, Arc::clone(&physical_table_path), - 1, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - &row_type, + uncompressed_config(&row_type, 2 * 1024 * 1024), 0, false, - 2 * 1024 * 1024, - Arc::new(ArrowCompressionRatioEstimator::default()), ) .unwrap(); let mut batch = WriteBatch::ArrowLog(arrow_batch); @@ -566,9 +548,6 @@ mod tests { #[test] fn test_arrow_log_write_batch_estimated_size() { use crate::client::WriteRecord; - use crate::compression::{ - ArrowCompressionInfo, ArrowCompressionType, DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }; use crate::metadata::{DataField, DataTypes, RowType}; use crate::row::GenericRow; use arrow::array::{Int32Array, RecordBatch, StringArray}; @@ -587,16 +566,9 @@ mod tests { let mut batch = ArrowLogWriteBatch::new( 1, Arc::clone(&physical_table_path), - 1, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - &row_type, + uncompressed_config(&row_type, 2 * 1024 * 1024), 0, false, - 2 * 1024 * 1024, - Arc::new(ArrowCompressionRatioEstimator::default()), ) .unwrap(); @@ -633,16 +605,9 @@ mod tests { let mut batch = ArrowLogWriteBatch::new( 1, physical_table_path.clone(), - 1, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - &row_type, + uncompressed_config(&row_type, 2 * 1024 * 1024), 0, true, - 2 * 1024 * 1024, - Arc::new(ArrowCompressionRatioEstimator::default()), ) .unwrap(); @@ -733,7 +698,6 @@ mod tests { use crate::client::WriteRecord; use crate::compression::{ ArrowCompressionInfo, ArrowCompressionRatioEstimator, ArrowCompressionType, - DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, }; use crate::metadata::{DataField, DataTypes, RowType}; use crate::row::GenericRow; @@ -752,16 +716,9 @@ mod tests { let mut batch = ArrowLogWriteBatch::new( 1, Arc::clone(&physical_table_path), - 1, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - &row_type, + uncompressed_config(&row_type, write_limit), 0, false, - write_limit, - Arc::new(ArrowCompressionRatioEstimator::default()), ) .unwrap(); @@ -802,16 +759,9 @@ mod tests { let mut batch = ArrowLogWriteBatch::new( 2, Arc::clone(&physical_table_path), - 1, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - &row_type_small, + uncompressed_config(&row_type_small, 2 * 1024 * 1024), 0, false, - 2 * 1024 * 1024, - Arc::new(ArrowCompressionRatioEstimator::default()), ) .unwrap(); @@ -846,13 +796,16 @@ mod tests { let mut batch1 = ArrowLogWriteBatch::new( 3, Arc::clone(&physical_table_path), - 1, - compression.clone(), - &row_type, + ArrowBatchConfig { + schema_id: 1, + row_type: row_type.clone(), + stats_index_mapping: None, + compression: compression.clone(), + write_limit, + compression_ratio_estimator: Arc::clone(&estimator), + }, 0, false, - write_limit, - Arc::clone(&estimator), ) .unwrap(); @@ -883,13 +836,16 @@ mod tests { let mut batch2 = ArrowLogWriteBatch::new( 4, Arc::clone(&physical_table_path), - 1, - compression, - &row_type, + ArrowBatchConfig { + schema_id: 1, + row_type: row_type.clone(), + stats_index_mapping: None, + compression, + write_limit, + compression_ratio_estimator: Arc::clone(&estimator), + }, 0, false, - write_limit, - Arc::clone(&estimator), ) .unwrap(); diff --git a/fluss-rust/crates/fluss/src/record/arrow.rs b/fluss-rust/crates/fluss/src/record/arrow.rs index 4ca2cc5de03..d54e53f0da7 100644 --- a/fluss-rust/crates/fluss/src/record/arrow.rs +++ b/fluss-rust/crates/fluss/src/record/arrow.rs @@ -21,7 +21,7 @@ use crate::compression::{ }; use crate::error::{Error, Result}; use crate::metadata::{DataField, DataType, RowType, UNEXIST_MAPPING}; -use crate::record::{ChangeType, ScanRecord}; +use crate::record::ScanRecord; use crate::row::column_vector::TypedBatch; use crate::row::column_writer::{ColumnWriter, round_up_to_8}; use crate::row::{ColumnarRow, InternalRow}; @@ -41,117 +41,18 @@ use arrow_schema::SchemaRef; use arrow_schema::{DataType as ArrowDataType, Field}; use byteorder::WriteBytesExt; use byteorder::{ByteOrder, LittleEndian}; -use bytes::Bytes; use crc32c::crc32c; use std::{ cell::Cell, collections::HashMap, - fs::File, - io::{Cursor, Read, Seek, SeekFrom, Write}, - path::PathBuf, + io::{Cursor, Write}, sync::Arc, }; +use super::log_record_batch::*; use crate::error::Error::IllegalArgument; +use crate::record::statistics::{estimated_serialized_size, serialize_statistics}; use arrow::ipc::writer::IpcWriteOptions; -/// 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; - -/// 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, -} - -/// 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 { - // 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) -} - -// NOTE: Rust layout/offsets currently match Java only for V0. -// TODO: Add V1 layout/offsets to keep parity with Java's V1 format. -pub const CURRENT_LOG_MAGIC_VALUE: u8 = LogMagicValue::V0 as u8; - -/// 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; pub const BUILDER_DEFAULT_OFFSET: i64 = 0; @@ -163,7 +64,7 @@ const INITIAL_ROW_CAPACITY: usize = 1024; /// Matching Java's `ArrowWriter.BUFFER_USAGE_RATIO`. const BUFFER_USAGE_RATIO: f32 = 0.95; -pub struct MemoryLogRecordsArrowBuilder { +pub(crate) struct MemoryLogRecordsArrowBuilder { base_log_offset: i64, schema_id: i32, magic: u8, @@ -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)>, + /// 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(crate) struct ArrowBatchConfig { + pub schema_id: i32, + pub row_type: RowType, + /// Statistics column mapping; `Some` upgrades batches to V1. + pub stats_index_mapping: Option>, + pub compression: ArrowCompressionInfo, + pub write_limit: usize, + pub compression_ratio_estimator: Arc, } pub trait ArrowRecordBatchInnerBuilder: Send { @@ -363,30 +280,54 @@ impl ArrowRecordBatchInnerBuilder for RowAppendRecordBatchBuilder { // the previous batch (recordsCount / 2) for a warm start, avoiding the first-record // size check on every new batch. impl MemoryLogRecordsArrowBuilder { - pub fn new( - schema_id: i32, - row_type: &RowType, - to_append_record_batch: bool, - arrow_compression_info: ArrowCompressionInfo, - write_limit: usize, - compression_ratio_estimator: Arc, - ) -> Result { + pub(crate) fn new(config: ArrowBatchConfig, to_append_record_batch: bool) -> Result { + let ArrowBatchConfig { + schema_id, + row_type, + stats_index_mapping, + compression: arrow_compression_info, + write_limit, + compression_ratio_estimator, + } = config; let arrow_batch_builder: Box = { if to_append_record_batch { Box::new(PrebuiltRecordBatchBuilder::default()) } else { - Box::new(RowAppendRecordBatchBuilder::new(row_type)?) + Box::new(RowAppendRecordBatchBuilder::new(&row_type)?) } }; - let schema = to_arrow_schema(row_type)?; + let schema = to_arrow_schema(&row_type)?; let ipc_overhead = estimate_arrow_ipc_overhead(&schema, arrow_compression_info.get_compression_type())?; let effective_limit = (write_limit as f32 * BUFFER_USAGE_RATIO) as usize; let estimated_compression_ratio = compression_ratio_estimator.estimation(); + + // Validate the mapping as Java does when building the collector's + // stats row type. + let field_count = row_type.fields().len(); + if let Some(mapping) = &stats_index_mapping { + if let Some(&index) = mapping.iter().find(|&&index| index >= field_count) { + return Err(IllegalArgument { + message: format!( + "Statistics column index {index} is out of range for {field_count} fields" + ), + }); + } + } + let magic = if stats_index_mapping.is_some() { + LOG_MAGIC_VALUE_V1 + } else { + LOG_MAGIC_VALUE_V0 + }; + let estimated_statistics_size = stats_index_mapping + .as_ref() + .map_or(0, |mapping| estimated_serialized_size(&row_type, mapping)); + let statistics = stats_index_mapping.map(|mapping| (row_type, mapping)); + Ok(MemoryLogRecordsArrowBuilder { base_log_offset: BUILDER_DEFAULT_OFFSET, schema_id, - magic: CURRENT_LOG_MAGIC_VALUE, + magic, writer_id: NO_WRITER_ID, batch_sequence: NO_BATCH_SEQUENCE, is_closed: false, @@ -397,9 +338,19 @@ impl MemoryLogRecordsArrowBuilder { estimated_max_records_count: Cell::new(-1), compression_ratio_estimator, estimated_compression_ratio, + statistics, + estimated_statistics_size, }) } + fn header_size(&self) -> usize { + if self.magic >= LOG_MAGIC_VALUE_V1 { + V1_RECORD_BATCH_HEADER_SIZE + } else { + RECORD_BATCH_HEADER_SIZE + } + } + pub fn append(&mut self, record: &WriteRecord) -> Result { match &record.record() { Record::Log(log_write_record) => match log_write_record { @@ -472,6 +423,9 @@ impl MemoryLogRecordsArrowBuilder { } pub fn build(&mut self) -> Result> { + // The header and CRC offsets below assume the V0/V1 layout without V2's leader epoch. + debug_assert!(self.magic < LOG_MAGIC_VALUE_V2); + // Capture uncompressed body size before serialization for compression ratio update. let uncompressed_body_size = self.arrow_record_batch_builder.estimated_size_in_bytes(); @@ -508,14 +462,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..]; @@ -527,7 +502,7 @@ impl MemoryLogRecordsArrowBuilder { Ok(batch_bytes.to_vec()) } - fn write_batch_header(&self, buffer: &mut [u8]) -> Result<()> { + fn write_batch_header(&self, buffer: &mut [u8], statistics_length: usize) -> Result<()> { let total_len = buffer.len(); let mut cursor = Cursor::new(buffer); cursor.write_i64::(self.base_log_offset)?; @@ -551,6 +526,10 @@ impl MemoryLogRecordsArrowBuilder { cursor.write_i64::(self.writer_id)?; cursor.write_i32::(self.batch_sequence)?; cursor.write_i32::(record_count)?; + + if self.magic >= LOG_MAGIC_VALUE_V1 { + cursor.write_i32::(statistics_length as i32)?; + } Ok(()) } @@ -559,13 +538,12 @@ impl MemoryLogRecordsArrowBuilder { self.batch_sequence = batch_base_sequence; } - /// Get an estimate of the number of bytes written to the underlying buffer. - /// Includes Fluss record batch header + Arrow IPC metadata + estimated - /// compressed body size. + /// Estimated bytes written: header, statistics estimate (V1), Arrow IPC + /// metadata and estimated compressed body. pub fn estimated_size_in_bytes(&self) -> usize { let body = self.arrow_record_batch_builder.estimated_size_in_bytes(); let estimated_body = self.estimated_compressed_size(body); - RECORD_BATCH_HEADER_SIZE + self.ipc_overhead + estimated_body + self.header_size() + self.estimated_statistics_size + self.ipc_overhead + estimated_body } /// Number of records appended so far. Used for writer throughput metrics. @@ -635,487 +613,6 @@ pub trait ToArrow { fn append_to(&self, builder: &mut dyn ArrayBuilder) -> Result<()>; } -/// In-memory log record source. -/// Used for local tablet server fetches (existing path). -struct MemorySource { - data: Bytes, -} - -impl MemorySource { - fn new(data: Vec) -> 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 { - 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, // 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 { - 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 { - 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 { - 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) -> 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 { - 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> { - 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; - - fn next(&mut self) -> Option { - 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] - } - - pub fn commit_timestamp(&self) -> i64 { - let offset = COMMIT_TIMESTAMP_OFFSET; - LittleEndian::read_i64(&self.data[offset..offset + COMMIT_TIMESTAMP_LENGTH]) - } - - pub fn writer_id(&self) -> i64 { - let offset = WRITE_CLIENT_ID_OFFSET; - LittleEndian::read_i64(&self.data[offset..offset + WRITE_CLIENT_ID_LENGTH]) - } - - pub fn batch_sequence(&self) -> i32 { - let offset = BATCH_SEQUENCE_OFFSET; - LittleEndian::read_i32(&self.data[offset..offset + BATCH_SEQUENCE_LENGTH]) - } - - pub fn ensure_valid(&self) -> Result<()> { - // TODO enable validation once checksum handling is corrected. - Ok(()) - } - - pub fn is_valid(&self) -> bool { - self.size_in_bytes() >= RECORD_BATCH_HEADER_SIZE - && self.checksum() == self.compute_checksum() - } - - fn compute_checksum(&self) -> u32 { - let start = SCHEMA_ID_OFFSET; - crc32c(&self.data[start..]) - } - - fn attributes(&self) -> u8 { - self.data[ATTRIBUTES_OFFSET] - } - - /// Whether this batch is append-only (see [`APPEND_ONLY_FLAG_MASK`]). - fn is_append_only(&self) -> bool { - self.attributes() & APPEND_ONLY_FLAG_MASK != 0 - } - - pub fn next_log_offset(&self) -> i64 { - self.last_log_offset() + 1 - } - - pub fn checksum(&self) -> u32 { - let offset = CRC_OFFSET; - LittleEndian::read_u32(&self.data[offset..offset + CRC_LENGTH]) - } - - pub fn schema_id(&self) -> i16 { - let offset = SCHEMA_ID_OFFSET; - LittleEndian::read_i16(&self.data[offset..offset + SCHEMA_ID_LENGTH]) - } - - pub fn base_log_offset(&self) -> i64 { - let offset = BASE_OFFSET_OFFSET; - LittleEndian::read_i64(&self.data[offset..offset + BASE_OFFSET_LENGTH]) - } - - pub fn last_log_offset(&self) -> i64 { - self.base_log_offset() + self.last_offset_delta() as i64 - } - - fn last_offset_delta(&self) -> i32 { - let offset = LAST_OFFSET_DELTA_OFFSET; - LittleEndian::read_i32(&self.data[offset..offset + LAST_OFFSET_DELTA_LENGTH]) - } - - pub fn size_in_bytes(&self) -> usize { - let offset = LENGTH_OFFSET; - LittleEndian::read_i32(&self.data[offset..offset + LENGTH_LENGTH]) as usize + LOG_OVERHEAD - } - - pub fn record_count(&self) -> i32 { - let offset = RECORDS_COUNT_OFFSET; - LittleEndian::read_i32(&self.data[offset..offset + RECORDS_COUNT_LENGTH]) - } - - /// Splits the batch body into its per-record change types and the trailing - /// Arrow IPC payload (see [`APPEND_ONLY_FLAG_MASK`] for the layout). - fn decode_change_types(&self) -> Result<(BatchChangeTypes, &[u8])> { - let body = self - .data - .get(RECORDS_OFFSET..) - .ok_or_else(|| Error::UnexpectedError { - message: format!( - "Corrupt log record batch: data length {} is less than RECORDS_OFFSET {}", - self.data.len(), - RECORDS_OFFSET - ), - source: None, - })?; - - if self.is_append_only() { - return Ok((BatchChangeTypes::Uniform(ChangeType::AppendOnly), body)); - } - - let record_count = self.record_count(); - if record_count < 0 { - return Err(Error::UnexpectedError { - message: format!("Corrupt changelog batch: negative record count {record_count}"), - source: None, - }); - } - let record_count = record_count as usize; - let (change_type_bytes, arrow_data) = - body.split_at_checked(record_count) - .ok_or_else(|| Error::UnexpectedError { - message: format!( - "Corrupt changelog batch: body length {} is smaller than its \ - {record_count}-record change-type vector", - body.len() - ), - source: None, - })?; - - let mut change_types = Vec::with_capacity(record_count); - for &byte in change_type_bytes { - let change_type = - ChangeType::from_byte_value(byte).map_err(|message| Error::UnexpectedError { - message, - source: None, - })?; - change_types.push(change_type); - } - - Ok((BatchChangeTypes::PerRecord(change_types), arrow_data)) - } - - pub fn records(&self, read_context: &ReadContext) -> Result { - if self.record_count() == 0 { - return Ok(LogRecordIterator::empty()); - } - - let (change_types, arrow_data) = self.decode_change_types()?; - let record_batch = read_context.record_batch(arrow_data)?; - let arrow_reader = ArrowReader::new_with_fluss_row_type( - Arc::new(record_batch), - read_context.row_type.clone(), - read_context.fluss_row_type().cloned(), - )?; - let iterator = ArrowLogRecordIterator::new( - arrow_reader, - self.base_log_offset(), - self.commit_timestamp(), - change_types, - )?; - - Ok(LogRecordIterator::Arrow(iterator)) - } - - pub fn records_for_remote_log(&self, read_context: &ReadContext) -> Result { - if self.record_count() == 0 { - return Ok(LogRecordIterator::empty()); - } - - let (change_types, arrow_data) = self.decode_change_types()?; - let record_batch = read_context.record_batch_for_remote_log(arrow_data)?; - let log_record_iterator = match record_batch { - None => LogRecordIterator::empty(), - Some(record_batch) => { - let arrow_reader = ArrowReader::new_with_fluss_row_type( - Arc::new(record_batch), - read_context.row_type.clone(), - read_context.fluss_row_type().cloned(), - )?; - let iterator = ArrowLogRecordIterator::new( - arrow_reader, - self.base_log_offset(), - self.commit_timestamp(), - change_types, - )?; - LogRecordIterator::Arrow(iterator) - } - }; - Ok(log_record_iterator) - } - - /// Returns the record batch directly without creating an iterator. - /// This is more efficient when you need the entire batch rather than - /// iterating row-by-row. - pub fn record_batch(&self, read_context: &ReadContext) -> Result { - if self.record_count() == 0 { - // Return empty batch with correct schema - return Ok(RecordBatch::new_empty(read_context.target_schema.clone())); - } - - // Batch access drops the change-type vector; use `records()` for CDC. - let (_, arrow_data) = self.decode_change_types()?; - read_context.record_batch(arrow_data) - } -} - /// Parse an Arrow IPC message from a byte slice. /// /// Server returns RecordBatch message (without Schema message) in the encapsulated message format. @@ -1758,50 +1255,6 @@ fn align_record_batch_to_schema( Ok(RecordBatch::try_new(target_schema, columns)?) } -pub enum LogRecordIterator { - Empty, - Arrow(ArrowLogRecordIterator), -} - -impl LogRecordIterator { - pub fn empty() -> Self { - LogRecordIterator::Empty - } -} - -impl Iterator for LogRecordIterator { - type Item = ScanRecord; - - fn next(&mut self) -> Option { - match self { - LogRecordIterator::Empty => None, - LogRecordIterator::Arrow(iter) => iter.next(), - } - } -} - -/// Per-record change types decoded from a log batch. -/// -/// Append-only batches carry no change-type vector on the wire, so a single -/// `AppendOnly` value covers every record without allocating. Changelog batches -/// (the CDC stream of a primary-key table) decode one change type per record, -/// in record order. -enum BatchChangeTypes { - /// Every record shares this change type (append-only batches). - Uniform(ChangeType), - /// One change type per record, indexed by row id (changelog batches). - PerRecord(Vec), -} - -impl BatchChangeTypes { - fn get(&self, row_id: usize) -> ChangeType { - match self { - BatchChangeTypes::Uniform(change_type) => *change_type, - BatchChangeTypes::PerRecord(change_types) => change_types[row_id], - } - } -} - pub struct ArrowLogRecordIterator { reader: ArrowReader, base_offset: i64, @@ -1811,7 +1264,7 @@ pub struct ArrowLogRecordIterator { } impl ArrowLogRecordIterator { - fn new( + pub(crate) fn new( reader: ArrowReader, base_offset: i64, timestamp: i64, @@ -1895,12 +1348,12 @@ pub struct MyVec(pub StreamReader); mod tests { use super::*; use crate::client::WriteRecord; - use crate::compression::{ - ArrowCompressionInfo, ArrowCompressionType, DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }; use crate::metadata::{DataField, DataTypes, PhysicalTablePath, RowType, TablePath}; use crate::row::{DataGetters, GenericRow}; - use crate::test_utils::build_table_info; + use crate::test_utils::{ + build_append_only_batch, build_table_info, uncompressed_arrow_batch_config, + }; + use bytes::Bytes; #[test] fn nonnullable_append_builder_roundtrips_for_both_append_modes() { @@ -1915,15 +1368,8 @@ mod tests { for to_append_record_batch in [false, true] { let mut builder = MemoryLogRecordsArrowBuilder::new( - 1, - &row_type, + uncompressed_arrow_batch_config(1, &row_type, usize::MAX), to_append_record_batch, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - usize::MAX, - Arc::new(ArrowCompressionRatioEstimator::default()), ) .expect("NOT NULL builder should construct"); @@ -2276,27 +1722,6 @@ mod tests { assert!(matches!(result, Err(IllegalArgument { .. }))); } - #[test] - fn checksum_and_schema_id_read_minimum_header() { - // Header-only batches with record_count == 0 are valid; this covers the minimal bytes - // needed for checksum/schema_id access. - let mut data = vec![0u8; SCHEMA_ID_OFFSET + SCHEMA_ID_LENGTH]; - let crc = 0xA1B2C3D4u32; - let schema_id = 42i16; - LittleEndian::write_u32(&mut data[CRC_OFFSET..CRC_OFFSET + CRC_LENGTH], crc); - LittleEndian::write_i16( - &mut data[SCHEMA_ID_OFFSET..SCHEMA_ID_OFFSET + SCHEMA_ID_LENGTH], - schema_id, - ); - - let batch = LogRecordBatch::new(Bytes::from(data)); - assert_eq!(batch.checksum(), crc); - assert_eq!(batch.schema_id(), schema_id); - - let expected = crc32c(&batch.data[SCHEMA_ID_OFFSET..]); - assert_eq!(batch.compute_checksum(), expected); - } - fn le_bytes(vals: &[u32]) -> Vec { let mut out = Vec::with_capacity(vals.len() * 4); for &v in vals { @@ -2382,49 +1807,6 @@ mod tests { Ok(()) } - // Tests for file-backed streaming - - #[test] - fn test_file_source_streaming() -> Result<()> { - use tempfile::NamedTempFile; - - // Test 1: Basic file reads work - let test_data = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10]; - let mut tmp_file = NamedTempFile::new()?; - tmp_file.write_all(&test_data)?; - tmp_file.flush()?; - - let file_path = tmp_file.path().to_path_buf(); - let file = File::open(&file_path)?; - let mut source = FileSource::new(file, 0, file_path)?; - - // Read full data - let data = source.read_batch_data(0, 10)?; - assert_eq!(data.to_vec(), test_data); - - // Read partial data - let partial = source.read_batch_data(2, 5)?; - assert_eq!(partial.to_vec(), vec![3, 4, 5, 6, 7]); - - // Test 2: base_offset works (critical for remote logs with pos_in_log_segment) - let prefix = vec![0xFF; 100]; - let actual_data = vec![1, 2, 3, 4, 5]; - let mut tmp_file2 = NamedTempFile::new()?; - tmp_file2.write_all(&prefix)?; - tmp_file2.write_all(&actual_data)?; - tmp_file2.flush()?; - - let file_path2 = tmp_file2.path().to_path_buf(); - let file2 = File::open(&file_path2)?; - let mut source2 = FileSource::new(file2, 100, file_path2)?; // Skip first 100 bytes - - assert_eq!(source2.total_size(), 5); // Only counts data after offset - let data2 = source2.read_batch_data(0, 5)?; - assert_eq!(data2.to_vec(), actual_data); - - Ok(()) - } - #[test] fn test_all_types_end_to_end() -> Result<()> { use crate::row::{Date, Datum, Decimal, GenericRow, Time, TimestampLtz, TimestampNtz}; @@ -2561,17 +1943,24 @@ mod tests { Ok(()) } - #[test] - fn test_log_records_batches_from_file() -> Result<()> { - use crate::client::WriteRecord; - use crate::compression::{ - ArrowCompressionInfo, ArrowCompressionType, DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }; - use crate::metadata::{PhysicalTablePath, TablePath}; - use crate::row::GenericRow; - use tempfile::NamedTempFile; + /// An `(id INT, name STRING)` builder collecting statistics for `mapping`. + fn builder_with_statistics( + row_type: &RowType, + mapping: Option>, + to_append_record_batch: bool, + ) -> MemoryLogRecordsArrowBuilder { + MemoryLogRecordsArrowBuilder::new( + ArrowBatchConfig { + stats_index_mapping: mapping, + ..uncompressed_arrow_batch_config(1, row_type, usize::MAX) + }, + to_append_record_batch, + ) + .expect("builder should construct") + } - // Integration test: Real log record batch streamed from file + #[test] + fn statistics_upgrade_the_batch_to_v1() -> Result<()> { let row_type = RowType::new(vec![ DataField::new("id".to_string(), DataTypes::int(), None), DataField::new("name".to_string(), DataTypes::string(), None), @@ -2580,60 +1969,82 @@ mod tests { let table_info = Arc::new(build_table_info(table_path.clone(), 1, 1)); let physical_table_path = Arc::new(PhysicalTablePath::of(Arc::new(table_path))); - let mut builder = MemoryLogRecordsArrowBuilder::new( - 1, - &row_type, - false, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - usize::MAX, - Arc::new(ArrowCompressionRatioEstimator::default()), + // The block the batch must carry, derived independently of the writer. + let expected_batch = RecordBatch::try_new( + to_arrow_schema(&row_type)?, + vec![ + Arc::new(arrow::array::Int32Array::from(vec![1, 2])) as ArrayRef, + Arc::new(arrow::array::StringArray::from(vec!["alice", "bob"])) as ArrayRef, + ], )?; + let expected_statistics = serialize_statistics(&expected_batch, &row_type, &[0, 1])? + .expect("two rows must produce statistics"); - let mut row = GenericRow::new(2); - row.set_field(0, 1_i32); - row.set_field(1, "alice"); - let record = WriteRecord::for_append( - Arc::clone(&table_info), - physical_table_path.clone(), - 1, - &row, - ); - builder.append(&record)?; - - let mut row2 = GenericRow::new(2); - row2.set_field(0, 2_i32); - row2.set_field(1, "bob"); - let record2 = - WriteRecord::for_append(Arc::clone(&table_info), physical_table_path, 2, &row2); - builder.append(&record2)?; - - let data = builder.build()?; - - // Write to file - let mut tmp_file = NamedTempFile::new()?; - tmp_file.write_all(&data)?; - tmp_file.flush()?; - - // Create file-backed LogRecordsBatches (should stream, not load all into memory) - let file_path = tmp_file.path().to_path_buf(); - let file = File::open(&file_path)?; - let mut batches = LogRecordsBatches::from_file(file, 0, file_path)?; + for to_append_record_batch in [false, true] { + let mut builder = + builder_with_statistics(&row_type, Some(vec![0, 1]), to_append_record_batch); + if to_append_record_batch { + let record = WriteRecord::for_append_record_batch( + Arc::clone(&table_info), + physical_table_path.clone(), + 1, + expected_batch.clone(), + ); + builder.append(&record)?; + } else { + for (id, name) in [(1, "alice"), (2, "bob")] { + let mut row = GenericRow::new(2); + row.set_field(0, id); + row.set_field(1, name); + let record = WriteRecord::for_append( + Arc::clone(&table_info), + physical_table_path.clone(), + 1, + &row, + ); + builder.append(&record)?; + } + } - // Iterate through batches (should work just like in-memory) - let batch = batches.next().expect("Should have at least one batch")?; - assert!(batch.size_in_bytes() > 0); - assert_eq!(batch.record_count(), 2); + let bytes = builder.build()?; + let batch = LogRecordBatch::new(Bytes::from(bytes.clone())); + assert_eq!(batch.magic(), LOG_MAGIC_VALUE_V1); + assert!(batch.is_valid(), "the CRC must cover the statistics"); + + let statistics_length = LittleEndian::read_i32( + &bytes[V1_STATISTICS_LENGTH_OFFSET..V1_STATISTICS_DATA_OFFSET], + ) as usize; + assert_eq!(statistics_length, expected_statistics.len()); + assert_eq!( + &bytes[V1_STATISTICS_DATA_OFFSET..V1_STATISTICS_DATA_OFFSET + statistics_length], + &expected_statistics[..] + ); + let read_context = ReadContext::new( + to_arrow_schema(&row_type)?, + Arc::new(row_type.clone()), + false, + ); + let records: Vec<_> = batch.records(&read_context)?.collect(); + let mut ids = Vec::new(); + for record in &records { + ids.push(record.row().get_int(0)?); + } + assert_eq!(ids, vec![1, 2]); + } Ok(()) } - /// Builds an append-only `(id INT, name STRING)` Arrow log batch from `rows`. - /// The writer always emits append-only batches, so changelog tests derive - /// their bytes from this with [`splice_change_type_vector`]. - fn build_append_only_batch(rows: &[(i32, &str)]) -> (RowType, Vec) { + #[test] + fn no_statistics_mapping_keeps_the_v0_format() { + let (_, append_only) = build_append_only_batch(&[(1, "alice")]); + let batch = LogRecordBatch::new(Bytes::from(append_only)); + assert_eq!(batch.magic(), LOG_MAGIC_VALUE_V0); + } + + #[test] + fn empty_statistics_mapping_writes_a_v1_batch_with_no_statistics() -> Result<()> { + // A table can enable statistics while no column supports them. let row_type = RowType::new(vec![ DataField::new("id".to_string(), DataTypes::int(), None), DataField::new("name".to_string(), DataTypes::string(), None), @@ -2642,151 +2053,56 @@ mod tests { let table_info = Arc::new(build_table_info(table_path.clone(), 1, 1)); let physical_table_path = Arc::new(PhysicalTablePath::of(Arc::new(table_path))); - let mut builder = MemoryLogRecordsArrowBuilder::new( - 1, - &row_type, - false, - ArrowCompressionInfo { - compression_type: ArrowCompressionType::None, - compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, - }, - usize::MAX, - Arc::new(ArrowCompressionRatioEstimator::default()), - ) - .unwrap(); - - for (id, name) in rows { - let mut row = GenericRow::new(2); - row.set_field(0, *id); - row.set_field(1, *name); - let record = WriteRecord::for_append( - Arc::clone(&table_info), - physical_table_path.clone(), - 1, - &row, - ); - builder.append(&record).unwrap(); - } - - (row_type, builder.build().unwrap()) - } - - /// Turns an append-only batch into a wire-valid changelog batch: clears the - /// append-only flag, splices one change-type byte per record between the - /// header and the Arrow payload, then fixes up the length field and CRC. - fn splice_change_type_vector(append_only: &[u8], change_types: &[ChangeType]) -> Vec { - let mut data = append_only.to_vec(); - data[ATTRIBUTES_OFFSET] &= !APPEND_ONLY_FLAG_MASK; - let change_bytes = change_types.iter().map(|ct| ct.to_byte_value()); - data.splice(RECORDS_OFFSET..RECORDS_OFFSET, change_bytes); - - let new_length = (data.len() - LOG_OVERHEAD) as i32; - data[LENGTH_OFFSET..LENGTH_OFFSET + LENGTH_LENGTH] - .copy_from_slice(&new_length.to_le_bytes()); + let mut builder = builder_with_statistics(&row_type, Some(Vec::new()), false); + let mut row = GenericRow::new(2); + row.set_field(0, 1_i32); + row.set_field(1, "alice"); + let record = WriteRecord::for_append(table_info, physical_table_path, 1, &row); + builder.append(&record)?; - let crc = crc32c(&data[SCHEMA_ID_OFFSET..]); - data[CRC_OFFSET..CRC_OFFSET + CRC_LENGTH].copy_from_slice(&crc.to_le_bytes()); - data - } + let bytes = builder.build()?; + let batch = LogRecordBatch::new(Bytes::from(bytes.clone())); + assert_eq!(batch.magic(), LOG_MAGIC_VALUE_V1); + let statistics_length = + LittleEndian::read_i32(&bytes[V1_STATISTICS_LENGTH_OFFSET..V1_STATISTICS_DATA_OFFSET]); + assert_eq!(statistics_length, 0); - #[test] - fn decode_changelog_batch_applies_per_record_change_types() -> Result<()> { - let (row_type, append_only) = - build_append_only_batch(&[(1, "alice"), (2, "bob"), (3, "carol")]); let read_context = ReadContext::new(to_arrow_schema(&row_type)?, Arc::new(row_type), false); - - // Append-only batch: every record decodes as AppendOnly (regression guard). - let batch = LogRecordsBatches::new(append_only.clone()) - .next() - .expect("append-only batch")?; - assert!(batch.is_append_only()); - let records: Vec<_> = batch.records(&read_context)?.collect(); - assert_eq!(records.len(), 3); - assert!( - records - .iter() - .all(|r| *r.change_type() == ChangeType::AppendOnly) - ); - - // Changelog variant: the spliced change-type vector drives per-record types. - let change_types = [ - ChangeType::Insert, - ChangeType::UpdateAfter, - ChangeType::Delete, - ]; - let changelog = splice_change_type_vector(&append_only, &change_types); - let batch = LogRecordsBatches::new(changelog) - .next() - .expect("changelog batch")?; - assert!(!batch.is_append_only()); - assert_eq!(batch.record_count(), 3); - - let records: Vec<_> = batch.records(&read_context)?.collect(); - let got: Vec = records.iter().map(|r| *r.change_type()).collect(); - assert_eq!(got, change_types.to_vec()); - - // The row payload and offsets survive the splice unchanged. - let mut ids = Vec::new(); - for record in &records { - ids.push(record.row().get_int(0)?); - } - assert_eq!(ids, vec![1, 2, 3]); - let offsets: Vec = records.iter().map(|r| r.offset()).collect(); - assert_eq!(offsets, vec![0, 1, 2]); - - // Batch-level access skips the change-type vector and still decodes rows. - let batch = LogRecordsBatches::new(splice_change_type_vector(&append_only, &change_types)) - .next() - .expect("changelog batch")?; - assert_eq!(batch.record_batch(&read_context)?.num_rows(), 3); - + assert_eq!(batch.record_batch(&read_context)?.num_rows(), 1); Ok(()) } #[test] - fn decode_changelog_batch_rejects_invalid_change_type_byte() { - let (row_type, append_only) = build_append_only_batch(&[(1, "a"), (2, "b")]); - let read_context = ReadContext::new( - to_arrow_schema(&row_type).unwrap(), - Arc::new(row_type), - false, + fn estimated_size_reserves_room_for_the_statistics() { + let row_type = RowType::new(vec![ + DataField::new("id".to_string(), DataTypes::int(), None), + DataField::new("name".to_string(), DataTypes::string(), None), + ]); + let with_statistics = builder_with_statistics(&row_type, Some(vec![0, 1]), false); + let without_statistics = builder_with_statistics(&row_type, None, false); + assert_eq!( + with_statistics.estimated_size_in_bytes() + - without_statistics.estimated_size_in_bytes(), + STATISTICS_LENGTH_LENGTH + estimated_serialized_size(&row_type, &[0, 1]) ); - - let mut changelog = - splice_change_type_vector(&append_only, &[ChangeType::Insert, ChangeType::Insert]); - // Corrupt the second change-type byte to an out-of-range value. - changelog[RECORDS_OFFSET + 1] = 99; - - let batch = LogRecordBatch::new(Bytes::from(changelog)); - let err = batch - .records(&read_context) - .err() - .expect("expected decode to reject an invalid change-type byte"); - assert!(matches!(err, Error::UnexpectedError { .. })); - assert!(err.to_string().contains("change type")); } #[test] - fn decode_changelog_batch_rejects_truncated_change_type_vector() { - let (row_type, append_only) = build_append_only_batch(&[(1, "a"), (2, "b")]); - let read_context = ReadContext::new( - to_arrow_schema(&row_type).unwrap(), - Arc::new(row_type), + fn builder_rejects_an_out_of_range_statistics_index() { + let row_type = RowType::new(vec![DataField::new( + "id".to_string(), + DataTypes::int(), + None, + )]); + let err = MemoryLogRecordsArrowBuilder::new( + ArrowBatchConfig { + stats_index_mapping: Some(vec![5]), + ..uncompressed_arrow_batch_config(1, &row_type, usize::MAX) + }, false, - ); - - // Clear the append-only flag, then cut the body shorter than the - // record_count change-type bytes the decoder now expects. - let mut data = append_only; - data[ATTRIBUTES_OFFSET] &= !APPEND_ONLY_FLAG_MASK; - data.truncate(RECORDS_OFFSET + 1); - - let batch = LogRecordBatch::new(Bytes::from(data)); - assert_eq!(batch.record_count(), 2); - let err = batch - .records(&read_context) - .err() - .expect("expected decode to reject a truncated change-type vector"); - assert!(matches!(err, Error::UnexpectedError { .. })); + ) + .err() + .expect("an out-of-range statistics index must be rejected"); + assert!(err.to_string().contains("out of range")); } } diff --git a/fluss-rust/crates/fluss/src/record/log_record_batch.rs b/fluss-rust/crates/fluss/src/record/log_record_batch.rs new file mode 100644 index 00000000000..f7c323364ad --- /dev/null +++ b/fluss-rust/crates/fluss/src/record/log_record_batch.rs @@ -0,0 +1,1298 @@ +// 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 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 field to the header. + 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 { + 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 { + // 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) +} + +#[allow( + dead_code, + reason = "mirrors Java's LogRecordBatchFormat default magic" +)] +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) -> 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 { + 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, // 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 { + 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 { + 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 { + 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) -> 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 { + 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> { + 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; + + fn next(&mut self) -> Option { + 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 leader epoch field of a V2 batch, or [`NO_LEADER_EPOCH`] for + /// magics before V2. + pub fn leader_epoch(&self) -> i32 { + if self.magic() < LOG_MAGIC_VALUE_V2 { + return NO_LEADER_EPOCH; + } + let offset = V2_LEADER_EPOCH_OFFSET; + LittleEndian::read_i32(&self.data[offset..offset + LEADER_EPOCH_LENGTH]) + } + + pub fn writer_id(&self) -> i64 { + let offset = WRITE_CLIENT_ID_OFFSET + self.header_field_shift(); + LittleEndian::read_i64(&self.data[offset..offset + WRITE_CLIENT_ID_LENGTH]) + } + + pub fn batch_sequence(&self) -> i32 { + let offset = BATCH_SEQUENCE_OFFSET + self.header_field_shift(); + LittleEndian::read_i32(&self.data[offset..offset + BATCH_SEQUENCE_LENGTH]) + } + + pub fn ensure_valid(&self) -> Result<()> { + // TODO enable validation once checksum handling is corrected. + Ok(()) + } + + pub fn is_valid(&self) -> bool { + match self.ensure_header_complete() { + Ok(header_size) => { + self.size_in_bytes() >= header_size && self.checksum() == self.compute_checksum() + } + Err(_) => false, + } + } + + /// Rejects unsupported magic versions and batches shorter than their + /// magic's fixed header, so the fixed-offset accessors cannot slice past + /// the buffer. Mirrors the guard in Java's `DefaultLogRecordBatch`. + fn ensure_header_complete(&self) -> Result { + if self.data.len() <= MAGIC_OFFSET { + return Err(Error::UnexpectedError { + message: format!( + "Corrupt log record batch: data length {} does not reach the magic byte", + self.data.len() + ), + source: None, + }); + } + let magic = self.magic(); + let header_size = record_batch_header_size(magic)?; + if self.data.len() < header_size { + return Err(Error::UnexpectedError { + message: format!( + "Corrupt log record batch: data length {} is less than the V{magic} header size {header_size}", + self.data.len() + ), + source: None, + }); + } + Ok(header_size) + } + + fn compute_checksum(&self) -> u32 { + let start = SCHEMA_ID_OFFSET + self.header_field_shift(); + crc32c(&self.data[start..]) + } + + fn attributes(&self) -> u8 { + self.data[ATTRIBUTES_OFFSET + self.header_field_shift()] + } + + /// Whether this batch is append-only (see [`APPEND_ONLY_FLAG_MASK`]). + fn is_append_only(&self) -> bool { + self.attributes() & APPEND_ONLY_FLAG_MASK != 0 + } + + pub fn next_log_offset(&self) -> i64 { + self.last_log_offset() + 1 + } + + pub fn checksum(&self) -> u32 { + let offset = CRC_OFFSET + self.header_field_shift(); + LittleEndian::read_u32(&self.data[offset..offset + CRC_LENGTH]) + } + + pub fn schema_id(&self) -> i16 { + let offset = SCHEMA_ID_OFFSET + self.header_field_shift(); + LittleEndian::read_i16(&self.data[offset..offset + SCHEMA_ID_LENGTH]) + } + + pub fn base_log_offset(&self) -> i64 { + let offset = BASE_OFFSET_OFFSET; + LittleEndian::read_i64(&self.data[offset..offset + BASE_OFFSET_LENGTH]) + } + + pub fn last_log_offset(&self) -> i64 { + self.base_log_offset() + self.last_offset_delta() as i64 + } + + fn last_offset_delta(&self) -> i32 { + let offset = LAST_OFFSET_DELTA_OFFSET + self.header_field_shift(); + LittleEndian::read_i32(&self.data[offset..offset + LAST_OFFSET_DELTA_LENGTH]) + } + + pub fn size_in_bytes(&self) -> usize { + let offset = LENGTH_OFFSET; + LittleEndian::read_i32(&self.data[offset..offset + LENGTH_LENGTH]) as usize + LOG_OVERHEAD + } + + pub fn record_count(&self) -> i32 { + let offset = RECORDS_COUNT_OFFSET + self.header_field_shift(); + LittleEndian::read_i32(&self.data[offset..offset + RECORDS_COUNT_LENGTH]) + } + + /// Offset where the records data starts, mirroring Java's + /// `DefaultLogRecordBatch.recordsDataOffset`. + fn records_data_offset(&self) -> Result { + let magic = self.magic(); + let header_size = record_batch_header_size(magic)?; + if magic < LOG_MAGIC_VALUE_V1 { + return Ok(header_size); + } + let offset = V1_STATISTICS_LENGTH_OFFSET + self.header_field_shift(); + let statistics_length = self + .data + .get(offset..offset + STATISTICS_LENGTH_LENGTH) + .map(LittleEndian::read_i32) + .ok_or_else(|| Error::UnexpectedError { + message: format!( + "Corrupt log record batch: data length {} is less than the V{magic} header size {header_size}", + self.data.len(), + ), + source: None, + })?; + if statistics_length < 0 { + return Err(Error::UnexpectedError { + message: format!( + "Corrupt log record batch: negative statistics length {statistics_length}" + ), + source: None, + }); + } + Ok(header_size + statistics_length as usize) + } + + /// Splits the batch body into its per-record change types and the trailing + /// Arrow IPC payload (see [`APPEND_ONLY_FLAG_MASK`] for the layout). + fn decode_change_types(&self) -> Result<(BatchChangeTypes, &[u8])> { + let records_offset = self.records_data_offset()?; + let body = self + .data + .get(records_offset..) + .ok_or_else(|| Error::UnexpectedError { + message: format!( + "Corrupt log record batch: data length {} is less than the records offset {records_offset}", + self.data.len(), + ), + source: None, + })?; + + if self.is_append_only() { + return Ok((BatchChangeTypes::Uniform(ChangeType::AppendOnly), body)); + } + + let record_count = self.record_count(); + if record_count < 0 { + return Err(Error::UnexpectedError { + message: format!("Corrupt changelog batch: negative record count {record_count}"), + source: None, + }); + } + let record_count = record_count as usize; + let (change_type_bytes, arrow_data) = + body.split_at_checked(record_count) + .ok_or_else(|| Error::UnexpectedError { + message: format!( + "Corrupt changelog batch: body length {} is smaller than its \ + {record_count}-record change-type vector", + body.len() + ), + source: None, + })?; + + let mut change_types = Vec::with_capacity(record_count); + for &byte in change_type_bytes { + let change_type = + ChangeType::from_byte_value(byte).map_err(|message| Error::UnexpectedError { + message, + source: None, + })?; + change_types.push(change_type); + } + + Ok((BatchChangeTypes::PerRecord(change_types), arrow_data)) + } + + pub fn records(&self, read_context: &ReadContext) -> Result { + // Gate on the magic and length first: header offsets of an unsupported + // version or a truncated batch must not be interpreted with this layout. + self.ensure_header_complete()?; + if self.record_count() == 0 { + return Ok(LogRecordIterator::empty()); + } + + let (change_types, arrow_data) = self.decode_change_types()?; + let record_batch = read_context.record_batch(arrow_data)?; + let arrow_reader = ArrowReader::new_with_fluss_row_type( + Arc::new(record_batch), + read_context.row_type_arc(), + read_context.fluss_row_type().cloned(), + )?; + let iterator = ArrowLogRecordIterator::new( + arrow_reader, + self.base_log_offset(), + self.commit_timestamp(), + change_types, + )?; + + Ok(LogRecordIterator::Arrow(iterator)) + } + + pub fn records_for_remote_log(&self, read_context: &ReadContext) -> Result { + self.ensure_header_complete()?; + if self.record_count() == 0 { + return Ok(LogRecordIterator::empty()); + } + + let (change_types, arrow_data) = self.decode_change_types()?; + let record_batch = read_context.record_batch_for_remote_log(arrow_data)?; + let log_record_iterator = match record_batch { + None => LogRecordIterator::empty(), + Some(record_batch) => { + let arrow_reader = ArrowReader::new_with_fluss_row_type( + Arc::new(record_batch), + read_context.row_type_arc(), + read_context.fluss_row_type().cloned(), + )?; + let iterator = ArrowLogRecordIterator::new( + arrow_reader, + self.base_log_offset(), + self.commit_timestamp(), + change_types, + )?; + LogRecordIterator::Arrow(iterator) + } + }; + Ok(log_record_iterator) + } + + /// Returns the record batch directly without creating an iterator. + /// This is more efficient when you need the entire batch rather than + /// iterating row-by-row. + pub fn record_batch(&self, read_context: &ReadContext) -> Result { + self.ensure_header_complete()?; + if self.record_count() == 0 { + // Return empty batch with correct schema + return Ok(RecordBatch::new_empty(read_context.target_schema())); + } + + // Batch access drops the change-type vector; use `records()` for CDC. + let (_, arrow_data) = self.decode_change_types()?; + read_context.record_batch(arrow_data) + } +} + +pub enum LogRecordIterator { + Empty, + Arrow(ArrowLogRecordIterator), +} + +impl LogRecordIterator { + pub fn empty() -> Self { + LogRecordIterator::Empty + } +} + +impl Iterator for LogRecordIterator { + type Item = ScanRecord; + + fn next(&mut self) -> Option { + match self { + LogRecordIterator::Empty => None, + LogRecordIterator::Arrow(iter) => iter.next(), + } + } +} + +/// Per-record change types decoded from a log batch. +/// +/// Append-only batches carry no change-type vector on the wire, so a single +/// `AppendOnly` value covers every record without allocating. Changelog batches +/// (the CDC stream of a primary-key table) decode one change type per record, +/// in record order. +pub(crate) enum BatchChangeTypes { + /// Every record shares this change type (append-only batches). + Uniform(ChangeType), + /// One change type per record, indexed by row id (changelog batches). + PerRecord(Vec), +} + +impl BatchChangeTypes { + pub(crate) fn get(&self, row_id: usize) -> ChangeType { + match self { + BatchChangeTypes::Uniform(change_type) => *change_type, + BatchChangeTypes::PerRecord(change_types) => change_types[row_id], + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::error::Result; + use crate::metadata::{DataField, DataTypes, RowType}; + use crate::record::{MemoryLogRecordsArrowBuilder, to_arrow_schema}; + use crate::row::DataGetters; + use std::io::Write; + + use crate::test_utils::{ + build_append_only_batch, build_table_info, splice_change_type_vector, + splice_statistics_section, uncompressed_arrow_batch_config, + }; + #[test] + fn checksum_and_schema_id_read_minimum_header() { + // Header-only batches with record_count == 0 are valid; this covers the minimal bytes + // needed for checksum/schema_id access. + let mut data = vec![0u8; SCHEMA_ID_OFFSET + SCHEMA_ID_LENGTH]; + let crc = 0xA1B2C3D4u32; + let schema_id = 42i16; + LittleEndian::write_u32(&mut data[CRC_OFFSET..CRC_OFFSET + CRC_LENGTH], crc); + LittleEndian::write_i16( + &mut data[SCHEMA_ID_OFFSET..SCHEMA_ID_OFFSET + SCHEMA_ID_LENGTH], + schema_id, + ); + + let batch = LogRecordBatch::new(Bytes::from(data)); + assert_eq!(batch.checksum(), crc); + assert_eq!(batch.schema_id(), schema_id); + + let expected = crc32c(&batch.data[SCHEMA_ID_OFFSET..]); + assert_eq!(batch.compute_checksum(), expected); + } + + // Tests for file-backed streaming + + #[test] + fn test_file_source_streaming() -> Result<()> { + use tempfile::NamedTempFile; + + // Test 1: Basic file reads work + let test_data = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10]; + let mut tmp_file = NamedTempFile::new()?; + tmp_file.write_all(&test_data)?; + tmp_file.flush()?; + + let file_path = tmp_file.path().to_path_buf(); + let file = File::open(&file_path)?; + let mut source = FileSource::new(file, 0, file_path)?; + + // Read full data + let data = source.read_batch_data(0, 10)?; + assert_eq!(data.to_vec(), test_data); + + // Read partial data + let partial = source.read_batch_data(2, 5)?; + assert_eq!(partial.to_vec(), vec![3, 4, 5, 6, 7]); + + // Test 2: base_offset works (critical for remote logs with pos_in_log_segment) + let prefix = vec![0xFF; 100]; + let actual_data = vec![1, 2, 3, 4, 5]; + let mut tmp_file2 = NamedTempFile::new()?; + tmp_file2.write_all(&prefix)?; + tmp_file2.write_all(&actual_data)?; + tmp_file2.flush()?; + + let file_path2 = tmp_file2.path().to_path_buf(); + let file2 = File::open(&file_path2)?; + let mut source2 = FileSource::new(file2, 100, file_path2)?; // Skip first 100 bytes + + assert_eq!(source2.total_size(), 5); // Only counts data after offset + let data2 = source2.read_batch_data(0, 5)?; + assert_eq!(data2.to_vec(), actual_data); + + Ok(()) + } + + #[test] + fn test_log_records_batches_from_file() -> Result<()> { + use crate::client::WriteRecord; + use crate::metadata::{PhysicalTablePath, TablePath}; + use crate::row::GenericRow; + use tempfile::NamedTempFile; + + // Integration test: Real log record batch streamed from file + let row_type = RowType::new(vec![ + DataField::new("id".to_string(), DataTypes::int(), None), + DataField::new("name".to_string(), DataTypes::string(), None), + ]); + let table_path = TablePath::new("db".to_string(), "tbl".to_string()); + let table_info = Arc::new(build_table_info(table_path.clone(), 1, 1)); + let physical_table_path = Arc::new(PhysicalTablePath::of(Arc::new(table_path))); + + let mut builder = MemoryLogRecordsArrowBuilder::new( + uncompressed_arrow_batch_config(1, &row_type, usize::MAX), + false, + )?; + + let mut row = GenericRow::new(2); + row.set_field(0, 1_i32); + row.set_field(1, "alice"); + let record = WriteRecord::for_append( + Arc::clone(&table_info), + physical_table_path.clone(), + 1, + &row, + ); + builder.append(&record)?; + + let mut row2 = GenericRow::new(2); + row2.set_field(0, 2_i32); + row2.set_field(1, "bob"); + let record2 = + WriteRecord::for_append(Arc::clone(&table_info), physical_table_path, 2, &row2); + builder.append(&record2)?; + + let data = builder.build()?; + + // Write to file + let mut tmp_file = NamedTempFile::new()?; + tmp_file.write_all(&data)?; + tmp_file.flush()?; + + // Create file-backed LogRecordsBatches (should stream, not load all into memory) + let file_path = tmp_file.path().to_path_buf(); + let file = File::open(&file_path)?; + let mut batches = LogRecordsBatches::from_file(file, 0, file_path)?; + + // Iterate through batches (should work just like in-memory) + let batch = batches.next().expect("Should have at least one batch")?; + assert!(batch.size_in_bytes() > 0); + assert_eq!(batch.record_count(), 2); + + Ok(()) + } + + #[test] + fn decode_changelog_record_batch_applies_per_record_change_types() -> Result<()> { + let (row_type, append_only) = + build_append_only_batch(&[(1, "alice"), (2, "bob"), (3, "carol")]); + let read_context = ReadContext::new(to_arrow_schema(&row_type)?, Arc::new(row_type), false); + + // Append-only batch: every record decodes as AppendOnly (regression guard). + let batch = LogRecordsBatches::new(append_only.clone()) + .next() + .expect("append-only batch")?; + assert!(batch.is_append_only()); + let records: Vec<_> = batch.records(&read_context)?.collect(); + assert_eq!(records.len(), 3); + assert!( + records + .iter() + .all(|r| *r.change_type() == ChangeType::AppendOnly) + ); + + // Changelog variant: the spliced change-type vector drives per-record types. + let change_types = [ + ChangeType::Insert, + ChangeType::UpdateAfter, + ChangeType::Delete, + ]; + let changelog = splice_change_type_vector(&append_only, &change_types); + let batch = LogRecordsBatches::new(changelog) + .next() + .expect("changelog batch")?; + assert!(!batch.is_append_only()); + assert_eq!(batch.record_count(), 3); + + let records: Vec<_> = batch.records(&read_context)?.collect(); + let got: Vec = records.iter().map(|r| *r.change_type()).collect(); + assert_eq!(got, change_types.to_vec()); + + // The row payload and offsets survive the splice unchanged. + let mut ids = Vec::new(); + for record in &records { + ids.push(record.row().get_int(0)?); + } + assert_eq!(ids, vec![1, 2, 3]); + let offsets: Vec = records.iter().map(|r| r.offset()).collect(); + assert_eq!(offsets, vec![0, 1, 2]); + + // Batch-level access skips the change-type vector and still decodes rows. + let batch = LogRecordsBatches::new(splice_change_type_vector(&append_only, &change_types)) + .next() + .expect("changelog batch")?; + assert_eq!(batch.record_batch(&read_context)?.num_rows(), 3); + + Ok(()) + } + + #[test] + fn decode_changelog_record_batch_rejects_invalid_change_type_byte() { + let (row_type, append_only) = build_append_only_batch(&[(1, "a"), (2, "b")]); + let read_context = ReadContext::new( + to_arrow_schema(&row_type).unwrap(), + Arc::new(row_type), + false, + ); + + let mut changelog = + splice_change_type_vector(&append_only, &[ChangeType::Insert, ChangeType::Insert]); + // Corrupt the second change-type byte to an out-of-range value. + changelog[RECORDS_OFFSET + 1] = 99; + + let batch = LogRecordBatch::new(Bytes::from(changelog)); + let err = batch + .records(&read_context) + .err() + .expect("expected decode to reject an invalid change-type byte"); + assert!(matches!(err, Error::UnexpectedError { .. })); + assert!(err.to_string().contains("change type")); + } + + #[test] + fn decode_changelog_record_batch_rejects_truncated_change_type_vector() { + let (row_type, append_only) = build_append_only_batch(&[(1, "a"), (2, "b")]); + let read_context = ReadContext::new( + to_arrow_schema(&row_type).unwrap(), + Arc::new(row_type), + false, + ); + + // Clear the append-only flag, then cut the body shorter than the + // record_count change-type bytes the decoder now expects. + let mut data = append_only; + data[ATTRIBUTES_OFFSET] &= !APPEND_ONLY_FLAG_MASK; + data.truncate(RECORDS_OFFSET + 1); + + let batch = LogRecordBatch::new(Bytes::from(data)); + assert_eq!(batch.record_count(), 2); + let err = batch + .records(&read_context) + .err() + .expect("expected decode to reject a truncated change-type vector"); + assert!(matches!(err, Error::UnexpectedError { .. })); + } + + #[test] + fn header_size_follows_the_magic_version() { + assert_eq!(record_batch_header_size(LOG_MAGIC_VALUE_V0).unwrap(), 48); + assert_eq!(record_batch_header_size(LOG_MAGIC_VALUE_V1).unwrap(), 52); + assert_eq!(record_batch_header_size(LOG_MAGIC_VALUE_V2).unwrap(), 56); + let err = record_batch_header_size(3).expect_err("V3 is not supported"); + assert!(err.to_string().contains("Unsupported magic value 3")); + } + + #[test] + fn decode_v1_batch_skips_the_statistics_section() -> Result<()> { + let (row_type, append_only) = build_append_only_batch(&[(1, "alice"), (2, "bob")]); + let read_context = ReadContext::new(to_arrow_schema(&row_type)?, Arc::new(row_type), false); + + // The reader must step over the statistics blindly, whatever they hold. + let statistics = vec![0xAB_u8; 37]; + let v1 = splice_statistics_section(&append_only, &statistics); + let batch = LogRecordsBatches::new(v1).next().expect("V1 batch")?; + assert_eq!(batch.magic(), LOG_MAGIC_VALUE_V1); + assert!( + batch.is_valid(), + "the CRC must cover the spliced statistics" + ); + + let records: Vec<_> = batch.records(&read_context)?.collect(); + let mut ids = Vec::new(); + for record in &records { + ids.push(record.row().get_int(0)?); + } + assert_eq!(ids, vec![1, 2]); + assert_eq!(batch.record_batch(&read_context)?.num_rows(), 2); + + // The server's projection path emits V1 with an empty statistics section. + let v1_empty = splice_statistics_section(&append_only, &[]); + let batch = LogRecordsBatches::new(v1_empty).next().expect("V1 batch")?; + assert_eq!(batch.magic(), LOG_MAGIC_VALUE_V1); + assert_eq!(batch.record_batch(&read_context)?.num_rows(), 2); + Ok(()) + } + + #[test] + fn decode_v1_changelog_record_batch_reads_change_types_after_statistics() -> Result<()> { + let (row_type, append_only) = build_append_only_batch(&[(1, "a"), (2, "b")]); + let read_context = ReadContext::new(to_arrow_schema(&row_type)?, Arc::new(row_type), false); + + // Change types first (a V0 changelog), then the statistics in front of + // them, giving the V1 layout [header][statistics][changeTypes][arrow]. + let change_types = [ChangeType::Insert, ChangeType::Delete]; + let changelog = splice_change_type_vector(&append_only, &change_types); + let v1 = splice_statistics_section(&changelog, &[0xCD_u8; 21]); + + let batch = LogRecordsBatches::new(v1).next().expect("V1 changelog")?; + assert!(!batch.is_append_only()); + let records: Vec<_> = batch.records(&read_context)?.collect(); + let got: Vec = records.iter().map(|r| *r.change_type()).collect(); + assert_eq!(got, change_types.to_vec()); + Ok(()) + } + + #[test] + fn decode_rejects_an_unsupported_magic_version() { + let (row_type, append_only) = build_append_only_batch(&[(1, "a")]); + let read_context = ReadContext::new( + to_arrow_schema(&row_type).unwrap(), + Arc::new(row_type), + false, + ); + + let mut data = append_only; + data[MAGIC_OFFSET] = 3; + // A future magic shifting the header would leave a stale value at this + // offset, so a zero here made the batch read as empty and get silently + // dropped before the magic gate. + data[RECORDS_COUNT_OFFSET..RECORDS_COUNT_OFFSET + RECORDS_COUNT_LENGTH] + .copy_from_slice(&0_i32.to_le_bytes()); + let batch = LogRecordBatch::new(Bytes::from(data)); + let err = batch + .records(&read_context) + .err() + .expect("V3 batches must be rejected, not misparsed"); + assert!(err.to_string().contains("Unsupported magic value 3")); + let err = batch + .record_batch(&read_context) + .expect_err("batch mode must reject V3 too"); + assert!(err.to_string().contains("Unsupported magic value 3")); + } + + #[test] + fn decode_rejects_a_batch_shorter_than_its_header() { + let (row_type, append_only) = build_append_only_batch(&[(1, "a")]); + let read_context = ReadContext::new( + to_arrow_schema(&row_type).unwrap(), + Arc::new(row_type), + false, + ); + + // A corrupt length field can hand the iterator a batch shorter than + // the fixed header; decoding must error instead of panicking. + let mut truncated = append_only[..RECORD_BATCH_HEADER_SIZE - 8].to_vec(); + let declared = (truncated.len() - LOG_OVERHEAD) as i32; + truncated[LENGTH_OFFSET..LENGTH_OFFSET + LENGTH_LENGTH] + .copy_from_slice(&declared.to_le_bytes()); + let batch = LogRecordsBatches::new(truncated) + .next() + .expect("the iterator must yield the truncated batch") + .expect("reading the truncated batch bytes must succeed"); + let err = batch + .records(&read_context) + .err() + .expect("a batch shorter than its header must be rejected"); + assert!(err.to_string().contains("less than the V0 header size")); + assert!(!batch.is_valid()); + + // A V1 batch cut between the V0 and V1 header sizes. + let v1 = splice_statistics_section(&append_only, &[]); + let batch = LogRecordBatch::new(Bytes::from(v1[..RECORD_BATCH_HEADER_SIZE + 2].to_vec())); + let err = batch + .records(&read_context) + .err() + .expect("a truncated V1 batch must be rejected"); + assert!(err.to_string().contains("less than the V1 header size")); + assert!(!batch.is_valid()); + + // A buffer that does not even reach the magic byte. + let batch = LogRecordBatch::new(Bytes::from(vec![0_u8; MAGIC_OFFSET])); + assert!(!batch.is_valid()); + let err = batch + .records(&read_context) + .err() + .expect("a batch without a magic byte must be rejected"); + assert!(err.to_string().contains("does not reach the magic byte")); + } + + #[test] + fn is_valid_is_false_for_an_unsupported_magic() { + let (_, append_only) = build_append_only_batch(&[(1, "a")]); + let mut data = append_only; + data[MAGIC_OFFSET] = 3; + assert!(!LogRecordBatch::new(Bytes::from(data)).is_valid()); + } + + /// Turns a V1 batch into a wire-valid V2 batch by inserting the leader + /// epoch before the CRC and fixing up the magic, length field and CRC. + fn splice_leader_epoch(v1_batch: &[u8], leader_epoch: i32) -> Vec { + let mut data = v1_batch.to_vec(); + data[MAGIC_OFFSET] = LOG_MAGIC_VALUE_V2; + data.splice( + V2_LEADER_EPOCH_OFFSET..V2_LEADER_EPOCH_OFFSET, + leader_epoch.to_le_bytes(), + ); + + let new_length = (data.len() - LOG_OVERHEAD) as i32; + data[LENGTH_OFFSET..LENGTH_OFFSET + LENGTH_LENGTH] + .copy_from_slice(&new_length.to_le_bytes()); + let crc_offset = CRC_OFFSET + LEADER_EPOCH_LENGTH; + let crc = crc32c(&data[SCHEMA_ID_OFFSET + LEADER_EPOCH_LENGTH..]); + data[crc_offset..crc_offset + CRC_LENGTH].copy_from_slice(&crc.to_le_bytes()); + data + } + + #[test] + fn decode_v2_batch_reads_records_after_the_leader_epoch() -> Result<()> { + let (row_type, append_only) = build_append_only_batch(&[(1, "alice"), (2, "bob")]); + let read_context = ReadContext::new(to_arrow_schema(&row_type)?, Arc::new(row_type), false); + + let v1 = splice_statistics_section(&append_only, &[0xAB_u8; 19]); + let v2 = splice_leader_epoch(&v1, 7); + let batch = LogRecordsBatches::new(v2).next().expect("V2 batch")?; + assert_eq!(batch.magic(), LOG_MAGIC_VALUE_V2); + assert_eq!(batch.leader_epoch(), 7); + assert!( + batch.is_valid(), + "the CRC must be read from its shifted V2 offset" + ); + // Every post-CRC header field must come from its shifted offset. + assert_eq!(batch.record_count(), 2); + assert_eq!(batch.schema_id(), 1); + assert_eq!(batch.writer_id(), NO_WRITER_ID); + assert_eq!(batch.batch_sequence(), NO_BATCH_SEQUENCE); + assert_eq!(batch.last_log_offset(), 1); + + let records: Vec<_> = batch.records(&read_context)?.collect(); + let mut ids = Vec::new(); + for record in &records { + ids.push(record.row().get_int(0)?); + } + assert_eq!(ids, vec![1, 2]); + assert_eq!(batch.record_batch(&read_context)?.num_rows(), 2); + + // The server can also emit V2 with an empty statistics section. + let v2_empty = splice_leader_epoch(&splice_statistics_section(&append_only, &[]), 7); + let batch = LogRecordsBatches::new(v2_empty).next().expect("V2 batch")?; + assert_eq!(batch.record_batch(&read_context)?.num_rows(), 2); + Ok(()) + } + + #[test] + fn decode_v2_changelog_batch_reads_change_types_after_statistics() -> Result<()> { + let (row_type, append_only) = build_append_only_batch(&[(1, "a"), (2, "b")]); + let read_context = ReadContext::new(to_arrow_schema(&row_type)?, Arc::new(row_type), false); + + let change_types = [ChangeType::Insert, ChangeType::Delete]; + let changelog = splice_change_type_vector(&append_only, &change_types); + let v1 = splice_statistics_section(&changelog, &[0xCD_u8; 11]); + let v2 = splice_leader_epoch(&v1, 3); + + let batch = LogRecordsBatches::new(v2).next().expect("V2 changelog")?; + assert!(!batch.is_append_only()); + let records: Vec<_> = batch.records(&read_context)?.collect(); + let got: Vec = records.iter().map(|r| *r.change_type()).collect(); + assert_eq!(got, change_types.to_vec()); + Ok(()) + } + + #[test] + fn pre_v2_batches_report_no_leader_epoch() { + let (_, append_only) = build_append_only_batch(&[(1, "a")]); + let batch = LogRecordBatch::new(Bytes::from(append_only.clone())); + assert_eq!(batch.leader_epoch(), NO_LEADER_EPOCH); + + let v1 = splice_statistics_section(&append_only, &[]); + let batch = LogRecordBatch::new(Bytes::from(v1)); + assert_eq!(batch.leader_epoch(), NO_LEADER_EPOCH); + } + + #[test] + fn decode_rejects_a_negative_statistics_length() { + let (row_type, append_only) = build_append_only_batch(&[(1, "a")]); + let read_context = ReadContext::new( + to_arrow_schema(&row_type).unwrap(), + Arc::new(row_type), + false, + ); + + let mut v1 = splice_statistics_section(&append_only, &[]); + v1[V1_STATISTICS_LENGTH_OFFSET..V1_STATISTICS_LENGTH_OFFSET + STATISTICS_LENGTH_LENGTH] + .copy_from_slice(&(-1_i32).to_le_bytes()); + let batch = LogRecordBatch::new(Bytes::from(v1)); + let err = batch + .records(&read_context) + .err() + .expect("a negative statistics length must be rejected"); + assert!(err.to_string().contains("negative statistics length")); + } + + #[test] + fn decode_rejects_a_statistics_length_past_the_batch_end() { + let (row_type, append_only) = build_append_only_batch(&[(1, "a")]); + let read_context = ReadContext::new( + to_arrow_schema(&row_type).unwrap(), + Arc::new(row_type), + false, + ); + + let mut v1 = splice_statistics_section(&append_only, &[]); + let past_the_end = v1.len() as i32; + v1[V1_STATISTICS_LENGTH_OFFSET..V1_STATISTICS_LENGTH_OFFSET + STATISTICS_LENGTH_LENGTH] + .copy_from_slice(&past_the_end.to_le_bytes()); + let batch = LogRecordBatch::new(Bytes::from(v1)); + let err = batch + .records(&read_context) + .err() + .expect("a statistics length past the batch end must be rejected"); + assert!(err.to_string().contains("records offset")); + } +} diff --git a/fluss-rust/crates/fluss/src/record/mod.rs b/fluss-rust/crates/fluss/src/record/mod.rs index c30d49a20ed..fa50664bb09 100644 --- a/fluss-rust/crates/fluss/src/record/mod.rs +++ b/fluss-rust/crates/fluss/src/record/mod.rs @@ -24,13 +24,13 @@ use std::collections::HashMap; mod arrow; mod error; pub mod kv; -// Reachable once the Arrow builder emits V1 batches. -#[allow(dead_code, reason = "consumed by the V1 batch builder")] +mod log_record_batch; mod statistics; pub(crate) use statistics::is_supported_statistics_type; pub use arrow::*; +pub(crate) use log_record_batch::*; #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum ChangeType { diff --git a/fluss-rust/crates/fluss/src/record/statistics.rs b/fluss-rust/crates/fluss/src/record/statistics.rs index 6e24a9ee103..80f4ceeb5a6 100644 --- a/fluss-rust/crates/fluss/src/record/statistics.rs +++ b/fluss-rust/crates/fluss/src/record/statistics.rs @@ -47,7 +47,7 @@ use crate::error::{Error, Result}; use crate::metadata::{DataType, RowType}; -use crate::row::aligned::AlignedRowWriter; +use crate::row::aligned::{AlignedRowWriter, calculate_fix_part_size_in_bytes}; use crate::row::binary::BinaryWriter; use crate::row::{Decimal, TimestampLtz, TimestampNtz}; use arrow::array::{Array, RecordBatch}; @@ -62,6 +62,52 @@ use arrow::datatypes::{ /// `LogRecordBatchFormat.STATISTICS_VERSION`. const STATISTICS_VERSION: u8 = 1; +/// Matches Java's `LogRecordBatchStatisticsWriter.VARIABLE_LENGTH_FIELD_ESTIMATE`. +const VARIABLE_LENGTH_FIELD_ESTIMATE: usize = 16; + +/// Rough serialized statistics size for `mapping`, mirroring Java's +/// `LogRecordBatchStatisticsWriter.estimatedSizeInBytes` with both bound rows +/// assumed present. +pub(crate) fn estimated_serialized_size(row_type: &RowType, mapping: &[usize]) -> usize { + // Version, column count, indexes and null counts, then the two + // length-prefixed bound rows. + let header = 3 + mapping.len() * (2 + 4); + header + 2 * (4 + estimated_row_size(row_type, mapping)) +} + +/// Mirrors Java's `LogRecordBatchStatisticsWriter.getRowSizeEstimate`. +fn estimated_row_size(row_type: &RowType, mapping: &[usize]) -> usize { + let mut estimate = calculate_fix_part_size_in_bytes(mapping.len()); + for &index in mapping { + if !is_in_fixed_length_part(row_type.fields()[index].data_type()) { + estimate += VARIABLE_LENGTH_FIELD_ESTIMATE; + } + } + estimate +} + +/// Whether an aligned row stores this type inline in its 8-byte slot, +/// mirroring Java's `AlignedRow.isInFixedLengthPart`. +fn is_in_fixed_length_part(data_type: &DataType) -> bool { + match data_type { + DataType::Boolean(_) + | DataType::TinyInt(_) + | DataType::SmallInt(_) + | DataType::Int(_) + | DataType::BigInt(_) + | DataType::Float(_) + | DataType::Double(_) + | DataType::Date(_) + | DataType::Time(_) => true, + DataType::Decimal(decimal_type) => Decimal::is_compact_precision(decimal_type.precision()), + DataType::Timestamp(timestamp_type) => TimestampNtz::is_compact(timestamp_type.precision()), + DataType::TimestampLTz(timestamp_type) => { + TimestampLtz::is_compact(timestamp_type.precision()) + } + _ => false, + } +} + /// Whether statistics can be collected for `data_type`, mirroring Java's /// `DataTypeChecks.isSupportedStatisticsType`. pub(crate) fn is_supported_statistics_type(data_type: &DataType) -> bool { @@ -254,6 +300,50 @@ fn column_bounds(column: &dyn Array, data_type: &DataType) -> Result {{ + let array = column + .as_any() + .downcast_ref::>() + .ok_or_else(|| unexpected_array(column, data_type))?; + let java_cmp = + |a: <$arrow_ty as arrow::datatypes::ArrowPrimitiveType>::Native, + b: <$arrow_ty as arrow::datatypes::ArrowPrimitiveType>::Native| { + match (a.is_nan(), b.is_nan()) { + (true, true) => std::cmp::Ordering::Equal, + (true, false) => std::cmp::Ordering::Greater, + (false, true) => std::cmp::Ordering::Less, + (false, false) => a.total_cmp(&b), + } + }; + let mut bounds = None; + for value in array.iter().flatten() { + bounds = Some(match bounds { + None => (value, value), + Some((min, max)) => ( + if java_cmp(value, min).is_lt() { + value + } else { + min + }, + if java_cmp(value, max).is_gt() { + value + } else { + max + }, + ), + }); + } + Ok(bounds.map(|(min, max)| ColumnBounds::$variant(min, max))) + }}; + } + match data_type { DataType::Boolean(_) => { let array = column @@ -270,8 +360,8 @@ fn column_bounds(column: &dyn Array, data_type: &DataType) -> Result primitive!(Int32Type, Int32), DataType::Date(_) => primitive!(Date32Type, Int32), DataType::BigInt(_) => primitive!(Int64Type, Int64), - DataType::Float(_) => primitive!(Float32Type, Float32), - DataType::Double(_) => primitive!(Float64Type, Float64), + DataType::Float(_) => float!(Float32Type, Float32), + DataType::Double(_) => float!(Float64Type, Float64), // Fluss stores TIME as millis of day, so every unit but millisecond // has to be converted back from what the Arrow array holds. DataType::Time(_) => match column.data_type() { @@ -350,9 +440,11 @@ fn column_bounds(column: &dyn Array, data_type: &DataType) -> Result Err(Error::IllegalArgument { - message: format!("Statistics are not supported for column type {other:?}"), - }), + // Java's collector skips unsupported types per column (null bounds, + // block still emitted), so a server-side whitelist that grows before + // this client's does degrades gracefully instead of dropping the + // statistics for every column. + _ => Ok(None), } } @@ -688,6 +780,78 @@ mod tests { assert_eq!(f64::from_le_bytes(max[8..16].try_into().unwrap()), 2.5); } + #[test] + fn orders_nan_bounds_like_java() { + // Java's `Float.compare` treats every NaN as one largest value, so a + // hardware-produced negative NaN must become the maximum, not the + // minimum the way arrow's totalOrder aggregate would make it. The + // retained NaN keeps its raw bits, matching Java's keep-first ties. + let neg_nan = f32::from_bits(0xFFC0_0000); + let (min, max) = single_column_rows( + DataTypes::float(), + ArrowType::Float32, + Arc::new(Float32Array::from(vec![ + Some(-0.0), + Some(neg_nan), + Some(1.0), + ])), + ); + assert_eq!( + u32::from_le_bytes(min[8..12].try_into().unwrap()), + (-0.0_f32).to_bits(), + "-0.0 must stay the minimum, below 0.0 and NaN" + ); + assert_eq!( + u32::from_le_bytes(max[8..12].try_into().unwrap()), + 0xFFC0_0000, + "the NaN bound must keep the raw bits of the NaN it saw" + ); + + // An all-NaN column has NaN as both bounds. + let neg_nan = f64::from_bits(0xFFF8_0000_0000_0000); + let (min, max) = single_column_rows( + DataTypes::double(), + ArrowType::Float64, + Arc::new(Float64Array::from(vec![Some(neg_nan)])), + ); + assert_eq!( + u64::from_le_bytes(min[8..16].try_into().unwrap()), + 0xFFF8_0000_0000_0000 + ); + assert_eq!( + u64::from_le_bytes(max[8..16].try_into().unwrap()), + 0xFFF8_0000_0000_0000 + ); + } + + #[test] + fn skips_an_unsupported_column_type_with_null_bounds() { + // Mirrors Java's per-column degradation: the block is still emitted + // and the unsupported column just carries null bounds, so a server + // whitelist that grows before this client's degrades gracefully. + let rt = RowType::new(vec![DataField::new("v", DataTypes::bytes(), None)]); + let schema = Schema::new(vec![Field::new("v", ArrowType::Binary, true)]); + let batch = RecordBatch::try_new( + Arc::new(schema), + vec![Arc::new(arrow::array::BinaryArray::from(vec![Some( + &b"ab"[..], + )]))], + ) + .expect("batch"); + let bytes = serialize_statistics(&batch, &rt, &[0]) + .expect("an unsupported column type must not fail the block") + .expect("statistics"); + let (_, _, _, nulls) = parse_prefix(&bytes, 1); + assert_eq!(nulls, vec![0]); + let min_len = i32::from_le_bytes(bytes[9..13].try_into().unwrap()) as usize; + let min_row = &bytes[13..13 + min_len]; + assert_eq!( + min_row[1] & 0x01, + 0x01, + "the unsupported column must carry a null bound" + ); + } + #[test] fn collects_bounds_for_char_like_a_string() { let (min, max) = single_column_rows( @@ -921,4 +1085,144 @@ 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, a copy of fluss-common's checked-in + /// `encoding/statistics_block.hex` fixture that + /// `LogRecordBatchStatisticsCompatibilityTest` generates and asserts, so + /// both languages pin to one set of bytes. Embedded so the test also runs + /// outside the monorepo; in the monorepo the copies are asserted identical. + fn java_statistics_block_hex() -> String { + let embedded = include_str!("testdata/statistics_block.hex").trim(); + let java_path = concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../../fluss-common/src/test/resources/encoding/statistics_block.hex" + ); + if let Ok(java_fixture) = std::fs::read_to_string(java_path) { + assert_eq!( + java_fixture.trim(), + embedded, + "testdata/statistics_block.hex is out of sync with fluss-common's fixture" + ); + } + embedded.to_string() + } + + #[test] + fn matches_the_java_writer_byte_for_byte() { + // CHAR is excluded because Java's collector records no CHAR bounds. + let row_type = RowType::new(vec![ + DataField::new("bool", DataTypes::boolean(), None), + DataField::new("i8", DataTypes::tinyint(), None), + DataField::new("i16", DataTypes::smallint(), None), + DataField::new("i32", DataTypes::int(), None), + DataField::new("i64", DataTypes::bigint(), None), + DataField::new("f32", DataTypes::float(), None), + DataField::new("f64", DataTypes::double(), None), + DataField::new("str", DataTypes::string(), None), + DataField::new("dec5", DataTypes::decimal(5, 2), None), + DataField::new("dec20", DataTypes::decimal(20, 3), None), + DataField::new("date", DataTypes::date(), None), + DataField::new("time", DataTypes::time(), None), + DataField::new("ts3", DataTypes::timestamp_with_precision(3), None), + DataField::new("ts6", DataTypes::timestamp_with_precision(6), None), + DataField::new("ltz3", DataTypes::timestamp_ltz_with_precision(3), None), + DataField::new("ltz6", DataTypes::timestamp_ltz_with_precision(6), None), + DataField::new("strnull", DataTypes::string(), None), + DataField::new("f32x", DataTypes::float(), None), + DataField::new("f64x", DataTypes::double(), None), + ]); + let millis = ArrowType::Timestamp(arrow::datatypes::TimeUnit::Millisecond, None); + let micros = ArrowType::Timestamp(arrow::datatypes::TimeUnit::Microsecond, None); + let schema = Schema::new(vec![ + Field::new("bool", ArrowType::Boolean, true), + Field::new("i8", ArrowType::Int8, true), + Field::new("i16", ArrowType::Int16, true), + Field::new("i32", ArrowType::Int32, true), + Field::new("i64", ArrowType::Int64, true), + Field::new("f32", ArrowType::Float32, true), + Field::new("f64", ArrowType::Float64, true), + Field::new("str", ArrowType::Utf8, true), + Field::new("dec5", ArrowType::Decimal128(5, 2), true), + Field::new("dec20", ArrowType::Decimal128(20, 3), true), + Field::new("date", ArrowType::Date32, true), + Field::new( + "time", + ArrowType::Time32(arrow::datatypes::TimeUnit::Millisecond), + true, + ), + Field::new("ts3", millis.clone(), true), + Field::new("ts6", micros.clone(), true), + Field::new("ltz3", millis, true), + Field::new("ltz6", micros, true), + Field::new("strnull", ArrowType::Utf8, true), + Field::new("f32x", ArrowType::Float32, true), + Field::new("f64x", ArrowType::Float64, true), + ]); + let batch = RecordBatch::try_new( + Arc::new(schema), + vec![ + Arc::new(BooleanArray::from(vec![true, false, true])), + Arc::new(Int8Array::from(vec![1, -3, 7])), + Arc::new(Int16Array::from(vec![100, 200, -50])), + Arc::new(Int32Array::from(vec![Some(10), None, Some(30)])), + Arc::new(Int64Array::from(vec![1000, -2000, 3000])), + Arc::new(Float32Array::from(vec![1.5, -2.5, 0.5])), + Arc::new(Float64Array::from(vec![3.25, 1.25, 9.75])), + Arc::new(StringArray::from(vec!["banana", "apple", "cherry"])), + Arc::new( + Decimal128Array::from(vec![12345_i128, 6789, 50000]) + .with_precision_and_scale(5, 2) + .expect("dec5"), + ), + Arc::new( + Decimal128Array::from(vec![12345678901_i128, 1234, 99999999999999999]) + .with_precision_and_scale(20, 3) + .expect("dec20"), + ), + Arc::new(Date32Array::from(vec![19000, 18000, 20000])), + Arc::new(Time32MillisecondArray::from(vec![ + 3600000, 7200000, 1800000, + ])), + Arc::new(TimestampMillisecondArray::from(vec![ + 1700000000123, + 1600000000000, + 1800000000999, + ])), + Arc::new(TimestampMicrosecondArray::from(vec![ + 1700000000123456, + 1600000000000001, + 1800000000999999, + ])), + Arc::new(TimestampMillisecondArray::from(vec![ + 1700000000123, + 1600000000000, + 1800000000999, + ])), + Arc::new(TimestampMicrosecondArray::from(vec![ + 1700000000123456, + 1600000000000001, + 1800000000999999, + ])), + Arc::new(StringArray::from(vec![None::<&str>, None, None])), + Arc::new(Float32Array::from(vec![ + -0.0, + f32::from_bits(0xFFC0_0000), + 0.0, + ])), + Arc::new(Float64Array::from(vec![ + 0.0, + -0.0, + f64::from_bits(0xFFF8_0000_0000_0000), + ])), + ], + ) + .expect("batch"); + + let mapping: Vec = (0..19).collect(); + let bytes = serialize_statistics(&batch, &row_type, &mapping) + .expect("serialize") + .expect("statistics"); + let hex: String = bytes.iter().map(|b| format!("{b:02x}")).collect(); + assert_eq!(hex, java_statistics_block_hex()); + } } diff --git a/fluss-rust/crates/fluss/src/record/testdata/statistics_block.hex b/fluss-rust/crates/fluss/src/record/testdata/statistics_block.hex new file mode 100644 index 00000000000..bce645214fe --- /dev/null +++ b/fluss-rust/crates/fluss/src/record/testdata/statistics_block.hex @@ -0,0 +1 @@ +01130000000100020003000400050006000700080009000a000b000c000d000e000f0010001100120000000000000000000000000001000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000030000000000000000000000c000000000000001000000000000000000000000fd00000000000000ceff0000000000000a0000000000000030f8ffffffffffff000020c000000000000000000000f43f6170706c65000085851a00000000000002000000a0000000504600000000000040771b000000000000806e8774010000e8030000b000000000806e8774010000e8030000b800000000000000000000000000008000000000000000000000008004d2000000000000000000000000000000806e877401000000806e8774010000c0000000000000010000000001000000000000000700000000000000c8000000000000001e00000000000000b80b0000000000000000c03f000000000000000000802340636865727279008650c300000000000008000000a0000000204e00000000000000dd6d0000000000e7535c18a3010000583e0f00b0000000e7535c18a3010000583e0f00b800000000000000000000000000c0ff00000000000000000000f8ff016345785d89ffff0000000000000000e7535c18a3010000e7535c18a3010000 \ No newline at end of file diff --git a/fluss-rust/crates/fluss/src/row/aligned/aligned_row_writer.rs b/fluss-rust/crates/fluss/src/row/aligned/aligned_row_writer.rs index 04247750a48..046dd9e6769 100644 --- a/fluss-rust/crates/fluss/src/row/aligned/aligned_row_writer.rs +++ b/fluss-rust/crates/fluss/src/row/aligned/aligned_row_writer.rs @@ -81,10 +81,6 @@ impl AlignedRowWriter { Bytes::copy_from_slice(&self.buffer[..self.cursor]) } - pub fn size_in_bytes(&self) -> usize { - self.cursor - } - fn field_offset(&self, pos: usize) -> usize { self.null_bits_size_in_bytes + 8 * pos } @@ -118,6 +114,11 @@ fn calculate_bit_set_width_in_bytes(arity: usize) -> usize { ((arity + 63 + HEADER_SIZE_IN_BITS) / 64) * 8 } +/// Mirrors Java's `AlignedRow.calculateFixPartSizeInBytes`. +pub(crate) fn calculate_fix_part_size_in_bytes(arity: usize) -> usize { + calculate_bit_set_width_in_bytes(arity) + 8 * arity +} + impl BinaryWriter for AlignedRowWriter { fn reset(&mut self) { self.cursor = self.fixed_size; diff --git a/fluss-rust/crates/fluss/src/row/aligned/mod.rs b/fluss-rust/crates/fluss/src/row/aligned/mod.rs index 4ccb5b760d8..bd3b9fa3055 100644 --- a/fluss-rust/crates/fluss/src/row/aligned/mod.rs +++ b/fluss-rust/crates/fluss/src/row/aligned/mod.rs @@ -17,4 +17,5 @@ mod aligned_row_writer; -pub use aligned_row_writer::AlignedRowWriter; +pub(crate) use aligned_row_writer::AlignedRowWriter; +pub(crate) use aligned_row_writer::calculate_fix_part_size_in_bytes; diff --git a/fluss-rust/crates/fluss/src/row/mod.rs b/fluss-rust/crates/fluss/src/row/mod.rs index 5d606deae13..9e9307350d0 100644 --- a/fluss-rust/crates/fluss/src/row/mod.rs +++ b/fluss-rust/crates/fluss/src/row/mod.rs @@ -25,7 +25,7 @@ pub mod view; pub(crate) mod datum; mod decimal; -pub mod aligned; +pub(crate) mod aligned; pub mod binary; pub(crate) mod column_writer; pub mod compacted; diff --git a/fluss-rust/crates/fluss/src/test_utils.rs b/fluss-rust/crates/fluss/src/test_utils.rs index e1f31bf93a7..c05d3637d09 100644 --- a/fluss-rust/crates/fluss/src/test_utils.rs +++ b/fluss-rust/crates/fluss/src/test_utils.rs @@ -15,15 +15,47 @@ // specific language governing permissions and limitations // under the License. +use crate::client::WriteRecord; use crate::cluster::{BucketLocation, Cluster, ServerNode, ServerType}; +use crate::compression::{ + ArrowCompressionInfo, ArrowCompressionRatioEstimator, ArrowCompressionType, + DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, +}; use crate::metadata::{ - DataField, DataTypes, PhysicalTablePath, Schema, TableBucket, TableDescriptor, TableInfo, - TablePath, + DataField, DataTypes, PhysicalTablePath, RowType, Schema, TableBucket, TableDescriptor, + TableInfo, TablePath, }; use crate::metrics::{LABEL_DATABASE, LABEL_TABLE, ScannerMetrics}; +use crate::record::{ + APPEND_ONLY_FLAG_MASK, ATTRIBUTES_OFFSET, ArrowBatchConfig, CRC_LENGTH, CRC_OFFSET, ChangeType, + LENGTH_LENGTH, LENGTH_OFFSET, LOG_MAGIC_VALUE_V1, LOG_OVERHEAD, MAGIC_OFFSET, + MemoryLogRecordsArrowBuilder, RECORDS_OFFSET, SCHEMA_ID_OFFSET, +}; +use crate::row::GenericRow; +use crc32c::crc32c; use std::collections::HashMap; use std::sync::Arc; +/// An uncompressed [`ArrowBatchConfig`] with no statistics and a fresh +/// compression ratio estimator. +pub(crate) fn uncompressed_arrow_batch_config( + schema_id: i32, + row_type: &RowType, + write_limit: usize, +) -> ArrowBatchConfig { + ArrowBatchConfig { + schema_id, + row_type: row_type.clone(), + stats_index_mapping: None, + compression: ArrowCompressionInfo { + compression_type: ArrowCompressionType::None, + compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL, + }, + write_limit, + compression_ratio_estimator: Arc::new(ArrowCompressionRatioEstimator::default()), + } +} + pub(crate) fn build_table_info(table_path: TablePath, table_id: i64, buckets: i32) -> TableInfo { build_table_info_with_columns( table_path, @@ -178,3 +210,74 @@ pub(crate) fn assert_scanner_entries_labeled( ); } } + +/// Builds an append-only `(id INT, name STRING)` Arrow log batch from `rows`. +/// The writer always emits append-only batches, so changelog tests derive +/// their bytes from this with [`splice_change_type_vector`]. +pub(crate) fn build_append_only_batch(rows: &[(i32, &str)]) -> (RowType, Vec) { + let row_type = RowType::new(vec![ + DataField::new("id".to_string(), DataTypes::int(), None), + DataField::new("name".to_string(), DataTypes::string(), None), + ]); + let table_path = TablePath::new("db".to_string(), "tbl".to_string()); + let table_info = Arc::new(build_table_info(table_path.clone(), 1, 1)); + let physical_table_path = Arc::new(PhysicalTablePath::of(Arc::new(table_path))); + + let mut builder = MemoryLogRecordsArrowBuilder::new( + uncompressed_arrow_batch_config(1, &row_type, usize::MAX), + false, + ) + .unwrap(); + + for (id, name) in rows { + let mut row = GenericRow::new(2); + row.set_field(0, *id); + row.set_field(1, *name); + let record = WriteRecord::for_append( + Arc::clone(&table_info), + physical_table_path.clone(), + 1, + &row, + ); + builder.append(&record).unwrap(); + } + + (row_type, builder.build().unwrap()) +} + +/// Turns a V0 batch into a wire-valid V1 batch by splicing a +/// length-prefixed statistics section before the records data and fixing +/// up the magic, length field and CRC. +pub(crate) fn splice_statistics_section(v0_batch: &[u8], statistics: &[u8]) -> Vec { + let mut data = v0_batch.to_vec(); + data[MAGIC_OFFSET] = LOG_MAGIC_VALUE_V1; + let mut section = (statistics.len() as i32).to_le_bytes().to_vec(); + section.extend_from_slice(statistics); + data.splice(RECORDS_OFFSET..RECORDS_OFFSET, section); + + let new_length = (data.len() - LOG_OVERHEAD) as i32; + data[LENGTH_OFFSET..LENGTH_OFFSET + LENGTH_LENGTH].copy_from_slice(&new_length.to_le_bytes()); + let crc = crc32c(&data[SCHEMA_ID_OFFSET..]); + data[CRC_OFFSET..CRC_OFFSET + CRC_LENGTH].copy_from_slice(&crc.to_le_bytes()); + data +} + +/// Turns an append-only batch into a wire-valid changelog batch: clears the +/// append-only flag, splices one change-type byte per record between the +/// header and the Arrow payload, then fixes up the length field and CRC. +pub(crate) fn splice_change_type_vector( + append_only: &[u8], + change_types: &[ChangeType], +) -> Vec { + let mut data = append_only.to_vec(); + data[ATTRIBUTES_OFFSET] &= !APPEND_ONLY_FLAG_MASK; + let change_bytes = change_types.iter().map(|ct| ct.to_byte_value()); + data.splice(RECORDS_OFFSET..RECORDS_OFFSET, change_bytes); + + let new_length = (data.len() - LOG_OVERHEAD) as i32; + data[LENGTH_OFFSET..LENGTH_OFFSET + LENGTH_LENGTH].copy_from_slice(&new_length.to_le_bytes()); + + let crc = crc32c(&data[SCHEMA_ID_OFFSET..]); + data[CRC_OFFSET..CRC_OFFSET + CRC_LENGTH].copy_from_slice(&crc.to_le_bytes()); + data +} diff --git a/pom.xml b/pom.xml index c765972876e..d3b621a174d 100644 --- a/pom.xml +++ b/pom.xml @@ -676,6 +676,8 @@ tools/releasing/release/** **/fluss-bin/conf/servers + + **/src/test/resources/encoding/*.hex website/**/_category_.json website/package.json