feat(io): let FileIO wrap a caller-provided opendal operator

crates/paimon/src/io/file_io.rs

@@ -372,6 +372,7 @@ pub struct FileIOBuilder {
     scheme_str: Option<String>,
     props: HashMap<String, String>,
     cache: Option<Arc<LocalCache>>,
+    operator: Option<Operator>,
 }
 
 impl FileIOBuilder {
@@ -380,11 +381,22 @@ impl FileIOBuilder {
             scheme_str: Some(scheme_str.to_string()),
             props: HashMap::default(),
             cache: None,
+            operator: None,
         }
     }
 
-    pub(crate) fn into_parts(self) -> (String, HashMap<String, String>) {
-        (self.scheme_str.unwrap_or_default(), self.props)
+    pub(crate) fn into_parts(self) -> (String, HashMap<String, String>, Option<Operator>) {
+        (self.scheme_str.unwrap_or_default(), self.props, self.operator)
+    }
+
+    /// Uses a caller-provided opendal operator instead of building one from the scheme:
+    /// embedders bring their own storage backend (for example a customized local-filesystem
+    /// service) without registering a scheme. Paths are handed to the operator in
+    /// scheme-relative form, exactly as the built-in local-filesystem backend receives them,
+    /// so the operator's root decides what they resolve against.
+    pub fn with_operator(mut self, operator: Operator) -> Self {
+        self.operator = Some(operator);
+        self
     }
 
     pub fn with_prop(mut self, key: impl ToString, value: impl ToString) -> Self {

crates/paimon/src/io/storage.rs

@@ -68,6 +68,8 @@ use super::FileIOBuilder;
 /// The storage carries all supported storage services in paimon
 #[derive(Debug)]
 pub enum Storage {
+    /// A caller-provided opendal operator (see `FileIOBuilder::with_operator`).
+    Custom { op: Operator },
     #[cfg(feature = "storage-memory")]
     Memory { op: Operator },
     #[cfg(feature = "storage-fs")]
@@ -111,7 +113,10 @@ pub enum Storage {
 
 impl Storage {
     pub(crate) fn build(file_io_builder: FileIOBuilder) -> crate::Result<Self> {
-        let (scheme_str, props) = file_io_builder.into_parts();
+        let (scheme_str, props, operator) = file_io_builder.into_parts();
+        if let Some(op) = operator {
+            return Ok(Self::Custom { op });
+        }
         let scheme = scheme_str.to_ascii_lowercase();
         match scheme.as_str() {
             #[cfg(feature = "storage-memory")]
@@ -186,6 +191,7 @@ impl Storage {
 
     pub(crate) fn create<'a>(&self, path: &'a str) -> crate::Result<(Operator, Cow<'a, str>)> {
         match self {
+            Storage::Custom { op } => Ok((op.clone(), Self::fs_relative_path(path)?)),
             #[cfg(feature = "storage-memory")]
             Storage::Memory { op } => {
                 Ok((op.clone(), Cow::Borrowed(Self::memory_relative_path(path)?)))
@@ -289,7 +295,6 @@ impl Storage {
     /// does `PathBuf::from("/").join("C:/dir")`, and because the argument
     /// carries a drive prefix `Path::join` replaces the base, yielding the real
     /// `C:\dir` on Windows.
-    #[cfg(feature = "storage-fs")]
     fn fs_relative_path(path: &str) -> crate::Result<Cow<'_, str>> {
         // A `file://` / `file:/` URL is already in scheme-relative form.
         if let Some(stripped) = path.strip_prefix("file:/") {