sundapeng commented on code in PR #1014: URL: https://github.com/apache/paimon-rust/pull/1014#discussion_r4218111008
########## 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: Following up: after another look I reverted 129fad9 in b85e26b and leave this decision to the catalog that vends the targets. A Format Table is a Hive-style table whose files scheduled jobs may replace in place, but whether that returns old rows depends on the target. One that forwards every request to OSS, such as an OSS accelerator, or a cache that revalidates with OSS, never returns the replaced content, and routing Format Table reads through it is useful. Only a target that serves cached content without asking OSS can return the old copy, and FileIO cannot tell the two kinds apart, while the REST catalog knows which kind it vends. It already resolves `io-cache.policy` and `io-cache.whitelist` per table, database and catalog, so it can leave the io-cache options out for such tables, the same way `exists` is only vended for targets that answer it as OSS does. The getting-started section now says that the catalog decides per table whether a Format Table uses a target. The changelog manifest fix (b63dae4) and the docs fix (6ed10b4) stay. -- 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]
