From 7000243dedc2e3cc19afa3a022030c04a6b298cf Mon Sep 17 00:00:00 2001 From: Socrates Date: Thu, 5 Nov 2026 16:15:12 +0800 Subject: [PATCH] feat(c): add bounded commit-message range export --- bindings/c/src/tests/commit_messages.rs | 106 ++++++++++++++++++++++ bindings/c/src/write.rs | 84 ++++++++++++++++- crates/paimon/src/table/commit_message.rs | 92 ++++++++++++++++--- docs/src/c-binding.md | 41 +++++++++ 4 files changed, 304 insertions(+), 19 deletions(-) diff --git a/bindings/c/src/tests/commit_messages.rs b/bindings/c/src/tests/commit_messages.rs index ac2905c3a..af373ccfe 100644 --- a/bindings/c/src/tests/commit_messages.rs +++ b/bindings/c/src/tests/commit_messages.rs @@ -119,6 +119,107 @@ unsafe fn copy_bodies(array: &paimon_bytes_array) -> Vec> { .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::(); + 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], @@ -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); diff --git a/bindings/c/src/write.rs b/bindings/c/src/write.rs index 9294bb4bc..04d89d892 100644 --- a/bindings/c/src/write.rs +++ b/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>, *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, diff --git a/crates/paimon/src/table/commit_message.rs b/crates/paimon/src/table/commit_message.rs index db24ee9cf..5d0b1994e 100644 --- a/crates/paimon/src/table/commit_message.rs +++ b/crates/paimon/src/table/commit_message.rs @@ -30,16 +30,27 @@ fn invalid(message: impl Into) -> crate::Error { } } +fn append(out: &mut Vec, 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( out: &mut Vec, values: &[T], encode: impl Fn(&T) -> crate::Result>, + 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> { + 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> { 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 diff --git a/docs/src/c-binding.md b/docs/src/c-binding.md index b32a739ea..c31cb59b6 100644 --- a/docs/src/c-binding.md +++ b/docs/src/c-binding.md @@ -733,6 +733,47 @@ of those bodies. Do not concatenate bodies or add a header inside a body. An empty message list exports an empty array. The buffers are independent of the source handle and remain valid after that handle is freed. +For a size-limited transport, query the message count and export ranges. The +limit applies to the sum of body lengths, excluding array overhead and the +application's framing. Reserve space for that framing in the transport budget. +This example reduces a range after `ResourceExhausted` (code `6`): + +```c +size_t total = 0; +paimon_error *count_error = paimon_commit_messages_count(messages, &total); +if (count_error != NULL) { + /* Report the error before releasing it. */ + paimon_error_free(count_error); + return -1; +} +for (size_t offset = 0; offset < total; ) { + size_t count = total - offset; + paimon_result_bytes_array chunk; + for (;;) { + chunk = paimon_commit_messages_serialize_range( + messages, offset, count, body_limit); + if (chunk.error == NULL) break; + int retry = chunk.error->code == 6 && count > 1; + paimon_error_free(chunk.error); + paimon_bytes_array_free(chunk.bytes); /* Empty on error. */ + if (!retry) return -1; /* A single body may exceed the limit. */ + count /= 2; + } + /* Send or copy chunk.bytes, preserving the bodies' order and boundaries. */ + paimon_bytes_array_free(chunk.bytes); + offset += count; +} +``` + +Errors do not consume messages or expose a partial array, so a smaller range +can be retried against the same handle. `offset + count` must not overflow or +exceed the message count. Empty ranges at or before the end succeed even with +a zero-byte limit. The existing full-list export remains available. + +The byte limit is not a peak-memory or RSS bound: individual metadata rows +still use temporary encoding buffers, and vector capacity and result ownership +conversion can allocate additional memory. + On the coordinator, create a standard committer for the intended table and operation. Enable overwrite on its builder before creating it if the workers prepared an overwrite. Pass the received bodies as borrowed