leaves12138 commented on code in PR #600: URL: https://github.com/apache/paimon-rust/pull/600#discussion_r3638518933
########## crates/paimon/src/io/cache/mod.rs: ########## @@ -0,0 +1,549 @@ +// 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. + +mod disk; +mod file_type; +mod reader; + +use self::file_type::FileType; +use crate::common::{CatalogOptions, Options}; +use indexmap::IndexMap; +use sha2::{Digest, Sha256}; +use std::collections::{HashMap, HashSet}; +use std::path::PathBuf; +use std::sync::{Arc, Mutex, Weak}; + +use disk::{BlockKey, DiskCache}; +pub(super) use reader::CachedFileReader; + +const CACHE_DIRECTORY_NAME: &str = "paimon-local-cache-v2"; +const DEFAULT_FILE_SIZE_CAPACITY: usize = 65_536; + +#[derive(Debug)] +pub(crate) struct LocalCache { + disk: DiskCache, + namespace: String, + block_size: u64, + whitelist: HashSet<FileType>, + file_sizes: Mutex<IndexMap<String, u64>>, + file_size_capacity: usize, + in_flight: tokio::sync::Mutex<HashMap<BlockKey, Weak<tokio::sync::Mutex<()>>>>, + path_states: Mutex<HashMap<String, Weak<PathCacheState>>>, +} + +#[derive(Debug)] +struct PathCacheState { + generation: std::sync::atomic::AtomicU64, + publish_gate: tokio::sync::RwLock<()>, +} + +#[derive(Clone)] +pub(super) struct CacheReadToken { + generation: u64, + state: Arc<PathCacheState>, +} + +impl LocalCache { + pub(super) fn new(config: LocalCacheConfig) -> crate::Result<Self> { + let file_size_capacity = config + .max_size + .map(|max_size| max_size / config.block_size) + .and_then(|capacity| usize::try_from(capacity).ok()) + .unwrap_or(DEFAULT_FILE_SIZE_CAPACITY) + .clamp(1, DEFAULT_FILE_SIZE_CAPACITY); + Ok(Self { + disk: DiskCache::new(config.dir.join(CACHE_DIRECTORY_NAME), config.max_size)?, Review Comment: Each catalog creates an independent `DiskCache` and `CacheState` here, even when multiple catalogs in the same process share the same configured `local-cache.dir`. If two catalogs are constructed before either writes, both scan an empty directory; with a one-block `local-cache.max-size`, each can subsequently retain one block, so total encoded disk usage becomes twice the configured limit. Their LRU views also become stale relative to each other. The documentation explicitly allows catalogs to share the base directory, so namespace isolation alone does not preserve the size/LRU contract. Could we share the disk cache state per canonical cache root within the process (and handle conflicting limits), or physically isolate capacity accounting per catalog? A two-cache-instance test with one shared directory would catch this. ########## crates/paimon/src/io/cache/disk.rs: ########## @@ -0,0 +1,889 @@ +// 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. + +use bytes::Bytes; +use indexmap::IndexMap; +use sha2::{Digest, Sha256}; +use std::collections::{HashMap, HashSet}; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; +use tokio::io::AsyncWriteExt; + +const CACHE_MAGIC: &[u8; 8] = b"PAIMONLC"; +const CACHE_FORMAT_VERSION: u8 = 2; +const FIXED_HEADER_LEN: usize = CACHE_MAGIC.len() + 1 + 4 + 4 + 8 + 8 + 8; +const CHECKSUM_LEN: usize = 4; + +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +pub(super) struct BlockKey { + namespace: String, + path: String, + block_size: u64, + block_index: u64, +} + +impl BlockKey { + #[cfg(test)] + pub(super) fn new(path: impl Into<String>, block_size: u64, block_index: u64) -> Self { + Self::with_namespace("", path, block_size, block_index) + } + + pub(super) fn with_namespace( + namespace: impl Into<String>, + path: impl Into<String>, + block_size: u64, + block_index: u64, + ) -> Self { + Self { + namespace: namespace.into(), + path: path.into(), + block_size, + block_index, + } + } + + pub(super) fn cache_relative_path(&self) -> PathBuf { + let mut digest = Sha256::new(); + digest.update([CACHE_FORMAT_VERSION]); + digest.update((self.namespace.len() as u64).to_le_bytes()); + digest.update(self.namespace.as_bytes()); + digest.update((self.path.len() as u64).to_le_bytes()); + digest.update(self.path.as_bytes()); + digest.update(self.block_size.to_le_bytes()); + digest.update(self.block_index.to_le_bytes()); + let hex = hex::encode(digest.finalize()); + PathBuf::from(&hex[..2]).join(hex) + } +} + +#[derive(Debug)] +pub(super) struct BlockDecodeError(&'static str); + +impl std::fmt::Display for BlockDecodeError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str(self.0) + } +} + +#[derive(Debug)] +pub(super) struct DiskCache { + root: PathBuf, + max_size: Option<u64>, + state: Mutex<CacheState>, +} + +#[derive(Debug, Default)] +struct CacheState { + entries: IndexMap<BlockKey, u64>, + paths: HashMap<LogicalPath, HashSet<BlockKey>>, + current_size: u64, +} + +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +struct LogicalPath { + namespace: String, + path: String, +} + +impl DiskCache { + pub(super) fn new(root: impl AsRef<Path>, max_size: Option<u64>) -> crate::Result<Self> { + let root = root.as_ref().to_path_buf(); + initialize_cache_directory(&root)?; + if let Err(error) = cleanup_temporary_files(&root) { + log::warn!( + "Failed to clean temporary files in local cache directory '{}': {error}", + root.display() + ); + } + let state = scan_existing_blocks(&root, max_size); + Ok(Self { + root, + max_size, + state: Mutex::new(state), + }) + } + + pub(super) async fn get_block(&self, key: &BlockKey) -> Option<Bytes> { + if !self.is_active(key) { + return None; + } + let path = self.root.join(key.cache_relative_path()); + let encoded = match tokio::fs::read(&path).await { + Ok(encoded) => encoded, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + self.forget_entry(key); + return None; + } + Err(error) => { + log::debug!( + "Failed to read local cache block '{}': {error}", + path.display() + ); + return None; + } + }; + match decode_block(key, &encoded) { + Ok(payload) if self.touch_entry(key) => Some(payload), + Ok(_) => None, + Err(error) => { + log::debug!( + "Discarding invalid local cache block '{}': {error}", + path.display() + ); + self.forget_entry(key); + let _ = tokio::fs::remove_file(path).await; + None + } + } + } + + pub(super) async fn put_block(&self, key: &BlockKey, payload: Bytes) { + let encoded = encode_block(key, &payload); + let encoded_size = encoded.len() as u64; + if self + .max_size + .is_some_and(|max_size| encoded_size > max_size) + { + return; + } + + let path = self.root.join(key.cache_relative_path()); + let Some(parent) = path.parent() else { + return; + }; + if let Err(error) = tokio::fs::create_dir_all(parent).await { + log::debug!( + "Failed to create local cache shard '{}': {error}", + parent.display() + ); + return; + } + + let file_name = path + .file_name() + .map(|name| name.to_string_lossy()) + .unwrap_or_default(); + let temporary = parent.join(format!(".{file_name}.tmp.{}", uuid::Uuid::new_v4())); + let mut options = tokio::fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + options.mode(0o600); + } + let mut temporary_file = match options.open(&temporary).await { + Ok(file) => file, + Err(error) => { + log::debug!( + "Failed to create local cache temporary block '{}': {error}", + temporary.display() + ); + return; + } + }; + if let Err(error) = temporary_file.write_all(&encoded).await { + log::debug!( + "Failed to write local cache temporary block '{}': {error}", + temporary.display() + ); + drop(temporary_file); + let _ = tokio::fs::remove_file(&temporary).await; + return; + } + drop(temporary_file); + if let Err(error) = tokio::fs::rename(&temporary, &path).await { + log::debug!( + "Failed to publish local cache block '{}': {error}", + path.display() + ); + let _ = tokio::fs::remove_file(temporary).await; + return; + } + self.record_entry_and_evict(key.clone(), encoded_size).await; + } + + pub(super) async fn invalidate_path(&self, namespace: &str, path: &str) { + self.invalidate_matching(|logical_path| { + logical_path.namespace == namespace && logical_path.path == path + }) + .await; + } + + pub(super) async fn remove_block(&self, key: &BlockKey) { + self.forget_entry(key); + let path = self.root.join(key.cache_relative_path()); + if let Err(error) = tokio::fs::remove_file(&path).await { + if error.kind() != std::io::ErrorKind::NotFound { + log::debug!( + "Failed to remove local cache block '{}': {error}", + path.display() + ); + } + } + } + + pub(super) async fn invalidate_prefix(&self, namespace: &str, prefix: &str) { + let prefix = prefix.trim_end_matches('/'); + self.invalidate_matching(|logical_path| { + logical_path.namespace == namespace + && (logical_path.path == prefix + || logical_path + .path + .strip_prefix(prefix) + .is_some_and(|suffix| suffix.starts_with('/'))) + }) + .await; + } + + async fn invalidate_matching(&self, matches: impl Fn(&LogicalPath) -> bool) { + let keys = { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + let keys = state + .paths + .keys() + .filter(|path| matches(path)) + .filter_map(|path| state.paths.get(path)) + .flat_map(|keys| keys.iter().cloned()) + .collect::<Vec<_>>(); + for key in &keys { + remove_state_entry(&mut state, key); + } + keys + }; + for key in keys { + let cache_path = self.root.join(key.cache_relative_path()); + if let Err(error) = tokio::fs::remove_file(&cache_path).await { + if error.kind() != std::io::ErrorKind::NotFound { + log::debug!( + "Failed to invalidate local cache block '{}': {error}", + cache_path.display() + ); + } + } + } + } + + async fn record_entry_and_evict(&self, key: BlockKey, encoded_size: u64) { + let to_evict = { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + insert_state_entry(&mut state, key, encoded_size); + + let mut to_evict = Vec::new(); + if let Some(max_size) = self.max_size { + while state.current_size > max_size { + let Some((eldest, size)) = state.entries.shift_remove_index(0) else { + break; + }; + state.current_size = state.current_size.saturating_sub(size); + remove_path_index_entry(&mut state, &eldest); + to_evict.push(eldest); + } + } + to_evict + }; + + for key in to_evict { + let path = self.root.join(key.cache_relative_path()); + if let Err(error) = tokio::fs::remove_file(&path).await { + if error.kind() != std::io::ErrorKind::NotFound { + log::debug!( + "Failed to evict local cache block '{}': {error}", + path.display() + ); + } + } + } + } + + fn is_active(&self, key: &BlockKey) -> bool { + self.state + .lock() + .unwrap_or_else(|error| error.into_inner()) + .entries + .contains_key(key) + } + + fn touch_entry(&self, key: &BlockKey) -> bool { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + let Some(encoded_size) = state.entries.shift_remove(key) else { + return false; + }; + state.entries.insert(key.clone(), encoded_size); + true + } + + fn forget_entry(&self, key: &BlockKey) { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + remove_state_entry(&mut state, key); + } +} + +fn logical_path(key: &BlockKey) -> LogicalPath { + LogicalPath { + namespace: key.namespace.clone(), + path: key.path.clone(), + } +} + +fn insert_state_entry(state: &mut CacheState, key: BlockKey, encoded_size: u64) { + if let Some(previous_size) = state.entries.shift_remove(&key) { + state.current_size = state.current_size.saturating_sub(previous_size); + } + state + .paths + .entry(logical_path(&key)) + .or_default() + .insert(key.clone()); + state.entries.insert(key, encoded_size); + state.current_size = state.current_size.saturating_add(encoded_size); +} + +fn remove_state_entry(state: &mut CacheState, key: &BlockKey) { + if let Some(encoded_size) = state.entries.shift_remove(key) { + state.current_size = state.current_size.saturating_sub(encoded_size); + remove_path_index_entry(state, key); + } +} + +fn remove_path_index_entry(state: &mut CacheState, key: &BlockKey) { + let path = logical_path(key); + if let Some(keys) = state.paths.get_mut(&path) { + keys.remove(key); + if keys.is_empty() { + state.paths.remove(&path); + } + } +} + +fn initialize_cache_directory(root: &Path) -> crate::Result<()> { + if let Ok(metadata) = std::fs::symlink_metadata(root) { + if metadata.file_type().is_symlink() || !metadata.is_dir() { + return Err(crate::Error::ConfigInvalid { + message: format!( + "Local cache path '{}' must be a directory and not a symlink", + root.display() + ), + }); + } + } else { + let mut builder = std::fs::DirBuilder::new(); + builder.recursive(true); + #[cfg(unix)] + { + use std::os::unix::fs::DirBuilderExt; + builder.mode(0o700); + } + builder + .create(root) + .map_err(|error| crate::Error::ConfigInvalid { + message: format!( + "Failed to initialize local cache directory '{}': {error}", + root.display() + ), + })?; + } + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(root, std::fs::Permissions::from_mode(0o700)).map_err( + |error| crate::Error::ConfigInvalid { + message: format!( + "Failed to secure local cache directory '{}': {error}", + root.display() + ), + }, + )?; + } + Ok(()) +} + +fn cleanup_temporary_files(directory: &Path) -> std::io::Result<()> { + for shard in std::fs::read_dir(directory)? { + let shard = shard?; + let shard_name = shard.file_name(); + let shard_name = shard_name.to_string_lossy(); + if !shard.file_type()?.is_dir() || !is_lower_hex(&shard_name, 2) { + continue; + } + for entry in std::fs::read_dir(shard.path())? { + let entry = entry?; + let name = entry.file_name(); + let name = name.to_string_lossy(); + if entry.file_type()?.is_file() && is_cache_temporary_name(&name, &shard_name) { + std::fs::remove_file(entry.path())?; + } + } + } + Ok(()) +} + +fn scan_existing_blocks(root: &Path, max_size: Option<u64>) -> CacheState { + let mut discovered = Vec::new(); + collect_cache_files(root, &mut discovered); + discovered.sort_by_key(|(modified, _, _, _)| *modified); + + let mut state = CacheState::default(); + for (_, path, key, encoded_size) in discovered { + if root.join(key.cache_relative_path()) != path { + let _ = std::fs::remove_file(path); + continue; + } + insert_state_entry(&mut state, key, encoded_size); + } + + if let Some(max_size) = max_size { + while state.current_size > max_size { + let Some((key, encoded_size)) = state.entries.shift_remove_index(0) else { + break; + }; + state.current_size = state.current_size.saturating_sub(encoded_size); + remove_path_index_entry(&mut state, &key); + let path = root.join(key.cache_relative_path()); + if let Err(error) = std::fs::remove_file(&path) { + if error.kind() != std::io::ErrorKind::NotFound { + log::debug!( + "Failed to evict local cache block '{}' during startup: {error}", + path.display() + ); + } + } + } + } + + state +} + +fn collect_cache_files( + directory: &Path, + discovered: &mut Vec<(std::time::SystemTime, PathBuf, BlockKey, u64)>, +) { + let shards = match std::fs::read_dir(directory) { + Ok(entries) => entries, + Err(error) => { + log::debug!( + "Failed to scan local cache directory '{}': {error}", + directory.display() + ); + return; + } + }; + for shard in shards.flatten() { + let shard_name = shard.file_name(); + let shard_name = shard_name.to_string_lossy(); + let file_type = match shard.file_type() { + Ok(file_type) => file_type, + Err(_) => continue, + }; + if !file_type.is_dir() || !is_lower_hex(&shard_name, 2) { + continue; + } + let entries = match std::fs::read_dir(shard.path()) { + Ok(entries) => entries, + Err(_) => continue, + }; + for entry in entries.flatten() { + let name = entry.file_name(); + let name = name.to_string_lossy(); + if !entry.file_type().is_ok_and(|file_type| file_type.is_file()) + || !is_cache_block_name(&name, &shard_name) + { + continue; + } + collect_cache_file(&entry, discovered); + } + } +} + +fn collect_cache_file( + entry: &std::fs::DirEntry, + discovered: &mut Vec<(std::time::SystemTime, PathBuf, BlockKey, u64)>, +) { + let path = entry.path(); + let encoded = match std::fs::read(&path) { Review Comment: `DiskCache::new` reaches this path synchronously, and startup reads every cached block in full; `decode_block_any` then validates the CRC over the complete payload. With the documented 20 GiB cache example, catalog construction can synchronously read roughly 20 GiB and block the async caller before the catalog is usable. Could startup recover the key/size from bounded metadata and validate the payload lazily on the first hit, or otherwise move the full scan off the blocking construction path? -- 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]
