JingsongLi commented on code in PR #1014: URL: https://github.com/apache/paimon-rust/pull/1014#discussion_r4244835208
########## crates/paimon/src/io/cache_routing.rs: ########## @@ -0,0 +1,1082 @@ +// 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. + +//! Per-request choice between io-cache targets and the origin OSS endpoint, following the +//! `io-cache.*` options that a REST catalog may vend with a table token. + +#![cfg_attr(not(feature = "storage-oss"), allow(dead_code))] + +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, LazyLock}; + +use opendal::Operator; +use regex::Regex; + +use crate::common::CatalogOptions; +use crate::io::cache::FileType; + +/// OSS endpoint, compatible with paimon-java's `fs.oss.endpoint`. +pub(crate) const OSS_ENDPOINT: &str = "fs.oss.endpoint"; +pub(crate) const IO_CACHE_ENABLED: &str = "io-cache.enabled"; +pub(crate) const IO_CACHE_ORIGIN_ENDPOINT: &str = "io-cache.origin.endpoint"; +pub(crate) const IO_CACHE_POLICY: &str = "io-cache.policy"; +pub(crate) const IO_CACHE_WHITELIST: &str = "io-cache.whitelist"; +/// Single-target shorthand for a target named `default`, ignored when `io-cache.targets` is set. +pub(crate) const IO_CACHE_ENDPOINT: &str = "io-cache.endpoint"; +pub(crate) const IO_CACHE_TARGETS: &str = "io-cache.targets"; +/// Prefix of `io-cache.target.<name>.{endpoint,path-style-access,region}`. +pub(crate) const IO_CACHE_TARGET_PREFIX: &str = "io-cache.target."; +pub(crate) const IO_CACHE_ROUTES: &str = "io-cache.routes"; + +const DEFAULT_TARGET: &str = "default"; +const DATA_FILE_PREFIX: &str = "data-file.prefix"; +const CHANGELOG_FILE_PREFIX: &str = "changelog-file.prefix"; +const DEFAULT_DATA_PREFIXES: [&str; 2] = ["data-", "changelog-"]; +const MANIFEST_SIDECAR_SUFFIX: &str = ".avro.sidecar"; + +// Data files are named {prefix}{uuid}-{count}.{extension}; this matches what follows the prefix. +const UUID: &str = "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"; +static DATA_FILE_SUFFIX: LazyLock<Regex> = LazyLock::new(|| { + Regex::new(&format!(r"^{UUID}-[0-9]+\..+$")).expect("valid data file pattern") +}); +// Other types are routed only under the names Paimon writes; Format Table files may be replaced +// in place. +static META_FILE: LazyLock<Regex> = LazyLock::new(|| { + Regex::new(&format!( + r"^(?:(?:manifest|manifest-list|index-manifest|stat)-{UUID}-[0-9]+|manifest-{UUID}-[0-9]+\.avro\.sidecar)$" + )) + .expect("valid metadata file pattern") +}); +static BUCKET_INDEX_FILE: LazyLock<Regex> = LazyLock::new(|| { + Regex::new(&format!(r"^index-{UUID}-[0-9]+$")).expect("valid bucket index file pattern") +}); +static GLOBAL_INDEX_FILE: LazyLock<Regex> = LazyLock::new(|| { + Regex::new(&format!(r"^[a-z0-9_-]+-global-index-{UUID}\.index$")) + .expect("valid global index file pattern") +}); + +/// Operation class of a FileIO request; only `Meta`, `Exists`, `Read` and `Write` can leave origin. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub(crate) enum OpClass { + /// File status lookups, including the size lookup before a cached read. + Meta, + /// Existence checks, which ask whether a file is still there. + Exists, + Read, + /// Writes of new files, including streaming and multipart writes. + Write, + /// A writer's existence checks, listing, deletes, renames, mkdirs, copies. + Origin, +} + +impl OpClass { + fn bit(self) -> u8 { + match self { + Self::Meta => 1, + Self::Read => 2, + Self::Write => 4, + Self::Exists => 8, + Self::Origin => 0, + } + } +} + +/// A set of [`OpClass`]es; `Origin` is never a member. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub(crate) struct OpClasses(u8); + +impl OpClasses { + fn insert(&mut self, op: OpClass) { + self.0 |= op.bit(); + } + + pub(crate) fn contains(self, op: OpClass) -> bool { + self.0 & op.bit() != 0 + } + + pub(crate) fn is_empty(self) -> bool { + self.0 == 0 + } +} + +/// Endpoint chosen for a request. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum Target { + /// `fs.oss.endpoint`, what clients without io-cache routing use. + Default, + /// `dlf.oss-endpoint` set on the client. + Override, + /// `io-cache.origin.endpoint`, defaulting to `fs.oss.endpoint`. + Origin, + /// The io-cache target at this index of [`IoCacheRouting::targets`]. + Cache(usize), +} + +/// A declared io-cache target. +#[derive(Debug)] +pub(crate) struct CacheTarget { + pub(crate) name: String, + /// `None` when the endpoint is missing or empty. + pub(crate) endpoint: Option<String>, + pub(crate) path_style_access: bool, +} + +/// The `io-cache.*` keys of merged table-token options. +#[derive(Debug)] +pub(crate) struct IoCacheRouting { + override_endpoint: Option<String>, + default_endpoint: Option<String>, + origin_endpoint: Option<String>, + enabled: bool, + policy: OpClasses, + whitelist: HashSet<FileType>, + targets: Vec<CacheTarget>, + /// Ordered `(file types, target name)` rules, when `io-cache.routes` is set. + routes: Option<Vec<(HashSet<FileType>, String)>>, + data_prefixes: Vec<String>, +} + +impl IoCacheRouting { + pub(crate) fn from_props(props: &HashMap<String, String>) -> Self { + let non_empty = |key: &str| props.get(key).filter(|v| !v.trim().is_empty()).cloned(); + let default_endpoint = non_empty(OSS_ENDPOINT); + Self { + override_endpoint: non_empty(CatalogOptions::DLF_OSS_ENDPOINT), + origin_endpoint: non_empty(IO_CACHE_ORIGIN_ENDPOINT) + .or_else(|| default_endpoint.clone()), + default_endpoint, + enabled: flag(props, IO_CACHE_ENABLED, false), + policy: parse_policy(props.get(IO_CACHE_POLICY).map(String::as_str)), + whitelist: FileType::parse_whitelist( + &props + .get(IO_CACHE_WHITELIST) + .map_or("*".to_string(), |value| value.to_ascii_lowercase()), + ), + targets: parse_targets(props), + routes: props.get(IO_CACHE_ROUTES).map(|value| parse_routes(value)), + data_prefixes: data_prefixes(props), + } + } + + /// Whether requests may go anywhere other than the endpoint of clients without routing. + pub(crate) fn enabled(&self) -> bool { + self.override_endpoint.is_none() && self.has_targets() + } + + fn has_targets(&self) -> bool { + self.enabled && !self.targets.is_empty() && !self.policy.is_empty() + } + + pub(crate) fn targets(&self) -> &[CacheTarget] { + &self.targets + } + + /// Applies the routing rules in order, the first match winning. + pub(crate) fn route(&self, op: OpClass, path: &str) -> Target { + if self.override_endpoint.is_some() { + return Target::Override; + } + if !self.has_targets() { + return Target::Default; + } + // Only OSS paths can use cache targets. + if !self.policy.contains(op) || !path.starts_with("oss://") { + return Target::Origin; + } + let Some(file_type) = + routable_type(path, &self.data_prefixes).filter(|t| self.whitelist.contains(t)) + else { + return Target::Origin; + }; + let chosen = match &self.routes { + Some(routes) => routes + .iter() + .find(|(types, _)| types.contains(&file_type)) + .and_then(|(_, name)| self.targets.iter().position(|t| t.name == *name)), + None => self.targets.iter().position(|t| t.endpoint.is_some()), + }; + match chosen { + Some(index) if self.targets[index].endpoint.is_some() => Target::Cache(index), + _ => Target::Origin, + } + } + + /// The target that [`Self::route`] picks for `path`, and the operation classes it serves. + pub(crate) fn cache_target(&self, path: &str) -> Option<(usize, OpClasses)> { + let mut found = None; + let mut classes = OpClasses::default(); + for op in [ + OpClass::Meta, + OpClass::Exists, + OpClass::Read, + OpClass::Write, + ] { + if let Target::Cache(index) = self.route(op, path) { + found = Some(index); + classes.insert(op); + } + } + found.map(|index| (index, classes)) + } + + /// Endpoint of requests that never leave origin. + pub(crate) fn origin(&self) -> Option<&str> { + self.endpoint(self.route(OpClass::Origin, "")) + } + + pub(crate) fn endpoint(&self, target: Target) -> Option<&str> { + match target { + Target::Default => self.default_endpoint.as_deref(), + Target::Override => self.override_endpoint.as_deref(), + Target::Origin => self.origin_endpoint.as_deref(), + Target::Cache(index) => self.targets.get(index)?.endpoint.as_deref(), + } + } +} + +/// The type a cache may serve, or `None` when the file must be read from the origin. +pub(crate) fn routable_type(path: &str, data_prefixes: &[String]) -> Option<FileType> { + let path = path.trim_end_matches('/'); + let name = path.rsplit('/').next().unwrap_or(path); + // Temp files are renamed away once written, whatever prefix they carry. + let temp = name.ends_with(".tmp") || name.contains(".tmp-") || name.contains(".tmp."); + if temp || FileType::is_mutable(path) { + return None; + } + if is_data_file_name(name, data_prefixes) { Review Comment: I rechecked this on b85e26be through the actual FileSystemCatalog create/load/scan APIs with catalog-level `io-cache.enabled=true` and the documented `meta,read` policy. After replacing an external Parquet object on origin from row 7 to row 9 while the target retains row 7, the table scan still returns row 7; listing reaches origin, but the data GETs reach the target. The same table with origin-only FileIO returns row 9. No local block cache is enabled. The proposed per-table catalog decision is absent on the built-in paths: FileSystemCatalog shares its catalog FileIO across tables (`filesystem.rs:241–245`), and external REST tables explicitly skip table tokens (`data_token_enabled && !is_external`, `rest_env.rs:373`) and build FileIO from catalog options (`513–518`). Thus a mixed catalog using the advertised read cache cannot safely load these external files through a per-table token override. Please retain a safe default for mutable external sources or wire an actual per-table decision before enabling their cache reads, with a catalog load/replace/read regression. ########## crates/paimon/src/io/cache_routing.rs: ########## @@ -0,0 +1,1088 @@ +// 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. + +//! Per-request choice between io-cache targets and the origin OSS endpoint, following the +//! `io-cache.*` options that a REST catalog may vend with a table token. + +#![cfg_attr(not(feature = "storage-oss"), allow(dead_code))] + +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, LazyLock}; + +use opendal::Operator; +use regex::Regex; + +use crate::common::CatalogOptions; +use crate::io::cache::FileType; + +/// OSS endpoint, compatible with paimon-java's `fs.oss.endpoint`. +pub(crate) const OSS_ENDPOINT: &str = "fs.oss.endpoint"; +pub(crate) const IO_CACHE_ENABLED: &str = "io-cache.enabled"; +pub(crate) const IO_CACHE_ORIGIN_ENDPOINT: &str = "io-cache.origin.endpoint"; +pub(crate) const IO_CACHE_POLICY: &str = "io-cache.policy"; +pub(crate) const IO_CACHE_WHITELIST: &str = "io-cache.whitelist"; +/// Single-target shorthand for a target named `default`, ignored when `io-cache.targets` is set. +pub(crate) const IO_CACHE_ENDPOINT: &str = "io-cache.endpoint"; +pub(crate) const IO_CACHE_TARGETS: &str = "io-cache.targets"; +/// Prefix of `io-cache.target.<name>.{endpoint,path-style-access,region}`. +pub(crate) const IO_CACHE_TARGET_PREFIX: &str = "io-cache.target."; +pub(crate) const IO_CACHE_ROUTES: &str = "io-cache.routes"; + +const DEFAULT_TARGET: &str = "default"; +const DATA_FILE_PREFIX: &str = "data-file.prefix"; +const CHANGELOG_FILE_PREFIX: &str = "changelog-file.prefix"; +const DEFAULT_DATA_PREFIXES: [&str; 2] = ["data-", "changelog-"]; +const MANIFEST_SIDECAR_SUFFIX: &str = ".avro.sidecar"; + +// Data files are named {prefix}{uuid}-{count}.{extension}; this matches what follows the prefix. +const UUID: &str = "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"; +static DATA_FILE_SUFFIX: LazyLock<Regex> = LazyLock::new(|| { + Regex::new(&format!(r"^{UUID}-[0-9]+\..+$")).expect("valid data file pattern") +}); +// Other types are routed only under the names Paimon writes; Format Table files may be replaced +// in place. TableCommit names changelog manifests manifest-{uuid}-changelog-{count}. +static META_FILE: LazyLock<Regex> = LazyLock::new(|| { + Regex::new(&format!( + r"^(?:(?:manifest-list|index-manifest|stat)-{UUID}-[0-9]+|manifest-{UUID}(?:-changelog)?-[0-9]+(?:\.avro\.sidecar)?)$" + )) + .expect("valid metadata file pattern") +}); +static BUCKET_INDEX_FILE: LazyLock<Regex> = LazyLock::new(|| { + Regex::new(&format!(r"^index-{UUID}-[0-9]+$")).expect("valid bucket index file pattern") +}); +static GLOBAL_INDEX_FILE: LazyLock<Regex> = LazyLock::new(|| { + Regex::new(&format!(r"^[a-z0-9_-]+-global-index-{UUID}\.index$")) + .expect("valid global index file pattern") +}); + +/// Operation class of a FileIO request; only `Meta`, `Exists`, `Read` and `Write` can leave origin. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub(crate) enum OpClass { + /// File status lookups, including the size lookup before a cached read. + Meta, + /// Existence checks, which ask whether a file is still there. + Exists, + Read, + /// Writes of new files, including streaming and multipart writes. + Write, + /// A writer's existence checks, listing, deletes, renames, mkdirs, copies. + Origin, +} + +impl OpClass { + fn bit(self) -> u8 { + match self { + Self::Meta => 1, + Self::Read => 2, + Self::Write => 4, + Self::Exists => 8, + Self::Origin => 0, + } + } +} + +/// A set of [`OpClass`]es; `Origin` is never a member. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub(crate) struct OpClasses(u8); + +impl OpClasses { + fn insert(&mut self, op: OpClass) { + self.0 |= op.bit(); + } + + pub(crate) fn contains(self, op: OpClass) -> bool { + self.0 & op.bit() != 0 + } + + pub(crate) fn is_empty(self) -> bool { + self.0 == 0 + } +} + +/// Endpoint chosen for a request. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum Target { + /// `fs.oss.endpoint`, what clients without io-cache routing use. + Default, + /// `dlf.oss-endpoint` set on the client. + Override, + /// `io-cache.origin.endpoint`, defaulting to `fs.oss.endpoint`. + Origin, + /// The io-cache target at this index of [`IoCacheRouting::targets`]. + Cache(usize), +} + +/// A declared io-cache target. +#[derive(Debug)] +pub(crate) struct CacheTarget { + pub(crate) name: String, + /// `None` when the endpoint is missing or empty. + pub(crate) endpoint: Option<String>, + pub(crate) path_style_access: bool, +} + +/// The `io-cache.*` keys of merged table-token options. +#[derive(Debug)] +pub(crate) struct IoCacheRouting { + override_endpoint: Option<String>, + default_endpoint: Option<String>, + origin_endpoint: Option<String>, + enabled: bool, + policy: OpClasses, + whitelist: HashSet<FileType>, + targets: Vec<CacheTarget>, + /// Ordered `(file types, target name)` rules, when `io-cache.routes` is set. + routes: Option<Vec<(HashSet<FileType>, String)>>, + data_prefixes: Vec<String>, +} + +impl IoCacheRouting { + pub(crate) fn from_props(props: &HashMap<String, String>) -> Self { + let non_empty = |key: &str| props.get(key).filter(|v| !v.trim().is_empty()).cloned(); + let default_endpoint = non_empty(OSS_ENDPOINT); + Self { + override_endpoint: non_empty(CatalogOptions::DLF_OSS_ENDPOINT), + origin_endpoint: non_empty(IO_CACHE_ORIGIN_ENDPOINT) + .or_else(|| default_endpoint.clone()), + default_endpoint, + enabled: flag(props, IO_CACHE_ENABLED, false), + policy: parse_policy(props.get(IO_CACHE_POLICY).map(String::as_str)), + whitelist: FileType::parse_whitelist( + &props + .get(IO_CACHE_WHITELIST) + .map_or("*".to_string(), |value| value.to_ascii_lowercase()), + ), + targets: parse_targets(props), + routes: props.get(IO_CACHE_ROUTES).map(|value| parse_routes(value)), + data_prefixes: data_prefixes(props), + } + } + + /// Whether requests may go anywhere other than the endpoint of clients without routing. + pub(crate) fn enabled(&self) -> bool { + self.override_endpoint.is_none() && self.has_targets() + } + + fn has_targets(&self) -> bool { + self.enabled && !self.targets.is_empty() && !self.policy.is_empty() + } + + pub(crate) fn targets(&self) -> &[CacheTarget] { + &self.targets + } + + /// Applies the routing rules in order, the first match winning. + pub(crate) fn route(&self, op: OpClass, path: &str) -> Target { + if self.override_endpoint.is_some() { + return Target::Override; + } + if !self.has_targets() { + return Target::Default; + } + // Only OSS paths can use cache targets. + if !self.policy.contains(op) || !path.starts_with("oss://") { + return Target::Origin; + } + let Some(file_type) = + routable_type(path, &self.data_prefixes).filter(|t| self.whitelist.contains(t)) + else { + return Target::Origin; + }; + let chosen = match &self.routes { + Some(routes) => routes + .iter() + .find(|(types, _)| types.contains(&file_type)) + .and_then(|(_, name)| self.targets.iter().position(|t| t.name == *name)), + None => self.targets.iter().position(|t| t.endpoint.is_some()), + }; + match chosen { + Some(index) if self.targets[index].endpoint.is_some() => Target::Cache(index), + _ => Target::Origin, + } + } + + /// The target that [`Self::route`] picks for `path`, and the operation classes it serves. + pub(crate) fn cache_target(&self, path: &str) -> Option<(usize, OpClasses)> { + let mut found = None; + let mut classes = OpClasses::default(); + for op in [ + OpClass::Meta, + OpClass::Exists, + OpClass::Read, + OpClass::Write, + ] { + if let Target::Cache(index) = self.route(op, path) { + found = Some(index); + classes.insert(op); + } + } + found.map(|index| (index, classes)) + } + + /// Endpoint of requests that never leave origin. + pub(crate) fn origin(&self) -> Option<&str> { + self.endpoint(self.route(OpClass::Origin, "")) + } + + pub(crate) fn endpoint(&self, target: Target) -> Option<&str> { + match target { + Target::Default => self.default_endpoint.as_deref(), + Target::Override => self.override_endpoint.as_deref(), + Target::Origin => self.origin_endpoint.as_deref(), + Target::Cache(index) => self.targets.get(index)?.endpoint.as_deref(), + } + } +} + +/// The type a cache may serve, or `None` when the file must be read from the origin. +pub(crate) fn routable_type(path: &str, data_prefixes: &[String]) -> Option<FileType> { + let path = path.trim_end_matches('/'); + let name = path.rsplit('/').next().unwrap_or(path); + // Temp files are renamed away once written, whatever prefix they carry. + let temp = name.ends_with(".tmp") || name.contains(".tmp-") || name.contains(".tmp."); + if temp || FileType::is_mutable(path) { + return None; + } + if is_data_file_name(name, data_prefixes) { + return Some(if name.ends_with(".index") { + FileType::FileIndex + } else { + FileType::Data + }); + } + if is_sequential(path) { + return None; + } + let file_type = FileType::classify(path); + let paimon_name = match file_type { + FileType::Meta => META_FILE.is_match(name), + FileType::BucketIndex => BUCKET_INDEX_FILE.is_match(name), + FileType::GlobalIndex => GLOBAL_INDEX_FILE.is_match(name), + // Data files and their file indexes are recognized by `is_data_file_name`. + _ => false, + }; + paimon_name.then_some(file_type) +} + +// Manifests, indexes and statistics share the uuid-count shape but have no extension. +fn is_data_file_name(name: &str, data_prefixes: &[String]) -> bool { + !name.ends_with(MANIFEST_SIDECAR_SUFFIX) + && data_prefixes.iter().any(|prefix| { + name.strip_prefix(prefix.as_str()) + .is_some_and(|rest| DATA_FILE_SUFFIX.is_match(rest)) + }) +} + +// Sequential metadata may be read before it exists; cached NotFound would hide new commits. +fn is_sequential(path: &str) -> bool { + let mut segments = path.rsplit('/'); + let name = segments.next().unwrap_or(path); + name.starts_with("snapshot-") + || name.starts_with("schema-") + || (name.starts_with("changelog-") && segments.next() == Some("changelog")) +} + +fn flag(props: &HashMap<String, String>, key: &str, default: bool) -> bool { + props + .get(key) + .map_or(default, |value| value.trim().eq_ignore_ascii_case("true")) +} + +/// Unknown tokens, including `write`, are ignored, and `none` empties the policy. +fn parse_policy(value: Option<&str>) -> OpClasses { + let mut policy = OpClasses::default(); + for token in value.unwrap_or_default().split(',') { + match token.trim().to_ascii_lowercase().as_str() { + "none" => return OpClasses::default(), + "meta" => policy.insert(OpClass::Meta), + "exists" => policy.insert(OpClass::Exists), + "read" => policy.insert(OpClass::Read), + "write" => policy.insert(OpClass::Write), + _ => {} + } + } + policy +} + +/// Declared targets in order; `io-cache.targets` wins over the `io-cache.endpoint` shorthand. +fn parse_targets(props: &HashMap<String, String>) -> Vec<CacheTarget> { + // OpenDAL's OSS service signs without a region, so `io-cache.target.<name>.region` is ignored. + let target = |name: String, endpoint_key: &str| { + let path_style_key = format!("{IO_CACHE_TARGET_PREFIX}{name}.path-style-access"); + CacheTarget { + endpoint: props + .get(endpoint_key) + .map(|endpoint| endpoint.trim()) + .filter(|endpoint| !endpoint.is_empty()) + .map(String::from), + path_style_access: flag(props, &path_style_key, false), + name, + } + }; + let Some(names) = props.get(IO_CACHE_TARGETS) else { + return match props.get(IO_CACHE_ENDPOINT) { + Some(_) => vec![target(DEFAULT_TARGET.to_string(), IO_CACHE_ENDPOINT)], + None => Vec::new(), + }; + }; + let mut unique: Vec<String> = Vec::new(); + for name in names.split(',') { + let name = name.trim().to_ascii_lowercase(); + if is_valid_target_name(&name) && !unique.contains(&name) { + unique.push(name); + } + } + unique + .into_iter() + .map(|name| { + let endpoint_key = format!("{IO_CACHE_TARGET_PREFIX}{name}.endpoint"); + target(name, &endpoint_key) + }) + .collect() +} + +/// `[a-z][a-z0-9-]*`. +fn is_valid_target_name(name: &str) -> bool { + name.starts_with(|c: char| c.is_ascii_lowercase()) + && name + .chars() + .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-') +} + +/// Rules `types=target` separated by `;`; malformed rules are skipped. +fn parse_routes(value: &str) -> Vec<(HashSet<FileType>, String)> { + value + .split(';') + .filter_map(|rule| { + let (types, target) = rule.trim().split_once('=')?; + let target = target.trim().to_ascii_lowercase(); + let types = FileType::parse_whitelist(&types.to_ascii_lowercase()); + (!target.is_empty() && !types.is_empty()).then_some((types, target)) + }) + .collect() +} + +/// `data-` and `changelog-`, plus the table's own prefixes when they are in the options. +fn data_prefixes(props: &HashMap<String, String>) -> Vec<String> { + let mut prefixes: Vec<String> = DEFAULT_DATA_PREFIXES.map(String::from).to_vec(); + for key in [DATA_FILE_PREFIX, CHANGELOG_FILE_PREFIX] { + if let Some(prefix) = props.get(key).filter(|prefix| !prefix.trim().is_empty()) { Review Comment: [P2] Pass the loaded table's custom prefixes into the router These props come from FileIO/catalog or REST-token options, but normal table `data-file.prefix` / `changelog-file.prefix` settings live in the loaded schema. FileSystemCatalog passes its shared FileIO unchanged to Table::new, and TableWrite obtains the prefixes from that schema, so the table-specific values never reach this parser. I created and loaded a real catalog table with schema `data-file.prefix=custom-`, enabled `meta,read,write` on a data target, and wrote/committed/scanned a row: all PUT/GETs for `custom-<uuid>-0.parquet` went to origin, while the identical default-prefix control used the target. This silently defeats the advertised custom-prefix/data routing unless each table prefix is manually duplicated into catalog/token properties. Please propagate the resolved table prefixes into its FileIO routing context and cover a schema-only custom prefix through the actual catalog API. ########## crates/paimon/src/io/cache_routing.rs: ########## @@ -0,0 +1,1088 @@ +// 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. + +//! Per-request choice between io-cache targets and the origin OSS endpoint, following the +//! `io-cache.*` options that a REST catalog may vend with a table token. + +#![cfg_attr(not(feature = "storage-oss"), allow(dead_code))] + +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, LazyLock}; + +use opendal::Operator; +use regex::Regex; + +use crate::common::CatalogOptions; +use crate::io::cache::FileType; + +/// OSS endpoint, compatible with paimon-java's `fs.oss.endpoint`. +pub(crate) const OSS_ENDPOINT: &str = "fs.oss.endpoint"; +pub(crate) const IO_CACHE_ENABLED: &str = "io-cache.enabled"; +pub(crate) const IO_CACHE_ORIGIN_ENDPOINT: &str = "io-cache.origin.endpoint"; +pub(crate) const IO_CACHE_POLICY: &str = "io-cache.policy"; +pub(crate) const IO_CACHE_WHITELIST: &str = "io-cache.whitelist"; +/// Single-target shorthand for a target named `default`, ignored when `io-cache.targets` is set. +pub(crate) const IO_CACHE_ENDPOINT: &str = "io-cache.endpoint"; +pub(crate) const IO_CACHE_TARGETS: &str = "io-cache.targets"; +/// Prefix of `io-cache.target.<name>.{endpoint,path-style-access,region}`. +pub(crate) const IO_CACHE_TARGET_PREFIX: &str = "io-cache.target."; +pub(crate) const IO_CACHE_ROUTES: &str = "io-cache.routes"; + +const DEFAULT_TARGET: &str = "default"; +const DATA_FILE_PREFIX: &str = "data-file.prefix"; +const CHANGELOG_FILE_PREFIX: &str = "changelog-file.prefix"; +const DEFAULT_DATA_PREFIXES: [&str; 2] = ["data-", "changelog-"]; +const MANIFEST_SIDECAR_SUFFIX: &str = ".avro.sidecar"; + +// Data files are named {prefix}{uuid}-{count}.{extension}; this matches what follows the prefix. +const UUID: &str = "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"; +static DATA_FILE_SUFFIX: LazyLock<Regex> = LazyLock::new(|| { + Regex::new(&format!(r"^{UUID}-[0-9]+\..+$")).expect("valid data file pattern") Review Comment: [P2] Recognize the immutable names emitted by the supported writers The `{uuid}-{count}.{extension}` restriction excludes actual Paimon data files, even with the default `data-` prefix: the postpone writer emits `data--u-<commit-user>-s-<write-id>-w-<uuid>-0.parquet` (`TableWrite::create_postpone_writer`), and ManagedBlobWriter emits `data-<uuid>.managed.blob`. Through real catalog tables with `meta,read,write` and a data target, a `bucket=-2` primary-key write sent its Parquet PUT to origin, and a primary-key BLOB table sent its managed-pack PUT and subsequent GET to origin; the ordinary Parquet control used the target. These unique, immutable payloads therefore miss both the cache reads and the extra write path/quotas this feature is intended to provide. Please recognize the existing writer formats (and their relevant companions) with routing tests driven by real writers, while preserving the separate protection needed for mutable external Format Table files. -- 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]
