diff --git a/crates/paimon/src/io/cache/file_type.rs b/crates/paimon/src/io/cache/file_type.rs index e01a6e15e..2e8b66e50 100644 --- a/crates/paimon/src/io/cache/file_type.rs +++ b/crates/paimon/src/io/cache/file_type.rs @@ -24,22 +24,24 @@ const TEMP_FILE_SUFFIX_LEN: usize = 41; #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub(super) enum FileType { Meta, - Data, + DedicatedData, + StructuredData, BucketIndex, GlobalIndex, FileIndex, } impl FileType { - const ALL: [Self; 5] = [ + const ALL: [Self; 6] = [ Self::Meta, - Self::Data, + Self::DedicatedData, + Self::StructuredData, Self::BucketIndex, Self::GlobalIndex, Self::FileIndex, ]; - /// Mirrors Java `org.apache.paimon.utils.FileType#classify`, including its check order. + /// Keeps metadata/index classification ahead of dedicated data detection. pub(super) fn classify(path: &str) -> Self { let mut segments = path.rsplit('/'); let name = unwrap_temp_file_name(segments.next().unwrap_or(path)); @@ -78,7 +80,17 @@ impl FileType { return Self::Meta; } - Self::Data + if Self::is_dedicated_file(name) { + Self::DedicatedData + } else { + Self::StructuredData + } + } + + /// Matches Java's BLOB/video and vector-store file naming rules. + fn is_dedicated_file(name: &str) -> bool { + let name = name.to_ascii_lowercase(); + name.ends_with(".blob") || name.ends_with(".video") || name.contains(".vector.") } pub(super) fn is_mutable(path: &str) -> bool { @@ -107,13 +119,15 @@ impl FileType { "meta" => &[Self::Meta], "global-index" => &[Self::GlobalIndex], "bucket-index" => &[Self::BucketIndex], - "data" => &[Self::Data], + // Preserve the existing broad data selector. + "data" => &[Self::DedicatedData, Self::StructuredData], + "structured-data" => &[Self::StructuredData], "file-index" => &[Self::FileIndex], "" => &[], unknown => { log::warn!( "Unknown local-cache.whitelist value '{}'; supported values are \ - meta, global-index, bucket-index, data, file-index, \ + meta, global-index, bucket-index, data, structured-data, file-index, \ or * for all of them", unknown ); @@ -168,7 +182,7 @@ mod tests { ), ( "s3://bucket/table/bucket-0/data-abc.parquet", - FileType::Data, + FileType::StructuredData, ), ( "s3://bucket/table/manifest/manifest-123e4567-e89b-12d3-a456-426614174000-0.avro.sidecar", @@ -182,7 +196,7 @@ mod tests { ("s3://bucket/table/changelog/changelog-5", FileType::Meta), ( "s3://bucket/table/bucket-0/changelog-123e4567-e89b-12d3-a456-426614174000-0.parquet", - FileType::Data, + FileType::StructuredData, ), ( "s3://bucket/table/snapshot/.snapshot-13.123e4567-e89b-12d3-a456-426614174000.tmp", @@ -194,7 +208,7 @@ mod tests { ), ( "s3://bucket/table/bucket-0/.data-abc.parquet.123e4567-e89b-12d3-a456-426614174000.tmp", - FileType::Data, + FileType::StructuredData, ), ]; @@ -248,13 +262,61 @@ mod tests { } } + #[test] + fn test_dedicated_file_classification_keeps_metadata_and_indexes_separate() { + for name in [ + "data.parquet", + "data.ORC", + "data.avro", + "data.row", + "data.mosaic", + "data.vortex", + "data.csv", + "data.json", + "data.text", + "data.csv.gz", + "data.JSON.ZST", + "data.text.snappy", + "data.unknown", + "dir.parquet/data", + ] { + assert_eq!(FileType::classify(name), FileType::StructuredData); + } + for name in [ + "data.blob", + "data.BLOB", + "data.parquet.blob", + "data.video", + "data.VIDEO", + "data.vector.vortex", + "data.VECTOR.index-data", + ] { + assert_eq!(FileType::classify(name), FileType::DedicatedData); + } + assert_eq!(FileType::classify("manifest-1.avro"), FileType::Meta); + assert_eq!(FileType::classify("snapshot-1.parquet"), FileType::Meta); + assert_eq!( + FileType::classify("data.parquet.index"), + FileType::FileIndex + ); + assert_eq!( + FileType::parse_whitelist("data"), + HashSet::from([FileType::DedicatedData, FileType::StructuredData]) + ); + assert_eq!( + FileType::parse_whitelist("structured-data"), + HashSet::from([FileType::StructuredData]) + ); + } + #[test] fn test_file_type_parses_whitelist() { let all = HashSet::from([ FileType::Meta, FileType::GlobalIndex, FileType::BucketIndex, - FileType::Data, + FileType::DedicatedData, + FileType::StructuredData, FileType::FileIndex, ]); diff --git a/crates/paimon/src/io/cache/mod.rs b/crates/paimon/src/io/cache/mod.rs index 061cec191..74c636d01 100644 --- a/crates/paimon/src/io/cache/mod.rs +++ b/crates/paimon/src/io/cache/mod.rs @@ -306,7 +306,6 @@ impl LocalCacheConfig { .get(CatalogOptions::LOCAL_CACHE_WHITELIST) .map(String::as_str) .unwrap_or("meta,global-index"); - Ok(Some(Self { dir, namespace: catalog_namespace(namespace_options), @@ -409,7 +408,11 @@ mod tests { assert_eq!(config.block_size, 64 * 1024); assert_eq!( config.whitelist, - std::collections::HashSet::from([FileType::Meta, FileType::Data]) + std::collections::HashSet::from([ + FileType::Meta, + FileType::DedicatedData, + FileType::StructuredData + ]) ); } @@ -485,6 +488,28 @@ mod tests { assert_eq!(cache.file_size(path, &token).await, None); } + #[test] + fn test_structured_data_excludes_dedicated_and_mutable_files() { + let mut options = Options::new(); + options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true"); + options.set(CatalogOptions::LOCAL_CACHE_WHITELIST, "structured-data"); + let cache = create_local_cache(&options).unwrap().unwrap(); + assert!(cache.is_cacheable("s3://bucket/data.parquet")); + assert!(cache.is_cacheable("s3://bucket/data.unknown")); + for path in [ + "s3://bucket/data.parquet.tmp", + "s3://bucket/.data.parquet.123e4567-e89b-12d3-a456-426614174000.tmp", + "s3://bucket/data.tmp.parquet", + "s3://bucket/manifest-1.avro", + "s3://bucket/data.parquet.index", + "s3://bucket/data.blob", + "s3://bucket/data.video", + "s3://bucket/data.vector.vortex", + ] { + assert!(!cache.is_cacheable(path), "{path}"); + } + } + #[tokio::test] async fn test_local_cache_file_size_is_invalidated_across_shared_instances() { let directory = tempfile::tempdir().unwrap(); diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index e65ffadda..41cc2adbc 100644 --- a/crates/paimon/src/io/file_io.rs +++ b/crates/paimon/src/io/file_io.rs @@ -268,7 +268,7 @@ impl FileIO { /// `block_size` controls the aligned ranges presented to the cache. /// `whitelist` uses the same comma-separated values as /// `local-cache.whitelist`: `meta`, `global-index`, `bucket-index`, `data`, - /// and `file-index`, or `*` for all of them. + /// `structured-data`, and `file-index`, or `*` for all of them. pub fn with_file_block_cache( mut self, cache: Arc, @@ -2426,6 +2426,92 @@ mod input_output_test { } } + #[tokio::test] + async fn test_structured_data_cache_preserves_existing_selectors() { + for disk in [false, true] { + for whitelist in [ + "meta,global-index", + "data", + "structured-data", + "meta,global-index,structured-data", + "*", + ] { + let directory = tempfile::tempdir().unwrap(); + let mut options = Options::new(); + options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true"); + options.set(CatalogOptions::LOCAL_CACHE_WHITELIST, whitelist); + options.set(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE, "4"); + if disk { + options.set( + CatalogOptions::LOCAL_CACHE_DIR, + directory.path().to_string_lossy(), + ); + } + let cache = Arc::new( + LocalCache::new(LocalCacheConfig::from_options(&options).unwrap().unwrap()) + .unwrap(), + ); + let file_io = FileIOBuilder::new("memory") + .with_local_cache(cache) + .build() + .unwrap(); + for name in [ + "data.parquet", + "data.blob", + "data.orc", + "data.avro", + "data.row", + "data.mosaic", + "data.vortex", + "data.csv", + "data.json.zst", + "data.text.snappy", + "data.video", + "data.vector.vortex", + "data.PARQUET", + "data.unknown", + "data.parquet.index", + "snapshot-1", + "snapshot-1.blob", + ] { + let path = format!("memory:/{name}"); + file_io + .new_output(&path) + .unwrap() + .write(Bytes::from_static(b"abcdefgh")) + .await + .unwrap(); + let input = file_io.new_input(&path).unwrap(); + let reader = input.reader().await.unwrap(); + assert_eq!(reader.read(1..7).await.unwrap(), b"bcdefg"[..]); + drop(reader); + let (op, relative_path, _) = input.source.resolve(&path).await.unwrap(); + op.delete(&relative_path).await.unwrap(); + let data_file = name.starts_with("data.") && !name.ends_with(".index"); + let dedicated = + matches!(name, "data.blob" | "data.video" | "data.vector.vortex"); + let cached = match whitelist { + "*" => true, + "data" => data_file, + "structured-data" => data_file && !dedicated, + "meta,global-index,structured-data" => { + name.starts_with("snapshot-") || (data_file && !dedicated) + } + _ => name.starts_with("snapshot-"), + }; + let result = input.read().await; + if cached { + assert_eq!(result.unwrap(), b"abcdefgh"[..], "{whitelist}: {name}"); + let reader = input.reader().await.unwrap(); + assert_eq!(reader.read(1..7).await.unwrap(), b"bcdefg"[..]); + } else { + assert!(result.is_err(), "{whitelist}: {name}"); + } + } + } + } + } + async fn common_test_output_file_write_and_read(file_io: &FileIO, path: &str) { let output = file_io.new_output(path).unwrap(); let mut writer = output.writer().await.unwrap(); diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md index 40216050f..49fca6a18 100644 --- a/docs/src/getting-started.md +++ b/docs/src/getting-started.md @@ -403,7 +403,11 @@ let catalog = CatalogFactory::create(options).await?; | `local-cache.dir` | none | Optional base directory. When set, Paimon uses a persistent disk cache in a private versioned child directory; otherwise it uses memory. | | `local-cache.max-size` | unlimited | Maximum cache size. Memory caches count payload bytes; disk caches count encoded bytes. Values accept byte units such as `512 MiB` or `20 GiB`. | | `local-cache.block-size` | `1 MiB` | Block size used for cached range reads. | -| `local-cache.whitelist` | `meta,global-index` | Comma-separated eligible types: `meta`, `global-index`, `bucket-index`, `data`, and `file-index`, or `*` for all of them. | +| `local-cache.whitelist` | `meta,global-index` | Comma-separated eligible types: `meta`, `global-index`, `bucket-index`, `data`, `structured-data`, and `file-index`, or `*` for all of them. | + +Use `local-cache.whitelist=meta,global-index,structured-data` to cache data +blocks without dedicated BLOB, video, or vector-store files. `data` and `*` +retain their broader behavior. Each catalog owns its in-memory cache for the catalog's lifetime. Disk caches are reused after process restarts. Cache keys include a catalog-configuration