Add cross-language Split.serialize()

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -39,6 +39,8 @@ class TableSchema:
 class Split:
     def __init__(self, state: bytes) -> None: ...
     def row_count(self) -> int: ...
+    # Java SplitSerializer v1 binary: a DataSplit (v8), or an IndexedSplit when the split has row ranges.
+    def serialize(self) -> bytes: ...
 
 class Plan:
     def splits(self) -> List[Split]: ...

bindings/python/src/read.rs

@@ -310,6 +310,14 @@ impl PySplit {
         self.inner.row_count()
     }
 
+    /// Serialize this planned split to the Java `SplitSerializer` (v1) binary, so pypaimon (or
+    /// any Paimon reader) can rebuild it without re-planning. A split carrying row ranges is
+    /// serialized as an `IndexedSplit`.
+    fn serialize<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
+        let bytes = self.inner.serialize_split_v1().map_err(to_py_err)?;
+        Ok(PyBytes::new(py, &bytes))
+    }
+
     /// Reduce to `Split(bytes)` for pickle/copy. The bytes are an opaque,
     /// implementation-detail encoding; only same/compatible-version round-trip
     /// is guaranteed.

bindings/python/tests/test_read.py

@@ -657,3 +657,58 @@ def test_time_travel_conflicting_selectors_raises():
         # both offending keys are named
         assert "scan.snapshot-id" in str(exc.value)
         assert "scan.tag-name" in str(exc.value)
+
+
+def test_split_serialize_produces_split_v1_binary():
+    import struct
+
+    with tempfile.TemporaryDirectory() as warehouse:
+        table = _make_table_with_data(warehouse)
+        splits = table.new_read_builder().new_scan().plan().splits()
+        assert splits
+        data = splits[0].serialize()
+        assert isinstance(data, (bytes, bytearray))
+        # SplitSerializer frame: "SPLIT_V1" + version(1) + type-id(1=DATA_SPLIT)
+        assert data[:8] == b"SPLIT_V1"
+        version, type_id = struct.unpack_from(">ii", data, 8)
+        assert version == 1
+        assert type_id == 1
+        # DataSplit v8 body follows: MAGIC + VERSION(8)
+        magic, body_version = struct.unpack_from(">qi", data, 16)
+        assert magic == -2394839472490812314
+        assert body_version == 8
+        assert splits[0].serialize() == data  # deterministic
+
+
+def test_split_serialize_carries_partition_and_reads():
+    with tempfile.TemporaryDirectory() as warehouse:
+        table = _make_partitioned_table(warehouse)
+        b = table.new_read_builder().with_filter(
+            {"method": "equal", "field": "dt", "literals": ["p1"]})
+        splits = b.new_scan().plan().splits()
+        assert splits
+        assert b"p1" in splits[0].serialize()  # partition value encoded
+        # partition filter pruned to p1
+        t = pa.Table.from_batches(b.new_read().read(splits))
+        assert sorted(t.column("id").to_pylist()) == [1, 2]
+
+
+def test_split_serialize_encodes_deletions_and_external_path():
+    # plan() can't produce these; inject them via the serde payload.
+    import json
+
+    with tempfile.TemporaryDirectory() as warehouse:
+        table = _make_string_table(warehouse)  # two files
+        split = table.new_read_builder().new_scan().plan().splits()[0]
+        cls, (raw,) = split.__reduce__()
+        payload = json.loads(bytes(raw))
+        n = len(payload["data_files"])
+        assert n >= 2
+        payload["data_deletion_files"] = [
+            {"path": "dv/idx-0", "offset": 4, "length": 20, "cardinality": 3}
+        ] + [None] * (n - 1)
+        payload["data_files"][0]["_EXTERNAL_PATH"] = "s3://ext/data-0.parquet"
+
+        data = cls(json.dumps(payload).encode()).serialize()
+        assert b"dv/idx-0" in data                 # deletion file path
+        assert b"s3://ext/data-0.parquet" in data  # external path

crates/paimon/src/spec/binary_row.rs

@@ -533,6 +533,16 @@ impl BinaryRowBuilder {
         self.data[offset + 7] = 0x80 | (value.len() as u8);
     }
 
