From 706c691e5fbbbbf5cbe608605164bbf784a18b97 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 17:20:51 -0700 Subject: [PATCH 1/2] Reduce global index planning IO --- .../paimon/src/table/global_index_scanner.rs | 152 ++++++++++++++---- crates/paimon/src/table/table_scan.rs | 98 +++++++++-- 2 files changed, 207 insertions(+), 43 deletions(-) diff --git a/crates/paimon/src/table/global_index_scanner.rs b/crates/paimon/src/table/global_index_scanner.rs index 80cf6c683..7fc807a54 100644 --- a/crates/paimon/src/table/global_index_scanner.rs +++ b/crates/paimon/src/table/global_index_scanner.rs @@ -122,7 +122,7 @@ struct GlobalIndexScanResult { /// The scanner filters index manifest entries for global index files, /// uses BTreeIndexMeta for file-level pruning, then reads matching /// BTree or bitmap files to evaluate predicates and collect row IDs. -/// Opened BTreeIndexReaders are cached for reuse across evaluations. +/// A bounded number of opened BTreeIndexReaders are cached for reuse across evaluations. pub(crate) struct GlobalIndexScanner { file_io: FileIO, table_path: String, @@ -139,6 +139,7 @@ pub(crate) struct GlobalIndexScanner { schema_fields: Vec, /// Cache of opened BTree readers, keyed by file name. reader_cache: Mutex>>, + reader_cache_capacity: usize, #[cfg(test)] query_io_probe: Option>, } @@ -384,6 +385,7 @@ impl GlobalIndexScanner { coverage_by_field, schema_fields: schema_fields.to_vec(), reader_cache: Mutex::new(HashMap::new()), + reader_cache_capacity: global_index_thread_num, #[cfg(test)] query_io_probe: None, })) @@ -826,22 +828,26 @@ impl GlobalIndexScanner { /// Get a cached reader or open a new one for the given file. async fn get_or_open_reader( &self, - file_name: &str, + entry: &GlobalIndexEntry, meta: &BTreeIndexMeta, data_type: &DataType, ) -> Result { // Try to take from cache { let mut cache = self.reader_cache.lock().unwrap(); - if let Some(reader) = cache.remove(file_name) { + if let Some(reader) = cache.remove(&entry.file_name) { return Ok(OpenedGlobalIndexReader::BTree(reader)); } } // Open new reader - let path = format!("{}/{INDEX_DIR}/{}", self.table_path, file_name); + let path = format!("{}/{INDEX_DIR}/{}", self.table_path, entry.file_name); let input = self.file_io.new_input(&path)?; - let file_size = input.metadata().await?.size; + let file_size = if entry.file_size > 0 { + entry.file_size as u64 + } else { + input.metadata().await?.size + }; let file_reader = input.reader().await?; let cmp = make_key_comparator(data_type); @@ -849,7 +855,7 @@ impl GlobalIndexScanner { .await .map(OpenedGlobalIndexReader::BTree) .map_err(|e| crate::Error::DataInvalid { - message: format!("Failed to open BTree index file: {file_name}"), + message: format!("Failed to open BTree index file: {}", entry.file_name), source: Some(Box::new(e)), }) } @@ -861,12 +867,9 @@ impl GlobalIndexScanner { data_type: &DataType, ) -> Result { match entry.index_type { - GlobalIndexFileKind::BTree => { - self.get_or_open_reader(&entry.file_name, meta, data_type) - .await - } + GlobalIndexFileKind::BTree => self.get_or_open_reader(entry, meta, data_type).await, GlobalIndexFileKind::Bitmap => self - .open_bitmap_reader(&entry.file_name) + .open_bitmap_reader(entry) .await .map(OpenedGlobalIndexReader::Bitmap) .map_err(|e| crate::Error::DataInvalid { @@ -881,18 +884,22 @@ impl GlobalIndexScanner { async fn open_bitmap_reader( &self, - file_name: &str, + entry: &GlobalIndexEntry, ) -> std::io::Result { - let path = format!("{}/{INDEX_DIR}/{}", self.table_path, file_name); + let path = format!("{}/{INDEX_DIR}/{}", self.table_path, entry.file_name); let input = self .file_io .new_input(&path) .map_err(|e| std::io::Error::other(e.to_string()))?; - let file_size = input - .metadata() - .await - .map_err(|e| std::io::Error::other(e.to_string()))? - .size; + let file_size = if entry.file_size > 0 { + entry.file_size as u64 + } else { + input + .metadata() + .await + .map_err(|e| std::io::Error::other(e.to_string()))? + .size + }; let file_reader = input .reader() .await @@ -941,7 +948,9 @@ impl GlobalIndexScanner { /// Return a reader to the cache for future reuse. fn return_reader(&self, file_name: String, reader: BTreeIndexReader) { let mut cache = self.reader_cache.lock().unwrap(); - cache.insert(file_name, reader); + if cache.len() < self.reader_cache_capacity || cache.contains_key(&file_name) { + cache.insert(file_name, reader); + } } fn find_field_id_by_name(&self, column: &str) -> Result> { @@ -2168,10 +2177,14 @@ mod tests { #[tokio::test] async fn test_evaluate_global_index_eq() { - let (file_io, table_path, file_name, _tmp) = + let (file_io, table_path, file_name, tmp) = setup_testdata_table("btree_int_100_no_compress.bin"); let meta = BTreeIndexMeta::new(Some(le_int_key(0)), Some(le_int_key(198)), false); - let entries = vec![make_global_index_entry(&file_name, 1, 0, 99, &meta)]; + let mut entry = make_global_index_entry(&file_name, 1, 0, 99, &meta); + entry.index_file.file_size = std::fs::metadata(tmp.path().join("index").join(&file_name)) + .unwrap() + .len() as i64; + let entries = vec![entry]; let fields = int_schema_fields(); // key=50 -> row_id=25, offset by row_range_start=0 -> global row_id=25 @@ -2191,6 +2204,69 @@ mod tests { assert_eq!(ranges, vec![RowRange::new(25, 25)]); } + #[tokio::test] + async fn test_evaluate_global_index_uses_known_file_size() { + let (file_io, table_path, file_name, _tmp) = + setup_testdata_table("btree_int_100_no_compress.bin"); + let meta = BTreeIndexMeta::new(Some(le_int_key(0)), Some(le_int_key(198)), false); + let mut entry = make_global_index_entry(&file_name, 1, 0, 99, &meta); + entry.index_file.file_size = 1; + + let error = evaluate_global_index_fast( + &file_io, + &table_path, + &[entry], + &[int_eq("id", 0, 50)], + &int_schema_fields(), + ) + .await + .expect_err("the known file size should be used without a metadata lookup"); + + assert!(matches!( + error, + crate::Error::DataInvalid { message, .. } + if message.contains("Failed to open BTree index file") + )); + } + + #[tokio::test] + async fn test_btree_reader_cache_is_bounded() { + let (file_io, table_path, file_name, tmp) = + setup_testdata_table("btree_int_100_no_compress.bin"); + let source = tmp.path().join("index").join(&file_name); + let file_size = std::fs::metadata(&source).unwrap().len() as i64; + let meta = BTreeIndexMeta::new(Some(le_int_key(0)), Some(le_int_key(198)), false); + let mut entries = Vec::new(); + for shard in 0..5 { + let shard_name = format!("btree-{shard}.bin"); + std::fs::copy(&source, tmp.path().join("index").join(&shard_name)).unwrap(); + let mut entry = + make_global_index_entry(&shard_name, 1, shard * 100, shard * 100 + 99, &meta); + entry.index_file.file_size = file_size; + entries.push(entry); + } + + let scanner = GlobalIndexScanner::create( + &file_io, + &table_path, + 2, + i64::MAX, + i64::MAX, + &entries, + &int_schema_fields(), + ) + .unwrap() + .unwrap(); + let result = scanner + .evaluate(&int_eq("id", 0, 50)) + .await + .unwrap() + .unwrap(); + + assert_eq!(result.row_ranges.len(), 5); + assert!(scanner.reader_cache.lock().unwrap().len() <= 2); + } + #[tokio::test] async fn test_missing_index_meta_returns_error() { let (file_io, table_path, file_name, tmp) = @@ -2304,15 +2380,19 @@ mod tests { #[tokio::test] async fn test_evaluate_java_bitmap_golden_index_eq_and_null() { let data_type = DataType::VarChar(crate::spec::VarCharType::string_type()); - let (file_io, table_path, file_name, meta, _tmp) = setup_java_bitmap_testdata_table(); - let entries = vec![make_global_index_entry_with_type( + let (file_io, table_path, file_name, meta, tmp) = setup_java_bitmap_testdata_table(); + let mut entry = make_global_index_entry_with_type( BITMAP_GLOBAL_INDEX_TYPE, &file_name, 1, 100, 109, &meta, - )]; + ); + entry.index_file.file_size = std::fs::metadata(tmp.path().join("index").join(&file_name)) + .unwrap() + .len() as i64; + let entries = vec![entry]; let fields = string_schema_fields(); assert_eq!(meta.first_key, Some(b"alpha".to_vec())); assert_eq!(meta.last_key, Some(b"office".to_vec())); @@ -2659,15 +2739,20 @@ mod tests { #[tokio::test] async fn test_evaluate_java_bitmap_golden_index_string_fallback_scan() { let data_type = DataType::VarChar(crate::spec::VarCharType::string_type()); - let (file_io, table_path, file_name, meta, _tmp) = setup_java_bitmap_testdata_table(); - let entries = vec![make_global_index_entry_with_type( + let (file_io, table_path, file_name, meta, tmp) = setup_java_bitmap_testdata_table(); + let file_size = std::fs::metadata(tmp.path().join("index").join(&file_name)) + .unwrap() + .len() as i64; + let mut entry = make_global_index_entry_with_type( BITMAP_GLOBAL_INDEX_TYPE, &file_name, 1, 100, 109, &meta, - )]; + ); + entry.index_file.file_size = file_size; + let entries = vec![entry]; let fields = string_schema_fields(); let ends_with_predicates = vec![Predicate::Leaf { @@ -2746,15 +2831,16 @@ mod tests { .unwrap(); assert_eq!(less_than_result.unwrap(), vec![RowRange::new(100, 102)]); - let mut over_limit_entries = vec![make_global_index_entry_with_type( + let mut over_limit_entry = make_global_index_entry_with_type( BITMAP_GLOBAL_INDEX_TYPE, &file_name, 1, 100, 109, &meta, - )]; - over_limit_entries[0].index_file.file_size = 2; + ); + over_limit_entry.index_file.file_size = file_size; + let over_limit_entries = vec![over_limit_entry]; let over_limit_less_than = evaluate_global_index_fast_with_fallback_size( &file_io, &table_path, @@ -2762,7 +2848,7 @@ mod tests { &less_than_predicates, &fields, i64::MAX, - 1, + file_size - 1, ) .await .unwrap(); @@ -2785,7 +2871,7 @@ mod tests { &no_match_contains, &fields, i64::MAX, - 1, + file_size - 1, ) .await .unwrap(); @@ -2817,7 +2903,7 @@ mod tests { &direct_with_over_limit_fallback, &fields, i64::MAX, - 1, + file_size - 1, ) .await .unwrap(); diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs index 471606a15..7712eb820 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -400,6 +400,25 @@ fn retain_index_manifest_entry( && normalize_sorted_global_index_type(&entry.index_file.index_type).is_some()) } +fn retain_index_manifest_entry_for_scan( + entry: &IndexManifestEntry, + global_index_needed: bool, + deletion_vectors_needed: bool, + partition_filter: Option<&PartitionFilter>, +) -> crate::Result { + if !retain_index_manifest_entry(entry, global_index_needed, deletion_vectors_needed) { + return Ok(false); + } + // Deletion vectors are selected by partition and bucket when splits are built. + if normalize_sorted_global_index_type(&entry.index_file.index_type).is_none() { + return Ok(true); + } + partition_filter + .map(|filter| filter.matches_entry(&entry.partition)) + .transpose() + .map(|matched| matched.unwrap_or(true)) +} + /// Builds a map from (partition, bucket) to (data_file_name -> DeletionFile) from index manifest entries. /// Only considers ADD entries with index_type "DELETION_VECTORS" and their deletion_vectors_ranges. fn build_deletion_files_map( @@ -1229,13 +1248,17 @@ impl<'a> PaimonTableScan<'a> { }; let table_path = self.table.location().trim_end_matches('/'); let path = format!("{table_path}/{MANIFEST_DIR}/{index_manifest_name}"); - let entries = IndexManifest::read(self.table.file_io(), &path) - .await? - .into_iter() - .filter(|entry| { - retain_index_manifest_entry(entry, global_index_needed, deletion_vectors_needed) - }) - .collect(); + let mut entries = Vec::new(); + for entry in IndexManifest::read(self.table.file_io(), &path).await? { + if retain_index_manifest_entry_for_scan( + &entry, + global_index_needed, + deletion_vectors_needed, + self.partition_filter.as_ref(), + )? { + entries.push(entry); + } + } Ok(Some(entries)) } @@ -2034,9 +2057,9 @@ mod tests { data_evolution_row_range_groups, data_file_overlaps_row_range_index, group_data_files_by_partition_bucket, manifest_file_overlaps_row_range_index, prune_data_evolution_group_by_read_fields, retain_index_manifest_entry, - retain_manifest_entry_row_ranges, retain_manifest_row_ranges, - should_skip_level_zero_for_scan, split_row_ranges_for_files, LimitPushdownAccumulator, - PaimonTableScan, RowRangeIndex, TableScan, + retain_index_manifest_entry_for_scan, retain_manifest_entry_row_ranges, + retain_manifest_row_ranges, should_skip_level_zero_for_scan, split_row_ranges_for_files, + LimitPushdownAccumulator, PaimonTableScan, RowRangeIndex, TableScan, }; use crate::catalog::Identifier; use crate::io::FileIOBuilder; @@ -3998,6 +4021,61 @@ mod tests { ); } + #[test] + fn test_retain_index_manifest_entries_for_selected_partitions() { + let partition = |value| { + let mut builder = BinaryRowBuilder::new(1); + builder.write_int(0, value); + builder.build_serialized() + }; + let matching_partition = partition(7); + let filter = PartitionFilter::from_partition_set( + HashSet::from([matching_partition.clone()]), + &[DataField::new( + 0, + "dt".to_string(), + DataType::Int(IntType::new()), + )], + ) + .unwrap(); + let entry = |partition, index_type: &str| IndexManifestEntry { + version: 1, + kind: FileKind::Add, + partition, + bucket: 0, + index_file: IndexFileMeta { + index_type: index_type.to_string(), + file_name: "btree.idx".to_string(), + file_size: 1, + row_count: 1, + deletion_vectors_ranges: None, + global_index_meta: None, + }, + }; + + assert!(retain_index_manifest_entry_for_scan( + &entry(matching_partition, "btree"), + true, + false, + Some(&filter), + ) + .unwrap()); + assert!(!retain_index_manifest_entry_for_scan( + &entry(partition(8), "btree"), + true, + false, + Some(&filter), + ) + .unwrap()); + assert!(retain_index_manifest_entry_for_scan( + &entry(partition(8), "DELETION_VECTORS"), + true, + true, + Some(&filter), + ) + .unwrap()); + } + #[tokio::test] async fn test_skip_index_manifest_without_active_consumer() { let table = Table::new( From 06bbe0a0b444db48499c3b223372e57c054822d2 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 9 Aug 2026 17:24:01 -0700 Subject: [PATCH 2/2] Keep global index reader cache behavior unchanged --- .../paimon/src/table/global_index_scanner.rs | 46 +------------------ 1 file changed, 2 insertions(+), 44 deletions(-) diff --git a/crates/paimon/src/table/global_index_scanner.rs b/crates/paimon/src/table/global_index_scanner.rs index 7fc807a54..9af69b593 100644 --- a/crates/paimon/src/table/global_index_scanner.rs +++ b/crates/paimon/src/table/global_index_scanner.rs @@ -122,7 +122,7 @@ struct GlobalIndexScanResult { /// The scanner filters index manifest entries for global index files, /// uses BTreeIndexMeta for file-level pruning, then reads matching /// BTree or bitmap files to evaluate predicates and collect row IDs. -/// A bounded number of opened BTreeIndexReaders are cached for reuse across evaluations. +/// Opened BTreeIndexReaders are cached for reuse across evaluations. pub(crate) struct GlobalIndexScanner { file_io: FileIO, table_path: String, @@ -139,7 +139,6 @@ pub(crate) struct GlobalIndexScanner { schema_fields: Vec, /// Cache of opened BTree readers, keyed by file name. reader_cache: Mutex>>, - reader_cache_capacity: usize, #[cfg(test)] query_io_probe: Option>, } @@ -385,7 +384,6 @@ impl GlobalIndexScanner { coverage_by_field, schema_fields: schema_fields.to_vec(), reader_cache: Mutex::new(HashMap::new()), - reader_cache_capacity: global_index_thread_num, #[cfg(test)] query_io_probe: None, })) @@ -948,9 +946,7 @@ impl GlobalIndexScanner { /// Return a reader to the cache for future reuse. fn return_reader(&self, file_name: String, reader: BTreeIndexReader) { let mut cache = self.reader_cache.lock().unwrap(); - if cache.len() < self.reader_cache_capacity || cache.contains_key(&file_name) { - cache.insert(file_name, reader); - } + cache.insert(file_name, reader); } fn find_field_id_by_name(&self, column: &str) -> Result> { @@ -2229,44 +2225,6 @@ mod tests { )); } - #[tokio::test] - async fn test_btree_reader_cache_is_bounded() { - let (file_io, table_path, file_name, tmp) = - setup_testdata_table("btree_int_100_no_compress.bin"); - let source = tmp.path().join("index").join(&file_name); - let file_size = std::fs::metadata(&source).unwrap().len() as i64; - let meta = BTreeIndexMeta::new(Some(le_int_key(0)), Some(le_int_key(198)), false); - let mut entries = Vec::new(); - for shard in 0..5 { - let shard_name = format!("btree-{shard}.bin"); - std::fs::copy(&source, tmp.path().join("index").join(&shard_name)).unwrap(); - let mut entry = - make_global_index_entry(&shard_name, 1, shard * 100, shard * 100 + 99, &meta); - entry.index_file.file_size = file_size; - entries.push(entry); - } - - let scanner = GlobalIndexScanner::create( - &file_io, - &table_path, - 2, - i64::MAX, - i64::MAX, - &entries, - &int_schema_fields(), - ) - .unwrap() - .unwrap(); - let result = scanner - .evaluate(&int_eq("id", 0, 50)) - .await - .unwrap() - .unwrap(); - - assert_eq!(result.row_ranges.len(), 5); - assert!(scanner.reader_cache.lock().unwrap().len() <= 2); - } - #[tokio::test] async fn test_missing_index_meta_returns_error() { let (file_io, table_path, file_name, tmp) =