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 f0cbb845 feat: deserialize DataSplit from Paimon-native bytes (core +
C binding) (#565)
f0cbb845 is described below
commit f0cbb84519b0cf358e6b0997bf2be3d238550914
Author: Junrui Lee <[email protected]>
AuthorDate: Tue Jul 21 20:51:03 2026 +0800
feat: deserialize DataSplit from Paimon-native bytes (core + C binding)
(#565)
---
bindings/c/src/table.rs | 53 ++-
bindings/c/src/tests.rs | 56 ++++
crates/paimon/src/spec/binary_row.rs | 153 +++++++++
crates/paimon/src/spec/data_file.rs | 169 ++++++++++
crates/paimon/src/spec/stats.rs | 19 ++
crates/paimon/src/table/source.rs | 632 ++++++++++++++++++++++++++++++++++-
6 files changed, 1077 insertions(+), 5 deletions(-)
diff --git a/bindings/c/src/table.rs b/bindings/c/src/table.rs
index a17ba3fc..05f82e31 100644
--- a/bindings/c/src/table.rs
+++ b/bindings/c/src/table.rs
@@ -22,7 +22,7 @@ use arrow_array::ffi::{FFI_ArrowArray, FFI_ArrowSchema};
use arrow_array::{Array, StructArray};
use futures::StreamExt;
use paimon::spec::{DataField, DataType, Datum, Predicate, PredicateBuilder};
-use paimon::table::{ArrowRecordBatchStream, Table};
+use paimon::table::{ArrowRecordBatchStream, DataSplit, Table};
use paimon::Plan;
use crate::error::{check_non_null, paimon_error, validate_cstr,
PaimonErrorCode};
@@ -472,10 +472,54 @@ pub unsafe extern "C" fn paimon_table_scan_plan(
// ======================= Plan ===============================
+/// Build a one-split `paimon_plan` from a serialized Paimon-native `DataSplit`
+/// byte buffer (the wire form produced by `DataSplit::serialize` / Java
+/// `DataSplit#serialize`). `data` must be raw bytes (Base64 already decoded by
+/// the caller).
+///
+/// The returned plan is usable with `paimon_table_read_to_arrow` and must be
+/// freed with `paimon_plan_free`.
+///
+/// # Safety
+/// `data` must point to `len` valid bytes, or be null when `len == 0`.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_plan_from_split_bytes(
+ data: *const u8,
+ len: usize,
+) -> paimon_result_plan {
+ if data.is_null() || len == 0 {
+ return paimon_result_plan {
+ plan: std::ptr::null_mut(),
+ error: paimon_error::new(
+ PaimonErrorCode::InvalidInput,
+ "paimon_plan_from_split_bytes: null or empty
buffer".to_string(),
+ ),
+ };
+ }
+ let bytes = std::slice::from_raw_parts(data, len);
+ match DataSplit::deserialize(bytes) {
+ Ok(split) => {
+ let plan = Plan::new(vec![split]);
+ let wrapper = Box::new(paimon_plan {
+ inner: Box::into_raw(Box::new(plan)) as *mut c_void,
+ });
+ paimon_result_plan {
+ plan: Box::into_raw(wrapper),
+ error: std::ptr::null_mut(),
+ }
+ }
+ Err(e) => paimon_result_plan {
+ plan: std::ptr::null_mut(),
+ error: paimon_error::from_paimon(e),
+ },
+ }
+}
+
/// Free a paimon_plan.
///
/// # Safety
/// Only call with a plan returned from `paimon_table_scan_plan`.
+/// A plan returned from `paimon_plan_from_split_bytes` is also a valid source.
#[no_mangle]
pub unsafe extern "C" fn paimon_plan_free(plan: *mut paimon_plan) {
if !plan.is_null() {
@@ -490,6 +534,7 @@ pub unsafe extern "C" fn paimon_plan_free(plan: *mut
paimon_plan) {
///
/// # Safety
/// `plan` must be a valid pointer from `paimon_table_scan_plan`, or null
(returns 0).
+/// A plan returned from `paimon_plan_from_split_bytes` is also a valid source.
#[no_mangle]
pub unsafe extern "C" fn paimon_plan_num_splits(plan: *const paimon_plan) ->
usize {
if plan.is_null() {
@@ -1736,6 +1781,12 @@ const _: unsafe extern "C" fn(
usize,
) -> paimon_result_read_builder = paimon_table_new_read_builder_with_options;
+// Plan constructor ABI signature guard. Pins the symbol that builds a plan
from
+// serialized split bytes so an accidental signature change fails to compile
+// rather than silently breaking header consumers.
+const _: unsafe extern "C" fn(*const u8, usize) -> paimon_result_plan =
+ paimon_plan_from_split_bytes;
+
#[cfg(test)]
mod tests {
use super::*;
diff --git a/bindings/c/src/tests.rs b/bindings/c/src/tests.rs
index 2ced3245..3897189c 100644
--- a/bindings/c/src/tests.rs
+++ b/bindings/c/src/tests.rs
@@ -2418,3 +2418,59 @@ fn vector_search_empty_result_is_eof_stream() {
unwrap_table(handle);
}
}
+
+// =========================================================================
+// paimon_plan_from_split_bytes
+// =========================================================================
+
+#[test]
+fn plan_from_split_bytes_round_trips() {
+ let path = "memory:/test_plan_from_split_bytes";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ write_data_rust(&table, &[make_batch(vec![1, 2, 3], vec!["a", "b", "c"])]);
+
+ // Obtain a real DataSplit from a plan via the Rust API, then serialize it.
+ let splits = crate::runtime().block_on(async {
+ let rb = table.new_read_builder();
+ let scan = rb.new_scan();
+ scan.plan().await.unwrap().splits().to_vec()
+ });
+ assert!(!splits.is_empty(), "expected at least one planned split");
+ let bytes = splits[0].serialize().unwrap();
+
+ let result = unsafe { paimon_plan_from_split_bytes(bytes.as_ptr(),
bytes.len()) };
+ assert!(result.error.is_null());
+ assert_eq!(unsafe { paimon_plan_num_splits(result.plan) }, 1);
+ unsafe { paimon_plan_free(result.plan) };
+}
+
+#[test]
+fn plan_from_split_bytes_rejects_null_and_empty() {
+ let r = unsafe { paimon_plan_from_split_bytes(std::ptr::null(), 0) };
+ assert!(r.plan.is_null());
+ assert!(!r.error.is_null());
+ unsafe { paimon_error_free(r.error) };
+
+ let dummy = [0u8; 1];
+ let r2 = unsafe { paimon_plan_from_split_bytes(dummy.as_ptr(), 0) };
+ assert!(r2.plan.is_null());
+ assert!(!r2.error.is_null());
+ unsafe { paimon_error_free(r2.error) };
+}
+
+#[test]
+fn plan_from_split_bytes_rejects_garbage() {
+ let garbage = [1u8, 2, 3, 4, 5, 6, 7, 8, 9, 10];
+ let r = unsafe { paimon_plan_from_split_bytes(garbage.as_ptr(),
garbage.len()) };
+ assert!(r.plan.is_null());
+ assert!(!r.error.is_null());
+ unsafe { paimon_error_free(r.error) };
+}
diff --git a/crates/paimon/src/spec/binary_row.rs
b/crates/paimon/src/spec/binary_row.rs
index 4a4017f3..ce08ad8b 100644
--- a/crates/paimon/src/spec/binary_row.rs
+++ b/crates/paimon/src/spec/binary_row.rs
@@ -757,6 +757,104 @@ pub fn serialize_binary_array_long(values:
&[Option<i64>]) -> Vec<u8> {
data
}
+/// Reverse of [`serialize_binary_array_str`].
+pub fn deserialize_binary_array_str(data: &[u8]) -> crate::Result<Vec<String>>
{
+ let n = read_binary_array_len(data)?;
+ let header = binary_array_header(n);
+ // The fixed element region is `n * 8` bytes after the header; reject any
+ // count whose slots cannot fit in the buffer up front. This bounds both
the
+ // reservation and the loop, so a forged large count (with or without
+ // element slots) cannot amplify memory before per-element validation.
+ check_binary_array_fits(n, header, data.len())?;
+ let mut out = Vec::with_capacity(n);
+ for k in 0..n {
+ let eo = header + k * 8;
+ let slot = data
+ .get(eo..eo + 8)
+ .ok_or_else(|| bin_arr_err("string element slot out of range"))?;
+ let marker = slot[7];
+ let bytes = if marker & 0x80 != 0 {
+ let len = (marker & 0x7F) as usize;
+ slot.get(..len)
+ .ok_or_else(|| bin_arr_err("inline string length out of
range"))?
+ } else {
+ let encoded = u64::from_le_bytes(slot.try_into().unwrap());
+ let var_off = (encoded >> 32) as usize;
+ let len = (encoded & 0xFFFF_FFFF) as usize;
+ let end = var_off
+ .checked_add(len)
+ .ok_or_else(|| bin_arr_err("variable string bytes out of
range"))?;
+ data.get(var_off..end)
+ .ok_or_else(|| bin_arr_err("variable string bytes out of
range"))?
+ };
+ out.push(
+ std::str::from_utf8(bytes)
+ .map_err(|_| bin_arr_err("string element is not valid UTF-8"))?
+ .to_string(),
+ );
+ }
+ Ok(out)
+}
+
+/// Reverse of [`serialize_binary_array_long`].
+pub fn deserialize_binary_array_long(data: &[u8]) ->
crate::Result<Vec<Option<i64>>> {
+ let n = read_binary_array_len(data)?;
+ let header = binary_array_header(n);
+ // See `deserialize_binary_array_str`: reject a count whose fixed element
+ // region overflows the buffer before allocating. Null elements skip the
+ // per-slot read, so this up-front check is what prevents a forged
+ // "large count + all-null bitmap + no slots" input from amplifying memory.
+ check_binary_array_fits(n, header, data.len())?;
+ let mut out = Vec::with_capacity(n);
+ for k in 0..n {
+ let null = data
+ .get(4 + k / 8)
+ .map(|b| b & (1 << (k % 8)) != 0)
+ .unwrap_or(false);
+ if null {
+ out.push(None);
+ } else {
+ let eo = header + k * 8;
+ let slot = data
+ .get(eo..eo + 8)
+ .ok_or_else(|| bin_arr_err("long element slot out of range"))?;
+ out.push(Some(i64::from_le_bytes(slot.try_into().unwrap())));
+ }
+ }
+ Ok(out)
+}
+
+fn read_binary_array_len(data: &[u8]) -> crate::Result<usize> {
+ let raw = data
+ .get(0..4)
+ .ok_or_else(|| bin_arr_err("binary array too short for length
prefix"))?;
+ let n = i32::from_le_bytes(raw.try_into().unwrap());
+ if n < 0 {
+ return Err(bin_arr_err("binary array has negative length"));
+ }
+ Ok(n as usize)
+}
+
+/// Reject a binary array whose `n` fixed 8-byte element slots cannot fit in
the
+/// buffer after its `header`. Computed without overflow so a forged count
+/// cannot wrap; guards allocation and iteration for both decoders (`None`
+/// elements otherwise skip the per-slot bounds check).
+fn check_binary_array_fits(n: usize, header: usize, data_len: usize) ->
crate::Result<()> {
+ if n > data_len.saturating_sub(header) / 8 {
+ return Err(bin_arr_err(
+ "binary array element region exceeds buffer length",
+ ));
+ }
+ Ok(())
+}
+
+fn bin_arr_err(msg: &str) -> crate::Error {
+ crate::Error::DataInvalid {
+ message: msg.to_string(),
+ source: None,
+ }
+}
+
/// Extract a Datum from an Arrow RecordBatch column at the given row index.
pub fn extract_datum_from_arrow(
batch: &RecordBatch,
@@ -1917,4 +2015,59 @@ mod tests {
"binary-row write path must store euclidean timestamp parts"
);
}
+
+ #[test]
+ fn binary_array_str_round_trips() {
+ for v in [
+ vec![],
+ vec!["a".to_string()],
+ vec!["".to_string(), "short".to_string(), "x".repeat(20)],
+ vec!["1234567".to_string(), "12345678".to_string()], // 7-byte
inline vs 8-byte pointer
+ ] {
+ let bytes = serialize_binary_array_str(&v);
+ assert_eq!(deserialize_binary_array_str(&bytes).unwrap(), v);
+ }
+ }
+
+ #[test]
+ fn binary_array_long_round_trips() {
+ for v in [
+ vec![],
+ vec![Some(1i64), None, Some(-5), Some(i64::MAX)],
+ vec![None, None],
+ ] {
+ let bytes = serialize_binary_array_long(&v);
+ assert_eq!(deserialize_binary_array_long(&bytes).unwrap(), v);
+ }
+ }
+
+ #[test]
+ fn binary_array_str_rejects_truncated() {
+ assert!(deserialize_binary_array_str(&[1, 0]).is_err()); // < 4 header
bytes
+ }
+
+ #[test]
+ fn binary_array_rejects_huge_length_prefix() {
+ // A 4-byte buffer whose length prefix decodes to i32::MAX must return
an
+ // Err rather than eagerly reserving a huge Vec (capacity-overflow /
OOM).
+ let huge = [0xFF, 0xFF, 0xFF, 0x7F];
+ assert!(deserialize_binary_array_str(&huge).is_err());
+ assert!(deserialize_binary_array_long(&huge).is_err());
+ }
+
+ #[test]
+ fn binary_array_long_rejects_all_null_amplification() {
+ // Forged input: a large element count with an all-ones null bitmap and
+ // NO element slots. Null elements skip the per-slot bounds check, so
+ // without an up-front `count * 8 <= remaining` guard the loop would
+ // push `count` `None`s from a tiny buffer (~128x memory
amplification),
+ // reachable through the C entry point (OOM risk). Must error, not
+ // allocate.
+ let count: i32 = 8000;
+ let bitmap_len = (count as usize).div_ceil(8); // 1000 bytes
+ let mut buf = Vec::with_capacity(4 + bitmap_len);
+ buf.extend_from_slice(&count.to_le_bytes());
+ buf.extend(std::iter::repeat_n(0xFFu8, bitmap_len)); // every element
null
+ assert!(deserialize_binary_array_long(&buf).is_err());
+ }
}
diff --git a/crates/paimon/src/spec/data_file.rs
b/crates/paimon/src/spec/data_file.rs
index c6745ad3..43a23b43 100644
--- a/crates/paimon/src/spec/data_file.rs
+++ b/crates/paimon/src/spec/data_file.rs
@@ -169,6 +169,24 @@ fn opt_str_array(b: &mut BinaryRowBuilder, pos: usize, v:
&Option<Vec<String>>)
}
}
+fn data_file_err(msg: &str) -> crate::Error {
+ crate::Error::DataInvalid {
+ message: msg.to_string(),
+ source: None,
+ }
+}
+
+/// Reverse of `BinaryRowBuilder::write_timestamp_compact`: read the compact
+/// (precision-3) timestamp at `pos` back as epoch-millis and map it to `Utc`.
+fn read_compact_millis_as_utc(
+ row: &BinaryRow,
+ pos: usize,
+) -> crate::Result<chrono::DateTime<chrono::Utc>> {
+ let (millis, _nanos) = row.get_timestamp_raw(pos, 3)?;
+ chrono::DateTime::from_timestamp_millis(millis)
+ .ok_or_else(|| data_file_err("creation_time out of range"))
+}
+
impl DataFileMeta {
/// Decode this file's manifest value statistics for a field in the
current schema.
///
@@ -288,6 +306,93 @@ impl DataFileMeta {
Ok(b.build_row_data())
}
+ /// Reverse of [`DataFileMeta::to_serialized_row_data`]: decode the fixed
+ /// 20-field `DataFileMeta` BinaryRow (version 8 layout).
+ pub fn from_serialized_row_data(data: &[u8]) ->
crate::Result<DataFileMeta> {
+ use crate::spec::deserialize_binary_array_str;
+ let row = BinaryRow::from_bytes(20, data.to_vec());
+
+ let file_name = String::from_utf8(row.get_binary(0)?.to_vec())
+ .map_err(|_| data_file_err("file_name is not valid UTF-8"))?;
+ let file_size = row.get_long(1)?;
+ let row_count = row.get_long(2)?;
+ let min_key = row.get_binary(3)?.to_vec();
+ let max_key = row.get_binary(4)?.to_vec();
+ let key_stats =
BinaryTableStats::from_simple_stats_row_data(row.get_binary(5)?)?;
+ let value_stats =
BinaryTableStats::from_simple_stats_row_data(row.get_binary(6)?)?;
+ let min_sequence_number = row.get_long(7)?;
+ let max_sequence_number = row.get_long(8)?;
+ let schema_id = row.get_long(9)?;
+ let level = row.get_int(10)?;
+ let extra_files = deserialize_binary_array_str(row.get_binary(11)?)?;
+ let creation_time = if row.is_null_at(12) {
+ None
+ } else {
+ Some(read_compact_millis_as_utc(&row, 12)?)
+ };
+ let delete_row_count = if row.is_null_at(13) {
+ None
+ } else {
+ Some(row.get_long(13)?)
+ };
+ let embedded_index = if row.is_null_at(14) {
+ None
+ } else {
+ Some(row.get_binary(14)?.to_vec())
+ };
+ let file_source = if row.is_null_at(15) {
+ None
+ } else {
+ Some(row.get_byte(15)? as i32)
+ };
+ let value_stats_cols = if row.is_null_at(16) {
+ None
+ } else {
+ Some(deserialize_binary_array_str(row.get_binary(16)?)?)
+ };
+ let external_path = if row.is_null_at(17) {
+ None
+ } else {
+ Some(
+ String::from_utf8(row.get_binary(17)?.to_vec())
+ .map_err(|_| data_file_err("external_path is not valid
UTF-8"))?,
+ )
+ };
+ let first_row_id = if row.is_null_at(18) {
+ None
+ } else {
+ Some(row.get_long(18)?)
+ };
+ let write_cols = if row.is_null_at(19) {
+ None
+ } else {
+ Some(deserialize_binary_array_str(row.get_binary(19)?)?)
+ };
+
+ Ok(DataFileMeta {
+ file_name,
+ file_size,
+ row_count,
+ min_key,
+ max_key,
+ key_stats,
+ value_stats,
+ min_sequence_number,
+ max_sequence_number,
+ schema_id,
+ level,
+ extra_files,
+ creation_time,
+ delete_row_count,
+ embedded_index,
+ file_source,
+ value_stats_cols,
+ external_path,
+ first_row_id,
+ write_cols,
+ })
+ }
+
/// Full path for this data file.
///
/// Mirrors Java `DataFilePathFactory#toPath(DataFileMeta)`: use
@@ -415,6 +520,70 @@ mod tests {
DataField::new(id, name.to_string(), DataType::Int(IntType::new()))
}
+ fn sample_full_data_file_meta() -> DataFileMeta {
+ let stats = BinaryTableStats::empty();
+ DataFileMeta {
+ file_name: "data-full.parquet".to_string(),
+ file_size: 1024,
+ row_count: 42,
+ min_key: vec![1, 2, 3],
+ max_key: vec![9, 8, 7],
+ key_stats: stats.clone(),
+ value_stats: stats,
+ min_sequence_number: 10,
+ max_sequence_number: 20,
+ schema_id: 5,
+ level: 3,
+ extra_files: vec!["extra-1".to_string(), "extra-2".to_string()],
+ creation_time: DateTime::from_timestamp_millis(1_700_000_000_123),
+ delete_row_count: Some(4),
+ embedded_index: Some(vec![7, 8, 9]),
+ file_source: Some(0),
+ value_stats_cols: Some(vec!["a".to_string(), "b".to_string()]),
+ external_path: Some("s3://bucket/data-full.parquet".to_string()),
+ first_row_id: Some(1_000),
+ write_cols: Some(vec!["a".to_string(), "b".to_string(),
"c".to_string()]),
+ }
+ }
+
+ fn sample_minimal_data_file_meta() -> DataFileMeta {
+ let stats = BinaryTableStats::empty();
+ DataFileMeta {
+ file_name: "data-min.parquet".to_string(),
+ file_size: 0,
+ row_count: 0,
+ min_key: Vec::new(),
+ max_key: Vec::new(),
+ key_stats: stats.clone(),
+ value_stats: stats,
+ min_sequence_number: 0,
+ max_sequence_number: 0,
+ schema_id: 0,
+ level: 0,
+ extra_files: Vec::new(),
+ creation_time: None,
+ delete_row_count: None,
+ embedded_index: None,
+ file_source: None,
+ value_stats_cols: None,
+ external_path: None,
+ first_row_id: None,
+ write_cols: None,
+ }
+ }
+
+ #[test]
+ fn data_file_meta_row_data_round_trips() {
+ for meta in [
+ sample_full_data_file_meta(),
+ sample_minimal_data_file_meta(),
+ ] {
+ let bytes = meta.to_serialized_row_data().unwrap();
+ let back = DataFileMeta::from_serialized_row_data(&bytes).unwrap();
+ assert_eq!(back, meta);
+ }
+ }
+
#[test]
fn value_stats_for_field_resolves_dense_columns_and_schema() {
let fields = vec![int_field(0, "id"), int_field(1, "v")];
diff --git a/crates/paimon/src/spec/stats.rs b/crates/paimon/src/spec/stats.rs
index 5503a6f7..7660366a 100644
--- a/crates/paimon/src/spec/stats.rs
+++ b/crates/paimon/src/spec/stats.rs
@@ -123,6 +123,17 @@ impl BinaryTableStats {
b.write_bytes(2, &serialize_binary_array_long(&self.null_counts));
b.build_row_data()
}
+
+ /// Reverse of [`BinaryTableStats::to_simple_stats_row_data`]: a 3-field
+ /// `SimpleStats` BinaryRow body (min_values bytes, max_values bytes,
+ /// null_counts array<bigint>).
+ pub fn from_simple_stats_row_data(data: &[u8]) ->
crate::Result<BinaryTableStats> {
+ let row = crate::spec::BinaryRow::from_bytes(3, data.to_vec());
+ let min_values = row.get_binary(0)?.to_vec();
+ let max_values = row.get_binary(1)?.to_vec();
+ let null_counts =
crate::spec::deserialize_binary_array_long(row.get_binary(2)?)?;
+ Ok(BinaryTableStats::new(min_values, max_values, null_counts))
+ }
}
impl Display for BinaryTableStats {
@@ -217,4 +228,12 @@ mod tests {
assert_eq!(min_row.arity(), 0);
assert_eq!(max_row.arity(), 0);
}
+
+ #[test]
+ fn simple_stats_row_data_round_trips() {
+ let stats = BinaryTableStats::empty();
+ let bytes = stats.to_simple_stats_row_data();
+ let back =
BinaryTableStats::from_simple_stats_row_data(&bytes).unwrap();
+ assert_eq!(back.to_simple_stats_row_data(), bytes);
+ }
}
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index 49ab30eb..e852630f 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -471,7 +471,7 @@ impl PartitionBucket {
/// Input split for reading: partition + bucket + list of data files and
optional deletion files.
///
/// Reference:
[org.apache.paimon.table.source.DataSplit](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java)
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct DataSplit {
snapshot_id: i64,
partition: BinaryRow,
@@ -664,6 +664,126 @@ impl DataSplit {
Ok(out)
}
+ /// Reverse of [`DataSplit::serialize`]: parse a raw v8
`DataSplit#serialize` body.
+ /// Consumes the entire buffer; trailing bytes are an error.
+ pub fn deserialize(data: &[u8]) -> crate::Result<DataSplit> {
+ let mut cur = data;
+ let split = Self::read_v8_body(&mut cur)?;
+ if !cur.is_empty() {
+ return Err(crate::Error::DataInvalid {
+ message: format!("{} trailing bytes after DataSplit",
cur.len()),
+ source: None,
+ });
+ }
+ Ok(split)
+ }
+
+ /// Read a v8 `DataSplit` body from the cursor, leaving it positioned
after the body
+ /// (used both by `deserialize` and the SPLIT_V1 frame reader).
+ fn read_v8_body(cur: &mut &[u8]) -> crate::Result<DataSplit> {
+ let magic = read_i64(cur)?;
+ if magic != SPLIT_MAGIC {
+ return Err(crate::Error::DataInvalid {
+ message: format!("invalid DataSplit magic: {magic:#018x}"),
+ source: None,
+ });
+ }
+ match read_i32(cur)? {
+ SPLIT_VERSION => Self::read_v8_body_after_version(cur),
+ version => Err(crate::Error::Unsupported {
+ message: format!(
+ "DataSplit version {version} not supported (only
v{SPLIT_VERSION})"
+ ),
+ }),
+ }
+ }
+
+ /// Read the fields following the magic + version header of a v8
`DataSplit` body.
+ fn read_v8_body_after_version(cur: &mut &[u8]) -> crate::Result<DataSplit>
{
+ let snapshot_id = read_i64(cur)?;
+
+ let part_len = read_i32(cur)?;
+ if part_len < 0 {
+ return Err(crate::Error::DataInvalid {
+ message: "negative partition length".into(),
+ source: None,
+ });
+ }
+ let part_bytes = take(cur, part_len as usize)?;
+ let partition = BinaryRow::from_serialized_bytes(part_bytes)?;
+
+ let bucket = read_i32(cur)?;
+ let bucket_path = read_java_utf(cur)?;
+
+ let total_buckets = if read_u8(cur)? == 0 {
+ None
+ } else {
+ Some(read_i32(cur)?)
+ };
+
+ let before_files_n = read_i32(cur)?; // deprecated, always 0 on write
+ if before_files_n != 0 {
+ return Err(crate::Error::Unsupported {
+ message: format!("non-empty beforeFiles ({before_files_n}) not
supported"),
+ });
+ }
+ // `beforeDeletionFiles` is always a null list on write; a non-null
list
+ // is rejected rather than silently dropped (Java treats such a split
as
+ // invalid), matching the `beforeFiles` handling above.
+ if read_deletion_list(cur)?.is_some() {
+ return Err(crate::Error::Unsupported {
+ message: "non-null beforeDeletionFiles not
supported".to_string(),
+ });
+ }
+
+ let data_files_n = read_i32(cur)?;
+ if data_files_n < 0 {
+ return Err(crate::Error::DataInvalid {
+ message: "negative data_files count".into(),
+ source: None,
+ });
+ }
+ // Bound the eager reservation by the remaining bytes: each entry is
at least a
+ // 4-byte length prefix, so a crafted huge count cannot reserve more
than the
+ // buffer could ever hold. Underrun is still caught per-element below.
+ let mut data_files = Vec::with_capacity((data_files_n as
usize).min(cur.len() / 4));
+ for _ in 0..data_files_n {
+ let len = read_i32(cur)?;
+ if len < 0 {
+ return Err(crate::Error::DataInvalid {
+ message: "negative DataFileMeta length".into(),
+ source: None,
+ });
+ }
+ let row = take(cur, len as usize)?;
+ data_files.push(DataFileMeta::from_serialized_row_data(row)?);
+ }
+
+ let data_deletion_files = read_deletion_list(cur)?;
+ // Rust only produces and serves batch splits (isStreaming = false); a
+ // streaming split carries a semantic bit that Java readers branch on,
so
+ // reject it rather than silently dropping it.
+ if read_u8(cur)? != 0 {
+ return Err(crate::Error::Unsupported {
+ message: "streaming DataSplit (isStreaming = true) not
supported".to_string(),
+ });
+ }
+ let raw_convertible = read_u8(cur)? != 0;
+
+ let mut builder = DataSplitBuilder::new()
+ .with_snapshot(snapshot_id)
+ .with_partition(partition)
+ .with_bucket(bucket)
+ .with_bucket_path(bucket_path)
+ .with_total_buckets(total_buckets.unwrap_or(1))
+ .with_data_files(data_files)
+ .with_raw_convertible(raw_convertible);
+ if let Some(dels) = data_deletion_files {
+ builder = builder.with_data_deletion_files(dels);
+ }
+ builder.build()
+ }
+
/// 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
@@ -693,6 +813,88 @@ impl DataSplit {
}
Ok(out)
}
+
+ /// Reverse of [`DataSplit::serialize_split_v1`]: parse a Java
`SplitSerializer` frame into a
+ /// `DataSplit` (type 1) or an `IndexedSplit` (type 3, which carries
`row_ranges`). Other split
+ /// type ids, and IndexedSplit vector scores, are unsupported. Consumes
the entire buffer.
+ pub fn deserialize_split_v1(data: &[u8]) -> crate::Result<DataSplit> {
+ let mut cur = data;
+ let magic = read_i64(&mut cur)?;
+ if magic != SPLIT_SER_MAGIC {
+ return Err(crate::Error::DataInvalid {
+ message: format!("invalid SPLIT_V1 magic: {magic:#018x}"),
+ source: None,
+ });
+ }
+ let version = read_i32(&mut cur)?;
+ if version != SPLIT_SER_VERSION {
+ return Err(crate::Error::Unsupported {
+ message: format!("SplitSerializer version {version} not
supported"),
+ });
+ }
+ let type_id = read_i32(&mut cur)?;
+ let split = match type_id {
+ SPLIT_SER_TYPE_DATA_SPLIT => Self::read_v8_body(&mut cur)?,
+ SPLIT_SER_TYPE_INDEXED_SPLIT => {
+ let im = read_i64(&mut cur)?;
+ if im != INDEXED_SPLIT_MAGIC {
+ return Err(crate::Error::DataInvalid {
+ message: format!("invalid IndexedSplit magic:
{im:#018x}"),
+ source: None,
+ });
+ }
+ let iv = read_i32(&mut cur)?;
+ if iv != INDEXED_SPLIT_VERSION {
+ return Err(crate::Error::Unsupported {
+ message: format!("IndexedSplit version {iv} not
supported"),
+ });
+ }
+ let body = Self::read_v8_body(&mut cur)?;
+ let ranges_n = read_i32(&mut cur)?;
+ if ranges_n < 0 {
+ return Err(crate::Error::DataInvalid {
+ message: "negative row_ranges count".into(),
+ source: None,
+ });
+ }
+ // Bound the eager reservation by the remaining bytes: each
RowRange is
+ // 16 bytes (two i64), so a crafted huge count cannot
over-reserve.
+ let mut ranges = Vec::with_capacity((ranges_n as
usize).min(cur.len() / 16));
+ for _ in 0..ranges_n {
+ let from = read_i64(&mut cur)?;
+ let to = read_i64(&mut cur)?;
+ if from > to {
+ return Err(crate::Error::DataInvalid {
+ message: format!("row range from {from} > to
{to}"),
+ source: None,
+ });
+ }
+ ranges.push(RowRange::new(from, to));
+ }
+ let scores_flag = read_u8(&mut cur)?;
+ if scores_flag != 0 {
+ return Err(crate::Error::Unsupported {
+ message: "IndexedSplit scores are not
supported".into(),
+ });
+ }
+ let mut body = body;
+ body.row_ranges = Some(ranges);
+ body
+ }
+ other => {
+ return Err(crate::Error::Unsupported {
+ message: format!("unsupported split type id: {other}"),
+ });
+ }
+ };
+ if !cur.is_empty() {
+ return Err(crate::Error::DataInvalid {
+ message: format!("{} trailing bytes after SPLIT_V1 frame",
cur.len()),
+ source: None,
+ });
+ }
+ Ok(split)
+ }
}
/// Java `DataSplit#MAGIC` / `VERSION` for the serialize format.
@@ -775,6 +977,116 @@ fn write_deletion_list(
Ok(())
}
+/// Advances `cur` by `n` bytes, returning the consumed slice. Errors on
underrun.
+fn take<'a>(cur: &mut &'a [u8], n: usize) -> crate::Result<&'a [u8]> {
+ if cur.len() < n {
+ return Err(crate::Error::DataInvalid {
+ message: format!("split buffer underrun: need {n}, have {}",
cur.len()),
+ source: None,
+ });
+ }
+ let (head, tail) = cur.split_at(n);
+ *cur = tail;
+ Ok(head)
+}
+
+fn read_u8(cur: &mut &[u8]) -> crate::Result<u8> {
+ Ok(take(cur, 1)?[0])
+}
+
+fn read_i16(cur: &mut &[u8]) -> crate::Result<i16> {
+ Ok(i16::from_be_bytes(take(cur, 2)?.try_into().unwrap()))
+}
+
+fn read_i32(cur: &mut &[u8]) -> crate::Result<i32> {
+ Ok(i32::from_be_bytes(take(cur, 4)?.try_into().unwrap()))
+}
+
+fn read_i64(cur: &mut &[u8]) -> crate::Result<i64> {
+ Ok(i64::from_be_bytes(take(cur, 8)?.try_into().unwrap()))
+}
+
+fn utf_err() -> crate::Error {
+ crate::Error::DataInvalid {
+ message: "invalid modified UTF-8 in split".into(),
+ source: None,
+ }
+}
+
+/// Reverse of [`write_java_utf`]: `u16` byte-length prefix + modified UTF-8.
Each UTF-16 code
+/// unit is encoded independently, so a supplementary char arrives as two
3-byte surrogate units;
+/// collect the raw `u16` units and let [`String::from_utf16`] pair the
surrogates.
+fn read_java_utf(cur: &mut &[u8]) -> crate::Result<String> {
+ let len = read_i16(cur)? as u16 as usize;
+ let bytes = take(cur, len)?;
+ let mut units: Vec<u16> = Vec::new();
+ let mut i = 0;
+ while i < bytes.len() {
+ let b = bytes[i];
+ let (unit, adv) = if b & 0x80 == 0 {
+ // 1-byte: 0xxxxxxx (includes raw 0x00)
+ (b as u16, 1)
+ } else if b & 0xE0 == 0xC0 {
+ // 2-byte: 110xxxxx 10xxxxxx (0xC0 0x80 is Java's encoding of '\0')
+ if i + 1 >= bytes.len() || bytes[i + 1] & 0xC0 != 0x80 {
+ return Err(utf_err());
+ }
+ ((((b & 0x1F) as u16) << 6) | (bytes[i + 1] & 0x3F) as u16, 2)
+ } else if b & 0xF0 == 0xE0 {
+ // 3-byte: 1110xxxx 10xxxxxx 10xxxxxx
+ if i + 2 >= bytes.len() || bytes[i + 1] & 0xC0 != 0x80 || bytes[i
+ 2] & 0xC0 != 0x80 {
+ return Err(utf_err());
+ }
+ (
+ (((b & 0x0F) as u16) << 12)
+ | (((bytes[i + 1] & 0x3F) as u16) << 6)
+ | (bytes[i + 2] & 0x3F) as u16,
+ 3,
+ )
+ } else {
+ return Err(utf_err());
+ };
+ units.push(unit);
+ i += adv;
+ }
+ String::from_utf16(&units).map_err(|_| utf_err())
+}
+
+/// Reverse of [`write_deletion_list`]: `0` = null list; else `1` + count +
per entry
+/// (`0` = null, or `1` + path + offset + length + cardinality, where `-1`
decodes to `None`).
+fn read_deletion_list(cur: &mut &[u8]) ->
crate::Result<Option<Vec<Option<DeletionFile>>>> {
+ if read_u8(cur)? == 0 {
+ return Ok(None);
+ }
+ let count = read_i32(cur)?;
+ if count < 0 {
+ return Err(crate::Error::DataInvalid {
+ message: format!("deletion list count {count} < 0"),
+ source: None,
+ });
+ }
+ // Bound the eager reservation by the remaining bytes: each entry is at
least a
+ // 1-byte null flag, so a crafted huge count cannot over-reserve.
+ let mut out = Vec::with_capacity((count as usize).min(cur.len()));
+ for _ in 0..count {
+ if read_u8(cur)? == 0 {
+ out.push(None);
+ } else {
+ let path = read_java_utf(cur)?;
+ let offset = read_i64(cur)?;
+ let length = read_i64(cur)?;
+ let cardinality = read_i64(cur)?;
+ let cardinality = if cardinality == -1 {
+ None
+ } else {
+ Some(cardinality)
+ };
+ out.push(Some(DeletionFile::new(path, offset, length,
cardinality)));
+ }
+ }
+ Ok(Some(out))
+}
+
/// Builder for [DataSplit].
///
/// Reference:
[DataSplit.Builder](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java)
@@ -1192,9 +1504,16 @@ mod tests {
#[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 split = sample_v8_split();
+ assert_eq!(split.serialize().unwrap().as_slice(), &expected[..]);
+ }
+
+ /// The fixture split whose Java-generated bytes live in
`goldens/datasplit_v8.bin`.
+ /// Shared by the serialize and deserialize golden tests.
+ fn sample_v8_split() -> DataSplit {
+ use chrono::DateTime;
let mut pb = crate::spec::BinaryRowBuilder::new(1);
pb.write_bytes(0, b"aaaaa");
@@ -1235,7 +1554,7 @@ mod tests {
),
};
- let split = DataSplitBuilder::new()
+ DataSplitBuilder::new()
.with_snapshot(18)
.with_partition(pb.build())
.with_bucket(20)
@@ -1250,9 +1569,145 @@ mod tests {
))])
.with_raw_convertible(false)
.build()
+ .unwrap()
+ }
+
+ #[test]
+ fn deserialize_matches_datasplit_v8_golden() {
+ let golden = include_bytes!("goldens/datasplit_v8.bin");
+ let split = DataSplit::deserialize(golden).unwrap();
+ assert_eq!(split, sample_v8_split());
+ }
+
+ #[test]
+ fn deserialize_round_trips_serialize() {
+ let split = sample_v8_split();
+ assert_eq!(
+ DataSplit::deserialize(&split.serialize().unwrap()).unwrap(),
+ split
+ );
+ }
+
+ #[test]
+ fn deserialize_rejects_trailing_bytes() {
+ let mut bytes = sample_v8_split().serialize().unwrap();
+ bytes.push(0xFF);
+ assert!(DataSplit::deserialize(&bytes).is_err());
+ }
+
+ #[test]
+ fn deserialize_rejects_bad_magic() {
+ let bytes = [0u8; 12];
+ assert!(DataSplit::deserialize(&bytes).is_err());
+ }
+
+ // A non-null `beforeDeletionFiles` list is rejected rather than silently
+ // discarded (Java treats such a split as invalid). Walk the wire layout
with
+ // the same readers to locate the null-list flag, then flip it to a
non-null
+ // (empty) list and confirm deserialize fails loudly.
+ #[test]
+ fn deserialize_rejects_non_null_before_deletion_files() {
+ let split = sample_v8_split();
+ let bytes = split.serialize().unwrap();
+ assert!(DataSplit::deserialize(&bytes).is_ok());
+
+ let mut cur = bytes.as_slice();
+ read_i64(&mut cur).unwrap(); // magic
+ read_i32(&mut cur).unwrap(); // version
+ read_i64(&mut cur).unwrap(); // snapshot_id
+ let part_len = read_i32(&mut cur).unwrap() as usize;
+ take(&mut cur, part_len).unwrap(); // partition bytes
+ read_i32(&mut cur).unwrap(); // bucket
+ read_java_utf(&mut cur).unwrap(); // bucket_path
+ if read_u8(&mut cur).unwrap() != 0 {
+ read_i32(&mut cur).unwrap(); // total_buckets
+ }
+ read_i32(&mut cur).unwrap(); // beforeFiles count (0)
+ // `cur` now points at the
beforeDeletionFiles null-list flag byte.
+ let offset = bytes.len() - cur.len();
+
+ let mut patched = bytes.clone();
+ // null-list marker (0x00) -> non-null empty list (0x01 + i32 count 0).
+ patched.splice(offset..offset + 1, [1u8, 0, 0, 0, 0]);
+ match DataSplit::deserialize(&patched) {
+ Err(crate::Error::Unsupported { .. }) => {}
+ other => panic!("expected Unsupported, got {other:?}"),
+ }
+ }
+
+ // A streaming split (isStreaming = true) is rejected rather than silently
+ // dropping the bit: Rust only writes/serves batch splits (isStreaming =
+ // false), and Java readers branch on this flag. The isStreaming byte is
the
+ // second-to-last byte on the wire (isStreaming, then raw_convertible).
+ #[test]
+ fn deserialize_rejects_streaming_split() {
+ let bytes = sample_v8_split().serialize().unwrap();
+ assert!(DataSplit::deserialize(&bytes).is_ok());
+
+ let mut patched = bytes.clone();
+ let pos = patched.len() - 2; // isStreaming flag
+ assert_eq!(
+ patched[pos], 0,
+ "fixture must serialize isStreaming = false"
+ );
+ patched[pos] = 1;
+ match DataSplit::deserialize(&patched) {
+ Err(crate::Error::Unsupported { .. }) => {}
+ other => panic!("expected Unsupported, got {other:?}"),
+ }
+ }
+
+ #[test]
+ fn deserialize_rejects_huge_data_files_count_without_aborting() {
+ // Build a well-formed split with an empty data-file list so the
trailing layout
+ // after `data_files_n` is fixed: deletion-null (1) + isStreaming (1) +
+ // raw_convertible (1) = 3 bytes. Thus `data_files_n` (i32) sits at
`len - 7`.
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(crate::spec::BinaryRowBuilder::new(0).build())
+ .with_bucket(0)
+ .with_bucket_path("p".to_string())
+ .with_total_buckets(1)
+ .with_data_files(vec![])
+ .with_raw_convertible(false)
+ .build()
.unwrap();
+ let mut bytes = split.serialize().unwrap();
+ // Sanity: unpatched buffer round-trips.
+ assert!(DataSplit::deserialize(&bytes).is_ok());
+
+ let pos = bytes.len() - 7;
+ bytes[pos..pos + 4].copy_from_slice(&i32::MAX.to_be_bytes());
+ match DataSplit::deserialize(&bytes) {
+ Err(crate::Error::DataInvalid { .. }) => {}
+ other => panic!("expected DataInvalid, got {other:?}"),
+ }
+ }
- assert_eq!(split.serialize().unwrap().as_slice(), &expected[..]);
+ // Same hardening for the IndexedSplit row-ranges count in the SPLIT_V1
frame.
+ #[test]
+ fn deserialize_split_v1_rejects_huge_ranges_count_without_aborting() {
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(crate::spec::BinaryRowBuilder::new(0).build())
+ .with_bucket(0)
+ .with_bucket_path("p".to_string())
+ .with_total_buckets(1)
+ .with_data_files(vec![])
+ .with_row_ranges(vec![RowRange::new(0, 5)])
+ .with_raw_convertible(false)
+ .build()
+ .unwrap();
+ let mut bytes = split.serialize_split_v1().unwrap();
+ assert!(DataSplit::deserialize_split_v1(&bytes).is_ok());
+
+ // Trailing after `ranges_n` (i32): one RowRange (16) + scores flag
(1) = 17 bytes.
+ let pos = bytes.len() - 21;
+ bytes[pos..pos + 4].copy_from_slice(&i32::MAX.to_be_bytes());
+ match DataSplit::deserialize_split_v1(&bytes) {
+ Err(crate::Error::DataInvalid { .. }) => {}
+ other => panic!("expected DataInvalid, got {other:?}"),
+ }
}
#[test]
@@ -1280,6 +1735,74 @@ mod tests {
assert!(write_java_utf(&mut sink, &"a".repeat(70000)).is_err());
}
+ #[test]
+ fn java_utf_round_trips() {
+ for s in [
+ "",
+ "ascii",
+ "caf\u{e9}",
+ "\0",
+ "\u{800}",
+ "\u{1F600}emoji",
+ &"x".repeat(1000),
+ ] {
+ let mut out = Vec::new();
+ write_java_utf(&mut out, s).unwrap();
+ let mut cur = out.as_slice();
+ assert_eq!(read_java_utf(&mut cur).unwrap(), s);
+ assert!(cur.is_empty(), "should consume exactly the encoded
bytes");
+ }
+ }
+
+ #[test]
+ fn java_utf_rejects_truncated() {
+ let mut out = Vec::new();
+ write_java_utf(&mut out, "hello").unwrap();
+ out.truncate(out.len() - 2); // drop tail bytes
+ let mut cur = out.as_slice();
+ assert!(read_java_utf(&mut cur).is_err());
+ }
+
+ #[test]
+ fn java_utf_rejects_malformed_continuation() {
+ // len=2, lead 0xC0 (2-byte form) followed by 0x41 ('A'), which is NOT
a
+ // 10xxxxxx continuation byte. Must error, not silently decode to
"\u{1}".
+ let bytes = [0x00u8, 0x02, 0xC0, 0x41];
+ let mut cur = bytes.as_slice();
+ assert!(read_java_utf(&mut cur).is_err());
+ }
+
+ #[test]
+ fn deletion_list_round_trips() {
+ for list in [
+ None,
+ Some(vec![None]),
+ Some(vec![
+ Some(DeletionFile::new("f.idx".into(), 1, 2, Some(3))),
+ None,
+ ]),
+ ] {
+ let mut out = Vec::new();
+ write_deletion_list(&mut out, list.as_deref()).unwrap();
+ let mut cur = out.as_slice();
+ assert_eq!(read_deletion_list(&mut cur).unwrap(), list);
+ assert!(cur.is_empty());
+ }
+ }
+
+ #[test]
+ fn deletion_list_cardinality_sentinel_maps_to_none() {
+ // A -1 cardinality on the wire must decode back to `None`, matching
the
+ // forward `unwrap_or(-1)`.
+ let list = Some(vec![Some(DeletionFile::new("f.idx".into(), 4, 5,
None))]);
+ let mut out = Vec::new();
+ write_deletion_list(&mut out, list.as_deref()).unwrap();
+ let mut cur = out.as_slice();
+ let back = read_deletion_list(&mut cur).unwrap().unwrap();
+ assert_eq!(back[0].as_ref().unwrap().cardinality(), None);
+ assert!(cur.is_empty());
+ }
+
#[test]
fn binary_array_str_var_element() {
// Element > 7 bytes takes the offset+length pointer branch, which
datasplit-v8
@@ -1395,4 +1918,105 @@ mod tests {
.unwrap();
assert_eq!(split.serialize_split_v1().unwrap(), expected);
}
+
+ // The scores-stripped IndexedSplit frame the Rust serializer emits (see
+ // `serialize_indexed_split_v1_matches_golden`): the Java golden's
trailing vector scores are
+ // replaced by a `false` scores flag, which is the exact byte form
`deserialize_split_v1` supports.
+ fn indexed_split_v1_no_scores() -> Vec<u8> {
+ let golden = include_bytes!("goldens/split_v1_indexed.bin");
+ let scores_len = 1 + 4 + 3 * 4; // writeBoolean(true) + count + 3
floats
+ let mut bytes = golden[..golden.len() - scores_len].to_vec();
+ bytes.push(0); // scores = false
+ bytes
+ }
+
+ #[test]
+ fn deserialize_split_v1_data_golden() {
+ // `split_v1_data.bin` is the SPLIT_V1 (type 1) frame produced by
`v1_data_split_builder`.
+ let golden = include_bytes!("goldens/split_v1_data.bin");
+ assert_eq!(
+ DataSplit::deserialize_split_v1(golden).unwrap(),
+ v1_data_split_builder().build().unwrap()
+ );
+ // Round-trip gate: bytes -> split -> bytes.
+ assert_eq!(
+ DataSplit::deserialize_split_v1(golden)
+ .unwrap()
+ .serialize_split_v1()
+ .unwrap()
+ .as_slice(),
+ &golden[..]
+ );
+ }
+
+ #[test]
+ fn deserialize_split_v1_indexed_golden() {
+ // IndexedSplit (type 3) without vector scores, carrying row_ranges.
+ let frame = indexed_split_v1_no_scores();
+ let split = DataSplit::deserialize_split_v1(&frame).unwrap();
+ assert_eq!(
+ split.row_ranges(),
+ Some([RowRange::new(1, 4), RowRange::new(11, 13)].as_slice())
+ );
+ // Round-trips back to the same bytes.
+ assert_eq!(split.serialize_split_v1().unwrap(), frame);
+ }
+
+ #[test]
+ fn deserialize_split_v1_rejects_indexed_scores() {
+ // The raw Java golden carries vector scores, which the Rust split
cannot model.
+ let golden = include_bytes!("goldens/split_v1_indexed.bin");
+ assert!(matches!(
+ DataSplit::deserialize_split_v1(golden),
+ Err(crate::Error::Unsupported { .. })
+ ));
+ }
+
+ #[test]
+ fn deserialize_split_v1_rejects_unsupported_type() {
+ // hand-build a frame with type id 2
+ let mut b = Vec::new();
+ b.extend_from_slice(&SPLIT_SER_MAGIC.to_be_bytes());
+ b.extend_from_slice(&SPLIT_SER_VERSION.to_be_bytes());
+ b.extend_from_slice(&2i32.to_be_bytes());
+ assert!(matches!(
+ DataSplit::deserialize_split_v1(&b),
+ Err(crate::Error::Unsupported { .. })
+ ));
+ }
+
+ #[test]
+ fn deserialize_split_v1_rejects_trailing_bytes() {
+ let golden = include_bytes!("goldens/split_v1_data.bin");
+ let mut b = golden.to_vec();
+ b.push(0xff); // one trailing byte after a complete frame
+ assert!(matches!(
+ DataSplit::deserialize_split_v1(&b),
+ Err(crate::Error::DataInvalid { .. })
+ ));
+ }
+
+ #[test]
+ fn deserialize_split_v1_rejects_inverted_row_range() {
+ // Take a valid IndexedSplit frame (first range [1,4]) and flip its
bounds to from=5,to=2.
+ // `RowRange::new` asserts from <= to, so the loop must reject this
before constructing it.
+ let mut frame = indexed_split_v1_no_scores();
+ let needle: Vec<u8> = 1i64
+ .to_be_bytes()
+ .iter()
+ .chain(4i64.to_be_bytes().iter())
+ .copied()
+ .collect();
+ let pos = frame
+ .windows(needle.len())
+ .position(|w| w == needle.as_slice())
+ .expect("range [1,4] present in indexed frame");
+ let mut replacement = 5i64.to_be_bytes().to_vec();
+ replacement.extend_from_slice(&2i64.to_be_bytes());
+ frame[pos..pos + needle.len()].copy_from_slice(&replacement);
+ assert!(matches!(
+ DataSplit::deserialize_split_v1(&frame),
+ Err(crate::Error::DataInvalid { .. })
+ ));
+ }
}
