From 80ff707033e0b350164a632b05741080902867d8 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Tue, 15 Sep 2026 22:18:43 +0800 Subject: [PATCH] feat(io): support URI-aware FileIO providers --- crates/paimon/src/io/file_io.rs | 299 +++++++---- .../paimon/src/io/file_io/provider_tests.rs | 489 ++++++++++++++++++ docs/src/getting-started.md | 71 +++ 3 files changed, 762 insertions(+), 97 deletions(-) create mode 100644 crates/paimon/src/io/file_io/provider_tests.rs diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index 1257f6328..720225ebc 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), + Provider(Arc), +} + #[derive(Clone)] pub struct FileIO { - storage: Arc, + backend: FileIOBackend, cache: Option>, - provider: Option>, } 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) -> 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) -> 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 { + 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: pub fn new_input(&self, path: &str) -> crate::Result { - 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: pub fn new_output(&self, path: &str) -> Result { - 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> { 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::::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 { + 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, cache: Option>, operator: Option, + provider: Option>, } 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) -> 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 { + pub fn build(mut self) -> crate::Result { 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>, } -#[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>, - provider: Option>, +#[derive(Clone, Debug)] +enum FileSource { + Static { + op: Operator, + relative_path: String, + cache_path: String, + }, + Provider(Arc), } -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>, +} +impl InputFile { pub fn location(&self) -> &str { &self.path } pub async fn exists(&self) -> crate::Result { - 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 { - 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 { - 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 { - 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>, - provider: Option>, } 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 { - 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> { - let (op, relative_path, cache_path) = self.resolve().await?; + let (op, relative_path, cache_path) = self.source.resolve(&self.path).await?; let writer: Box = 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> { - let (op, relative_path, cache_path) = self.resolve().await?; + let (op, relative_path, cache_path) = self.source.resolve(&self.path).await?; let writer: Box = 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 000000000..270dfcf93 --- /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) { + 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(result: Result) { + 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::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 5a8725006..026fdb5f8 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.