Skip to content

[core] Reduce global index planning IO #699

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 2 commits into from
Aug 10, 2026
+163 −41
Merged

[core] Reduce global index planning IO #699

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
706c691
Reduce global index planning IO
XiaoHongbo-Hope Aug 10, 2026
06bbe0a
Keep global index reader cache behavior unchanged
XiaoHongbo-Hope Aug 10, 2026
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
106 changes: 75 additions & 31 deletions crates/paimon/src/table/global_index_scanner.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -826,30 +826,34 @@ 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);
BTreeIndexReader::open(Box::new(file_reader), file_size, meta, cmp)
.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)),
})
}
Expand All @@ -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 {
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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) =
Expand Down Expand Up @@ -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()));
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -2746,23 +2789,24 @@ 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,
&over_limit_entries,
&less_than_predicates,
&fields,
i64::MAX,
1,
file_size - 1,
)
.await
.unwrap();
Expand All @@ -2785,7 +2829,7 @@ mod tests {
&no_match_contains,
&fields,
i64::MAX,
1,
file_size - 1,
)
.await
.unwrap();
Expand Down Expand Up @@ -2817,7 +2861,7 @@ mod tests {
&direct_with_over_limit_fallback,
&fields,
i64::MAX,
1,
file_size - 1,
)
.await
.unwrap();
Expand Down
98 changes: 88 additions & 10 deletions crates/paimon/src/table/table_scan.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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))
}

Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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(
Expand Down
Loading