Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 73 additions & 11 deletions crates/paimon/src/io/cache/file_type.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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));
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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
);
Expand Down Expand Up @@ -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",
Expand All @@ -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",
Expand All @@ -194,7 +208,7 @@ mod tests {
),
(
"s3://bucket/table/bucket-0/.data-abc.parquet.123e4567-e89b-12d3-a456-426614174000.tmp",
FileType::Data,
FileType::StructuredData,
),
];

Expand Down Expand Up @@ -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,
]);

Expand Down
29 changes: 27 additions & 2 deletions crates/paimon/src/io/cache/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down Expand Up @@ -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
])
);
}

Expand Down Expand Up @@ -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();
Expand Down
88 changes: 87 additions & 1 deletion crates/paimon/src/io/file_io.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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>,
Expand Down Expand Up @@ -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();
Expand Down
6 changes: 5 additions & 1 deletion docs/src/getting-started.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading