This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new edbe5061 feat(core): support structured-data local cache selection
(#972)
edbe5061 is described below
commit edbe50610b7c31b170f7d846a38fcc589a86b468
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Oct 20 17:40:54 2026 +0800
feat(core): support structured-data local cache selection (#972)
---
crates/paimon/src/io/cache/file_type.rs | 84 ++++++++++++++++++++++++++-----
crates/paimon/src/io/cache/mod.rs | 29 ++++++++++-
crates/paimon/src/io/file_io.rs | 88 ++++++++++++++++++++++++++++++++-
docs/src/getting-started.md | 6 ++-
4 files changed, 192 insertions(+), 15 deletions(-)
diff --git a/crates/paimon/src/io/cache/file_type.rs
b/crates/paimon/src/io/cache/file_type.rs
index e01a6e15..2e8b66e5 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 061cec19..74c636d0 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 e65ffadd..41cc2adb 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<dyn FileBlockCache>,
@@ -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 40216050..49fca6a1 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
