sundapeng commented on code in PR #1014: URL: https://github.com/apache/paimon-rust/pull/1014#discussion_r4253759289
########## 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: Agreed, the per-table decision has no way through these loads. 194d85d gives a Format Table the origin-only view of its FileIO again, in `Table::new`, `Table::from_resolved_schema` and `copy_with_resolved_schema`, so FileSystemCatalog tables, REST external tables and token-backed tables all read, stat and write their files on the origin whatever `io-cache.policy` says. `ObjectTable` keeps an origin-only view as well, since other writers manage its objects too. A Format Table that should go through an OSS accelerator gets it by making the accelerator the origin with `io-cache.origin.endpoint`. `format_table_reads_replaced_files_from_origin` follows your catalog probe: FileSystemCatalog create/load/scan with catalog-level `io-cache.enabled=true` and `meta,read`, the target keeping row 7 and the origin holding its replacement, row 9, under a Paimon-shaped name. The loaded table, a `copy_with_options` copy and a `from_resolved_schema` table all read 9 and the target sees no request; without the change they read 7. `object_table_reads_objects_from_origin` covers an Object Table's FileIO. ########## 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: Thanks, 6341ba4 gives each table a view of its FileIO that carries the `data-file.prefix` and `changelog-file.prefix` of its schema, built in `Table::new` and `Table::from_resolved_schema`, and every request passes them to the router, through `RESTTokenFileIO` as well. `copy_with_options` and `copy_with_resolved_schema` add the prefixes of the new schema, so files written under a dynamic prefix use the target while earlier files keep using it. `data_files_named_with_the_schema_prefix_use_the_target` creates and loads a FileSystemCatalog table with `data-file.prefix=custom-` only in its schema, under `meta,read,write` and the `data` whitelist, then writes, commits and scans it: the PUT and GET of `custom-<uuid>-0.parquet` go to the target and none to the origin, and the same holds for a copy with a `dynamic-` prefix. It fails without the change. `test_token_file_io_routes_files_named_with_table_prefixes` covers the REST token path. ########## 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: Thanks, 6c2f5c4 accepts the postpone writer's `{prefix}-u-{commit user}-s-{write id}-w-{uuid}-{count}.{extension}` and managed BLOB packs `{prefix}{uuid}.managed.blob` as data files, with the `.index` and `.blobref` files named after them. I also went through the other names the crate writes on current main (data, changelog and blob files, bucket and deletion vector indexes, global indexes including vector, full-text and Lumina, manifests, statistics); they already match. `postpone_bucket_files_use_the_target` writes a `bucket=-2` primary-key table and `managed_blob_packs_use_the_target` writes and reads a primary-key BLOB table through FileSystemCatalog under `meta,read,write`: the postpone file and the packs are written and read on the target and never on the origin. Both fail without the change. Format Table files keep using the origin whatever their names (194d85d). -- 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]
