JingsongLi commented on code in PR #1014:
URL: https://github.com/apache/paimon-rust/pull/1014#discussion_r4479096293
##########
crates/paimon/src/io/file_io.rs:
##########
@@ -494,13 +579,135 @@ impl FileIO {
}
}
- pub(crate) fn create_static(&self, path: &str) -> crate::Result<(Operator,
String)> {
+ /// A view of this FileIO that sends every request to origin, as copies
require.
+ pub(crate) fn origin_only(&self) -> Self {
+ let mut file_io = self.clone();
+ file_io.origin_only = true;
+ file_io.schema_file_prefixes = None;
+ file_io
+ }
+
+ /// A view of this FileIO whose io-cache routing also knows the prefixes
`schemas` loads,
+ /// on the first request that may use io-cache.
+ pub(crate) fn with_schema_file_prefixes(&self, schemas:
SchemaFilePrefixes) -> Self {
+ let mut file_io = self.clone();
+ if !self.origin_only {
+ file_io.schema_file_prefixes = Some(Arc::new(schemas));
+ }
+ file_io
+ }
+
+ /// Whether routing `path` needs the persisted prefixes, which are not
loaded yet.
+ fn awaits_schema_file_prefixes(&self, path: &str) -> bool {
+ self.schema_file_prefixes
+ .as_ref()
+ .is_some_and(|schemas| schemas.loaded.get().is_none())
+ && has_other_data_file_prefix(path, &self.file_prefixes)
+ }
+
+ /// Prefixes for io-cache routing, with the persisted ones loaded so far.
+ fn loaded_file_prefixes(&self) -> Cow<'_, [String]> {
+ match self
+ .schema_file_prefixes
+ .as_ref()
+ .and_then(|s| s.loaded.get())
+ {
+ Some(persisted) => with_persisted_prefixes(&self.file_prefixes,
persisted),
+ None => Cow::Borrowed(&self.file_prefixes),
+ }
+ }
+
+ /// Prefixes for io-cache routing of `path`, loading the persisted ones
first when only they
+ /// may make it routable.
+ async fn routing_file_prefixes(&self, path: &str) -> crate::Result<Cow<'_,
[String]>> {
+ if self.awaits_schema_file_prefixes(path) &&
self.routes_to_io_cache().await? {
+ if let Some(schemas) = &self.schema_file_prefixes {
+ schemas.load().await;
+ }
+ }
+ Ok(self.loaded_file_prefixes())
+ }
+
+ /// A view of this FileIO whose io-cache routing also knows data files
named with `prefixes`.
+ pub(crate) fn with_file_prefixes<'a>(
+ &self,
+ prefixes: impl IntoIterator<Item = &'a str>,
+ ) -> Self {
+ let mut file_prefixes = self.file_prefixes.to_vec();
+ for prefix in prefixes {
+ if !prefix.trim().is_empty() && !file_prefixes.iter().any(|known|
known == prefix) {
+ file_prefixes.push(prefix.to_string());
+ }
+ }
+ let mut file_io = self.clone();
+ file_io.file_prefixes = file_prefixes.into();
+ file_io
+ }
+
+ /// A view of this FileIO serving the table at `location`.
+ pub(crate) fn with_table_location(&self, location: &str) -> Self {
+ let mut file_io = self.clone();
+ file_io.table_location = Some(location.trim_end_matches('/').into());
+ file_io
+ }
+
+ /// The view to read URIs taken from row data through. A FileIO that
serves no table cannot
+ /// tell who owns them, and other writers may replace what they own, so it
uses origin.
+ pub(crate) fn for_row_data(&self) -> Self {
+ if self.table_location.is_some() {
+ self.clone()
+ } else {
+ self.origin_only()
+ }
+ }
+
+ /// Whether requests on `path` must use origin: a table's view routes only
its own files.
+ fn origin_only_for(&self, path: &str) -> bool {
+ self.origin_only
+ || self.table_location.as_deref().is_some_and(|location| {
+ !path
+ .strip_prefix(location)
+ .is_some_and(|rest| rest.starts_with('/'))
+ })
+ }
+
+ /// Whether some request through this view may use an io-cache target.
+ pub(crate) async fn routes_to_io_cache(&self) -> crate::Result<bool> {
+ if self.origin_only {
+ return Ok(false);
+ }
+ match &self.backend {
+ FileIOBackend::Provider(provider) =>
provider.routes_to_io_cache().await,
+ FileIOBackend::Storage(storage) =>
Ok(storage.routes_to_io_cache()),
+ }
+ }
+
+ fn storage(&self) -> crate::Result<&Storage> {
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(storage)
+ }
+
+ /// The operator for requests that always go to origin, and the relative
path.
+ pub(crate) fn create_static(&self, path: &str) -> crate::Result<(Operator,
String)> {
+ let (op, relative_path) = self.storage()?.create(path)?;
+ Ok((op, relative_path.into_owned()))
+ }
+
+ pub(crate) fn create_routed_static(
+ &self,
+ path: &str,
+ file_prefixes: &[String],
+ ) -> crate::Result<(RoutedOperator, String)> {
+ let (op, relative_path) = self.storage()?.create_routed(path,
file_prefixes)?;
Review Comment:
[P2] Choose origin before constructing a cache target operator
`create_routed_static` constructs both operators before applying
`origin_only_for(path)`. If a selected target has an invalid endpoint, the
target's configuration error escapes even for requests the policy sends
exclusively to origin.
I reproduced this with a working local OSS origin, `io-cache.policy=read`,
and `io-cache.endpoint=ftp://cache.invalid`. The plain FileIO wrote the object
successfully. With routing enabled, an origin-only view, an out-of-table object
read, `exists`, and an origin write all failed with `endpoint protocol is
invalid` before reaching the valid origin.
Please resolve the origin-only decision before building a target operator,
and construct the target only for an operation whose policy uses it. Add a test
where an unusable target leaves origin-only requests functional.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
