[rust] Wire batch statistics into the log batch write and read paths

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<Object[]> 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();
+        }
+    }
+}

fluss-common/src/test/resources/encoding/statistics_block.hex

@@ -0,0 +1 @@
+01130000000100020003000400050006000700080009000a000b000c000d000e000f0010001100120000000000000000000000000001000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000030000000000000000000000c000000000000001000000000000000000000000fd00000000000000ceff0000000000000a0000000000000030f8ffffffffffff000020c000000000000000000000f43f6170706c65000085851a00000000000002000000a0000000504600000000000040771b000000000000806e8774010000e8030000b000000000806e8774010000e8030000b800000000000000000000000000008000000000000000000000008004d2000000000000000000000000000000806e877401000000806e8774010000c0000000000000010000000001000000000000000700000000000000c8000000000000001e00000000000000b80b0000000000000000c03f000000000000000000802340636865727279008650c300000000000008000000a0000000204e00000000000000dd6d0000000000e7535c18a3010000583e0f00b0000000e7535c18a3010000583e0f00b800000000000000000000000000c0ff00000000000000000000f8ff016345785d89ffff0000000000000000e7535c18a3010000e7535c18a3010000

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

fluss-rust/crates/fluss/src/client/table/batch_scanner.rs

@@ -459,18 +459,14 @@ 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,
     };
     use crate::record::MemoryLogRecordsArrowBuilder;
     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() {

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