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 c120ac3b [core] Reduce global index planning IO (#699)
c120ac3b is described below
commit c120ac3b03c077c684bd9b194b1f4a92e2d6b898
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Aug 10 10:39:40 2026 +0800
[core] Reduce global index planning IO (#699)
---
crates/paimon/src/table/global_index_scanner.rs | 106 +++++++++++++++++-------
crates/paimon/src/table/table_scan.rs | 98 +++++++++++++++++++---
2 files changed, 163 insertions(+), 41 deletions(-)
diff --git a/crates/paimon/src/table/global_index_scanner.rs
b/crates/paimon/src/table/global_index_scanner.rs
index 80cf6c68..9af69b59 100644
--- a/crates/paimon/src/table/global_index_scanner.rs
+++ b/crates/paimon/src/table/global_index_scanner.rs
@@ -826,22 +826,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<OpenedGlobalIndexReader> {
// 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 +853,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 +865,9 @@ impl GlobalIndexScanner {
data_type: &DataType,
) -> Result<OpenedGlobalIndexReader> {
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 +882,22 @@ impl GlobalIndexScanner {
async fn open_bitmap_reader(
&self,
- file_name: &str,
+ entry: &GlobalIndexEntry,
) -> std::io::Result<BitmapGlobalIndexReader> {
- 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
@@ -2168,10 +2173,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 +2200,31 @@ 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_missing_index_meta_returns_error() {
let (file_io, table_path, file_name, tmp) =
@@ -2304,15 +2338,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 +2697,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 +2789,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 +2806,7 @@ mod tests {
&less_than_predicates,
&fields,
i64::MAX,
- 1,
+ file_size - 1,
)
.await
.unwrap();
@@ -2785,7 +2829,7 @@ mod tests {
&no_match_contains,
&fields,
i64::MAX,
- 1,
+ file_size - 1,
)
.await
.unwrap();
@@ -2817,7 +2861,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 471606a1..7712eb82 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<bool> {
+ 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(