+    /// Inline (len <= 7) or var-length, matching Java `BinaryRowWriter`. Also correct for
+    /// nested rows / arrays: always > 7 bytes, so they land in the var part like `writeRow`.
+    pub fn write_bytes(&mut self, pos: usize, value: &[u8]) {
+        if value.len() <= 7 {
+            self.write_binary_inline(pos, value);
+        } else {
+            self.write_binary(pos, value);
+        }
+    }
+
     /// Write a compact Decimal (precision <= 18) as its unscaled i64 value.
     pub fn write_decimal_compact(&mut self, pos: usize, unscaled: i64) {
         self.write_long(pos, unscaled);
@@ -594,6 +604,12 @@ impl BinaryRowBuilder {
         serialized
     }
 
+    /// Raw row data without the arity prefix, for embedding in a parent serializer that writes
+    /// its own length (e.g. `writeInt(size) + rowData`).
+    pub fn build_row_data(self) -> Vec<u8> {
+        self.data
+    }
+
     /// Write a Datum value at the given position, dispatching by type.
     pub fn write_datum(&mut self, pos: usize, datum: &Datum, data_type: &DataType) {
         match datum {
@@ -681,6 +697,66 @@ pub fn datums_to_binary_row(datums: &[(&Option<Datum>, &DataType)]) -> Vec<u8> {
     builder.build_serialized()
 }
 
+/// Round up to the nearest 8-byte word (Java `roundNumberOfBytesToNearestWord`).
+fn round_to_word(n: usize) -> usize {
+    let r = n & 7;
+    if r == 0 {
+        n
+    } else {
+        n + (8 - r)
+    }
+}
+
+/// `BinaryArray` header size: 4-byte element count + null bitset (4-byte aligned).
+fn binary_array_header(n: usize) -> usize {
+    4 + n.div_ceil(32) * 4
+}
+
+/// Serialize a `BinaryArray` of non-null UTF-8 strings, matching Java's `BinaryArray`
+/// layout: `[count(int)] [null bits] [8B element slots] [var-length part]`. Each element
+/// is inline (len <= 7) or an offset+length pointer, like `BinaryRowWriter`.
+pub fn serialize_binary_array_str(values: &[String]) -> Vec<u8> {
+    let n = values.len();
+    let header = binary_array_header(n);
+    let mut data = vec![0u8; round_to_word(header + n * 8)];
+    data[0..4].copy_from_slice(&(n as i32).to_le_bytes());
+    for (k, s) in values.iter().enumerate() {
+        let eo = header + k * 8;
+        let b = s.as_bytes();
+        if b.len() <= 7 {
+            data[eo..eo + b.len()].copy_from_slice(b);
+            data[eo + 7] = 0x80 | (b.len() as u8);
+        } else {
+            let var_off = data.len();
+            data.extend_from_slice(b);
+            let pad = (8 - (b.len() % 8)) % 8;
+            data.extend(std::iter::repeat_n(0u8, pad));
+            let encoded = ((var_off as u64) << 32) | (b.len() as u64);
+            data[eo..eo + 8].copy_from_slice(&encoded.to_le_bytes());
+        }
+    }
+    data
+}
+
+/// Serialize a `BinaryArray` of nullable i64 (Java `array<bigint>`): fixed 8-byte
+/// element slots, a set null bit for `None` elements.
+pub fn serialize_binary_array_long(values: &[Option<i64>]) -> Vec<u8> {
+    let n = values.len();
+    let header = binary_array_header(n);
+    let mut data = vec![0u8; round_to_word(header + n * 8)];
+    data[0..4].copy_from_slice(&(n as i32).to_le_bytes());
+    for (k, v) in values.iter().enumerate() {
+        match v {
+            None => data[4 + k / 8] |= 1 << (k % 8),
+            Some(x) => {
+                let eo = header + k * 8;
+                data[eo..eo + 8].copy_from_slice(&x.to_le_bytes());
+            }
+        }
+    }
+    data
+}
+
 /// Extract a Datum from an Arrow RecordBatch column at the given row index.
 pub fn extract_datum_from_arrow(
     batch: &RecordBatch,

crates/paimon/src/spec/data_file.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 use crate::spec::stats::BinaryTableStats;
+use crate::spec::{serialize_binary_array_str, BinaryRowBuilder};
 use chrono::serde::ts_milliseconds_option::deserialize as from_millis_opt;
 use chrono::serde::ts_milliseconds_option::serialize as to_millis_opt;
 use chrono::{DateTime, Utc};
@@ -144,12 +145,72 @@ impl Display for DataFileMeta {
     }
 }
 
+fn opt_long(b: &mut BinaryRowBuilder, pos: usize, v: Option<i64>) {
+    match v {
+        Some(x) => b.write_long(pos, x),
+        None => b.set_null_at(pos),
+    }
+}
+
+fn opt_str_array(b: &mut BinaryRowBuilder, pos: usize, v: &Option<Vec<String>>) {
+    match v {
+        Some(a) => b.write_bytes(pos, &serialize_binary_array_str(a)),
+        None => b.set_null_at(pos),
+    }
+}
+
 impl DataFileMeta {
     /// Returns the row ID range `[first_row_id, first_row_id + row_count - 1]` if `first_row_id` is set.
     pub fn row_id_range(&self) -> Option<(i64, i64)> {
         self.first_row_id.map(|fid| (fid, fid + self.row_count - 1))
     }
 
+    /// Serialize as a `DataFileMeta.SCHEMA` (version 8) BinaryRow, raw data without the
+    /// arity prefix -- the form `DataSplit#serialize` writes per file as `writeInt(len) + data`.
+    /// Fields, order and nullability mirror Java `DataFileMetaSerializer#toRow`.
+    pub fn to_serialized_row_data(&self) -> crate::Result<Vec<u8>> {
+        let mut b = BinaryRowBuilder::new(20);
+        b.write_bytes(0, self.file_name.as_bytes());
+        b.write_long(1, self.file_size);
+        b.write_long(2, self.row_count);
+        b.write_bytes(3, &self.min_key);
+        b.write_bytes(4, &self.max_key);
+        b.write_bytes(5, &self.key_stats.to_simple_stats_row_data());
+        b.write_bytes(6, &self.value_stats.to_simple_stats_row_data());
+        b.write_long(7, self.min_sequence_number);
+        b.write_long(8, self.max_sequence_number);
+        b.write_long(9, self.schema_id);
+        b.write_int(10, self.level);
+        b.write_bytes(11, &serialize_binary_array_str(&self.extra_files));
+        match self.creation_time {
+            Some(t) => b.write_timestamp_compact(12, t.timestamp_millis()),
+            None => b.set_null_at(12),
+        }
+        opt_long(&mut b, 13, self.delete_row_count);
+        match &self.embedded_index {
+            Some(v) => b.write_bytes(14, v),
+            None => b.set_null_at(14),
+        }
+        match self.file_source {
+            Some(v) => {
+                let byte = i8::try_from(v).map_err(|_| crate::Error::DataInvalid {
+                    message: format!("file_source {v} out of TINYINT range [-128, 127]"),
+                    source: None,
+                })?;
+                b.write_byte(15, byte);
+            }
+            None => b.set_null_at(15),
+        }
+        opt_str_array(&mut b, 16, &self.value_stats_cols);
+        match &self.external_path {
+            Some(s) => b.write_bytes(17, s.as_bytes()),
+            None => b.set_null_at(17),
+        }
+        opt_long(&mut b, 18, self.first_row_id);
+        opt_str_array(&mut b, 19, &self.write_cols);
+        Ok(b.build_row_data())
+    }
+
     /// Full path for this data file.
     ///
     /// Mirrors Java `DataFilePathFactory#toPath(DataFileMeta)`: use

crates/paimon/src/spec/stats.rs

@@ -18,7 +18,10 @@
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use std::fmt::{Display, Formatter};
 
-use super::{extract_datum_from_arrow, BinaryRowBuilder, DataType, Datum, EMPTY_SERIALIZED_ROW};
+use super::{
+    extract_datum_from_arrow, serialize_binary_array_long, BinaryRowBuilder, DataType, Datum,
+    EMPTY_SERIALIZED_ROW,
+};
 use arrow_array::RecordBatch;
 
 /// Deserialize `_NULL_COUNTS` which in Avro is `["null", {"type":"array","items":["null","long"]}]`.
@@ -109,6 +112,17 @@ impl BinaryTableStats {
             null_counts: Vec::new(),
         }
     }
+
+    /// Serialize as a `SimpleStats.SCHEMA` BinaryRow (raw data, no arity prefix), matching
+    /// Java `SimpleStats#toRow`: `[_MIN_VALUES bytes] [_MAX_VALUES bytes] [_NULL_COUNTS array<bigint>]`.
+    /// `min_values`/`max_values` are already serialized `BinaryRow`s, written as-is.
+    pub fn to_simple_stats_row_data(&self) -> Vec<u8> {
+        let mut b = BinaryRowBuilder::new(3);
+        b.write_bytes(0, &self.min_values);
+        b.write_bytes(1, &self.max_values);
+        b.write_bytes(2, &serialize_binary_array_long(&self.null_counts));
+        b.build_row_data()
+    }
 }
 
 impl Display for BinaryTableStats {