feat(c): add bounded commit-message range export

bindings/c/src/tests/commit_messages.rs

@@ -119,6 +119,107 @@ unsafe fn copy_bodies(array: &paimon_bytes_array) -> Vec<Vec<u8>> {
         .collect()
 }
 
+#[test]
+fn bounded_range_export_preserves_java_bytes_and_ownership() {
+    unsafe {
+        let table = memory_table("memory:/bounded-commit-export");
+        let commit = new_committer(table, "coordinator", false);
+        let bodies = java_bodies();
+        let messages = import(commit, &bodies);
+        let mut count = 0;
+        assert_ok(paimon_commit_messages_count(messages, &mut count));
+        assert_eq!(count, bodies.len());
+        let total = bodies.iter().map(Vec::len).sum::<usize>();
+        let all = paimon_commit_messages_serialize_range(messages, 0, count, total);
+        assert_ok(all.error);
+        assert_eq!(copy_bodies(&all.bytes), bodies);
+        paimon_bytes_array_free(all.bytes);
+
+        // Every prefix is rolled back on a cumulative limit error. The same
+        // source remains available for a smaller-range retry.
+        let rejected = paimon_commit_messages_serialize_range(messages, 0, count, total - 1);
+        assert_error(rejected.error, PaimonErrorCode::ResourceExhausted);
+        assert!(rejected.bytes.data.is_null());
+        assert_eq!(rejected.bytes.len, 0);
+        paimon_bytes_array_free(rejected.bytes);
+
+        for (offset, body) in bodies.iter().enumerate() {
+            let rejected =
+                paimon_commit_messages_serialize_range(messages, offset, 1, body.len() - 1);
+            assert_error(rejected.error, PaimonErrorCode::ResourceExhausted);
+            assert!(rejected.bytes.data.is_null());
+            assert_eq!(rejected.bytes.len, 0);
+            paimon_bytes_array_free(rejected.bytes);
+            let result = paimon_commit_messages_serialize_range(messages, offset, 1, body.len());
+            assert_ok(result.error);
+            assert_eq!(copy_bodies(&result.bytes), vec![body.clone()]);
+            paimon_bytes_array_free(result.bytes);
+        }
+        let tail = paimon_commit_messages_serialize_range(messages, 1, 2, total - bodies[0].len());
+        assert_ok(tail.error);
+        paimon_commit_messages_free(messages);
+        paimon_table_commit_free(commit);
+        unwrap_table(table);
+        assert_eq!(copy_bodies(&tail.bytes), bodies[1..]);
+        paimon_bytes_array_free(tail.bytes);
+    }
+}
+
+#[test]
+fn bounded_range_export_validates_ranges_and_pointers() {
+    unsafe {
+        let table = memory_table("memory:/invalid-commit-range");
+        let commit = new_committer(table, "coordinator", false);
+        let messages = import(commit, &java_bodies());
+        for (offset, count) in [(4, 0), (3, 1), (usize::MAX, 1), (1, usize::MAX)] {
+            let result =
+                paimon_commit_messages_serialize_range(messages, offset, count, usize::MAX);
+            assert_error(result.error, PaimonErrorCode::InvalidInput);
+            assert_eq!(result.bytes.len, 0);
+            assert!(result.bytes.data.is_null());
+            paimon_bytes_array_free(result.bytes);
+        }
+        for offset in [0, 1, 3] {
+            let result = paimon_commit_messages_serialize_range(messages, offset, 0, 0);
+            assert_ok(result.error);
+            assert_eq!(result.bytes.len, 0);
+            paimon_bytes_array_free(result.bytes);
+        }
+        let uninitialized = paimon_commit_messages {
+            inner: ptr::null_mut(),
+        };
+        for handle in [ptr::null(), &uninitialized] {
+            let result = paimon_commit_messages_serialize_range(handle, 0, 0, 0);
+            assert_error(result.error, PaimonErrorCode::InvalidInput);
+            assert_eq!(result.bytes.len, 0);
+            assert!(result.bytes.data.is_null());
+            paimon_bytes_array_free(result.bytes);
+            let mut count = 42;
+            assert_error(
+                paimon_commit_messages_count(handle, &mut count),
+                PaimonErrorCode::InvalidInput,
+            );
+            assert_eq!(count, 42);
+        }
+        assert_error(
+            paimon_commit_messages_count(messages, ptr::null_mut()),
+            PaimonErrorCode::InvalidInput,
+        );
+        let empty = import(commit, &[]);
+        let mut count = 42;
+        assert_ok(paimon_commit_messages_count(empty, &mut count));
+        assert_eq!(count, 0);
+        let result = paimon_commit_messages_serialize_range(empty, 0, 0, 0);
+        assert_ok(result.error);
+        assert_eq!(result.bytes.len, 0);
+        paimon_bytes_array_free(result.bytes);
+        paimon_commit_messages_free(empty);
+        paimon_commit_messages_free(messages);
+        paimon_table_commit_free(commit);
+        unwrap_table(table);
+    }
+}
+
 unsafe fn import(
     commit: *const paimon_table_commit,
     bodies: &[Vec<u8>],
@@ -473,6 +574,11 @@ fn format_table_messages_are_unsupported_and_remain_owned_by_caller() {
         assert!(exported.bytes.data.is_null());
         assert_eq!(exported.bytes.len, 0);
         paimon_bytes_array_free(exported.bytes);
+        let exported = paimon_commit_messages_serialize_range(messages, 0, 1, usize::MAX);
+        assert_error(exported.error, PaimonErrorCode::Unsupported);
+        assert!(exported.bytes.data.is_null());
+        assert_eq!(exported.bytes.len, 0);
+        paimon_bytes_array_free(exported.bytes);
         let commit = new_committer(table, "worker", false);
         let result = paimon_commit_messages_deserialize(commit, 14, ptr::null(), 0);
         assert_error(result.error, PaimonErrorCode::Unsupported);

bindings/c/src/write.rs

@@ -861,6 +861,31 @@ pub extern "C" fn paimon_commit_messages_serializer_version() -> i32 {
     COMMIT_MESSAGE_SERIALIZER_VERSION
 }
 
+/// Write the number of messages in a standard message handle to count.
+/// The handle is borrowed; count is unchanged on error.
+///
+/// # Safety
+/// msgs must be a live standard message handle and count must point to writable
+/// storage, or either may be null (returns an error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_commit_messages_count(
+    msgs: *const paimon_commit_messages,
+    count: *mut usize,
+) -> *mut paimon_error {
+    if let Err(error) = check_non_null(msgs, "msgs") {
+        return error;
+    }
+    if let Err(error) = check_non_null(count, "count") {
+        return error;
+    }
+    if let Err(error) = check_non_null((*msgs).inner, "msgs.inner") {
+        return error;
+    }
+    let state = &*((*msgs).inner as *const CommitMessagesState);
+    *count = state.messages.len();
+    ptr::null_mut()
+}
+
 /// Export one unframed Java v14 body per message, preserving message order.
 /// Table identity, commit_user and overwrite mode are not part of the bodies.
 /// Format Table messages are unsupported. No postpone-specific handle is accepted.
@@ -874,16 +899,57 @@ pub extern "C" fn paimon_commit_messages_serializer_version() -> i32 {
 #[no_mangle]
 pub unsafe extern "C" fn paimon_commit_messages_serialize(
     msgs: *const paimon_commit_messages,
+) -> paimon_result_bytes_array {
+    serialize_commit_messages(msgs, None, usize::MAX)
+}
+
+/// Export count messages starting at offset, in order, limiting the sum of
+/// their unframed Java body lengths to max_bytes. No list or RPC framing is added.
+/// The output uses the same version and ownership as paimon_commit_messages_serialize.
+///
+/// An empty range at or before the end is valid, including with a zero limit.
+/// Invalid ranges return InvalidInput; an exceeded limit returns ResourceExhausted.
+/// Errors return no partial output and leave the messages usable for a smaller
+/// range retry. This bounds output length, not temporary allocations or peak memory.
+///
+/// # Safety
+/// msgs must be a live standard message handle, or null (returns an error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_commit_messages_serialize_range(
+    msgs: *const paimon_commit_messages,
+    offset: usize,
+    count: usize,
+    max_bytes: usize,
+) -> paimon_result_bytes_array {
+    serialize_commit_messages(msgs, Some((offset, count)), max_bytes)
+}
+
+unsafe fn serialize_commit_messages(
+    msgs: *const paimon_commit_messages,
+    range: Option<(usize, usize)>,
+    max_bytes: usize,
 ) -> paimon_result_bytes_array {
     let encode = || -> Result<Vec<Vec<u8>>, *mut paimon_error> {
         check_non_null(msgs, "msgs")?;
         check_non_null((*msgs).inner, "msgs.inner")?;
         let state = &*((*msgs).inner as *const CommitMessagesState);
-        state
-            .messages
-            .iter()
-            .map(|message| message.serialize().map_err(paimon_error::from_paimon))
-            .collect()
+        let selected = match range {
+            Some((offset, count)) => offset
+                .checked_add(count)
+                .and_then(|end| state.messages.get(offset..end))
+                .ok_or_else(|| invalid_input("Invalid commit message range"))?,
+            None => &state.messages,
+        };
+        let mut remaining = max_bytes;
+        let mut bodies = Vec::new();
+        for message in selected {
+            let body = message
+                .serialize_with_limit(remaining)
+                .map_err(paimon_error::from_paimon)?;
+            remaining -= body.len();
+            bodies.push(body);
+        }
+        Ok(bodies)
     };
     match encode() {
         Ok(bytes) => paimon_result_bytes_array {
@@ -1513,8 +1579,16 @@ const _: unsafe extern "C" fn(
 const _: unsafe extern "C" fn(*mut paimon_table_write) -> paimon_result_prepare_commit =
     paimon_table_write_prepare_commit;
 const _: extern "C" fn() -> i32 = paimon_commit_messages_serializer_version;
+const _: unsafe extern "C" fn(*const paimon_commit_messages, *mut usize) -> *mut paimon_error =
+    paimon_commit_messages_count;
 const _: unsafe extern "C" fn(*const paimon_commit_messages) -> paimon_result_bytes_array =
     paimon_commit_messages_serialize;
+const _: unsafe extern "C" fn(
+    *const paimon_commit_messages,
+    usize,
+    usize,
+    usize,
+) -> paimon_result_bytes_array = paimon_commit_messages_serialize_range;
 const _: unsafe extern "C" fn(
     *const paimon_table_commit,
     i32,

crates/paimon/src/table/commit_message.rs

@@ -30,16 +30,27 @@ fn invalid(message: impl Into<String>) -> crate::Error {
     }
 }
 
+fn append(out: &mut Vec<u8>, bytes: &[u8], max_bytes: usize) -> crate::Result<()> {
+    if bytes.len() > max_bytes.saturating_sub(out.len()) {
+        return Err(crate::Error::ResourceExhausted {
+            message: "Serialized commit message exceeds its output limit".into(),
+        });
+    }
+    out.extend_from_slice(bytes);
+    Ok(())
+}
+
 fn write_rows<T>(
     out: &mut Vec<u8>,
     values: &[T],
     encode: impl Fn(&T) -> crate::Result<Vec<u8>>,
+    max_bytes: usize,
 ) -> crate::Result<()> {
-    out.extend_from_slice(&(values.len() as i32).to_be_bytes());
+    append(out, &(values.len() as i32).to_be_bytes(), max_bytes)?;
     for value in values {
         let row = encode(value)?;
-        out.extend_from_slice(&(row.len() as i32).to_be_bytes());
-        out.extend_from_slice(&row);
+        append(out, &(row.len() as i32).to_be_bytes(), max_bytes)?;
+        append(out, &row, max_bytes)?;
     }
     Ok(())
 }
@@ -137,6 +148,16 @@ impl CommitMessage {
 
     /// Write the unframed Java v14 `CommitMessageSerializer.serialize` body.
     pub fn serialize(&self) -> crate::Result<Vec<u8>> {
+        self.serialize_with_limit(usize::MAX)
+    }
+
+    /// Encode the same body as [`Self::serialize`], limiting its output length.
+    /// Returns [`crate::Error::ResourceExhausted`] before appending bytes that
+    /// exceed `max_bytes`, without modifying this message.
+    ///
+    /// This is not a peak-memory limit: temporary metadata rows are encoded by
+    /// their existing serializers, and vector capacity may exceed its length.
+    pub fn serialize_with_limit(&self, max_bytes: usize) -> crate::Result<Vec<u8>> {
         if self.format_file.is_some() {
             return Err(crate::Error::Unsupported {
                 message: "Format Table two-phase file messages use a different Java serializer"
@@ -153,39 +174,67 @@ impl CommitMessage {
             BinaryRow::from_serialized_bytes(&self.partition)?;
             self.partition.clone()
         };
-        out.extend_from_slice(&(partition.len() as i32).to_be_bytes());
-        out.extend_from_slice(&partition);
-        out.extend_from_slice(&self.bucket.to_be_bytes());
-        out.push(u8::from(self.total_buckets.is_some()));
+        append(&mut out, &(partition.len() as i32).to_be_bytes(), max_bytes)?;
+        append(&mut out, &partition, max_bytes)?;
+        append(&mut out, &self.bucket.to_be_bytes(), max_bytes)?;
+        append(
+            &mut out,
+            &[u8::from(self.total_buckets.is_some())],
+            max_bytes,
+        )?;
         if let Some(value) = self.total_buckets {
-            out.extend_from_slice(&value.to_be_bytes());
+            append(&mut out, &value.to_be_bytes(), max_bytes)?;
         }
         for files in [
             &self.new_files,
             &self.deleted_files,
             &self.new_changelog_files,
         ] {
-            write_rows(&mut out, files, DataFileMeta::to_serialized_row_data)?;
+            write_rows(
+                &mut out,
+                files,
+                DataFileMeta::to_serialized_row_data,
+                max_bytes,
+            )?;
         }
         for files in [&self.new_index_files, &self.deleted_index_files] {
-            write_rows(&mut out, files, IndexFileMeta::to_serialized_row_data)?;
+            write_rows(
+                &mut out,
+                files,
+                IndexFileMeta::to_serialized_row_data,
+                max_bytes,
+            )?;
         }
         for files in [
             &self.compact_before,
             &self.compact_after,
             &self.compact_changelog_files,
         ] {
-            write_rows(&mut out, files, DataFileMeta::to_serialized_row_data)?;
+            write_rows(
+                &mut out,
+                files,
+                DataFileMeta::to_serialized_row_data,
+                max_bytes,
+            )?;
         }
         for files in [
             &self.compact_new_index_files,
             &self.compact_deleted_index_files,
         ] {
-            write_rows(&mut out, files, IndexFileMeta::to_serialized_row_data)?;
+            write_rows(
+                &mut out,
+                files,
+                IndexFileMeta::to_serialized_row_data,
+                max_bytes,
+            )?;
         }
-        out.push(u8::from(self.check_from_snapshot.is_some()));
+        append(
+            &mut out,
+            &[u8::from(self.check_from_snapshot.is_some())],
+            max_bytes,
+        )?;
         if let Some(value) = self.check_from_snapshot {
-            out.extend_from_slice(&value.to_be_bytes());
+            append(&mut out, &value.to_be_bytes(), max_bytes)?;
         }
         Ok(out)
     }
@@ -370,6 +419,21 @@ mod tests {
         );
     }
 
+    #[test]
+    fn bounded_encoding_matches_the_upstream_protocol() {
+        for fixture in [EMPTY, INDEXED, RICH_INDEXED] {
+            let bytes = base64::engine::general_purpose::STANDARD
+                .decode(fixture)
+                .unwrap();
+            let message = CommitMessage::deserialize(14, &bytes).unwrap();
+            assert_eq!(message.serialize_with_limit(bytes.len()).unwrap(), bytes);
+            assert!(matches!(
+                message.serialize_with_limit(bytes.len() - 1),
+                Err(crate::Error::ResourceExhausted { .. })
+            ));
+        }
+    }
+
     #[test]
     fn rejects_truncated_or_unsupported_message() {
         let bytes = base64::engine::general_purpose::STANDARD