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 66f655b3 feat(io): support URI-aware FileIO providers (#846)
66f655b3 is described below
commit 66f655b30a3861a31b5a0abafd25a8633574049c
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 15 22:49:48 2026 +0800
feat(io): support URI-aware FileIO providers (#846)
---
crates/paimon/src/io/file_io.rs | 299 ++++++++++-----
crates/paimon/src/io/file_io/provider_tests.rs | 489 +++++++++++++++++++++++++
docs/src/getting-started.md | 71 ++++
3 files changed, 762 insertions(+), 97 deletions(-)
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index 1257f632..720225eb 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -28,7 +28,7 @@ use bytes::Bytes;
use chrono::{DateTime, Utc};
use futures::stream::BoxStream;
use futures::{StreamExt, TryStreamExt};
-use opendal::raw::normalize_root;
+use opendal::raw::{normalize_path, normalize_root};
use opendal::Operator;
use snafu::ResultExt;
use tokio_util::compat::FuturesAsyncWriteCompatExt;
@@ -37,6 +37,9 @@ use url::Url;
use super::cache::{CachedFileReader, LocalCache};
use super::Storage;
+#[cfg(all(test, feature = "storage-memory"))]
+mod provider_tests;
+
/// An externally managed block cache used by [`FileIO`].
///
/// Implementations store immutable file ranges. Cache failures must be handled
@@ -56,24 +59,44 @@ pub trait FileBlockCache: std::fmt::Debug + Send + Sync +
'static {
async fn invalidate_prefix(&self, prefix: &str);
}
+/// Resolves original paths to application-managed OpenDAL operators.
+///
+/// Providers own scheme/bucket routing, operator reuse, and credential
refresh.
+/// Errors propagate without falling back to built-in storage or credentials.
+/// Already-open readers and writers retain their operator, so its backend must
+/// refresh credentials internally if those handles need to outlive
credentials.
+///
+/// Custom services must report distinct `(scheme, name, root)` storage
identities
+/// for different namespaces, since these identities are used by the file
cache.
#[async_trait::async_trait]
-pub(crate) trait FileIOProvider: std::fmt::Debug + Send + Sync {
+pub trait FileIOProvider: std::fmt::Debug + Send + Sync + 'static {
+ /// Return an operator and its relative path, preserving literal object
keys.
+ ///
+ /// Empty paths and `/` denote the operator root. For directory listings,
the
+ /// relative path must be an unchanged suffix of the original path starting
+ /// at a component boundary; this permits reconstruction of reusable full
URIs.
+ /// Object paths that OpenDAL would trim or collapse are rejected by
FileIO.
+ /// Rename requires both paths to resolve to the same shared service
instance.
async fn create(&self, path: &str) -> crate::Result<(Operator, String)>;
}
+#[derive(Clone, Debug)]
+enum FileIOBackend {
+ Storage(Arc<Storage>),
+ Provider(Arc<dyn FileIOProvider>),
+}
+
#[derive(Clone)]
pub struct FileIO {
- storage: Arc<Storage>,
+ backend: FileIOBackend,
cache: Option<Arc<LocalCache>>,
- provider: Option<Arc<dyn FileIOProvider>>,
}
impl std::fmt::Debug for FileIO {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FileIO")
- .field("storage", &self.storage)
+ .field("backend", &self.backend)
.field("cache", &self.cache)
- .field("provider", &self.provider)
.finish()
}
}
@@ -97,20 +120,44 @@ impl FileIO {
Ok(self)
}
- pub(crate) fn with_provider(mut self, provider: Arc<dyn FileIOProvider>)
-> Self {
- self.provider = Some(provider);
+ /// Replace the storage backend with a provider, retaining the file cache.
+ ///
+ /// Resolution is deferred to async operations, including for file handles
+ /// subsequently created by [`Self::new_input`] and [`Self::new_output`].
+ pub fn with_provider(mut self, provider: Arc<dyn FileIOProvider>) -> Self {
+ self.backend = FileIOBackend::Provider(provider);
self
}
pub(crate) fn create_static(&self, path: &str) -> crate::Result<(Operator,
String)> {
- let (op, relative_path) = self.storage.create(path)?;
+ let FileIOBackend::Storage(storage) = &self.backend else {
+ return Err(Error::IoUnsupported {
+ message: "A FileIOProvider requires async path
resolution".to_string(),
+ });
+ };
+ let (op, relative_path) = storage.create(path)?;
Ok((op, relative_path.into_owned()))
}
async fn create(&self, path: &str) -> crate::Result<(Operator, String)> {
- match &self.provider {
- Some(provider) => provider.create(path).await,
- None => self.create_static(path),
+ match &self.backend {
+ FileIOBackend::Provider(provider) =>
resolve_provider(provider.as_ref(), path).await,
+ FileIOBackend::Storage(_) => self.create_static(path),
+ }
+ }
+
+ fn file_source(&self, path: &str) -> crate::Result<FileSource> {
+ match &self.backend {
+ FileIOBackend::Provider(provider) =>
Ok(FileSource::Provider(provider.clone())),
+ FileIOBackend::Storage(_) => {
+ let (op, relative_path) = self.create_static(path)?;
+ let cache_path = cache_object_path(&op, &relative_path);
+ Ok(FileSource::Static {
+ op,
+ relative_path,
+ cache_path,
+ })
+ }
}
}
@@ -157,42 +204,34 @@ impl FileIO {
}
/// Create a new input file to read data.
+ /// With a provider, path resolution and its errors are deferred to async
IO.
///
/// Reference:
<https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L76>
pub fn new_input(&self, path: &str) -> crate::Result<InputFile> {
- let (op, relative_path) = self.storage.create(path)?;
- let cache_path = cache_object_path(&op, relative_path.as_ref());
Ok(InputFile {
- op,
+ source: self.file_source(path)?,
path: path.to_string(),
- relative_path: relative_path.into_owned(),
- cache_path,
cache: self
.cache
.as_ref()
.filter(|cache| cache.is_cacheable(path))
.cloned(),
- provider: self.provider.clone(),
})
}
/// Create a new output file to write data.
+ /// With a provider, path resolution and its errors are deferred to async
IO.
///
/// Reference:
<https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java#L87>
pub fn new_output(&self, path: &str) -> Result<OutputFile> {
- let (op, relative_path) = self.storage.create(path)?;
- let cache_path = cache_object_path(&op, relative_path.as_ref());
Ok(OutputFile {
- op,
+ source: self.file_source(path)?,
path: path.to_string(),
- relative_path: relative_path.into_owned(),
- cache_path,
cache: self
.cache
.as_ref()
.filter(|cache| cache.is_cacheable(path))
.cloned(),
- provider: self.provider.clone(),
})
}
@@ -225,12 +264,7 @@ impl FileIO {
/// FIXME: how to handle large dir? Better to return a stream instead?
pub async fn list_status(&self, path: &str) -> Result<Vec<FileStatus>> {
let (op, relative_path) = self.create(path).await?;
- // `relative_path` is a byte-suffix of `path` for object stores and
POSIX
- // local paths, so this recovers the scheme/root prefix. For a Windows
- // local path the relative form only swaps `\`->`/`
(length-preserving),
- // so this is `""` and entries are reported in opendal's normalized
- // `/C:/...` form — which still round-trips back through `create`.
- let base_path = &path[..path.len() - relative_path.len()];
+ let base_path = listing_base_path(path, &relative_path)?;
// Opendal list() expects directory path to end with `/`.
// use normalize_root to make sure it end with `/`.
let list_path = normalize_root(relative_path.as_ref());
@@ -243,6 +277,9 @@ impl FileIO {
let list_path_normalized = list_path.trim_start_matches('/');
for entry in entries {
let entry_path = entry.path();
+ if matches!(self.backend, FileIOBackend::Provider(_)) {
+ validate_provider_path(path, entry_path)?;
+ }
if entry_path.trim_start_matches('/') == list_path_normalized {
continue;
}
@@ -250,7 +287,7 @@ impl FileIO {
statuses.push(FileStatus {
size: meta.content_length(),
is_dir: meta.is_dir(),
- path: status_path(base_path, entry_path),
+ path: status_path(&base_path, entry_path),
last_modified: meta
.last_modified()
.map(|v| DateTime::<Utc>::from(SystemTime::from(v))),
@@ -286,9 +323,8 @@ impl FileIO {
}
let (op, relative_path) = self.create(path).await?;
- // See `list_status`: `relative_path` is a byte-suffix of `path` except
- // for Windows local paths, where it only swaps separators (same
length).
- let base_path = path[..path.len() - relative_path.len()].to_string();
+ let base_path = listing_base_path(path, &relative_path)?;
+ let has_provider = matches!(self.backend, FileIOBackend::Provider(_));
let list_path = normalize_root(relative_path.as_ref());
let entries =
@@ -308,6 +344,9 @@ impl FileIO {
message: format!("Failed to list files recursively in
'{path}'"),
})? {
let entry_path = entry.path();
+ if has_provider {
+ validate_provider_path(&path, entry_path)?;
+ }
if entry_path.trim_start_matches('/') == list_path_normalized {
continue;
}
@@ -426,6 +465,14 @@ impl FileIO {
pub async fn rename(&self, src: &str, dst: &str) -> Result<()> {
let (op_src, relative_path_src) = self.create(src).await?;
let (op_dst, relative_path_dst) = self.create(dst).await?;
+ if matches!(self.backend, FileIOBackend::Provider(_))
+ && !Arc::ptr_eq(op_src.service(), op_dst.service())
+ {
+ return Err(Error::IoUnsupported {
+ message: "Rename through a FileIOProvider requires the same
shared storage service"
+ .to_string(),
+ });
+ }
let cache_path_src = cache_object_path(&op_src,
relative_path_src.as_ref());
let cache_path_dst = cache_object_path(&op_dst,
relative_path_dst.as_ref());
@@ -444,6 +491,57 @@ impl FileIO {
}
}
+async fn resolve_provider(
+ provider: &dyn FileIOProvider,
+ path: &str,
+) -> crate::Result<(Operator, String)> {
+ let (op, relative_path) = provider.create(path).await?;
+ validate_provider_path(path, &relative_path)?;
+ Ok((op, relative_path))
+}
+
+fn validate_provider_path(path: &str, relative_path: &str) -> Result<()> {
+ // Filesystem paths retain their existing separator normalization. Object
+ // keys must not silently resolve to another object through OpenDAL's path
+ // normalization (e.g. `a//b` -> `a/b` or `key ` -> `key`).
+ if path.contains("://")
+ && !path.starts_with("file:/")
+ && !relative_path.is_empty()
+ && normalize_path(relative_path) != relative_path
+ {
+ return Err(Error::ConfigInvalid {
+ message: "FileIOProvider returned an object path that OpenDAL
would normalize"
+ .to_string(),
+ });
+ }
+ Ok(())
+}
+
+fn listing_base_path(path: &str, relative_path: &str) -> Result<String> {
+ if relative_path.is_empty() || relative_path == "/" {
+ return Ok(path.to_string());
+ }
+ if let Some(base) = path.strip_suffix(relative_path) {
+ if base.is_empty() || base.ends_with('/') {
+ return Ok(base.to_string());
+ }
+ }
+ // Windows filesystem paths only change separators. Try the original path
+ // first, since backslashes are literal filename characters on POSIX.
+ if looks_like_windows_drive_path(path) || (cfg!(windows) &&
path.starts_with("file:/")) {
+ let normalized = path.replace('\\', "/");
+ if let Some(base) = normalized.strip_suffix(relative_path) {
+ if base.is_empty() || base.ends_with('/') {
+ return Ok(base.to_string());
+ }
+ }
+ }
+ Err(Error::ConfigInvalid {
+ message: "Cannot list a path whose resolved relative path is not a
component suffix"
+ .to_string(),
+ })
+}
+
fn status_path(base_path: &str, entry_path: &str) -> String {
if base_path.ends_with('/') || entry_path.starts_with('/') {
format!("{base_path}{entry_path}")
@@ -478,6 +576,7 @@ pub struct FileIOBuilder {
props: HashMap<String, String>,
cache: Option<Arc<LocalCache>>,
operator: Option<Operator>,
+ provider: Option<Arc<dyn FileIOProvider>>,
}
impl FileIOBuilder {
@@ -487,6 +586,7 @@ impl FileIOBuilder {
props: HashMap::default(),
cache: None,
operator: None,
+ provider: None,
}
}
@@ -510,6 +610,16 @@ impl FileIOBuilder {
self
}
+ /// Use an application-managed provider instead of built-in storage.
+ ///
+ /// The provider receives original paths, regardless of the builder's
scheme.
+ /// Storage properties are not parsed and no built-in storage feature is
+ /// required. Combining this with [`Self::with_fs_operator`] is an error.
+ pub fn with_provider(mut self, provider: Arc<dyn FileIOProvider>) -> Self {
+ self.provider = Some(provider);
+ self
+ }
+
pub fn with_prop(mut self, key: impl ToString, value: impl ToString) ->
Self {
self.props.insert(key.to_string(), value.to_string());
self
@@ -529,14 +639,19 @@ impl FileIOBuilder {
self
}
- pub fn build(self) -> crate::Result<FileIO> {
+ pub fn build(mut self) -> crate::Result<FileIO> {
let cache = self.cache.clone();
- let storage = Storage::build(self)?;
- Ok(FileIO {
- storage: Arc::new(storage),
- cache,
- provider: None,
- })
+ let backend = if let Some(provider) = self.provider.take() {
+ if self.operator.is_some() {
+ return Err(Error::ConfigInvalid {
+ message: "with_provider and with_fs_operator cannot be
combined".to_string(),
+ });
+ }
+ FileIOBackend::Provider(provider)
+ } else {
+ FileIOBackend::Storage(Arc::new(Storage::build(self)?))
+ };
+ Ok(FileIO { backend, cache })
}
}
@@ -670,45 +785,52 @@ pub struct FileStatus {
pub last_modified: Option<DateTime<Utc>>,
}
-#[derive(Debug)]
-pub struct InputFile {
- op: Operator,
- path: String,
- /// The opendal-relative path (see [`FileIO::new_input`]); not necessarily
a
- /// suffix of `path`, since local paths are separator-normalized.
- relative_path: String,
- cache_path: String,
- cache: Option<Arc<LocalCache>>,
- provider: Option<Arc<dyn FileIOProvider>>,
+#[derive(Clone, Debug)]
+enum FileSource {
+ Static {
+ op: Operator,
+ relative_path: String,
+ cache_path: String,
+ },
+ Provider(Arc<dyn FileIOProvider>),
}
-impl InputFile {
- async fn resolve(&self) -> crate::Result<(Operator, String, String)> {
- match &self.provider {
- Some(provider) => {
- let (op, relative_path) = provider.create(&self.path).await?;
+impl FileSource {
+ async fn resolve(&self, path: &str) -> crate::Result<(Operator, String,
String)> {
+ match self {
+ Self::Provider(provider) => {
+ let (op, relative_path) = resolve_provider(provider.as_ref(),
path).await?;
let cache_path = cache_object_path(&op, &relative_path);
Ok((op, relative_path, cache_path))
}
- None => Ok((
- self.op.clone(),
- self.relative_path.clone(),
- self.cache_path.clone(),
- )),
+ Self::Static {
+ op,
+ relative_path,
+ cache_path,
+ } => Ok((op.clone(), relative_path.clone(), cache_path.clone())),
}
}
+}
+
+#[derive(Debug)]
+pub struct InputFile {
+ source: FileSource,
+ path: String,
+ cache: Option<Arc<LocalCache>>,
+}
+impl InputFile {
pub fn location(&self) -> &str {
&self.path
}
pub async fn exists(&self) -> crate::Result<bool> {
- let (op, relative_path, _) = self.resolve().await?;
+ let (op, relative_path, _) = self.source.resolve(&self.path).await?;
Ok(op.exists(&relative_path).await?)
}
pub async fn metadata(&self) -> crate::Result<FileStatus> {
- let (op, relative_path, _) = self.resolve().await?;
+ let (op, relative_path, _) = self.source.resolve(&self.path).await?;
let meta = op.stat(&relative_path).await?;
Ok(FileStatus {
@@ -722,7 +844,7 @@ impl InputFile {
}
pub async fn read(&self) -> crate::Result<Bytes> {
- let (op, relative_path, cache_path) = self.resolve().await?;
+ let (op, relative_path, cache_path) =
self.source.resolve(&self.path).await?;
let Some(cache) = &self.cache else {
return Ok(op.read(&relative_path).await?.to_bytes());
};
@@ -741,7 +863,7 @@ impl InputFile {
}
pub async fn reader(&self) -> crate::Result<impl FileRead> {
- let (op, relative_path, cache_path) = self.resolve().await?;
+ let (op, relative_path, cache_path) =
self.source.resolve(&self.path).await?;
let reader = op.reader(&relative_path).await?;
let Some(cache) = &self.cache else {
return Ok(InputFileReader::Direct(reader));
@@ -766,50 +888,27 @@ impl InputFile {
#[derive(Debug, Clone)]
pub struct OutputFile {
- op: Operator,
+ source: FileSource,
path: String,
- /// The opendal-relative path (see [`FileIO::new_output`]); not
necessarily a
- /// suffix of `path`, since local paths are separator-normalized.
- relative_path: String,
- cache_path: String,
cache: Option<Arc<LocalCache>>,
- provider: Option<Arc<dyn FileIOProvider>>,
}
impl OutputFile {
- async fn resolve(&self) -> crate::Result<(Operator, String, String)> {
- match &self.provider {
- Some(provider) => {
- let (op, relative_path) = provider.create(&self.path).await?;
- let cache_path = cache_object_path(&op, &relative_path);
- Ok((op, relative_path, cache_path))
- }
- None => Ok((
- self.op.clone(),
- self.relative_path.clone(),
- self.cache_path.clone(),
- )),
- }
- }
-
pub fn location(&self) -> &str {
&self.path
}
pub async fn exists(&self) -> crate::Result<bool> {
- let (op, relative_path, _) = self.resolve().await?;
+ let (op, relative_path, _) = self.source.resolve(&self.path).await?;
Ok(op.exists(&relative_path).await?)
}
pub fn to_input_file(self) -> InputFile {
let cache = self.cache.filter(|cache| cache.is_cacheable(&self.path));
InputFile {
- op: self.op,
+ source: self.source,
path: self.path,
- relative_path: self.relative_path,
- cache_path: self.cache_path,
cache,
- provider: self.provider,
}
}
@@ -820,7 +919,7 @@ impl OutputFile {
}
pub async fn writer(&self) -> crate::Result<Box<dyn FileWrite>> {
- let (op, relative_path, cache_path) = self.resolve().await?;
+ let (op, relative_path, cache_path) =
self.source.resolve(&self.path).await?;
let writer: Box<dyn FileWrite> = Box::new(
op.writer_with(&relative_path)
.chunk(8 * 1024 * 1024)
@@ -838,7 +937,7 @@ impl OutputFile {
/// Get an async streaming writer for format-level writes (e.g. parquet).
pub(crate) async fn async_writer(&self) -> crate::Result<Box<dyn
AsyncFileWrite>> {
- let (op, relative_path, cache_path) = self.resolve().await?;
+ let (op, relative_path, cache_path) =
self.source.resolve(&self.path).await?;
let writer: Box<dyn AsyncFileWrite> = Box::new(
op.writer_with(&relative_path)
.chunk(8 * 1024 * 1024)
@@ -1332,14 +1431,20 @@ mod object_storage_path_test {
fn assert_relative_paths(file_io: &FileIO, path: &str,
expected_relative_path: &str) {
let input = file_io.new_input(path).unwrap();
assert_eq!(input.location(), path);
- assert_eq!(input.relative_path, expected_relative_path);
+ let FileSource::Static { relative_path, .. } = input.source else {
+ panic!("expected static input")
+ };
+ assert_eq!(relative_path, expected_relative_path);
let output = file_io.new_output(path).unwrap();
assert_eq!(output.location(), path);
- assert_eq!(output.relative_path, expected_relative_path);
+ let FileSource::Static { relative_path, .. } = output.source else {
+ panic!("expected static output")
+ };
+ assert_eq!(relative_path, expected_relative_path);
- let (_op, relative_path) = file_io.storage.create(path).unwrap();
- assert_eq!(relative_path.as_ref(), expected_relative_path);
+ let (_op, relative_path) = file_io.create_static(path).unwrap();
+ assert_eq!(relative_path, expected_relative_path);
let base_path = &path[..path.len() - relative_path.len()];
assert_eq!(format!("{base_path}{relative_path}"), path);
diff --git a/crates/paimon/src/io/file_io/provider_tests.rs
b/crates/paimon/src/io/file_io/provider_tests.rs
new file mode 100644
index 00000000..270dfcf9
--- /dev/null
+++ b/crates/paimon/src/io/file_io/provider_tests.rs
@@ -0,0 +1,489 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::*;
+use std::collections::BTreeSet;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+
+#[derive(Debug)]
+struct PrefixProvider {
+ routes: Vec<(String, Operator)>,
+ calls: AtomicUsize,
+ reject: AtomicBool,
+}
+
+#[async_trait::async_trait]
+impl FileIOProvider for PrefixProvider {
+ async fn create(&self, path: &str) -> Result<(Operator, String)> {
+ self.calls.fetch_add(1, Ordering::SeqCst);
+ if self.reject.load(Ordering::SeqCst) {
+ return RejectingProvider.create(path).await;
+ }
+ for (prefix, op) in &self.routes {
+ if path == prefix.trim_end_matches('/') {
+ return Ok((op.clone(), String::new()));
+ }
+ if let Some(relative) = path.strip_prefix(prefix) {
+ return Ok((op.clone(), relative.to_string()));
+ }
+ }
+ RejectingProvider.create(path).await
+ }
+}
+
+fn memory_operator() -> Operator {
+ Operator::from_config(opendal::services::MemoryConfig::default()).unwrap()
+}
+
+fn provider_io(routes: Vec<(String, Operator)>) -> (FileIO,
Arc<PrefixProvider>) {
+ let provider = Arc::new(PrefixProvider {
+ routes,
+ calls: AtomicUsize::new(0),
+ reject: AtomicBool::new(false),
+ });
+ let io = FileIOBuilder::new("application-storage")
+ .with_provider(provider.clone())
+ .build()
+ .unwrap();
+ (io, provider)
+}
+
+#[derive(Debug)]
+struct RejectingProvider;
+
+#[async_trait::async_trait]
+impl FileIOProvider for RejectingProvider {
+ async fn create(&self, _path: &str) -> Result<(Operator, String)> {
+ Err(Error::ConfigInvalid {
+ message: "provider denied access".to_string(),
+ })
+ }
+}
+
+#[tokio::test]
+async fn provider_handles_defer_resolution_without_using_static_storage() {
+ let operator =
Operator::from_config(opendal::services::MemoryConfig::default()).unwrap();
+ let io = FileIOBuilder::new("file")
+ .with_fs_operator(operator)
+ .build()
+ .unwrap()
+ .with_provider(Arc::new(RejectingProvider));
+
+ let input = io.new_input("s3://bucket/key").unwrap();
+ let output = io.new_output("s3://bucket/key").unwrap();
+ assert_denied(input.read().await);
+ assert_denied(output.write(Bytes::from_static(b"data")).await);
+}
+
+fn assert_denied<T>(result: Result<T>) {
+ assert!(matches!(
+ result,
+ Err(Error::ConfigInvalid { message }) if message == "provider denied
access"
+ ));
+}
+
+#[tokio::test]
+async fn provider_routes_buckets_and_preserves_literal_keys() {
+ let first = memory_operator();
+ let second = memory_operator();
+ let (io, provider) = provider_io(vec![
+ ("s3://first/".to_string(), first.clone()),
+ ("oss://second/".to_string(), second.clone()),
+ ]);
+ let key = "table/中文 a%2Fb?part=1#fragment.parquet";
+ let first_path = format!("s3://first/{key}");
+ let second_path = format!("oss://second/{key}");
+ let input = io.new_input(&first_path).unwrap();
+ let output = io.new_output(&first_path).unwrap();
+ assert_eq!(provider.calls.load(Ordering::SeqCst), 0);
+
+ output
+ .write(Bytes::from_static(b"first bucket"))
+ .await
+ .unwrap();
+ io.clone()
+ .new_output(&second_path)
+ .unwrap()
+ .write(Bytes::from_static(b"second bucket"))
+ .await
+ .unwrap();
+ // Check the backing services directly: matching read/write mangling cannot
+ // make this test pass by accidentally accessing the same wrong object key.
+ assert_eq!(first.read(key).await.unwrap().to_bytes(), "first bucket");
+ assert_eq!(second.read(key).await.unwrap().to_bytes(), "second bucket");
+ assert_eq!(input.read().await.unwrap(), "first bucket");
+ assert_eq!(
+ input.reader().await.unwrap().read(1..5).await.unwrap(),
+ "irst"
+ );
+ assert!(input.exists().await.unwrap());
+ assert!(output.exists().await.unwrap());
+ let meta = input.metadata().await.unwrap();
+ assert_eq!(meta.path, first_path);
+ assert_eq!(meta.size, 12);
+ assert_eq!(io.get_status(&second_path).await.unwrap().size, 13);
+ assert_eq!(output.to_input_file().read().await.unwrap(), "first bucket");
+
+ let copied = "oss://second/table/copied";
+ io.copy_file(&first_path, copied).await.unwrap();
+ assert_eq!(
+ second.read("table/copied").await.unwrap().to_bytes(),
+ "first bucket"
+ );
+ io.delete_file(&first_path).await.unwrap();
+ assert!(!first.exists(key).await.unwrap());
+ assert!(io.exists(&second_path).await.unwrap());
+ io.delete_dir("oss://second/table/").await.unwrap();
+ assert!(!second.exists(key).await.unwrap());
+ assert!(!second.exists("table/copied").await.unwrap());
+}
+
+#[tokio::test]
+async fn provider_listings_round_trip_roots_prefixes_and_literal_keys() {
+ // An operator root below the physical bucket needs a longer logical URI
+ // prefix. The provider strips only this prefix; listing must restore it.
+ let mut config = opendal::services::MemoryConfig::default();
+ config.root = Some("/tenant/".to_string());
+ let op = Operator::from_config(config).unwrap();
+ let (io, _) = provider_io(vec![("s3://bucket/tenant/".to_string(),
op.clone())]);
+ for key in ["table/a%2Fb", "table/中文 ?#", "table/sub/c"] {
+ op.write(key, key.to_string()).await.unwrap();
+ }
+ for dir in ["s3://bucket/tenant/table", "s3://bucket/tenant/table/"] {
+ let statuses = io.list_status(dir).await.unwrap();
+ assert_eq!(
+ statuses
+ .iter()
+ .map(|s| s.path.as_str())
+ .collect::<BTreeSet<_>>(),
+ BTreeSet::from([
+ "s3://bucket/tenant/table/a%2Fb",
+ "s3://bucket/tenant/table/中文 ?#",
+ "s3://bucket/tenant/table/sub/",
+ ])
+ );
+ for status in statuses {
+ if status.is_dir {
+ assert_eq!(io.list_status(&status.path).await.unwrap().len(),
1);
+ } else {
+ assert_eq!(
+ io.new_input(&status.path).unwrap().read().await.unwrap(),
+ status.path.strip_prefix("s3://bucket/tenant/").unwrap()
+ );
+ }
+ }
+ }
+ for root in ["s3://bucket/tenant", "s3://bucket/tenant/"] {
+ assert_eq!(
+ io.list_status(root).await.unwrap()[0].path,
+ "s3://bucket/tenant/table/"
+ );
+ let files = io.list_status_recursive(root).await.unwrap();
+ assert_eq!(files.len(), 3);
+ for file in files {
+ assert!(!file.is_dir);
+ assert_eq!(
+ io.new_input(&file.path).unwrap().read().await.unwrap(),
+ file.path.strip_prefix("s3://bucket/tenant/").unwrap()
+ );
+ }
+ }
+ let (bucket_io, _) = provider_io(vec![("s3://bucket/".to_string(), op)]);
+ for root in ["s3://bucket", "s3://bucket/"] {
+ assert_eq!(
+ bucket_io.list_status(root).await.unwrap()[0].path,
+ "s3://bucket/table/"
+ );
+ }
+}
+
+#[tokio::test]
+async fn provider_errors_propagate_across_all_io_entrypoints() {
+ let io = FileIOBuilder::new("oss")
+ .with_prop("fs.oss.retry.count", "not a number")
+ .with_provider(Arc::new(RejectingProvider))
+ .build()
+ .unwrap();
+ let path = "oss://bucket/key";
+ let input = io.new_input(path).unwrap();
+ let output = io.new_output(path).unwrap();
+ assert_denied(io.exists(path).await);
+ assert_denied(io.exists_dir(path).await);
+ assert_denied(io.get_status(path).await);
+ assert_denied(io.list_status(path).await);
+ assert_denied(io.list_status_recursive(path).await);
+ assert_denied(io.mkdirs(path).await);
+ assert_denied(io.delete_file(path).await);
+ assert_denied(io.delete_dir(path).await);
+ assert_denied(io.rename(path, "oss://bucket/target").await);
+ assert_denied(io.copy_file(path, "oss://bucket/target").await);
+ assert_denied(input.exists().await);
+ assert_denied(input.metadata().await);
+ assert_denied(input.read().await);
+ assert_denied(input.reader().await);
+ assert_denied(output.exists().await);
+ assert_denied(output.writer().await);
+ assert_denied(output.async_writer().await);
+ assert_denied(output.to_input_file().read().await);
+}
+
+#[derive(Debug)]
+struct FixedPathProvider {
+ op: Operator,
+ relative: String,
+}
+
+#[async_trait::async_trait]
+impl FileIOProvider for FixedPathProvider {
+ async fn create(&self, _path: &str) -> Result<(Operator, String)> {
+ Ok((self.op.clone(), self.relative.clone()))
+ }
+}
+
+#[tokio::test]
+async fn provider_rejects_unrepresentable_keys_and_listing_mappings() {
+ let op = memory_operator();
+ op.write("a/b", "untouched").await.unwrap();
+ let (io, _) = provider_io(vec![("s3://bucket/".to_string(), op.clone())]);
+ for path in [
+ "s3://bucket/a//b",
+ "s3://bucket//a/b",
+ "s3://bucket/a/b ",
+ "s3://bucket/ a/b",
+ ] {
+ assert!(matches!(
+ io.new_output(path)
+ .unwrap()
+ .write(Bytes::from_static(b"wrong"))
+ .await,
+ Err(Error::ConfigInvalid { .. })
+ ));
+ assert!(matches!(
+ io.exists(path).await,
+ Err(Error::ConfigInvalid { .. })
+ ));
+ }
+ assert_eq!(op.read("a/b").await.unwrap().to_bytes(), "untouched");
+
+ for (uri, relative) in [
+ ("s3://b/a", "much/longer/than/the/original/path"),
+ ("s3://b/中文", "wrong"),
+ ("s3://b/foobar", "bar"),
+ ] {
+ let io = FileIOBuilder::new("unused")
+ .with_provider(Arc::new(FixedPathProvider {
+ op: op.clone(),
+ relative: relative.to_string(),
+ }))
+ .build()
+ .unwrap();
+ assert!(matches!(
+ io.list_status(uri).await,
+ Err(Error::ConfigInvalid { .. })
+ ));
+ assert!(matches!(
+ io.list_status_recursive(uri).await,
+ Err(Error::ConfigInvalid { .. })
+ ));
+ }
+ let root_io = FileIOBuilder::new("unused")
+ .with_provider(Arc::new(FixedPathProvider {
+ op,
+ relative: "/".to_string(),
+ }))
+ .build()
+ .unwrap();
+ assert_eq!(
+ root_io.list_status("s3://b").await.unwrap()[0].path,
+ "s3://b/a/"
+ );
+}
+
+#[test]
+fn provider_and_fs_operator_are_mutually_exclusive() {
+ for builder in [
+ FileIOBuilder::new("file")
+ .with_provider(Arc::new(RejectingProvider))
+ .with_fs_operator(memory_operator()),
+ FileIOBuilder::new("file")
+ .with_fs_operator(memory_operator())
+ .with_provider(Arc::new(RejectingProvider)),
+ ] {
+ assert!(matches!(builder.build(), Err(Error::ConfigInvalid { .. })));
+ }
+}
+
+#[tokio::test]
+async fn opened_reader_and_writer_keep_the_shared_backend() {
+ let op = memory_operator();
+ op.write("key", "before").await.unwrap();
+ let (io, provider) = provider_io(vec![("s3://bucket/".to_string(),
op.clone())]);
+ let input = io.new_input("s3://bucket/key").unwrap();
+ let reader = input.reader().await.unwrap();
+ let mut writer = io
+ .new_output("s3://bucket/output")
+ .unwrap()
+ .writer()
+ .await
+ .unwrap();
+ assert_eq!(reader.read(0..6).await.unwrap(), "before");
+
+ // Open handles retain the injected backend and see its shared state. They
+ // cannot rely on another provider call to replace an expired credential.
+ provider.reject.store(true, Ordering::SeqCst);
+ let calls = provider.calls.load(Ordering::SeqCst);
+ op.write("key", "after!").await.unwrap();
+ assert_eq!(reader.read(0..6).await.unwrap(), "after!");
+ writer.write(Bytes::from_static(b"written")).await.unwrap();
+ writer.close().await.unwrap();
+ assert_eq!(op.read("output").await.unwrap().to_bytes(), "written");
+ assert_eq!(provider.calls.load(Ordering::SeqCst), calls);
+ assert_denied(input.read().await);
+}
+
+#[cfg(all(feature = "storage-fs", not(windows)))]
+#[tokio::test]
+async fn static_listing_preserves_literal_posix_backslashes() {
+ let root = tempfile::tempdir().unwrap();
+ let dir = root.path().join(r"literal\directory");
+ std::fs::create_dir(&dir).unwrap();
+ let file = dir.join("data");
+ std::fs::write(&file, "data").unwrap();
+ let io = FileIOBuilder::new("file").build().unwrap();
+ let statuses = io
+ .list_status(&format!("file:{}", dir.display()))
+ .await
+ .unwrap();
+ assert_eq!(statuses.len(), 1);
+ assert_eq!(statuses[0].path, format!("file:{}", file.display()));
+ assert_eq!(
+ io.new_input(&statuses[0].path)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ "data"
+ );
+}
+
+fn with_memory_cache(mut io: FileIO) -> FileIO {
+ use crate::io::cache::LocalCacheConfig;
+ use crate::{CatalogOptions, Options};
+
+ let mut options = Options::new();
+ options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
+ options.set(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE, "4");
+ let config = LocalCacheConfig::from_options(&options).unwrap().unwrap();
+ io.cache = Some(Arc::new(LocalCache::new(config).unwrap()));
+ io
+}
+
+#[tokio::test]
+async fn provider_cache_isolates_buckets_and_invalidates_aliases() {
+ use tokio::io::AsyncWriteExt;
+
+ let first = memory_operator();
+ let second = memory_operator();
+ let (io, provider) = provider_io(vec![
+ ("s3://first/".to_string(), first.clone()),
+ ("s3a://first/".to_string(), first.clone()),
+ ("oss://second/".to_string(), second.clone()),
+ ]);
+ let io = with_memory_cache(io);
+ let key = "table/snapshot/snapshot-1";
+ let a = io.new_input(&format!("s3://first/{key}")).unwrap();
+ let b = io.new_input(&format!("oss://second/{key}")).unwrap();
+ first.write(key, "aaaa").await.unwrap();
+ second.write(key, "bbbb").await.unwrap();
+ assert_eq!(a.read().await.unwrap(), "aaaa");
+ assert_eq!(b.read().await.unwrap(), "bbbb");
+ first.delete(key).await.unwrap();
+ second.delete(key).await.unwrap();
+ assert_eq!(a.read().await.unwrap(), "aaaa");
+ assert_eq!(b.read().await.unwrap(), "bbbb");
+
+ // Even an existing cached handle must ask its provider before serving
data.
+ provider.reject.store(true, Ordering::SeqCst);
+ assert_denied(a.read().await);
+ assert_denied(b.reader().await);
+ provider.reject.store(false, Ordering::SeqCst);
+
+ let output = io.new_output(&format!("s3a://first/{key}")).unwrap();
+ output.write(Bytes::from_static(b"cccc")).await.unwrap();
+ assert_eq!(a.read().await.unwrap(), "cccc");
+ let mut writer = output.async_writer().await.unwrap();
+ writer.write_all(b"dddd").await.unwrap();
+ writer.shutdown().await.unwrap();
+ assert_eq!(a.read().await.unwrap(), "dddd");
+
+ io.delete_dir("s3a://first/table/").await.unwrap();
+ assert!(a.read().await.is_err());
+ io.delete_file(&format!("oss://second/{key}"))
+ .await
+ .unwrap();
+ assert!(b.read().await.is_err());
+}
+
+#[cfg(feature = "storage-fs")]
+#[tokio::test]
+async fn provider_rename_uses_one_service_and_rejects_other_backends() {
+ fn fs_operator(root: &std::path::Path) -> Operator {
+ let mut config = opendal_service_fs::FsConfig::default();
+ config.root = Some(root.to_string_lossy().to_string());
+ Operator::from_config(config).unwrap()
+ }
+ let first_root = tempfile::tempdir().unwrap();
+ let second_root = tempfile::tempdir().unwrap();
+ let first = fs_operator(first_root.path());
+ let second = fs_operator(second_root.path());
+ let (io, _) = provider_io(vec![
+ ("s3://first/".to_string(), first.clone()),
+ ("oss://second/".to_string(), second.clone()),
+ ]);
+ let io = with_memory_cache(io);
+ io.mkdirs("s3://first/table/snapshot/").await.unwrap();
+ assert!(io.exists_dir("s3://first/table/snapshot").await.unwrap());
+ let src = "s3://first/table/snapshot/snapshot-1";
+ let dst = "s3://first/table/snapshot/snapshot-2";
+ io.new_output(src)
+ .unwrap()
+ .write(Bytes::from_static(b"source"))
+ .await
+ .unwrap();
+ io.new_output(dst)
+ .unwrap()
+ .write(Bytes::from_static(b"target"))
+ .await
+ .unwrap();
+ assert_eq!(io.new_input(src).unwrap().read().await.unwrap(), "source");
+ assert_eq!(io.new_input(dst).unwrap().read().await.unwrap(), "target");
+ io.rename(src, dst).await.unwrap();
+ assert!(io.new_input(src).unwrap().read().await.is_err());
+ assert_eq!(io.new_input(dst).unwrap().read().await.unwrap(), "source");
+
+ // A mistaken rename on the source operator would overwrite its local
+ // `protected` file while leaving the intended destination untouched.
+ first.write("protected", "local").await.unwrap();
+ second.write("protected", "remote").await.unwrap();
+ assert!(matches!(
+ io.rename(dst, "oss://second/protected").await,
+ Err(Error::IoUnsupported { .. })
+ ));
+ assert_eq!(first.read("protected").await.unwrap().to_bytes(), "local");
+ assert_eq!(second.read("protected").await.unwrap().to_bytes(), "remote");
+ assert!(io.exists(dst).await.unwrap());
+}
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index 5a872500..026fdb5f 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -51,6 +51,77 @@ Available storage features:
| `storage-hdfs` | HDFS |
| `storage-all` | All of the above |
+## Reusing a storage backend
+
+On the development branch, Rust embedders can implement
`paimon::io::FileIOProvider`
+to reuse an OpenDAL operator managed by their application. The provider
receives
+the original URI and returns a shared operator and its relative object path.
+It can route different schemes or buckets to different operators.
+
+```rust
+use std::sync::Arc;
+use opendal::Operator;
+use paimon::io::{FileIOBuilder, FileIOProvider};
+use paimon::{Error, Result};
+
+#[derive(Debug)]
+struct SharedStorage {
+ // An application-managed operator rooted at this bucket's root.
+ operator: Operator,
+}
+
+#[async_trait::async_trait]
+impl FileIOProvider for SharedStorage {
+ async fn create(&self, uri: &str) -> Result<(Operator, String)> {
+ let key = uri.strip_prefix("s3://my-bucket/").ok_or_else(|| {
+ Error::ConfigInvalid {
+ message: "URI is outside the configured bucket".to_string(),
+ }
+ })?;
+ Ok((self.operator.clone(), key.to_string()))
+ }
+}
+
+// `operator` is supplied by the embedding application.
+let file_io = FileIOBuilder::new("s3")
+ .with_provider(Arc::new(SharedStorage { operator }))
+ .build()?;
+let bytes = file_io.new_input("s3://my-bucket/table/schema/schema-0")?
+ .read().await?;
+```
+
+The application needs `async-trait` and a compatible `opendal-core` dependency
+(named `opendal` above). Injecting a provider does not require Paimon's
built-in
+feature for that storage service. Supply the resulting `FileIO` to `Table::new`
+or other APIs accepting a `FileIO`; option-based catalog constructors continue
+to construct their own storage backends.
+
+Provider configuration bypasses built-in storage construction. Provider errors
+are returned without trying properties, environment credentials, or another
+backend. `new_input` and `new_output` remain synchronous: they retain the URI
and
+provider, and resolution errors are returned by the subsequent async operation.
+
+For directory listings, the returned relative path must be an unchanged suffix
+of the URI, starting at a path-component boundary. An operator rooted at
+`/tenant/`, for example, can resolve `s3://my-bucket/tenant/table/` to
`table/`.
+Both listing APIs return full URIs that can be passed back to `FileIO`. Empty
+relative paths and `/` represent the operator root. Preserve literal percent
+escapes, Unicode, `?`, and `#` in object keys; do not decode or normalize them
as
+URL components. Object paths that OpenDAL would change by trimming whitespace
+or collapsing slashes are rejected instead of accessing a different object.
+
+Providers manage operator reuse and credential refresh. Resolution is repeated
+for each async file operation, but an already-open reader or writer keeps its
+operator: that backend must refresh credentials internally. Custom services
+must report distinct OpenDAL storage identities (`scheme`, `name`, `root`) for
+different storage namespaces so cached data cannot overlap. Rename through a
+provider requires both paths to resolve to the same shared service instance;
+cross-backend rename is rejected.
+
+`with_fs_operator` remains available for filesystem paths. It cannot be
combined
+with `with_provider` on the same builder. Without a provider, property-based
+storage configuration behaves as before.
+
## Mosaic File Format
Mosaic data file reads are always available. The current Mosaic support is
read-only: Paimon Rust can read existing `.mosaic` data files, including array
and map columns, in a Paimon table, but it does not write Mosaic data files yet.
