sundapeng commented on code in PR #1014: URL: https://github.com/apache/paimon-rust/pull/1014#discussion_r4216633409
########## 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: a Paimon-shaped name does not make a Format Table file immutable. 129fad9 gives a Format Table the origin-only view of its FileIO in `Table::new` and `Table::from_resolved_schema`, so its scans, file status lookups and writes all use the origin whatever `io-cache.policy` says; other tables keep per-request routing. `format_table_reads_replaced_files_from_origin_with_io_cache` mirrors your probe: the target keeps id=1 at `review_external/data-{uuid}-1.parquet` and the origin holds the replacement id=2. The scan returns [2], an append also stays on the origin, and the target sees no request. Without the change the same test returns [1]. ########## 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)$" Review Comment: Thanks, b63dae4 accepts `manifest-{uuid}-changelog-{count}` and its `.avro.sidecar` form (the infix only for manifests, not manifest lists). `test_input_changelog_manifests_use_the_io_cache_target` commits a primary-key table with `changelog-producer=input` and `manifest.sidecar.enabled=true` under `meta,read,write`, reads the changelog incrementally, and checks that the changelog manifest PUT and GET and the sidecar PUT go to the target and none reach the origin; it fails on the previous rule. apache/paimon#10350 (bd41c99320) and apache/paimon#10351 (b449c5e266) accept the same names, so Java and Python readers of these tables route them alike. The getting-started section now lists `exists` as its own policy token, with `meta` covering file status only (6ed10b4). -- 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]
