From 4f463cc29a2445ae1326377e8eb76907a5fcb1be Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Tue, 21 Jul 2026 01:04:29 +0800 Subject: [PATCH 1/6] feat(spec): decode DataFileMeta and BinaryRow primitives from serialized bytes Add the inverse of the existing serialization, built on the existing BinaryRow primitives: BinaryArray (string/bigint) decoders, BinaryTableStats from a SimpleStats row, and DataFileMeta from its fixed 20-field BinaryRow. Malformed input yields typed errors and length prefixes are bounded against the buffer, so crafted input cannot trigger unbounded allocation or a panic. --- crates/paimon/src/spec/binary_row.rs | 122 +++++++++++++++++++ crates/paimon/src/spec/data_file.rs | 169 +++++++++++++++++++++++++++ crates/paimon/src/spec/stats.rs | 19 +++ 3 files changed, 310 insertions(+) diff --git a/crates/paimon/src/spec/binary_row.rs b/crates/paimon/src/spec/binary_row.rs index 4a4017f3..60a31962 100644 --- a/crates/paimon/src/spec/binary_row.rs +++ b/crates/paimon/src/spec/binary_row.rs @@ -757,6 +757,89 @@ pub fn serialize_binary_array_long(values: &[Option]) -> Vec { data } +/// Reverse of [`serialize_binary_array_str`]. +pub fn deserialize_binary_array_str(data: &[u8]) -> crate::Result> { + let n = read_binary_array_len(data)?; + let header = binary_array_header(n); + // Bound the preallocation to what the buffer could actually hold (8 bytes per + // element slot) so an attacker-controlled length prefix cannot trigger a huge + // reservation (capacity-overflow panic / OOM) before per-element validation. + let cap = n.min(data.len().saturating_sub(header) / 8); + let mut out = Vec::with_capacity(cap); + 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>> { + let n = read_binary_array_len(data)?; + let header = binary_array_header(n); + // Bound the preallocation to what the buffer could actually hold (8 bytes per + // element slot) so an attacker-controlled length prefix cannot trigger a huge + // reservation (capacity-overflow panic / OOM) before per-element validation. + let cap = n.min(data.len().saturating_sub(header) / 8); + let mut out = Vec::with_capacity(cap); + 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 { + 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) +} + +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 +2000,43 @@ 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()); + } } 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>) } } +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> { + 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 { + 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). + pub fn from_simple_stats_row_data(data: &[u8]) -> crate::Result { + 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); + } } From 0b5e0a087db47fb94361484fd7fbb10f437ec4d5 Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Tue, 21 Jul 2026 01:04:29 +0800 Subject: [PATCH 2/6] feat(table): deserialize DataSplit from Paimon-native bytes Add DataSplit::deserialize (raw v8, the reverse of serialize) and DataSplit::deserialize_split_v1 (the reverse of serialize_split_v1: the SplitSerializer frame carrying a DataSplit or an IndexedSplit with row ranges). Includes the Java modified-UTF and deletion-file-list decoders and big-endian cursor readers. Whole-buffer consumption is enforced (trailing bytes rejected), unsupported type ids/versions return Unsupported, counts are bounded against the remaining buffer, and continuation bytes are validated. v8 only for now, with the version dispatch structured so a future v9 branch is a localized addition. --- crates/paimon/src/table/source.rs | 564 +++++++++++++++++++++++++++++- 1 file changed, 560 insertions(+), 4 deletions(-) diff --git a/crates/paimon/src/table/source.rs b/crates/paimon/src/table/source.rs index 49ab30eb..e7654c29 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,112 @@ 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 { + 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 { + 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 { + 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"), + }); + } + let _before_deletion = read_deletion_list(cur)?; // discard (always null on write) + + 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)?; + let _is_streaming = read_u8(cur)?; + 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 +799,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 { + 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 +963,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 { + Ok(take(cur, 1)?[0]) +} + +fn read_i16(cur: &mut &[u8]) -> crate::Result { + Ok(i16::from_be_bytes(take(cur, 2)?.try_into().unwrap())) +} + +fn read_i32(cur: &mut &[u8]) -> crate::Result { + Ok(i32::from_be_bytes(take(cur, 4)?.try_into().unwrap())) +} + +fn read_i64(cur: &mut &[u8]) -> crate::Result { + 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 { + let len = read_i16(cur)? as u16 as usize; + let bytes = take(cur, len)?; + let mut units: Vec = 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>>> { + 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 +1490,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 +1540,7 @@ mod tests { ), }; - let split = DataSplitBuilder::new() + DataSplitBuilder::new() .with_snapshot(18) .with_partition(pb.build()) .with_bucket(20) @@ -1250,9 +1555,91 @@ 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 valid v8 header followed by a huge data_files count must return an error, not + // abort the process via an unbounded `Vec::with_capacity` reservation. + #[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 +1667,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 +1850,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 { + 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 = 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 { .. }) + )); + } } From 8e8256ae99a6f991266b8901bfb000dc9466520b Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Tue, 21 Jul 2026 01:04:29 +0800 Subject: [PATCH 3/6] feat(c): build a plan from native DataSplit bytes Add paimon_plan_from_split_bytes(data, len): deserialize raw-v8 DataSplit bytes into a one-split plan, wrapped in the existing paimon_plan (usable by paimon_table_read_to_arrow, freed by paimon_plan_free). This lets C consumers such as Doris read splits planned in another process, without a catalog round-trip. Null/empty input returns InvalidInput; a compile-time ABI signature guard pins the new symbol. --- bindings/c/src/table.rs | 53 +++++++++++++++++++++++++++++++++++++- bindings/c/src/tests.rs | 56 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 108 insertions(+), 1 deletion(-) 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) }; +} From fc0f99c925b3de52427e0e10f8b6462a60c658de Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Tue, 21 Jul 2026 13:34:56 +0800 Subject: [PATCH 4/6] fix(table): reject amplifying and non-null-before-deletion split inputs - Binary-array decoders: validate the fixed element region (count * 8) fits the buffer up front. Null bigint elements skip the per-slot bounds check, so a forged large count with an all-null bitmap and no element slots would otherwise push count None values from a tiny buffer (~128x memory amplification), reachable through the C entry point. - DataSplit v8 body: reject a non-null beforeDeletionFiles list instead of silently discarding it, matching the beforeFiles handling and Java, which treats such a split as invalid. --- crates/paimon/src/spec/binary_row.rs | 51 ++++++++++++++++++++++------ crates/paimon/src/table/source.rs | 43 ++++++++++++++++++++++- 2 files changed, 83 insertions(+), 11 deletions(-) diff --git a/crates/paimon/src/spec/binary_row.rs b/crates/paimon/src/spec/binary_row.rs index 60a31962..99057e99 100644 --- a/crates/paimon/src/spec/binary_row.rs +++ b/crates/paimon/src/spec/binary_row.rs @@ -761,11 +761,12 @@ pub fn serialize_binary_array_long(values: &[Option]) -> Vec { pub fn deserialize_binary_array_str(data: &[u8]) -> crate::Result> { let n = read_binary_array_len(data)?; let header = binary_array_header(n); - // Bound the preallocation to what the buffer could actually hold (8 bytes per - // element slot) so an attacker-controlled length prefix cannot trigger a huge - // reservation (capacity-overflow panic / OOM) before per-element validation. - let cap = n.min(data.len().saturating_sub(header) / 8); - let mut out = Vec::with_capacity(cap); + // 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 @@ -799,11 +800,12 @@ pub fn deserialize_binary_array_str(data: &[u8]) -> crate::Result> { pub fn deserialize_binary_array_long(data: &[u8]) -> crate::Result>> { let n = read_binary_array_len(data)?; let header = binary_array_header(n); - // Bound the preallocation to what the buffer could actually hold (8 bytes per - // element slot) so an attacker-controlled length prefix cannot trigger a huge - // reservation (capacity-overflow panic / OOM) before per-element validation. - let cap = n.min(data.len().saturating_sub(header) / 8); - let mut out = Vec::with_capacity(cap); + // 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) @@ -833,6 +835,19 @@ fn read_binary_array_len(data: &[u8]) -> crate::Result { 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(), @@ -2039,4 +2054,20 @@ mod tests { 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) + 7) / 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/table/source.rs b/crates/paimon/src/table/source.rs index e7654c29..02074fa7 100644 --- a/crates/paimon/src/table/source.rs +++ b/crates/paimon/src/table/source.rs @@ -727,7 +727,14 @@ impl DataSplit { message: format!("non-empty beforeFiles ({before_files_n}) not supported"), }); } - let _before_deletion = read_deletion_list(cur)?; // discard (always null on write) + // `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 { @@ -1587,6 +1594,40 @@ mod tests { 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 valid v8 header followed by a huge data_files count must return an error, not // abort the process via an unbounded `Vec::with_capacity` reservation. #[test] From 748c7b3da66098b8fd952c7e39c3e42c1d3397a2 Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Tue, 21 Jul 2026 13:56:25 +0800 Subject: [PATCH 5/6] fix(table): reject streaming DataSplit instead of dropping the flag read_v8_body read the isStreaming flag and discarded it. Rust only produces and serves batch splits (isStreaming = false), and Java readers branch on this bit, so a streaming split deserialized here would silently lose semantics. Reject it with Unsupported, matching the beforeFiles / beforeDeletionFiles handling. The SPLIT_V1 frame path inherits this via the shared read_v8_body. --- crates/paimon/src/table/source.rs | 33 ++++++++++++++++++++++++++++--- 1 file changed, 30 insertions(+), 3 deletions(-) diff --git a/crates/paimon/src/table/source.rs b/crates/paimon/src/table/source.rs index 02074fa7..e852630f 100644 --- a/crates/paimon/src/table/source.rs +++ b/crates/paimon/src/table/source.rs @@ -760,7 +760,14 @@ impl DataSplit { } let data_deletion_files = read_deletion_list(cur)?; - let _is_streaming = read_u8(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() @@ -1628,8 +1635,28 @@ mod tests { } } - // A valid v8 header followed by a huge data_files count must return an error, not - // abort the process via an unbounded `Vec::with_capacity` reservation. + // 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 From 45f568074f026aa0d9f30ffb750d020e95c13bd2 Mon Sep 17 00:00:00 2001 From: JunRuiLee Date: Tue, 21 Jul 2026 14:22:54 +0800 Subject: [PATCH 6/6] test(spec): use div_ceil for bitmap length in amplification test The manual (n + 7) / 8 form trips clippy::manual_div_ceil under newer toolchains. usize::div_ceil is stable and clearer. --- crates/paimon/src/spec/binary_row.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/paimon/src/spec/binary_row.rs b/crates/paimon/src/spec/binary_row.rs index 99057e99..ce08ad8b 100644 --- a/crates/paimon/src/spec/binary_row.rs +++ b/crates/paimon/src/spec/binary_row.rs @@ -2064,7 +2064,7 @@ mod tests { // reachable through the C entry point (OOM risk). Must error, not // allocate. let count: i32 = 8000; - let bitmap_len = ((count as usize) + 7) / 8; // 1000 bytes + 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