This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new dd2f0688 fix(cpp): catch panics at the FFI boundary so they surface as
errors (#766)
dd2f0688 is described below
commit dd2f0688e7a9b1e7a43ebb1fdff7d8df7e39b335
Author: Sagar Sumit <[email protected]>
AuthorDate: Tue Oct 27 23:06:38 2026 +0530
fix(cpp): catch panics at the FFI boundary so they surface as errors (#766)
cxx's shims are extern "C", so a panic reachable from a read (an
unsupported Avro type, an internal invariant) unwinds out of the shim
and aborts the host process. For an engine that embeds the library in
a long-lived multi-tenant worker that takes every co-running query
down with it.
Wrap each extern "Rust" entry point in catch_unwind and report the
panic through the Result the bridge already throws as rust::Error, so
the one call fails and the reader stays usable. The default panic hook
still prints the message to stderr; embedders that find that noisy can
install their own.
hudi-test moves to dev-dependencies: it is only used by the new tests
and consumers should not pull the fixture crate.
Closes #636
Signed-off-by: codope <[email protected]>
---
cpp/Cargo.toml | 4 +-
cpp/src/lib.rs | 381 +++++++++++++++++++++++++++++++++++++++++++--------------
2 files changed, 291 insertions(+), 94 deletions(-)
diff --git a/cpp/Cargo.toml b/cpp/Cargo.toml
index 3323d41b..721a9feb 100644
--- a/cpp/Cargo.toml
+++ b/cpp/Cargo.toml
@@ -33,11 +33,13 @@ crate-type = ["cdylib", "rlib"]
[dependencies]
hudi = { path = "../crates/hudi", version = "0.6.0-dev" }
-hudi-test = { path = "../crates/test", version = "0.6.0-dev" }
cxx = { version = "1.0" }
arrow = { workspace = true , features = ["ffi"]}
arrow-array = { workspace = true , features = ["ffi"]}
tokio = { workspace = true }
+[dev-dependencies]
+hudi-test = { path = "../crates/test", version = "0.6.0-dev" }
+
[build-dependencies]
cxx-build = { version = "1.0" }
diff --git a/cpp/src/lib.rs b/cpp/src/lib.rs
index 08c14658..fd8d54e4 100644
--- a/cpp/src/lib.rs
+++ b/cpp/src/lib.rs
@@ -24,6 +24,8 @@ use hudi::file_group::FileGroup;
use hudi::file_group::file_slice::FileSlice;
use hudi::file_group::reader::FileGroupReader;
use hudi::table::ReadOptions;
+use std::any::Any;
+use std::panic::{AssertUnwindSafe, catch_unwind};
#[cxx::bridge]
mod ffi {
@@ -35,24 +37,36 @@ mod ffi {
extern "Rust" {
type HudiFileGroupReader;
+ /// Opens a reader for the table at `base_uri` with `key=value`
options.
+ ///
+ /// Throws `rust::Error` on failure, including a panic inside the
library.
fn new_file_group_reader_with_options(
base_uri: &CxxString,
options: &CxxVector<CxxString>,
) -> Result<Box<HudiFileGroupReader>>;
type HudiFileSlice;
+ /// Builds a file slice from file names within `partition_path`.
+ ///
+ /// Throws `rust::Error` on failure, including a panic inside the
library.
fn new_file_slice_from_file_names(
partition_path: &CxxString,
base_file_name: &CxxString,
log_file_names: &CxxVector<CxxString>,
) -> Result<Box<HudiFileSlice>>;
+ /// Reads and merges a file slice given paths relative to the table
base.
+ ///
+ /// Throws `rust::Error` on failure, including a panic inside the
library.
fn read_file_slice_from_paths(
self: &HudiFileGroupReader,
base_file_path: &CxxString,
log_file_paths: &CxxVector<CxxString>,
) -> Result<*mut ArrowArrayStream>;
+ /// Reads and merges a file slice.
+ ///
+ /// Throws `rust::Error` on failure, including a panic inside the
library.
fn read_file_slice(
self: &HudiFileGroupReader,
file_slice: &HudiFileSlice,
@@ -60,6 +74,44 @@ mod ffi {
}
}
+/// Converts a panic in `f` into the `Err` the bridge already reports.
+///
+/// cxx's shims are `extern "C"`, so a panic unwinding out of one aborts the
+/// process, and with it every query in a host that embeds this library.
+/// `AssertUnwindSafe` holds because the reader is read-only across calls;
+/// the only state a panic can leave mid-update is storage read counters,
+/// which nothing depends on for correctness.
+fn guard<T>(entry_point: &str, f: impl FnOnce() -> Result<T, String>) ->
Result<T, String> {
+ catch_unwind(AssertUnwindSafe(f)).unwrap_or_else(|payload| {
+ Err(format!(
+ "{entry_point} panicked: {}",
+ panic_message(payload.as_ref())
+ ))
+ })
+}
+
+/// The message a panic carried, for the two payload types `panic!` produces.
+fn panic_message(payload: &(dyn Any + Send)) -> String {
+ if let Some(message) = payload.downcast_ref::<&str>() {
+ (*message).to_string()
+ } else if let Some(message) = payload.downcast_ref::<String>() {
+ message.clone()
+ } else {
+ "non-string panic payload".to_string()
+ }
+}
+
+fn cxx_str(value: &CxxString) -> Result<&str, String> {
+ value
+ .to_str()
+ .map_err(|e| format!("Failed to convert CxxString to str: {e}"))
+}
+
+fn cxx_strs(values: &CxxVector<CxxString>) -> Result<Vec<&str>, String> {
+ values.iter().map(cxx_str).collect()
+}
+
+#[derive(Debug)]
pub struct HudiFileGroupReader {
inner: FileGroupReader,
rt: tokio::runtime::Runtime,
@@ -68,29 +120,24 @@ pub struct HudiFileGroupReader {
pub fn new_file_group_reader_with_options(
base_uri: &CxxString,
options: &CxxVector<CxxString>,
-) -> std::result::Result<Box<HudiFileGroupReader>, String> {
- let base_uri = base_uri
- .to_str()
- .map_err(|e| format!("Failed to convert CxxString to str: {e}"))?;
-
- let mut opt_vec = Vec::new();
- for opt in options.iter() {
- let opt_str = opt
- .to_str()
- .map_err(|e| format!("Failed to convert CxxString to str: {e}"))?;
- if let Some((key, value)) = opt_str.split_once('=') {
- opt_vec.push((key, value))
- }
- }
-
- let rt = tokio::runtime::Builder::new_current_thread()
- .enable_all()
- .build()
- .map_err(|e| format!("Failed to create tokio runtime: {e}"))?;
- let reader = rt
- .block_on(FileGroupReader::new_with_options(base_uri, opt_vec))
- .map_err(|e| format!("Failed to create FileGroupReader: {e}"))?;
- Ok(Box::new(HudiFileGroupReader { inner: reader, rt }))
+) -> Result<Box<HudiFileGroupReader>, String> {
+ guard("new_file_group_reader_with_options", || {
+ let base_uri = cxx_str(base_uri)?;
+
+ let opt_vec: Vec<(&str, &str)> = cxx_strs(options)?
+ .into_iter()
+ .filter_map(|opt| opt.split_once('='))
+ .collect();
+
+ let rt = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .map_err(|e| format!("Failed to create tokio runtime: {e}"))?;
+ let reader = rt
+ .block_on(FileGroupReader::new_with_options(base_uri, opt_vec))
+ .map_err(|e| format!("Failed to create FileGroupReader: {e}"))?;
+ Ok(Box::new(HudiFileGroupReader { inner: reader, rt }))
+ })
}
impl HudiFileGroupReader {
@@ -98,56 +145,51 @@ impl HudiFileGroupReader {
&self,
base_file_path: &CxxString,
log_file_paths: &CxxVector<CxxString>,
- ) -> std::result::Result<*mut ffi::ArrowArrayStream, String> {
- let base_file_path = base_file_path
- .to_str()
- .map_err(|e| format!("Failed to convert CxxString to str: {e}"))?;
-
- let log_file_paths = log_file_paths
- .iter()
- .map(|p| {
- p.to_str()
- .map(String::from)
- .map_err(|e| format!("Failed to convert CxxString to str:
{e}"))
- })
- .collect::<std::result::Result<Vec<_>, _>>()?;
-
- let record_batch = self
- .rt
- .block_on(self.inner.read_file_slice_from_paths(
- base_file_path,
- log_file_paths,
- &ReadOptions::new(),
- ))
- .map_err(|e| format!("Failed to read file batch: {e}"))?;
- let schema = record_batch.schema();
+ ) -> Result<*mut ffi::ArrowArrayStream, String> {
+ guard("read_file_slice_from_paths", || {
+ let base_file_path = cxx_str(base_file_path)?;
+ let log_file_paths = cxx_strs(log_file_paths)?;
- Ok(create_raw_pointer_for_record_batches(
- vec![record_batch],
- schema,
- ))
+ let record_batch = self
+ .rt
+ .block_on(self.inner.read_file_slice_from_paths(
+ base_file_path,
+ log_file_paths,
+ &ReadOptions::new(),
+ ))
+ .map_err(|e| format!("Failed to read file batch: {e}"))?;
+ let schema = record_batch.schema();
+
+ Ok(create_raw_pointer_for_record_batches(
+ vec![record_batch],
+ schema,
+ ))
+ })
}
pub fn read_file_slice(
&self,
file_slice: &HudiFileSlice,
- ) -> std::result::Result<*mut ffi::ArrowArrayStream, String> {
- let record_batch = self
- .rt
- .block_on(
- self.inner
- .read_file_slice(&file_slice.inner, &ReadOptions::new()),
- )
- .map_err(|e| format!("Failed to read file slice: {e}"))?;
- let schema = record_batch.schema();
-
- Ok(create_raw_pointer_for_record_batches(
- vec![record_batch],
- schema,
- ))
+ ) -> Result<*mut ffi::ArrowArrayStream, String> {
+ guard("read_file_slice", || {
+ let record_batch = self
+ .rt
+ .block_on(
+ self.inner
+ .read_file_slice(&file_slice.inner,
&ReadOptions::new()),
+ )
+ .map_err(|e| format!("Failed to read file slice: {e}"))?;
+ let schema = record_batch.schema();
+
+ Ok(create_raw_pointer_for_record_batches(
+ vec![record_batch],
+ schema,
+ ))
+ })
}
}
+#[derive(Debug)]
pub struct HudiFileSlice {
inner: FileSlice,
}
@@ -156,35 +198,188 @@ pub fn new_file_slice_from_file_names(
partition_path: &CxxString,
base_file_name: &CxxString,
log_file_names: &CxxVector<CxxString>,
-) -> std::result::Result<Box<HudiFileSlice>, String> {
- let partition_path = partition_path
- .to_str()
- .map_err(|e| format!("Failed to convert CxxString to str: {e}"))?;
- let base_file_name = base_file_name
- .to_str()
- .map_err(|e| format!("Failed to convert CxxString to str: {e}"))?;
+) -> Result<Box<HudiFileSlice>, String> {
+ guard("new_file_slice_from_file_names", || {
+ let partition_path = cxx_str(partition_path)?;
+ let base_file_name = cxx_str(base_file_name)?;
+ let log_file_names = cxx_strs(log_file_names)?;
+
+ let mut file_group =
FileGroup::new_with_base_file_name(base_file_name, partition_path)
+ .map_err(|e| format!("Failed to create FileGroup: {e}"))?;
+ file_group
+ .add_log_files_from_names(&log_file_names)
+ .map_err(|e| format!("Failed to add files to FileGroup: {e}"))?;
+
+ let (_, file_slice) =
+ file_group.file_slices.iter().next().ok_or_else(|| {
+ format!("Failed to get file slice from FileGroup:
{file_group:?}")
+ })?;
+
+ Ok(Box::new(HudiFileSlice {
+ inner: file_slice.clone(),
+ }))
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow_array::ffi_stream::{ArrowArrayStreamReader,
FFI_ArrowArrayStream};
+ use cxx::let_cxx_string;
+ use hudi_test::SampleTable;
+
+ /// Takes back the stream a bridge function leaked for C++ and reads it.
+ fn rows_in(raw: *mut ffi::ArrowArrayStream) -> usize {
+ // SAFETY: `raw` is the one box the bridge leaks for the C++ caller,
+ // reclaimed here instead.
+ let stream = unsafe { Box::from_raw(raw as *mut FFI_ArrowArrayStream)
};
+ ArrowArrayStreamReader::try_new(*stream)
+ .unwrap()
+ .map(|batch| batch.unwrap().num_rows())
+ .sum()
+ }
+
+ #[test]
+ fn test_guard_ok_returns_value() {
+ assert_eq!(guard("op", || Ok(7)), Ok(7));
+ }
+
+ #[test]
+ fn test_guard_err_passes_through_unchanged() {
+ let err = guard("op", || Err::<(), _>("bad
input".to_string())).unwrap_err();
+ assert_eq!(err, "bad input");
+ }
+
+ #[test]
+ fn test_guard_str_panic_becomes_error_naming_entry_point() {
+ let err = guard("read_file_slice", || -> Result<(), String> {
+ panic!("unsupported avro type")
+ })
+ .unwrap_err();
+ assert_eq!(err, "read_file_slice panicked: unsupported avro type");
+ }
- let log_file_names = log_file_names
- .iter()
- .map(|name| {
- name.to_str()
- .map_err(|e| format!("Failed to convert CxxString to str:
{e}"))
+ #[test]
+ fn test_guard_string_payload_keeps_message() {
+ let column = "map_col";
+ let err = guard("op", || -> Result<(), String> {
+ panic!("no support for {column}")
})
- .collect::<std::result::Result<Vec<_>, _>>()?;
-
- let mut file_group = FileGroup::new_with_base_file_name(base_file_name,
partition_path)
- .map_err(|e| format!("Failed to create FileGroup: {e}"))?;
- file_group
- .add_log_files_from_names(&log_file_names)
- .map_err(|e| format!("Failed to add files to FileGroup: {e}"))?;
-
- let (_, file_slice) = file_group
- .file_slices
- .iter()
- .next()
- .ok_or_else(|| format!("Failed to get file slice from FileGroup:
{file_group:?}"))?;
-
- Ok(Box::new(HudiFileSlice {
- inner: file_slice.clone(),
- }))
+ .unwrap_err();
+ assert_eq!(err, "op panicked: no support for map_col");
+ }
+
+ #[test]
+ fn test_guard_non_string_payload_is_still_an_error() {
+ let err = guard("op", || -> Result<(), String> {
+ std::panic::panic_any(42u8)
+ })
+ .unwrap_err();
+ assert_eq!(err, "op panicked: non-string panic payload");
+ }
+
+ #[test]
+ fn test_new_file_group_reader_with_options_unreadable_table_is_error() {
+ let_cxx_string!(base_uri = "/definitely/not/a/hudi/table");
+ let options = CxxVector::<CxxString>::new();
+ let err = new_file_group_reader_with_options(&base_uri,
&options).unwrap_err();
+ assert!(
+ err.starts_with("Failed to create FileGroupReader"),
+ "unexpected error: {err}"
+ );
+ }
+
+ #[test]
+ fn test_new_file_slice_from_file_names_bad_base_file_name_is_error() {
+ let_cxx_string!(partition_path = "");
+ let_cxx_string!(base_file_name = "not-a-hudi-base-file");
+ let log_file_names = CxxVector::<CxxString>::new();
+ let err = new_file_slice_from_file_names(&partition_path,
&base_file_name, &log_file_names)
+ .unwrap_err();
+ assert!(
+ err.starts_with("Failed to create FileGroup"),
+ "unexpected error: {err}"
+ );
+ }
+
+ fn first_base_file(table_path: &str) -> String {
+ let mut base_files: Vec<String> = std::fs::read_dir(table_path)
+ .unwrap()
+ .filter_map(|e| e.ok())
+ .map(|e| e.file_name().to_string_lossy().into_owned())
+ .filter(|name| name.ends_with(".parquet") &&
!name.starts_with('.'))
+ .collect();
+ base_files.sort();
+ base_files
+ .into_iter()
+ .next()
+ .expect("fixture has a base file")
+ }
+
+ /// The happy path through the same entry points the C++ caller uses, so
+ /// wrapping them in the guard is shown not to change a successful read.
+ #[test]
+ fn test_entry_points_through_bridge_read_base_file() {
+ let table_path = SampleTable::V8Nonpartitioned.path_to_cow();
+ let base_file_name = first_base_file(&table_path);
+
+ let_cxx_string!(base_uri = table_path.as_str());
+ let options = CxxVector::<CxxString>::new();
+ let reader = new_file_group_reader_with_options(&base_uri,
&options).unwrap();
+
+ let_cxx_string!(base_file_path = base_file_name.as_str());
+ let log_file_paths = CxxVector::<CxxString>::new();
+ let raw = reader
+ .read_file_slice_from_paths(&base_file_path, &log_file_paths)
+ .unwrap();
+ assert!(rows_in(raw) > 0);
+
+ let_cxx_string!(partition_path = "");
+ let file_slice =
+ new_file_slice_from_file_names(&partition_path, &base_file_path,
&log_file_paths)
+ .unwrap();
+ let raw = reader.read_file_slice(&file_slice).unwrap();
+ assert!(rows_in(raw) > 0);
+ }
+
+ /// A host keeps one reader across many queries, so a panic in one read
+ /// must leave the runtime and reader usable for the next.
+ #[test]
+ fn test_reader_usable_after_panic_inside_block_on() {
+ let table_path = SampleTable::V8Nonpartitioned.path_to_cow();
+ let_cxx_string!(base_uri = table_path.as_str());
+ let options = CxxVector::<CxxString>::new();
+ let reader = new_file_group_reader_with_options(&base_uri,
&options).unwrap();
+
+ let err = guard("read", || -> Result<(), String> {
+ reader.rt.block_on(async { panic!("mid-read") })
+ })
+ .unwrap_err();
+ assert_eq!(err, "read panicked: mid-read");
+
+ let_cxx_string!(base_file_path = first_base_file(&table_path));
+ let log_file_paths = CxxVector::<CxxString>::new();
+ let raw = reader
+ .read_file_slice_from_paths(&base_file_path, &log_file_paths)
+ .unwrap();
+ assert!(rows_in(raw) > 0);
+ }
+
+ #[test]
+ fn test_read_file_slice_from_paths_missing_base_file_is_error() {
+ let table_path = SampleTable::V8Nonpartitioned.path_to_cow();
+ let_cxx_string!(base_uri = table_path.as_str());
+ let options = CxxVector::<CxxString>::new();
+ let reader = new_file_group_reader_with_options(&base_uri,
&options).unwrap();
+
+ let_cxx_string!(base_file_path = "missing-file.parquet");
+ let log_file_paths = CxxVector::<CxxString>::new();
+ let err = reader
+ .read_file_slice_from_paths(&base_file_path, &log_file_paths)
+ .unwrap_err();
+ assert!(
+ err.starts_with("Failed to read file batch"),
+ "unexpected error: {err}"
+ );
+ }
}
