fix(table): bound split count reservations by remaining bytes

Three split decoders read an i32 count from untrusted wire bytes and
eagerly reserved Vec capacity from it before reading any element, so a
crafted buffer with a valid header and a huge count could reserve
hundreds of GB and abort the process via handle_alloc_error. Cap each
reservation by the remaining cursor length divided by the minimum
per-element wire size (data files: 4B length prefix; deletion entries:
1B null flag; row ranges: 16B), matching the leaf decoders. Capacity is
only a hint, so well-formed round-trips are unchanged. Also note
paimon_plan_from_split_bytes as a valid plan source in the C bindings.

Author: JunRuiLee

bindings/c/src/table.rs

@@ -519,6 +519,7 @@ pub unsafe extern "C" fn paimon_plan_from_split_bytes(
 ///
 /// # 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() {
@@ -533,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() {

crates/paimon/src/table/source.rs

@@ -736,7 +736,10 @@ impl DataSplit {
                 source: None,
             });
         }
-        let mut data_files = Vec::with_capacity(data_files_n as usize);
+        // 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 {
@@ -840,7 +843,9 @@ impl DataSplit {
                         source: None,
                     });
                 }
-                let mut ranges = Vec::with_capacity(ranges_n as usize);
+                // 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)?;
@@ -1038,7 +1043,9 @@ fn read_deletion_list(cur: &mut &[u8]) -> crate::Result<Option<Vec<Option<Deleti
             source: None,
         });
     }
-    let mut out = Vec::with_capacity(count as usize);
+    // 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);
@@ -1572,6 +1579,61 @@ mod tests {
         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:?}"),
+        }
+    }
+
+    // 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]
     fn write_java_utf_matches_java_modified_utf8() {
         let enc = |s: &str| {