This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new d57e85a9 [jindo] Fix unsafe fallback after process fork (#999)
d57e85a9 is described below
commit d57e85a9dff07fa62b0740792b7dc39bc811ab7a
Author: XiaoHongbo <[email protected]>
AuthorDate: Fri Oct 2 21:40:29 2026 +0800
[jindo] Fix unsafe fallback after process fork (#999)
---
Cargo.lock | 1 +
bindings/python/src/error.rs | 115 ++++++++++++++++++++++-
bindings/python/src/lib.rs | 1 +
bindings/python/tests/test_runtime_fork.py | 5 +
crates/paimon/Cargo.toml | 1 +
crates/paimon/src/arrow/format/blob.rs | 39 ++++++++
crates/paimon/src/arrow/format/orc.rs | 122 +++++++++++++++++++++---
crates/paimon/src/arrow/format/parquet.rs | 73 ++++++++++++---
crates/paimon/src/error.rs | 136 ++++++++++++++++++++++++++-
crates/paimon/src/io/file_io.rs | 144 +++++++++++++++++++++--------
crates/paimon/src/io/storage_jindo.rs | 130 +++++++++++++++++++++++++-
docs/src/getting-started.md | 4 +-
12 files changed, 695 insertions(+), 76 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 5a29dcdc..d9d01852 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4677,6 +4677,7 @@ dependencies = [
"http",
"indexmap 2.14.0",
"jiff",
+ "libc",
"libloading 0.9.0",
"log",
"lru 0.18.2",
diff --git a/bindings/python/src/error.rs b/bindings/python/src/error.rs
index 7255bf31..b6ca355a 100644
--- a/bindings/python/src/error.rs
+++ b/bindings/python/src/error.rs
@@ -15,10 +15,24 @@
// specific language governing permissions and limitations
// under the License.
-use pyo3::exceptions::{PyNotImplementedError, PyValueError};
-use pyo3::PyErr;
+use pyo3::exceptions::{PyNotImplementedError, PyRuntimeError, PyValueError};
+use pyo3::prelude::*;
+
+pyo3::create_exception!(
+ pypaimon_rust,
+ ForkSafetyError,
+ PyRuntimeError,
+ "Raised when native state inherited across process fork cannot be used
safely."
+);
+
+pub fn register_module(py: Python<'_>, module: &Bound<'_, PyModule>) ->
PyResult<()> {
+ module.add("ForkSafetyError", py.get_type::<ForkSafetyError>())
+}
pub fn to_py_err(err: paimon::Error) -> PyErr {
+ if err.is_process_fork_unsupported() {
+ return ForkSafetyError::new_err(err.to_string());
+ }
match err {
// Unimplemented scan semantics: distinct from malformed input so upper
// layers can catch NotImplementedError and decide on a fallback.
@@ -28,5 +42,102 @@ pub fn to_py_err(err: paimon::Error) -> PyErr {
}
pub fn df_to_py_err(err: datafusion::error::DataFusionError) -> PyErr {
+ let mut causes: Vec<&(dyn std::error::Error + 'static)> = vec![&err];
+ while let Some(cause) = causes.pop() {
+ if cause
+ .downcast_ref::<paimon::Error>()
+ .is_some_and(paimon::Error::is_process_fork_unsupported)
+ {
+ return ForkSafetyError::new_err(err.to_string());
+ }
+ if let Some(datafusion::error::DataFusionError::Collection(errors)) =
+ cause.downcast_ref::<datafusion::error::DataFusionError>()
+ {
+ causes.extend(
+ errors
+ .iter()
+ .map(|error| error as &(dyn std::error::Error + 'static)),
+ );
+ } else if let Some(source) = cause.source() {
+ causes.push(source);
+ }
+ }
PyValueError::new_err(err.to_string())
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn process_fork_error_has_distinct_python_type() {
+ Python::attach(|py| {
+ let error = to_py_err(paimon::Error::ProcessForkUnsupported {
+ message: "fork is not supported".to_string(),
+ });
+ assert!(error.is_instance_of::<ForkSafetyError>(py));
+ });
+ }
+
+ #[test]
+ fn wrapped_process_fork_error_has_distinct_python_type() {
+ Python::attach(|py| {
+ let error = to_py_err(paimon::Error::UnexpectedError {
+ message: "read failed".to_string(),
+ source: Some(Box::new(paimon::Error::ProcessForkUnsupported {
+ message: "fork is not supported".to_string(),
+ })),
+ });
+ assert!(error.is_instance_of::<ForkSafetyError>(py));
+ });
+ }
+
+ #[test]
+ fn datafusion_fork_error_has_distinct_python_type() {
+ Python::attach(|py| {
+ let error = datafusion::error::DataFusionError::External(Box::new(
+ paimon::Error::ProcessForkUnsupported {
+ message: "fork is not supported".to_string(),
+ },
+ ));
+ assert!(df_to_py_err(error).is_instance_of::<ForkSafetyError>(py));
+ });
+ }
+
+ #[test]
+ fn datafusion_context_preserves_fork_error_type() {
+ Python::attach(|py| {
+ let error = datafusion::error::DataFusionError::External(Box::new(
+ paimon::Error::ProcessForkUnsupported {
+ message: "fork is not supported".to_string(),
+ },
+ ))
+ .context("SQL collection failed");
+ assert!(df_to_py_err(error).is_instance_of::<ForkSafetyError>(py));
+ });
+ }
+
+ #[test]
+ fn datafusion_collection_preserves_fork_error_type() {
+ Python::attach(|py| {
+ let error = datafusion::error::DataFusionError::Collection(vec![
+ datafusion::error::DataFusionError::Plan("invalid
SQL".to_string()),
+ datafusion::error::DataFusionError::External(Box::new(
+ paimon::Error::ProcessForkUnsupported {
+ message: "fork is not supported".to_string(),
+ },
+ )),
+ ])
+ .context("SQL planning failed");
+ assert!(df_to_py_err(error).is_instance_of::<ForkSafetyError>(py));
+ });
+ }
+
+ #[test]
+ fn unrelated_datafusion_error_remains_value_error() {
+ Python::attach(|py| {
+ let error = datafusion::error::DataFusionError::Plan("invalid
SQL".to_string());
+ assert!(df_to_py_err(error).is_instance_of::<PyValueError>(py));
+ });
+ }
+}
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
index 310cf78a..6756fc7a 100644
--- a/bindings/python/src/lib.rs
+++ b/bindings/python/src/lib.rs
@@ -35,6 +35,7 @@ mod tag;
#[pymodule]
fn pypaimon_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
+ error::register_module(py, m)?;
context::register_module(py, m)?;
variant::register_module(py, m)?;
Ok(())
diff --git a/bindings/python/tests/test_runtime_fork.py
b/bindings/python/tests/test_runtime_fork.py
index 740e42fe..6b28c79e 100644
--- a/bindings/python/tests/test_runtime_fork.py
+++ b/bindings/python/tests/test_runtime_fork.py
@@ -19,9 +19,14 @@ import multiprocessing
import pytest
+from pypaimon_rust import ForkSafetyError
from pypaimon_rust.datafusion import PaimonCatalog, SQLContext
+def test_fork_safety_error_is_runtime_error():
+ assert issubclass(ForkSafetyError, RuntimeError)
+
+
def _plan_rows(catalog):
assert "fork_db" in catalog.list_databases()
builder = catalog.get_table("fork_db.t").new_read_builder()
diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml
index 865fd0d5..9ca18244 100644
--- a/crates/paimon/Cargo.toml
+++ b/crates/paimon/Cargo.toml
@@ -159,3 +159,4 @@ cc = "1"
[dev-dependencies]
axum = { version = "0.7", features = ["macros", "tokio", "http1", "http2"] }
+libc = "0.2"
diff --git a/crates/paimon/src/arrow/format/blob.rs
b/crates/paimon/src/arrow/format/blob.rs
index 6b992a66..fff61b82 100644
--- a/crates/paimon/src/arrow/format/blob.rs
+++ b/crates/paimon/src/arrow/format/blob.rs
@@ -2348,6 +2348,45 @@ mod tests {
include!(concat!(env!("CARGO_MANIFEST_DIR"), "/blob_test_utils.rs"));
}
+ struct ForkFailingBlobRead {
+ tail: Option<Bytes>,
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for ForkFailingBlobRead {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ if range == (4..4100) {
+ if let Some(tail) = &self.tail {
+ return Ok(tail.clone());
+ }
+ }
+ Err(Error::ProcessForkUnsupported {
+ message: crate::error::JINDO_FORK_ERROR.to_string(),
+ })
+ }
+ }
+
+ #[tokio::test]
+ async fn blob_index_reads_preserve_fork_safety_error() {
+ let error = BlobFileIndex::load(&ForkFailingBlobRead { tail: None },
4100)
+ .await
+ .unwrap_err();
+ assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
+
+ let mut tail = vec![0; 4096];
+ tail[4091..4095].copy_from_slice(&4092_i32.to_le_bytes());
+ tail[4095] = BLOB_FORMAT_VERSION;
+ let error = BlobFileIndex::load(
+ &ForkFailingBlobRead {
+ tail: Some(Bytes::from(tail)),
+ },
+ 4100,
+ )
+ .await
+ .unwrap_err();
+ assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
+ }
+
#[tokio::test]
async fn test_index_tail_prefetch_boundaries() {
for rows in [0, 1, 4090, 4091, 4092, 9000] {
diff --git a/crates/paimon/src/arrow/format/orc.rs
b/crates/paimon/src/arrow/format/orc.rs
index f8f96b2a..85c77780 100644
--- a/crates/paimon/src/arrow/format/orc.rs
+++ b/crates/paimon/src/arrow/format/orc.rs
@@ -30,6 +30,7 @@ use orc_rust::projection::ProjectionMask;
use orc_rust::reader::AsyncChunkReader;
use orc_rust::ArrowReaderBuilder;
use std::collections::HashMap;
+use std::error::Error as StdError;
use std::io::Write;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
@@ -39,6 +40,37 @@ const ORC_IN_PREDICATE_MAX_LITERALS: usize = 20;
pub(crate) struct OrcFormatReader;
+fn orc_error<E>(error: E, context: &str) -> Error
+where
+ E: StdError + Send + Sync + 'static,
+{
+ if let Some(error) = find_fork_error(&error) {
+ return error;
+ }
+ Error::UnexpectedError {
+ message: format!("{context}: {error}"),
+ source: Some(Box::new(error)),
+ }
+}
+
+fn find_fork_error(error: &(dyn StdError + 'static)) -> Option<Error> {
+ if let Some(Error::ProcessForkUnsupported { message }) =
error.downcast_ref::<Error>() {
+ return Some(Error::ProcessForkUnsupported {
+ message: message.clone(),
+ });
+ }
+ // `std::io::Error::source` does not expose the custom error stored by
+ // `Error::other`, so inspect it explicitly before following normal
sources.
+ if let Some(io_error) = error.downcast_ref::<std::io::Error>() {
+ if let Some(source) = io_error.get_ref() {
+ if let Some(error) = find_fork_error(source) {
+ return Some(error);
+ }
+ }
+ }
+ error.source().and_then(find_fork_error)
+}
+
const ORC_STRIPE_SIZE: usize = 64 * 1024 * 1024;
const ORC_WORKER_COUNT: usize = 4;
static NEXT_ORC_WRITER_ID: AtomicU64 = AtomicU64::new(1);
@@ -336,10 +368,7 @@ impl FormatFileReader for OrcFormatReader {
let builder = ArrowReaderBuilder::try_new_async(orc_reader)
.await
- .map_err(|e| Error::UnexpectedError {
- message: format!("Failed to open ORC file: {e}"),
- source: Some(Box::new(e)),
- })?;
+ .map_err(|error| orc_error(error, "Failed to open ORC file"))?;
// Widen the scan to include predicate columns so the residual filter
can
// see every column it references, even when a predicate column is not
part
@@ -394,10 +423,7 @@ impl FormatFileReader for OrcFormatReader {
});
Ok(stream
.map(move |r| {
- let batch = r.map_err(|e| Error::UnexpectedError {
- message: format!("ORC read error: {e}"),
- source: Some(Box::new(e)),
- })?;
+ let batch = r.map_err(|error| orc_error(error, "ORC read
error"))?;
match &residual {
Some((fp, scan_fields)) => {
crate::arrow::residual::filter_record_batch_by_predicates(
@@ -664,10 +690,7 @@ pub(crate) async fn read_row_count(
) -> crate::Result<i64> {
let builder =
ArrowReaderBuilder::try_new_async(OrcFileReader::new(file_size, reader))
.await
- .map_err(|error| Error::UnexpectedError {
- message: format!("Failed to open ORC file: {error}"),
- source: Some(Box::new(error)),
- })?;
+ .map_err(|error| orc_error(error, "Failed to open ORC file"))?;
let rows = builder
.file_metadata()
.stripe_metadatas()
@@ -709,7 +732,7 @@ impl AsyncChunkReader for OrcFileReader {
self.r
.read(offset_from_start..offset_from_start + length)
.await
- .map_err(|e| std::io::Error::other(e.to_string()))
+ .map_err(std::io::Error::other)
})
}
}
@@ -1039,6 +1062,79 @@ mod tests {
buf
}
+ #[derive(Clone)]
+ struct SwitchableForkFailingFileRead {
+ data: Bytes,
+ fail: Arc<std::sync::atomic::AtomicBool>,
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for SwitchableForkFailingFileRead {
+ async fn read(&self, range: std::ops::Range<u64>) ->
crate::Result<Bytes> {
+ if self.fail.load(std::sync::atomic::Ordering::SeqCst) {
+ return Err(Error::ProcessForkUnsupported {
+ message: crate::error::JINDO_FORK_ERROR.to_string(),
+ });
+ }
+ Ok(self.data.slice(range.start as usize..range.end as usize))
+ }
+ }
+
+ #[tokio::test]
+ async fn orc_read_paths_preserve_fork_safety_error() {
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "id",
+ ArrowDataType::Int32,
+ true,
+ )]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
+ )
+ .unwrap();
+ let data = Bytes::from(write_single_stripe_orc(schema, &batch));
+ let file_size = data.len() as u64;
+ let fail = Arc::new(std::sync::atomic::AtomicBool::new(true));
+ let file_read = SwitchableForkFailingFileRead {
+ data,
+ fail: Arc::clone(&fail),
+ };
+ let fields = vec![field(0, "id", DataType::Int(IntType::new()))];
+
+ let error = match OrcFormatReader
+ .read_batch_stream(
+ Box::new(file_read.clone()),
+ file_size,
+ &fields,
+ None,
+ None,
+ None,
+ )
+ .await
+ {
+ Ok(_) => panic!("ORC metadata read should propagate the storage
error"),
+ Err(error) => error,
+ };
+ assert!(
+ matches!(error, Error::ProcessForkUnsupported { .. }),
+ "{error:?}"
+ );
+
+ let error = super::read_row_count(Box::new(file_read.clone()),
file_size)
+ .await
+ .unwrap_err();
+ assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
+
+ fail.store(false, std::sync::atomic::Ordering::SeqCst);
+ let mut stream = OrcFormatReader
+ .read_batch_stream(Box::new(file_read), file_size, &fields, None,
None, None)
+ .await
+ .unwrap();
+ fail.store(true, std::sync::atomic::Ordering::SeqCst);
+ let error = stream.next().await.unwrap().unwrap_err();
+ assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
+ }
+
#[tokio::test]
async fn orc_writer_flushes_stripes_and_accepts_more_rows() {
use crate::io::FileIOBuilder;
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index 5ce567bf..0214582a 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -2567,13 +2567,9 @@ pub(crate) async fn read_row_count(
) -> crate::Result<i64> {
let mut reader =
ArrowFileReader::new(file_size,
Arc::from(reader)).with_metadata_cache_enabled(false);
- let metadata = reader
- .get_metadata(None)
- .await
- .map_err(|error| Error::UnexpectedError {
- message: format!("Failed to read the Parquet footer: {error}"),
- source: Some(Box::new(error)),
- })?;
+ let metadata = reader.get_metadata(None).await.map_err(|error| {
+ Error::from_parquet_with_context(error, "Failed to read the Parquet
footer")
+ })?;
Ok(metadata.file_metadata().num_rows())
}
@@ -2592,10 +2588,11 @@ impl ArrowFileReader {
}
fn read_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_,
parquet::errors::Result<Bytes>> {
- Box::pin(self.r.read(range.start..range.end).map_err(|err| {
- let err_msg = format!("{err}");
- parquet::errors::ParquetError::External(err_msg.into())
- }))
+ Box::pin(
+ self.r
+ .read(range.start..range.end)
+ .map_err(|error|
parquet::errors::ParquetError::External(Box::new(error))),
+ )
}
async fn load_metadata(
@@ -2655,15 +2652,15 @@ impl AsyncFileReader for ArrowFileReader {
// All ranges fit within the concurrency limit — fire them all
at once.
futures::future::try_join_all(fetch_ranges.iter().map(|range| {
r.read(range.clone())
- .map_err(|e|
parquet::errors::ParquetError::External(format!("{e}").into()))
+ .map_err(|error|
parquet::errors::ParquetError::External(Box::new(error)))
}))
.await?
} else {
// More ranges than concurrency slots — use buffered stream.
futures::stream::iter(fetch_ranges.iter().cloned())
.map(|range| async move {
- r.read(range).await.map_err(|e| {
-
parquet::errors::ParquetError::External(format!("{e}").into())
+ r.read(range).await.map_err(|error| {
+
parquet::errors::ParquetError::External(Box::new(error))
})
})
.buffered(concurrency)
@@ -3397,6 +3394,54 @@ mod tests {
.unwrap()
}
+ struct ForkFailingFileRead;
+
+ #[async_trait::async_trait]
+ impl crate::io::FileRead for ForkFailingFileRead {
+ async fn read(&self, _range: std::ops::Range<u64>) ->
crate::Result<Bytes> {
+ Err(Error::ProcessForkUnsupported {
+ message: crate::error::JINDO_FORK_ERROR.to_string(),
+ })
+ }
+ }
+
+ #[tokio::test]
+ async fn parquet_read_paths_preserve_fork_safety_error() {
+ let fields = vec![DataField::new(
+ 0,
+ "id".to_string(),
+ DataType::Int(IntType::new()),
+ )];
+ let error = match ParquetFormatReader::default()
+ .read_batch_stream(Box::new(ForkFailingFileRead), 8, &fields,
None, None, None)
+ .await
+ {
+ Ok(_) => panic!("Parquet reader should propagate the storage
error"),
+ Err(error) => error,
+ };
+ assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
+
+ let error = super::read_row_count(Box::new(ForkFailingFileRead), 8)
+ .await
+ .unwrap_err();
+ assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
+
+ for count in [2, super::RANGE_FETCH_CONCURRENCY + 1] {
+ let step = super::RANGE_COALESCE_BYTES + 2;
+ let ranges = (0..count)
+ .map(|index| {
+ let start = index as u64 * step;
+ start..start + 1
+ })
+ .collect::<Vec<_>>();
+ let mut reader =
+ super::ArrowFileReader::new(count as u64 * step + 1,
Arc::new(ForkFailingFileRead));
+ let error = reader.get_byte_ranges(ranges).await.unwrap_err();
+ let error = Error::from(error);
+ assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
+ }
+ }
+
#[derive(Clone)]
struct ConcurrentTrackingFileRead {
data: Bytes,
diff --git a/crates/paimon/src/error.rs b/crates/paimon/src/error.rs
index 44aa4a74..47cd9e6e 100644
--- a/crates/paimon/src/error.rs
+++ b/crates/paimon/src/error.rs
@@ -17,6 +17,9 @@
use snafu::prelude::*;
+pub(crate) const JINDO_FORK_ERROR: &str =
+ "Jindo SDK cannot be reused after process fork; use spawn or avoid
initializing Jindo in the parent process";
+
/// Result type used in paimon.
pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -66,6 +69,8 @@ pub enum Error {
display("Paimon hitting unsupported io error {}", message)
)]
IoUnsupported { message: String },
+ #[snafu(display("{}", message))]
+ ProcessForkUnsupported { message: String },
#[snafu(
visibility(pub(crate)),
display("Paimon hitting invalid config: {}", message)
@@ -136,12 +141,68 @@ pub enum Error {
impl From<opendal::Error> for Error {
fn from(source: opendal::Error) -> Self {
- // TODO: Simple use IoUnexpected for now
+ Error::from_opendal_with_context(source, "IO operation failed on
underlying storage")
+ }
+}
+
+impl Error {
+ pub(crate) fn from_opendal_with_context(
+ source: opendal::Error,
+ message: impl Into<String>,
+ ) -> Self {
+ if source.kind() == opendal::ErrorKind::Unsupported &&
source.message() == JINDO_FORK_ERROR
+ {
+ return Error::ProcessForkUnsupported {
+ message: source.message().to_string(),
+ };
+ }
Error::IoUnexpected {
- message: "IO operation failed on underlying storage".to_string(),
+ message: message.into(),
source: Box::new(source),
}
}
+
+ /// Whether this error or one of its wrapped causes rejects inherited
native state.
+ #[doc(hidden)]
+ pub fn is_process_fork_unsupported(&self) -> bool {
+ match self {
+ Error::ProcessForkUnsupported { .. } => true,
+ Error::DataInvalid { source, .. } | Error::UnexpectedError {
source, .. } => source
+ .as_deref()
+ .is_some_and(|source|
error_chain_contains_process_fork(source)),
+ Error::IoUnexpected { source, .. } => is_jindo_fork_error(source),
+ Error::DataUnexpected { source, .. } =>
error_chain_contains_process_fork(source),
+ Error::ParquetDataUnexpected { source, .. } => {
+ error_chain_contains_process_fork(source)
+ }
+ Error::RestApi { source } =>
error_chain_contains_process_fork(source),
+ _ => false,
+ }
+ }
+}
+
+fn is_jindo_fork_error(error: &opendal::Error) -> bool {
+ error.kind() == opendal::ErrorKind::Unsupported && error.message() ==
JINDO_FORK_ERROR
+}
+
+fn error_chain_contains_process_fork(error: &(dyn std::error::Error +
'static)) -> bool {
+ if let Some(error) = error.downcast_ref::<Error>() {
+ return error.is_process_fork_unsupported();
+ }
+ if let Some(error) = error.downcast_ref::<opendal::Error>() {
+ return is_jindo_fork_error(error);
+ }
+ if let Some(error) = error.downcast_ref::<std::io::Error>() {
+ if error
+ .get_ref()
+ .is_some_and(|source| error_chain_contains_process_fork(source))
+ {
+ return true;
+ }
+ }
+ error
+ .source()
+ .is_some_and(error_chain_contains_process_fork)
}
impl From<apache_avro::Error> for Error {
@@ -153,17 +214,82 @@ impl From<apache_avro::Error> for Error {
}
}
-impl From<parquet::errors::ParquetError> for Error {
- fn from(source: parquet::errors::ParquetError) -> Self {
+impl Error {
+ pub(crate) fn from_parquet_with_context(
+ source: parquet::errors::ParquetError,
+ context: &str,
+ ) -> Self {
+ let source = match source {
+ parquet::errors::ParquetError::External(source) => match
source.downcast::<Error>() {
+ Ok(source) => return *source,
+ Err(source) => parquet::errors::ParquetError::External(source),
+ },
+ source => source,
+ };
Error::ParquetDataUnexpected {
- message: format!("Failed to read a Parquet file: {source}"),
+ message: format!("{context}: {source}"),
source: Box::new(source),
}
}
}
+impl From<parquet::errors::ParquetError> for Error {
+ fn from(source: parquet::errors::ParquetError) -> Self {
+ Error::from_parquet_with_context(source, "Failed to read a Parquet
file")
+ }
+}
+
impl From<crate::api::rest_error::RestError> for Error {
fn from(source: crate::api::rest_error::RestError) -> Self {
Error::RestApi { source }
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn preserves_jindo_fork_error_kind() {
+ let error: Error =
+ opendal::Error::new(opendal::ErrorKind::Unsupported,
JINDO_FORK_ERROR).into();
+ assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
+ }
+
+ #[test]
+ fn preserves_paimon_error_through_parquet_external_error() {
+ let parquet_error =
+
parquet::errors::ParquetError::External(Box::new(Error::ProcessForkUnsupported {
+ message: JINDO_FORK_ERROR.to_string(),
+ }));
+
+ let error: Error = parquet_error.into();
+ assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
+ }
+
+ #[test]
+ fn keeps_non_paimon_parquet_external_error_wrapped() {
+ let parquet_error = parquet::errors::ParquetError::External(Box::new(
+ std::io::Error::other("external parquet failure"),
+ ));
+
+ let error: Error = parquet_error.into();
+ assert!(matches!(error, Error::ParquetDataUnexpected { .. }));
+ }
+
+ #[test]
+ fn detects_wrapped_process_fork_error() {
+ let error = Error::UnexpectedError {
+ message: "outer context".to_string(),
+ source: Some(Box::new(Error::IoUnexpected {
+ message: "storage context".to_string(),
+ source: Box::new(opendal::Error::new(
+ opendal::ErrorKind::Unsupported,
+ JINDO_FORK_ERROR,
+ )),
+ })),
+ };
+
+ assert!(error.is_process_fork_unsupported());
+ }
+}
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index 7dba9fe8..67038518 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -33,7 +33,6 @@ use futures::stream::BoxStream;
use futures::{StreamExt, TryStreamExt};
use opendal::raw::{normalize_path, normalize_root};
use opendal::Operator;
-use snafu::ResultExt;
use tokio_util::compat::FuturesAsyncWriteCompatExt;
use url::Url;
@@ -322,12 +321,12 @@ impl FileIO {
/// Reference:
<https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L97>
pub async fn get_status(&self, path: &str) -> Result<FileStatus> {
let (op, relative_path) = self.create(path).await?;
- let meta = op
- .stat(relative_path.as_ref())
- .await
- .context(IoUnexpectedSnafu {
- message: format!("Failed to get file status for '{path}'"),
- })?;
+ let meta = op.stat(relative_path.as_ref()).await.map_err(|error| {
+ Error::from_opendal_with_context(
+ error,
+ format!("Failed to get file status for '{path}'"),
+ )
+ })?;
Ok(FileStatus {
size: meta.content_length(),
@@ -351,8 +350,8 @@ impl FileIO {
// use normalize_root to make sure it end with `/`.
let list_path = normalize_root(relative_path.as_ref());
- let entries = op.list_with(&list_path).await.context(IoUnexpectedSnafu
{
- message: format!("Failed to list files in '{path}'"),
+ let entries = op.list_with(&list_path).await.map_err(|error| {
+ Error::from_opendal_with_context(error, format!("Failed to list
files in '{path}'"))
})?;
let mut statuses = Vec::new();
@@ -409,21 +408,27 @@ impl FileIO {
let has_provider = matches!(self.backend, FileIOBackend::Provider(_));
let list_path = normalize_root(relative_path.as_ref());
- let entries =
- op.lister_with(&list_path)
- .recursive(true)
- .await
- .context(IoUnexpectedSnafu {
- message: format!("Failed to list files recursively in
'{path}'"),
- })?;
+ let entries = op
+ .lister_with(&list_path)
+ .recursive(true)
+ .await
+ .map_err(|error| {
+ Error::from_opendal_with_context(
+ error,
+ format!("Failed to list files recursively in '{path}'"),
+ )
+ })?;
let path = path.to_string();
let list_path_normalized =
list_path.trim_start_matches('/').to_string();
Ok(Box::pin(async_stream::try_stream! {
let mut entries = entries;
let mut emitted = 0usize;
- while let Some(entry) =
entries.try_next().await.context(IoUnexpectedSnafu {
- message: format!("Failed to list files recursively in
'{path}'"),
+ while let Some(entry) = entries.try_next().await.map_err(|error| {
+ Error::from_opendal_with_context(
+ error,
+ format!("Failed to list files recursively in '{path}'"),
+ )
})? {
let entry_path = entry.path();
if has_provider {
@@ -458,11 +463,12 @@ impl FileIO {
pub async fn exists(&self, path: &str) -> Result<bool> {
let (op, relative_path) = self.create(path).await?;
- op.exists(relative_path.as_ref())
- .await
- .context(IoUnexpectedSnafu {
- message: format!("Failed to check existence of '{path}'"),
- })
+ op.exists(relative_path.as_ref()).await.map_err(|error| {
+ Error::from_opendal_with_context(
+ error,
+ format!("Failed to check existence of '{path}'"),
+ )
+ })
}
/// Check if a directory exists.
@@ -470,8 +476,11 @@ impl FileIO {
let (op, relative_path) = self.create(path).await?;
let dir_path = normalize_root(relative_path.as_ref());
- op.exists(&dir_path).await.context(IoUnexpectedSnafu {
- message: format!("Failed to check existence of directory
'{path}'"),
+ op.exists(&dir_path).await.map_err(|error| {
+ Error::from_opendal_with_context(
+ error,
+ format!("Failed to check existence of directory '{path}'"),
+ )
})
}
@@ -482,11 +491,9 @@ impl FileIO {
let (op, relative_path) = self.create(path).await?;
let cache_path = cache_object_path(&op, relative_path.as_ref());
- op.delete(relative_path.as_ref())
- .await
- .context(IoUnexpectedSnafu {
- message: format!("Failed to delete file '{path}'"),
- })?;
+ op.delete(relative_path.as_ref()).await.map_err(|error| {
+ Error::from_opendal_with_context(error, format!("Failed to delete
file '{path}'"))
+ })?;
if let Some(cache) = self.cache.as_ref().filter(|cache|
cache.is_cacheable(path)) {
cache.invalidate_path(&cache_path).await;
}
@@ -504,8 +511,11 @@ impl FileIO {
op.delete_with(relative_path.as_ref())
.recursive(true)
.await
- .context(IoUnexpectedSnafu {
- message: format!("Failed to delete directory '{path}'"),
+ .map_err(|error| {
+ Error::from_opendal_with_context(
+ error,
+ format!("Failed to delete directory '{path}'"),
+ )
})?;
if let Some(cache) = &self.cache {
cache.invalidate_prefix(&cache_path).await;
@@ -523,8 +533,8 @@ impl FileIO {
let (op, relative_path) = self.create(path).await?;
// Opendal create_dir expects the path to end with `/` to indicate a
directory.
let dir_path = normalize_root(relative_path.as_ref());
- op.create_dir(&dir_path).await.context(IoUnexpectedSnafu {
- message: format!("Failed to create directory '{path}'"),
+ op.create_dir(&dir_path).await.map_err(|error| {
+ Error::from_opendal_with_context(error, format!("Failed to create
directory '{path}'"))
})?;
Ok(())
@@ -592,8 +602,11 @@ impl FileIO {
op_src
.rename(relative_path_src.as_ref(), relative_path_dst.as_ref())
.await
- .context(IoUnexpectedSnafu {
- message: format!("Failed to rename '{src}' to '{dst}'"),
+ .map_err(|error| {
+ Error::from_opendal_with_context(
+ error,
+ format!("Failed to rename '{src}' to '{dst}'"),
+ )
})?;
if let Some(cache) = &self.cache {
cache.invalidate_prefix(&cache_path_src).await;
@@ -1188,6 +1201,7 @@ mod file_action_test {
async fn create(&self, _path: &str) -> crate::Result<(Operator,
String)> {
let service: Servicer = Arc::new(CountingListService {
pulls: Arc::clone(&self.pulls),
+ fail_with_fork_error: false,
});
Ok((
Operator::from_parts(OperationContext::default(), service),
@@ -1199,6 +1213,7 @@ mod file_action_test {
#[derive(Debug)]
struct CountingListService {
pulls: Arc<AtomicUsize>,
+ fail_with_fork_error: bool,
}
impl Service for CountingListService {
@@ -1214,6 +1229,7 @@ mod file_action_test {
fn capability(&self) -> Capability {
Capability {
+ stat: true,
list: true,
list_with_recursive: true,
..Default::default()
@@ -1235,6 +1251,9 @@ mod file_action_test {
_path: &str,
_args: OpStat,
) -> opendal::Result<RpStat> {
+ if self.fail_with_fork_error {
+ return Err(fork_test_error());
+ }
Err(unsupported_test_operation())
}
@@ -1269,6 +1288,7 @@ mod file_action_test {
Ok(CountingLister {
pulls: Arc::clone(&self.pulls),
next: 0,
+ fail_with_fork_error: self.fail_with_fork_error,
})
}
@@ -1310,14 +1330,22 @@ mod file_action_test {
)
}
+ fn fork_test_error() -> opendal::Error {
+ opendal::Error::new(opendal::ErrorKind::Unsupported, JINDO_FORK_ERROR)
+ }
+
struct CountingLister {
pulls: Arc<AtomicUsize>,
next: usize,
+ fail_with_fork_error: bool,
}
impl oio::List for CountingLister {
async fn next(&mut self) -> opendal::Result<Option<oio::Entry>> {
self.pulls.fetch_add(1, AtomicOrdering::SeqCst);
+ if self.fail_with_fork_error {
+ return Err(fork_test_error());
+ }
if self.next == 0 {
self.next += 1;
return Ok(Some(oio::Entry::new(
@@ -1460,6 +1488,50 @@ mod file_action_test {
assert_eq!(pulls.load(AtomicOrdering::SeqCst), 1);
}
+ #[tokio::test]
+ async fn file_status_reads_preserve_fork_safety_error() {
+ let pulls = Arc::new(AtomicUsize::new(0));
+ let service: Servicer = Arc::new(CountingListService {
+ pulls,
+ fail_with_fork_error: true,
+ });
+ let operator = Operator::from_parts(OperationContext::default(),
service);
+
+ #[derive(Debug)]
+ struct FixedProvider(Operator);
+
+ #[async_trait::async_trait]
+ impl FileIOProvider for FixedProvider {
+ async fn create(&self, path: &str) -> crate::Result<(Operator,
String)> {
+ Ok((
+ self.0.clone(),
+ path.strip_prefix("counting:/").unwrap().to_string(),
+ ))
+ }
+ }
+
+ let file_io =
setup_memory_file_io().with_provider(Arc::new(FixedProvider(operator)));
+ let error = file_io
+ .get_status("counting:/objects/file")
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(error, Error::ProcessForkUnsupported { .. }),
+ "{error:?}"
+ );
+
+ let error =
file_io.list_status("counting:/objects/").await.unwrap_err();
+ assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
+
+ let snapshots =
+ crate::table::SnapshotManager::new(file_io,
"counting:/objects/table".to_string());
+ let error = snapshots.get_latest_snapshot_id().await.unwrap_err();
+ assert!(
+ matches!(error, Error::ProcessForkUnsupported { .. }),
+ "{error:?}"
+ );
+ }
+
#[tokio::test]
async fn test_recursive_listing_stream_yields_before_polling_next_entry() {
let pulls = Arc::new(AtomicUsize::new(0));
diff --git a/crates/paimon/src/io/storage_jindo.rs
b/crates/paimon/src/io/storage_jindo.rs
index 699b41b7..bfda2a3d 100644
--- a/crates/paimon/src/io/storage_jindo.rs
+++ b/crates/paimon/src/io/storage_jindo.rs
@@ -19,6 +19,7 @@ use std::ffi::{CStr, CString, OsStr};
use std::fmt::{Debug, Formatter};
use std::os::raw::{c_char, c_void};
use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicU32, Ordering as AtomicOrdering};
use std::sync::mpsc::{self, Sender};
use std::sync::{Arc, Mutex, OnceLock};
@@ -28,7 +29,7 @@ use opendal::{Buffer, Builder, BytesRange, Capability,
EntryMode, ErrorKind, Met
use opendal::{Error as OpendalError, OperationContext, Operator, Result as
OpendalResult};
use tokio::sync::{OnceCell, Semaphore};
-use crate::error::Error;
+use crate::error::{Error, JINDO_FORK_ERROR};
use crate::Result;
const OSS_IMPL: &str = "fs.oss.impl";
@@ -48,7 +49,6 @@ const JDO_REST_HTTP_403_ERROR: i32 = 6403;
const JDO_REST_HTTP_404_ERROR: i32 = 6404;
const JDO_REST_HTTP_503_ERROR: i32 = 6503;
const JINDO_EXCEPTION_BUFFER_SIZE: usize = 1024;
-
type JdoPtr = *mut c_void;
type JindoTeardown = Box<dyn FnOnce() + Send + 'static>;
@@ -424,6 +424,7 @@ impl LazyJindoClient {
}
async fn get(&self) -> OpendalResult<Arc<JindoClient>> {
+ ensure_jindo_process()?;
let client = self
.client
.get_or_try_init(|| {
@@ -471,8 +472,20 @@ fn unsupported<T>(operation: &str) -> OpendalResult<T> {
))
}
+fn ensure_jindo_process() -> OpendalResult<()> {
+ static SDK_PID: AtomicU32 = AtomicU32::new(0);
+
+ let pid = std::process::id();
+ match SDK_PID.compare_exchange(0, pid, AtomicOrdering::AcqRel,
AtomicOrdering::Acquire) {
+ Ok(_) => Ok(()),
+ Err(owner) if owner == pid => Ok(()),
+ Err(_) => Err(OpendalError::new(ErrorKind::Unsupported,
JINDO_FORK_ERROR)),
+ }
+}
+
fn jindo_teardown_sender() -> OpendalResult<&'static Sender<JindoTeardown>> {
- static SENDER: OnceLock<std::result::Result<Sender<JindoTeardown>,
String>> = OnceLock::new();
+ static SENDER: OnceLock<std::result::Result<(u32, Sender<JindoTeardown>),
String>> =
+ OnceLock::new();
match SENDER.get_or_init(|| {
let (sender, receiver) = mpsc::channel::<JindoTeardown>();
@@ -483,10 +496,14 @@ fn jindo_teardown_sender() -> OpendalResult<&'static
Sender<JindoTeardown>> {
teardown();
}
})
- .map(|_| sender)
+ .map(|_| (std::process::id(), sender))
.map_err(|error| error.to_string())
}) {
- Ok(sender) => Ok(sender),
+ Ok((pid, sender)) if *pid == std::process::id() => Ok(sender),
+ Ok(_) => Err(OpendalError::new(
+ ErrorKind::Unsupported,
+ "Jindo teardown worker is not available after process fork",
+ )),
Err(message) => Err(OpendalError::new(
ErrorKind::Unexpected,
"failed to start Jindo teardown worker",
@@ -665,6 +682,7 @@ struct JindoClient {
options: JdoPtr,
store: JdoPtr,
root: String,
+ pid: u32,
initialized: bool,
}
@@ -682,6 +700,7 @@ impl Debug for JindoClient {
impl JindoClient {
fn new(config: &JindoStorageConfig, bucket: &str) -> OpendalResult<Self> {
+ ensure_jindo_process()?;
jindo_teardown_sender()?;
let api = JindoApi::load(config.library_path.as_deref())?;
let root = format!("oss://{bucket}/");
@@ -722,6 +741,7 @@ impl JindoClient {
options,
store,
root,
+ pid: std::process::id(),
initialized: false,
};
client.with_handle(|handle| {
@@ -1001,6 +1021,9 @@ impl JindoClient {
}
fn with_handle<T>(&self, call: impl FnOnce(JdoPtr) -> OpendalResult<T>) ->
OpendalResult<T> {
+ if self.pid != std::process::id() {
+ return Err(OpendalError::new(ErrorKind::Unsupported,
JINDO_FORK_ERROR));
+ }
let handle = unsafe { (self.api.create_handle)(self.store) };
if handle.is_null() {
return Err(OpendalError::new(
@@ -1051,6 +1074,14 @@ impl JindoClient {
impl Drop for JindoClient {
fn drop(&mut self) {
+ if self.pid != std::process::id() {
+ // These handles belong to the parent process. The child must not
+ // call SDK teardown for them.
+ self.options = std::ptr::null_mut();
+ self.store = std::ptr::null_mut();
+ self.initialized = false;
+ return;
+ }
let api = Arc::clone(&self.api);
let options = std::mem::replace(&mut self.options,
std::ptr::null_mut()) as usize;
let store = std::mem::replace(&mut self.store, std::ptr::null_mut())
as usize;
@@ -1158,6 +1189,17 @@ mod tests {
#[cfg(target_os = "linux")]
use axum::Router;
+ #[cfg(target_os = "linux")]
+ fn wait_for_jindo_teardown() -> bool {
+ let (done_tx, done_rx) = std::sync::mpsc::channel();
+ enqueue_jindo_teardown(Box::new(move || {
+ let _ = done_tx.send(());
+ }));
+ done_rx
+ .recv_timeout(std::time::Duration::from_secs(5))
+ .is_ok()
+ }
+
#[test]
fn test_use_jindo() {
assert!(!use_jindo(&HashMap::new()).unwrap());
@@ -1323,6 +1365,81 @@ mod tests {
assert!(returned_before_timeout.load(Ordering::SeqCst));
}
+ #[cfg(target_os = "linux")]
+ async fn assert_operator_rejects_reuse_after_fork(operator: Operator) {
+ operator.stat("objects/data.bin").await.unwrap();
+
+ let child_pid = unsafe { libc::fork() };
+ assert!(child_pid >= 0, "fork failed");
+ if child_pid == 0 {
+ let success = std::thread::spawn(move || {
+ let runtime = match
tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ {
+ Ok(runtime) => runtime,
+ Err(_) => return false,
+ };
+ let rejected = runtime.block_on(async {
+ match operator.stat("objects/data.bin").await {
+ Ok(_) => false,
+ Err(error) => {
+ error.kind() == ErrorKind::Unsupported
+ && error.to_string().contains("after process
fork")
+ }
+ }
+ });
+ drop(operator);
+ rejected
+ })
+ .join()
+ .unwrap_or(false);
+ unsafe { libc::_exit(i32::from(!success)) };
+ }
+
+ let (status, timed_out) = tokio::task::spawn_blocking(move || {
+ let deadline = std::time::Instant::now() +
std::time::Duration::from_secs(10);
+ loop {
+ let mut status = 0;
+ let waited = unsafe { libc::waitpid(child_pid, &mut status,
libc::WNOHANG) };
+ if waited == child_pid {
+ return std::io::Result::Ok((status, false));
+ }
+ if waited < 0 {
+ let error = std::io::Error::last_os_error();
+ if error.kind() != std::io::ErrorKind::Interrupted {
+ return Err(error);
+ }
+ }
+ if std::time::Instant::now() >= deadline {
+ unsafe { libc::kill(child_pid, libc::SIGKILL) };
+ loop {
+ let waited = unsafe { libc::waitpid(child_pid, &mut
status, 0) };
+ if waited == child_pid {
+ return Ok((status, true));
+ }
+ let error = std::io::Error::last_os_error();
+ if error.kind() != std::io::ErrorKind::Interrupted {
+ return Err(error);
+ }
+ }
+ }
+ std::thread::sleep(std::time::Duration::from_millis(10));
+ }
+ })
+ .await
+ .unwrap()
+ .unwrap();
+ assert!(!timed_out, "Jindo child process hung after fork");
+ assert!(libc::WIFEXITED(status));
+ assert_eq!(libc::WEXITSTATUS(status), 0);
+
+ // The parent-side client and teardown worker remain usable.
+ assert!(operator.stat("objects/data.bin").await.unwrap().is_file());
+ drop(operator);
+ assert!(wait_for_jindo_teardown());
+ }
+
#[test]
fn test_parse_config_preserves_endpoint_protocol() {
for (endpoint, configured, expected) in [
@@ -1670,6 +1787,9 @@ mod tests {
assert_eq!(state.list_page_requests.load(Ordering::SeqCst), 3);
assert_eq!(state.range_requests.load(Ordering::SeqCst), 1);
assert_eq!(state.unavailable_requests.load(Ordering::SeqCst), 1);
+
+ let fork_operator = jindo_config_build(&config,
"jindo-test-bucket").unwrap();
+ assert_operator_rejects_reuse_after_fork(fork_operator).await;
server.abort();
}
}
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index 5297c27a..fa5f6434 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -310,7 +310,9 @@ also discovers it from an installed `pyjindosdk` package.
This initial
integration supports object stat, reads, and listings, but not writes, deletes,
or copies. Jindo reads are limited to 8 concurrent requests per OSS operator by
default. Set `fs.jindo.max.concurrent.reads` to a positive integer to tune this
-limit for the available network and JindoSDK connection capacity.
+limit for the available network and JindoSDK connection capacity. JindoSDK
state
+initialized before `fork` cannot be reused in the child; use `spawn` or
initialize
+Jindo only after worker processes start.
Supported metastore types:
