This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new ca180ad [table] Add cross-language Split.serialize() (SplitSerializer
v1 binary) (#455)
ca180ad is described below
commit ca180ad69a59fe4d8030f749e5bf3d1d82f3764b
Author: XiaoHongbo <[email protected]>
AuthorDate: Thu Jul 9 10:38:49 2026 +0800
[table] Add cross-language Split.serialize() (SplitSerializer v1 binary)
(#455)
---
.../python/python/pypaimon_rust/datafusion.pyi | 2 +
bindings/python/src/read.rs | 8 +
bindings/python/tests/test_read.py | 55 ++++
crates/paimon/src/spec/binary_row.rs | 76 +++++
crates/paimon/src/spec/data_file.rs | 61 ++++
crates/paimon/src/spec/stats.rs | 16 +-
crates/paimon/src/table/goldens/datasplit_v8.bin | Bin 0 -> 690 bytes
crates/paimon/src/table/goldens/split_v1_data.bin | Bin 0 -> 896 bytes
.../paimon/src/table/goldens/split_v1_indexed.bin | Bin 0 -> 961 bytes
crates/paimon/src/table/source.rs | 351 +++++++++++++++++++++
10 files changed, 568 insertions(+), 1 deletion(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 8373789..6b66d1f 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/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]: ...
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index d49f8fd..b5d4008 100644
--- a/bindings/python/src/read.rs
+++ b/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.
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index c37f564..c0ea736 100644
--- a/bindings/python/tests/test_read.py
+++ b/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
diff --git a/crates/paimon/src/spec/binary_row.rs
b/crates/paimon/src/spec/binary_row.rs
index 98bb1c2..d96fa95 100644
--- a/crates/paimon/src/spec/binary_row.rs
+++ b/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,
diff --git a/crates/paimon/src/spec/data_file.rs
b/crates/paimon/src/spec/data_file.rs
index 7350c51..cc90824 100644
--- a/crates/paimon/src/spec/data_file.rs
+++ b/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
diff --git a/crates/paimon/src/spec/stats.rs b/crates/paimon/src/spec/stats.rs
index 4ae9b24..5503a6f 100644
--- a/crates/paimon/src/spec/stats.rs
+++ b/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 {
diff --git a/crates/paimon/src/table/goldens/datasplit_v8.bin
b/crates/paimon/src/table/goldens/datasplit_v8.bin
new file mode 100644
index 0000000..02daf1a
Binary files /dev/null and b/crates/paimon/src/table/goldens/datasplit_v8.bin
differ
diff --git a/crates/paimon/src/table/goldens/split_v1_data.bin
b/crates/paimon/src/table/goldens/split_v1_data.bin
new file mode 100644
index 0000000..6cbade9
Binary files /dev/null and b/crates/paimon/src/table/goldens/split_v1_data.bin
differ
diff --git a/crates/paimon/src/table/goldens/split_v1_indexed.bin
b/crates/paimon/src/table/goldens/split_v1_indexed.bin
new file mode 100644
index 0000000..0d20df1
Binary files /dev/null and
b/crates/paimon/src/table/goldens/split_v1_indexed.bin differ
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index c4254b6..49ab30e 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -634,6 +634,145 @@ impl DataSplit {
pub fn builder() -> DataSplitBuilder {
DataSplitBuilder::new()
}
+
+ /// Serialize the DataSplit fields to Java `DataSplit#serialize` (version
8) binary.
+ /// Byte-compatible with `compatibility/datasplit-v8`. Row ranges are not
part of the v8
+ /// format; `serialize_split_v1` wraps a row-range split as an
`IndexedSplit` instead.
+ pub fn serialize(&self) -> crate::Result<Vec<u8>> {
+ let mut out = Vec::new();
+ out.extend_from_slice(&SPLIT_MAGIC.to_be_bytes());
+ out.extend_from_slice(&SPLIT_VERSION.to_be_bytes());
+ out.extend_from_slice(&self.snapshot_id.to_be_bytes());
+ let p = self.partition.to_serialized_bytes(); // serializeBinaryRow:
writeInt(len) + [arity+data]
+ out.extend_from_slice(&(p.len() as i32).to_be_bytes());
+ out.extend_from_slice(&p);
+ out.extend_from_slice(&self.bucket.to_be_bytes());
+ write_java_utf(&mut out, &self.bucket_path)?;
+ out.push(1); // totalBuckets present (version >= 6)
+ out.extend_from_slice(&self.total_buckets.to_be_bytes());
+ out.extend_from_slice(&0i32.to_be_bytes()); // deprecated beforeFiles
count
+ out.push(0); // beforeDeletionFiles = null list
+ out.extend_from_slice(&(self.data_files.len() as i32).to_be_bytes());
+ for f in &self.data_files {
+ let d = f.to_serialized_row_data()?;
+ out.extend_from_slice(&(d.len() as i32).to_be_bytes());
+ out.extend_from_slice(&d);
+ }
+ write_deletion_list(&mut out, self.data_deletion_files.as_deref())?;
+ out.push(0); // isStreaming = false
+ out.push(u8::from(self.raw_convertible));
+ Ok(out)
+ }
+
+ /// Java `SplitSerializer#serialize` frame: magic + version + type id +
body. The cross-language
+ /// entry point. A plain split serializes as `DataSplit` (type 1); a split
carrying row ranges as
+ /// `IndexedSplit` (type 3) wrapping the DataSplit body plus the ranges.
Byte-compatible with
+ /// `compatibility/split-v1-data` / `split-v1-indexed`.
+ pub fn serialize_split_v1(&self) -> crate::Result<Vec<u8>> {
+ let mut out = Vec::new();
+ out.extend_from_slice(&SPLIT_SER_MAGIC.to_be_bytes());
+ out.extend_from_slice(&SPLIT_SER_VERSION.to_be_bytes());
+ match &self.row_ranges {
+ None => {
+
out.extend_from_slice(&SPLIT_SER_TYPE_DATA_SPLIT.to_be_bytes());
+ out.extend_from_slice(&self.serialize()?);
+ }
+ Some(ranges) => {
+
out.extend_from_slice(&SPLIT_SER_TYPE_INDEXED_SPLIT.to_be_bytes());
+ // IndexedSplit#serialize: magic + version + DataSplit body +
ranges + scores.
+ out.extend_from_slice(&INDEXED_SPLIT_MAGIC.to_be_bytes());
+ out.extend_from_slice(&INDEXED_SPLIT_VERSION.to_be_bytes());
+ out.extend_from_slice(&self.serialize()?);
+ out.extend_from_slice(&(ranges.len() as i32).to_be_bytes());
+ for r in ranges {
+ out.extend_from_slice(&r.from().to_be_bytes());
+ out.extend_from_slice(&r.to().to_be_bytes());
+ }
+ out.push(0); // scores = null (vector scores are not modeled
on the Rust split)
+ }
+ }
+ Ok(out)
+ }
+}
+
+/// Java `DataSplit#MAGIC` / `VERSION` for the serialize format.
+const SPLIT_MAGIC: i64 = -2394839472490812314;
+const SPLIT_VERSION: i32 = 8;
+
+/// Java `SplitSerializer` frame: magic "SPLIT_V1", version, and the
`DataSplit` type id.
+const SPLIT_SER_MAGIC: i64 = 0x53504C49545F5631; // "SPLIT_V1"
+const SPLIT_SER_VERSION: i32 = 1;
+const SPLIT_SER_TYPE_DATA_SPLIT: i32 = 1;
+const SPLIT_SER_TYPE_INDEXED_SPLIT: i32 = 3;
+
+/// Java `IndexedSplit#MAGIC` / `VERSION`.
+const INDEXED_SPLIT_MAGIC: i64 = -938472394838495695;
+const INDEXED_SPLIT_VERSION: i32 = 1;
+
+/// Java `DataOutput#writeUTF`: modified UTF-8 over UTF-16 code units,
prefixed by the u16
+/// byte length. NUL and chars >= U+0800 (incl. surrogates for supplementary
chars) take 2-3
+/// bytes. Errors if the encoded form exceeds 65535 bytes, as Java throws
UTFDataFormatException.
+fn write_java_utf(out: &mut Vec<u8>, s: &str) -> crate::Result<()> {
+ let byte_len: usize = s
+ .encode_utf16()
+ .map(|c| {
+ if (0x0001..=0x007F).contains(&c) {
+ 1
+ } else if c > 0x07FF {
+ 3
+ } else {
+ 2
+ }
+ })
+ .sum();
+ if byte_len > 0xFFFF {
+ return Err(crate::Error::DataInvalid {
+ message: format!("string too long for writeUTF: {byte_len} bytes
(max 65535)"),
+ source: None,
+ });
+ }
+ out.extend_from_slice(&(byte_len as u16).to_be_bytes());
+ for c in s.encode_utf16() {
+ if (0x0001..=0x007F).contains(&c) {
+ out.push(c as u8);
+ } else if c > 0x07FF {
+ out.push(0xE0 | (c >> 12) as u8);
+ out.push(0x80 | ((c >> 6) & 0x3F) as u8);
+ out.push(0x80 | (c & 0x3F) as u8);
+ } else {
+ out.push(0xC0 | (c >> 6) as u8);
+ out.push(0x80 | (c & 0x3F) as u8);
+ }
+ }
+ Ok(())
+}
+
+/// Java `DeletionFile#serializeList`: `0` = null list; else `1` + count + per
entry
+/// (`0` = null, or `1` + path + offset + length + cardinality, -1 when
absent).
+fn write_deletion_list(
+ out: &mut Vec<u8>,
+ files: Option<&[Option<DeletionFile>]>,
+) -> crate::Result<()> {
+ match files {
+ None => out.push(0),
+ Some(list) => {
+ out.push(1);
+ out.extend_from_slice(&(list.len() as i32).to_be_bytes());
+ for f in list {
+ match f {
+ None => out.push(0),
+ Some(df) => {
+ out.push(1);
+ write_java_utf(out, df.path())?;
+ out.extend_from_slice(&df.offset().to_be_bytes());
+ out.extend_from_slice(&df.length().to_be_bytes());
+
out.extend_from_slice(&df.cardinality().unwrap_or(-1).to_be_bytes());
+ }
+ }
+ }
+ }
+ }
+ Ok(())
}
/// Builder for [DataSplit].
@@ -1044,4 +1183,216 @@ mod tests {
vec![(0, 9)]
);
}
+
+ fn single_col(s: &str) -> Vec<u8> {
+ let mut b = crate::spec::BinaryRowBuilder::new(1);
+ b.write_bytes(0, s.as_bytes());
+ b.build_serialized()
+ }
+
+ #[test]
+ fn serialize_matches_datasplit_v8() {
+ use chrono::DateTime;
+ // Golden generated by Java (paimon-core DataSplitCompatibleTest) for
cross-language parity.
+ let expected = include_bytes!("goldens/datasplit_v8.bin");
+
+ let mut pb = crate::spec::BinaryRowBuilder::new(1);
+ pb.write_bytes(0, b"aaaaa");
+
+ let file = DataFileMeta {
+ file_name: "my_file".to_string(),
+ file_size: 1024 * 1024,
+ row_count: 1024,
+ min_key: single_col("min_key"),
+ max_key: single_col("max_key"),
+ key_stats: BinaryTableStats::new(
+ single_col("min_key"),
+ single_col("max_key"),
+ vec![Some(0)],
+ ),
+ value_stats: BinaryTableStats::new(
+ single_col("min_value"),
+ single_col("max_value"),
+ vec![Some(0)],
+ ),
+ min_sequence_number: 15,
+ max_sequence_number: 200,
+ schema_id: 5,
+ level: 3,
+ extra_files: vec!["extra1".to_string(), "extra2".to_string()],
+ creation_time: DateTime::from_timestamp_millis(1646252412000),
+ delete_row_count: Some(11),
+ embedded_index: Some(vec![1, 2, 4]),
+ first_row_id: Some(12),
+ write_cols: Some(["a", "b", "c", "f"].iter().map(|s|
s.to_string()).collect()),
+ external_path: Some("hdfs:///path/to/warehouse".to_string()),
+ file_source: Some(1),
+ value_stats_cols: Some(
+ ["field1", "field2", "field3"]
+ .iter()
+ .map(|s| s.to_string())
+ .collect(),
+ ),
+ };
+
+ let split = DataSplitBuilder::new()
+ .with_snapshot(18)
+ .with_partition(pb.build())
+ .with_bucket(20)
+ .with_total_buckets(32)
+ .with_bucket_path("my path".to_string())
+ .with_data_files(vec![file])
+ .with_data_deletion_files(vec![Some(DeletionFile::new(
+ "deletion_file".to_string(),
+ 100,
+ 22,
+ Some(33),
+ ))])
+ .with_raw_convertible(false)
+ .build()
+ .unwrap();
+
+ assert_eq!(split.serialize().unwrap().as_slice(), &expected[..]);
+ }
+
+ #[test]
+ fn write_java_utf_matches_java_modified_utf8() {
+ let enc = |s: &str| {
+ let mut b = Vec::new();
+ write_java_utf(&mut b, s).unwrap();
+ b
+ };
+ // ASCII: u16 length + raw bytes.
+ assert_eq!(enc("ab"), vec![0, 2, b'a', b'b']);
+ // NUL -> C0 80 (2 bytes), not 0x00.
+ assert_eq!(enc("\u{0}"), vec![0, 2, 0xC0, 0x80]);
+ // U+00E9 'é' -> 2 bytes C3 A9.
+ assert_eq!(enc("\u{00E9}"), vec![0, 2, 0xC3, 0xA9]);
+ // U+4E2D '中' -> 3 bytes E4 B8 AD.
+ assert_eq!(enc("\u{4E2D}"), vec![0, 3, 0xE4, 0xB8, 0xAD]);
+ // U+1F600 -> surrogate pair D83D DE00, each surrogate 3 bytes -> 6
bytes.
+ assert_eq!(
+ enc("\u{1F600}"),
+ vec![0, 6, 0xED, 0xA0, 0xBD, 0xED, 0xB8, 0x80]
+ );
+ // Overlong -> error (like Java UTFDataFormatException).
+ let mut sink = Vec::new();
+ assert!(write_java_utf(&mut sink, &"a".repeat(70000)).is_err());
+ }
+
+ #[test]
+ fn binary_array_str_var_element() {
+ // Element > 7 bytes takes the offset+length pointer branch, which
datasplit-v8
+ // (all elements <= 7, inline) never exercises. Check the pointer
resolves to the value.
+ let out =
crate::spec::serialize_binary_array_str(&["abcdefgh".to_string()]);
+ assert_eq!(&out[0..4], &1i32.to_le_bytes()); // element count
+ let slot = u64::from_le_bytes(out[8..16].try_into().unwrap());
+ let off = (slot >> 32) as usize;
+ let len = (slot & 0xFFFF_FFFF) as usize;
+ assert_eq!(len, 8);
+ assert_eq!(&out[off..off + len], b"abcdefgh");
+ }
+
+ #[test]
+ fn binary_array_long_multiword_null_bitset() {
+ // n > 32 spans a second 4-byte null word; a None at index >= 32 must
set the right bit.
+ let mut vals: Vec<Option<i64>> = (0..40).map(|i| Some(i as
i64)).collect();
+ vals[35] = None;
+ let out = crate::spec::serialize_binary_array_long(&vals);
+ assert_eq!(&out[0..4], &40i32.to_le_bytes()); // element count
+ let header = 4 + 40usize.div_ceil(32) * 4; // 12
+ assert_eq!(out[4 + 35 / 8] & (1 << (35 % 8)), 1 << (35 % 8)); // null
bit in 2nd word
+ assert_eq!(out[4] & 1, 0); // element 0 not null
+ assert_eq!(
+ i64::from_le_bytes(out[header..header + 8].try_into().unwrap()),
+ 0
+ );
+ }
+
+ fn int_binary_row(vals: &[i32]) -> BinaryRow {
+ let mut b = crate::spec::BinaryRowBuilder::new(vals.len() as i32);
+ for (i, v) in vals.iter().enumerate() {
+ b.write_int(i, *v);
+ }
+ b.build()
+ }
+
+ // Mirror SplitSerializerTest.dataFile(name, level, minKey, maxKey,
maxSequence).
+ fn v1_data_file(
+ name: &str,
+ level: i32,
+ min_key: i32,
+ max_key: i32,
+ max_seq: i64,
+ ) -> DataFileMeta {
+ let size = (max_key - min_key + 1) as i64;
+ DataFileMeta {
+ file_name: name.to_string(),
+ file_size: size,
+ row_count: size,
+ min_key: int_binary_row(&[min_key]).to_serialized_bytes(),
+ max_key: int_binary_row(&[max_key]).to_serialized_bytes(),
+ key_stats: BinaryTableStats::empty(),
+ value_stats: BinaryTableStats::empty(),
+ min_sequence_number: 0,
+ max_sequence_number: max_seq,
+ schema_id: 0,
+ level,
+ extra_files: Vec::new(),
+ creation_time: chrono::DateTime::from_timestamp_millis(100),
+ delete_row_count: Some(0),
+ embedded_index: None,
+ file_source: Some(0), // FileSource.APPEND
+ value_stats_cols: None,
+ external_path: None,
+ first_row_id: None,
+ write_cols: None,
+ }
+ }
+
+ // Mirrors SplitSerializerTest.dataSplit(): two files, EMPTY_STATS,
null-padded deletion list.
+ fn v1_data_split_builder() -> DataSplitBuilder {
+ DataSplitBuilder::new()
+ .with_snapshot(42)
+ .with_partition(int_binary_row(&[2026, 7]))
+ .with_bucket(3)
+ .with_total_buckets(8)
+ .with_bucket_path("dt=20260706/bucket-3".to_string())
+ .with_data_files(vec![
+ v1_data_file("file-a", 0, 1, 10, 100),
+ v1_data_file("file-b", 1, 11, 20, 200),
+ ])
+ .with_data_deletion_files(vec![
+ None,
+ Some(DeletionFile::new("dv/file-b".to_string(), 2, 10,
Some(3))),
+ ])
+ .with_raw_convertible(true)
+ }
+
+ #[test]
+ fn serialize_split_v1_matches_golden() {
+ // Golden generated by Java (paimon-core) for byte-for-byte
cross-language compatibility.
+ let expected = include_bytes!("goldens/split_v1_data.bin");
+ let split = v1_data_split_builder().build().unwrap();
+ assert_eq!(
+ split.serialize_split_v1().unwrap().as_slice(),
+ &expected[..]
+ );
+ }
+
+ #[test]
+ fn serialize_indexed_split_v1_matches_golden() {
+ // Row ranges -> IndexedSplit (type 3): same DataSplit as
split-v1-data + ranges [1,4],[11,13].
+ // The Java golden ends with vector scores, which the Rust split has
none of, so compare against
+ // the golden up to its scores tail plus a `false` scores flag.
+ let golden = include_bytes!("goldens/split_v1_indexed.bin");
+ let scores_len = 1 + 4 + 3 * 4; // writeBoolean(true) + count + 3
floats
+ let mut expected = golden[..golden.len() - scores_len].to_vec();
+ expected.push(0); // scores = false
+ let split = v1_data_split_builder()
+ .with_row_ranges(vec![RowRange::new(1, 4), RowRange::new(11, 13)])
+ .build()
+ .unwrap();
+ assert_eq!(split.serialize_split_v1().unwrap(), expected);
+ }
}
