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 afc8248f perf: optimize scalar index reads and support BTree V2 (#888)
afc8248f is described below
commit afc8248fe4a76ec1c857a2747260b7e595e908c6
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Sep 20 20:02:12 2026 +0800
perf: optimize scalar index reads and support BTree V2 (#888)
---
crates/integrations/datafusion/tests/procedures.rs | 31 +-
crates/paimon/src/arrow/format/row.rs | 127 +++-
crates/paimon/src/btree/footer.rs | 7 +
crates/paimon/src/btree/mod.rs | 1 +
crates/paimon/src/btree/posting_list.rs | 271 ++++++++
crates/paimon/src/btree/query.rs | 119 ++--
crates/paimon/src/btree/reader.rs | 36 +-
crates/paimon/src/btree/tests.rs | 336 ++++++++++
crates/paimon/src/btree/writer.rs | 35 +-
crates/paimon/src/table/global_index_scanner.rs | 4 +
.../src/table/global_index_scanner/all_match.rs | 103 +++
.../paimon/src/table/global_index_scanner/entry.rs | 1 +
.../src/table/global_index_scanner/evaluator.rs | 142 +++-
.../src/table/global_index_scanner/reader.rs | 12 +
.../src/table/global_index_scanner/row_ranges.rs | 39 +-
.../paimon/src/table/global_index_scanner/tests.rs | 733 +++++++++++++++++++++
.../sorted_global_index_build_builder/tests.rs | 248 +++++++
.../sorted_global_index_build_builder/writer.rs | 7 +-
.../src/table/sorted_global_index_options.rs | 32 +
crates/paimon/testdata/btree/GenerateBTreeV2.java | 68 ++
crates/paimon/testdata/btree/README.md | 44 ++
crates/paimon/testdata/btree/btree_v2_java_lz4.bin | Bin 0 -> 419 bytes
.../paimon/testdata/btree/btree_v2_java_none.bin | Bin 0 -> 8562 bytes
docs/src/sql.md | 20 +-
24 files changed, 2274 insertions(+), 142 deletions(-)
diff --git a/crates/integrations/datafusion/tests/procedures.rs
b/crates/integrations/datafusion/tests/procedures.rs
index cb0779b8..73e4f143 100644
--- a/crates/integrations/datafusion/tests/procedures.rs
+++ b/crates/integrations/datafusion/tests/procedures.rs
@@ -361,6 +361,12 @@ async fn
test_global_index_procedures_echo_raw_unsupported_type() {
#[tokio::test]
async fn test_create_global_index_builds_btree_and_filter_reads() {
+ for version in [1, 2] {
+ assert_btree_sql_version(version).await;
+ }
+}
+
+async fn assert_btree_sql_version(version: u32) {
let (_tmp, sql_context) =
setup_btree_global_index_table("btree_build").await;
exec(
&sql_context,
@@ -370,12 +376,12 @@ async fn
test_create_global_index_builds_btree_and_filter_reads() {
exec(
&sql_context,
- "CALL sys.create_global_index(\
+ &format!("CALL sys.create_global_index(\
table => 'test_db.btree_build', \
index_column => 'id', \
index_type => 'btree', \
- options =>
'btree-index.block-size=1kb,btree-index.compression=lz4,btree-index.compression-level=1'\
- )",
+ options =>
'btree-index.file-version={version},btree-index.block-size=1kb,btree-index.compression=lz4,btree-index.compression-level=1'\
+ )"),
)
.await;
@@ -394,6 +400,25 @@ async fn
test_create_global_index_builds_btree_and_filter_reads() {
)
.await;
assert_eq!(rows, vec![(2, "bob".to_string())]);
+ for (condition, expected) in [
+ ("id >= 0 AND id = 2", vec![(2, "bob".to_string())]),
+ (
+ "id >= 0 AND id > 1 AND id < 3 AND id <= 4",
+ vec![(2, "bob".to_string())],
+ ),
+ ("id > 2 AND id < 2", vec![]),
+ (
+ "(id > 2 AND id < 2) OR id = 3",
+ vec![(3, "carol".to_string())],
+ ),
+ ] {
+ let rows = collect_id_name(
+ &sql_context,
+ &format!("SELECT id, name FROM paimon.test_db.btree_build WHERE
{condition}"),
+ )
+ .await;
+ assert_eq!(rows, expected, "V{version}: {condition}");
+ }
}
#[tokio::test]
diff --git a/crates/paimon/src/arrow/format/row.rs
b/crates/paimon/src/arrow/format/row.rs
index 9419507e..56278108 100644
--- a/crates/paimon/src/arrow/format/row.rs
+++ b/crates/paimon/src/arrow/format/row.rs
@@ -293,6 +293,9 @@ impl FormatFileReader for RowFormatReader {
})?;
index.validate_for_file(total_rows, index_start)?;
validate_row_selection(total_rows, row_selection.as_deref())?;
+ // Public callers may supply unordered or overlapping ranges. Normalize
+ // once so every block can seek directly to its overlapping window.
+ let row_selection = row_selection.map(crate::table::merge_row_ranges);
let schema = build_target_arrow_schema(read_fields)?;
let row_type = read_fields.to_vec();
@@ -2294,11 +2297,8 @@ fn blocks_to_read(
} else {
total_rows
};
- let intersects = selection.is_none_or(|ranges| {
- ranges
- .iter()
- .any(|r| (r.from() as usize) < end && (r.to() as usize) >=
start)
- });
+ let intersects =
+ selection.is_none_or(|ranges| !overlapping_ranges(ranges, start,
end).is_empty());
if intersects {
result.push(block_idx);
}
@@ -2315,7 +2315,7 @@ fn selected_local_indices(
None => (0..block_end - block_start).collect(),
Some(ranges) => {
let mut result = Vec::new();
- for range in ranges {
+ for range in overlapping_ranges(ranges, block_start, block_end) {
let start = (range.from() as usize).max(block_start);
let end = ((range.to() as usize) + 1).min(block_end);
if start < end {
@@ -2327,6 +2327,15 @@ fn selected_local_indices(
}
}
+/// Input is sorted, disjoint and nonnegative; `end` is exclusive.
+/// Never rescan a large prefix for each successive data block.
+fn overlapping_ranges(ranges: &[RowRange], start: usize, end: usize) ->
&[RowRange] {
+ let first = ranges.partition_point(|range| (range.to() as usize) < start);
+ let suffix = &ranges[first..];
+ let count = suffix.partition_point(|range| (range.from() as usize) < end);
+ &suffix[..count]
+}
+
fn i128_to_java_bigint_bytes(value: i128) -> Vec<u8> {
let bytes = value.to_be_bytes();
let pad = if value < 0 { 0xff } else { 0x00 };
@@ -2642,7 +2651,11 @@ mod tests {
&fields,
None,
Some(8),
- Some(vec![RowRange::new(1, 2)]),
+ Some(vec![
+ RowRange::new(2, 2),
+ RowRange::new(1, 2),
+ RowRange::new(1, 1),
+ ]),
)
.await
.unwrap()
@@ -2979,6 +2992,106 @@ mod tests {
assert_eq!(ids, vec![2, 3]);
}
+ #[test]
+ #[ignore = "manual release benchmark; exact selection oracle, no timing
assertions"]
+ fn benchmark_scalar_row_selection() {
+ use std::hint::black_box;
+ use std::time::Instant;
+ let rows = 1_200_000;
+ let block_rows = 1024;
+ let starts: Vec<_> = (0..rows).step_by(block_rows).map(|id| id as
i64).collect();
+ let index =
+ RowBlockIndex::new(vec![1; starts.len()], vec![1; starts.len()],
starts).unwrap();
+ for (name, run, stride) in [
+ ("90pct", 9, 10),
+ ("10pct", 1, 10),
+ ("contiguous", rows, rows),
+ ] {
+ let ranges: Vec<_> = (0..rows)
+ .step_by(stride)
+ .map(|id| RowRange::new(id as i64, (id + run - 1) as i64))
+ .collect();
+ let run_selection = || {
+ let blocks = blocks_to_read(&index, rows, Some(&ranges));
+ let mut selected = Vec::new();
+ for block in blocks {
+ let start = index.block_row_start(block);
+ let end = (start + block_rows).min(rows);
+ selected.extend(
+ selected_local_indices(start, end, Some(&ranges))
+ .into_iter()
+ .map(|id| start + id),
+ );
+ }
+ selected
+ };
+ let expected: Vec<_> = ranges
+ .iter()
+ .flat_map(|r| r.from() as usize..=r.to() as usize)
+ .collect();
+ for _ in 0..3 {
+ assert_eq!(run_selection(), expected);
+ }
+ let mut samples = Vec::new();
+ for _ in 0..7 {
+ let start = Instant::now();
+ let selected = run_selection();
+ let elapsed = start.elapsed().as_secs_f64() * 1000.0;
+ assert_eq!(selected, expected);
+ black_box(selected);
+ samples.push(elapsed);
+ }
+ samples.sort_by(f64::total_cmp);
+ println!(
+ "ROW_SELECTION_BENCH case={name} ranges={} blocks={}
median_ms={:.3}",
+ ranges.len(),
+ index.block_count(),
+ samples[3]
+ );
+ }
+ }
+
+ #[test]
+ fn row_selection_windows_match_independent_row_oracle() {
+ let index = RowBlockIndex::new(vec![1; 4], vec![1; 4], vec![0, 10, 20,
30]).unwrap();
+ for selection in [
+ None,
+ Some(vec![]),
+ Some(vec![RowRange::new(9, 11), RowRange::new(29, 39)]),
+ Some(vec![RowRange::new(0, 0), RowRange::new(39, 39)]),
+ Some(vec![RowRange::new(0, 39)]),
+ ] {
+ let expected: Vec<_> = (0..40)
+ .filter(|row| {
+ selection.as_ref().is_none_or(|ranges| {
+ ranges
+ .iter()
+ .any(|range| range.from() <= *row as i64 && *row
as i64 <= range.to())
+ })
+ })
+ .collect();
+ let blocks = blocks_to_read(&index, 40, selection.as_deref());
+ let mut actual = Vec::new();
+ for block in blocks {
+ actual.extend(
+ selected_local_indices(block * 10, (block + 1) * 10,
selection.as_deref())
+ .into_iter()
+ .map(|row| block * 10 + row),
+ );
+ }
+ assert_eq!(actual, expected);
+ }
+ let ranges: Vec<_> = (0..1_200_000)
+ .step_by(10)
+ .map(|row| RowRange::new(row, row))
+ .collect();
+ assert_eq!(
+ overlapping_ranges(&ranges, 1_199_989, 1_200_000),
+ &[RowRange::new(1_199_990, 1_199_990)]
+ );
+ assert!(overlapping_ranges(&ranges, 1_199_991, 1_200_000).is_empty());
+ }
+
#[tokio::test]
async fn row_reader_prefetches_blocks_with_bounded_parallelism() {
let fields = vec![
diff --git a/crates/paimon/src/btree/footer.rs
b/crates/paimon/src/btree/footer.rs
index 027802a7..7e625011 100644
--- a/crates/paimon/src/btree/footer.rs
+++ b/crates/paimon/src/btree/footer.rs
@@ -140,6 +140,13 @@ impl BTreeFileFooter {
));
}
+ if !matches!(version, 1 | 2) {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!("Unsupported BTree index version: {version}"),
+ ));
+ }
+
let mut pos = 0;
// bloom filter handle
diff --git a/crates/paimon/src/btree/mod.rs b/crates/paimon/src/btree/mod.rs
index 6cc4ea01..5b7fc6fa 100644
--- a/crates/paimon/src/btree/mod.rs
+++ b/crates/paimon/src/btree/mod.rs
@@ -41,6 +41,7 @@ mod bloom_filter;
mod footer;
pub(crate) mod key_serde;
mod meta;
+mod posting_list;
pub(crate) mod query;
mod reader;
mod sst_file;
diff --git a/crates/paimon/src/btree/posting_list.rs
b/crates/paimon/src/btree/posting_list.rs
new file mode 100644
index 00000000..b2eb40e7
--- /dev/null
+++ b/crates/paimon/src/btree/posting_list.rs
@@ -0,0 +1,271 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Java-compatible BTree V1 lists and adaptive V2 postings.
+
+use super::var_len::{encode_var_int, encode_var_long};
+use roaring::{RoaringBitmap, RoaringTreemap};
+use std::io::{self, Cursor, Read};
+
+const SINGLE: u8 = 0;
+const DELTA: u8 = 1;
+const ROARING: u8 = 2;
+
+fn invalid(message: &str) -> io::Error {
+ io::Error::new(io::ErrorKind::InvalidData, message)
+}
+
+// Reject overlong/overflowing encodings instead of allowing high bits to wrap.
+fn read_non_negative(input: &mut impl Read, bits: u32) -> io::Result<u64> {
+ let mut value = 0u64;
+ for shift in (0..bits).step_by(7) {
+ let mut byte = [0];
+ input.read_exact(&mut byte)?;
+ let payload = u64::from(byte[0] & 0x7f);
+ if payload >= (1u64 << (bits - shift).min(7)) {
+ return Err(invalid("BTree posting integer overflow"));
+ }
+ value |= payload << shift;
+ if byte[0] & 0x80 == 0 {
+ return Ok(value);
+ }
+ }
+ Err(invalid("Malformed BTree posting integer"))
+}
+
+pub(super) fn add_to(data: &[u8], version: u32, target: &mut RoaringTreemap)
-> io::Result<()> {
+ let mut input = Cursor::new(data);
+ if version == 1 {
+ let count = read_non_negative(&mut input, 31)?;
+ for _ in 0..count {
+ target.insert(read_non_negative(&mut input, 63)?);
+ }
+ } else {
+ let mut tag = [0];
+ input.read_exact(&mut tag)?;
+ match tag[0] {
+ SINGLE => {
+ target.insert(read_non_negative(&mut input, 63)?);
+ }
+ DELTA => {
+ let count = read_non_negative(&mut input, 31)?;
+ if count < 2 {
+ return Err(invalid("Invalid BTree delta posting count"));
+ }
+ let mut row = read_non_negative(&mut input, 63)?;
+ target.insert(row);
+ for _ in 1..count {
+ let delta = read_non_negative(&mut input, 63)?;
+ if delta == 0 {
+ return Err(invalid("BTree row ID delta must be
positive"));
+ }
+ row = row
+ .checked_add(delta)
+ .filter(|id| *id <= i64::MAX as u64)
+ .ok_or_else(|| invalid("BTree delta row ID
overflow"))?;
+ target.insert(row);
+ }
+ }
+ ROARING => {
+ // The generic treemap decoder overwrites duplicate high keys.
+ // Validate the portable outer map so corrupt input cannot
lose rows silently.
+ let mut count = [0; 8];
+ input.read_exact(&mut count)?;
+ let count = u64::from_le_bytes(count);
+ if count == 0 {
+ return Err(invalid("Empty BTree Roaring posting"));
+ }
+ let mut bitmaps = Vec::new();
+ let mut previous = None;
+ for _ in 0..count {
+ let mut high = [0; 4];
+ input.read_exact(&mut high)?;
+ let high = u32::from_le_bytes(high);
+ if high > i32::MAX as u32 || previous.is_some_and(|last|
high <= last) {
+ return Err(invalid("Invalid BTree Roaring high keys"));
+ }
+ let bitmap = RoaringBitmap::deserialize_from(&mut input)?;
+ if bitmap.is_empty() {
+ return Err(invalid("Empty BTree Roaring partition"));
+ }
+ bitmaps.push((high, bitmap));
+ previous = Some(high);
+ }
+ *target |= RoaringTreemap::from_bitmaps(bitmaps);
+ }
+ _ => return Err(invalid("Unknown BTree posting encoding")),
+ }
+ }
+ if input.position() != data.len() as u64 {
+ return Err(invalid("Trailing bytes in BTree posting"));
+ }
+ Ok(())
+}
+
+fn var_len(value: u64) -> usize {
+ ((64 - value.leading_zeros()).max(1) as usize).div_ceil(7)
+}
+
+fn container_lower_bound(count: u64, runs: u64) -> u64 {
+ 4 + (2 * count).min(8192).min(2 + 4 * runs)
+}
+
+pub(super) fn serialize(rows: &[i64], version: u32) -> io::Result<Vec<u8>> {
+ let count = i32::try_from(rows.len()).map_err(|_| invalid("Too many BTree
row IDs"))?;
+ if rows.is_empty() || rows.iter().any(|id| *id < 0) {
+ return Err(invalid("BTree posting requires non-negative row IDs"));
+ }
+ if version == 1 {
+ let mut result = Vec::new();
+ encode_var_int(&mut result, count)?;
+ for &row in rows {
+ encode_var_long(&mut result, row)?;
+ }
+ return Ok(result);
+ }
+ if rows.len() == 1 {
+ let mut result = vec![SINGLE];
+ encode_var_long(&mut result, rows[0])?;
+ return Ok(result);
+ }
+
+ let mut delta_size = 1 + var_len(rows.len() as u64) + var_len(rows[0] as
u64);
+ // Tag, high-key count, first high key, and nested bitmap cookie. As in
Java,
+ // a lower bound avoids building a bitmap when a delta list must be
smaller.
+ let mut roaring_min = 1 + 8 + 4 + 4;
+ let mut count_in_container = 1;
+ let mut runs = 1;
+ for pair in rows.windows(2) {
+ let (previous, row) = (pair[0], pair[1]);
+ if row <= previous {
+ return Err(invalid("BTree V2 row IDs must be strictly
increasing"));
+ }
+ delta_size += var_len((row - previous) as u64);
+ if row >> 16 != previous >> 16 {
+ roaring_min += container_lower_bound(count_in_container, runs);
+ if row >> 32 != previous >> 32 {
+ roaring_min += 8;
+ }
+ count_in_container = 1;
+ runs = 1;
+ } else {
+ count_in_container += 1;
+ if row != previous + 1 {
+ runs += 1;
+ }
+ }
+ }
+ roaring_min += container_lower_bound(count_in_container, runs);
+ if roaring_min < delta_size as u64 {
+ let mut bitmap = RoaringTreemap::new();
+ let mut start = rows[0] as u64;
+ let mut end = start;
+ for &row in &rows[1..] {
+ let row = row as u64;
+ if row != end + 1 {
+ bitmap.insert_range(start..=end);
+ start = row;
+ }
+ end = row;
+ }
+ bitmap.insert_range(start..=end);
+ bitmap.optimize();
+ if 1 + bitmap.serialized_size() < delta_size {
+ let mut result = Vec::with_capacity(1 + bitmap.serialized_size());
+ result.push(ROARING);
+ bitmap.serialize_into(&mut result)?;
+ return Ok(result);
+ }
+ }
+ let mut result = Vec::with_capacity(delta_size);
+ result.push(DELTA);
+ encode_var_int(&mut result, count)?;
+ encode_var_long(&mut result, rows[0])?;
+ for pair in rows.windows(2) {
+ encode_var_long(&mut result, pair[1] - pair[0])?;
+ }
+ Ok(result)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn encodings_and_boundaries() {
+ for (rows, tag) in [
+ (vec![i64::MAX], SINGLE),
+ (vec![0, 128, (1i64 << 32) + 5, i64::MAX], DELTA),
+ ((60000..140000).collect(), ROARING),
+ (((1i64 << 32) - 10..(1i64 << 32) + 10000).collect(), ROARING),
+ ] {
+ let expected: RoaringTreemap = rows.iter().map(|id| *id as
u64).collect();
+ for version in [1, 2] {
+ let bytes = serialize(&rows, version).unwrap();
+ if version == 2 {
+ assert_eq!(bytes[0], tag);
+ }
+ let mut actual = RoaringTreemap::new();
+ add_to(&bytes, version, &mut actual).unwrap();
+ assert_eq!(actual, expected);
+ }
+ }
+ }
+
+ #[test]
+ fn rejects_corrupt_postings() {
+ for bytes in [
+ vec![],
+ vec![3],
+ vec![SINGLE],
+ vec![SINGLE, 1, 2],
+ vec![DELTA, 0],
+ vec![DELTA, 1, 0],
+ vec![DELTA, 2, 1, 0],
+ vec![DELTA, 3, 0, 1],
+ vec![ROARING, 0, 0, 0, 0, 0, 0, 0, 0],
+ vec![SINGLE, 255, 255, 255, 255, 255, 255, 255, 255, 255, 1],
+ ] {
+ assert!(
+ add_to(&bytes, 2, &mut RoaringTreemap::new()).is_err(),
+ "{bytes:?}"
+ );
+ }
+ let mut overflow = vec![DELTA, 2];
+ encode_var_long(&mut overflow, i64::MAX).unwrap();
+ overflow.push(1);
+ assert!(add_to(&overflow, 2, &mut RoaringTreemap::new()).is_err());
+ for rows in [vec![], vec![-1], vec![1, 1], vec![2, 1]] {
+ assert!(serialize(&rows, 2).is_err());
+ }
+ }
+
+ #[test]
+ fn rejects_duplicate_unordered_and_unsigned_roaring_partitions() {
+ for keys in [[0, 0], [1, 0], [0, 1 << 31]] {
+ let mut bytes = vec![ROARING];
+ bytes.extend_from_slice(&2u64.to_le_bytes());
+ for (high, row) in keys.into_iter().zip([1, 2]) {
+ bytes.extend_from_slice(&u32::to_le_bytes(high));
+ RoaringBitmap::from_iter([row])
+ .serialize_into(&mut bytes)
+ .unwrap();
+ }
+ assert!(add_to(&bytes, 2, &mut RoaringTreemap::new()).is_err());
+ }
+ }
+}
diff --git a/crates/paimon/src/btree/query.rs b/crates/paimon/src/btree/query.rs
index 2573dc16..cb7688c0 100644
--- a/crates/paimon/src/btree/query.rs
+++ b/crates/paimon/src/btree/query.rs
@@ -189,80 +189,89 @@ pub(crate) struct BetweenInfo<'a> {
pub data_type: &'a DataType,
}
+impl BetweenInfo<'_> {
+ pub(crate) fn is_empty(&self) -> bool {
+ let cmp = crate::btree::make_key_comparator(self.data_type);
+ match cmp(
+ &serialize_datum(self.from, self.data_type),
+ &serialize_datum(self.to, self.data_type),
+ ) {
+ Ordering::Greater => true,
+ Ordering::Equal => !self.from_inclusive || !self.to_inclusive,
+ Ordering::Less => false,
+ }
+ }
+}
+
pub(crate) type ExtractBetweenResult<'a> = (
Option<BetweenInfo<'a>>,
Vec<(PredicateOperator, &'a [Datum], &'a DataType)>,
);
-/// Try to extract a between pattern (lower + upper bound) from predicates.
+/// Tighten all lower/upper bounds on one field into a single range query.
/// Returns (between_info, remaining_predicates).
///
-/// Recognizes two shapes:
-/// 1. A native `Between` leaf with two literals — preferred and emitted by the
-/// DataFusion translator since Stage 3.
-/// 2. A `GtEq` / `Gt` paired with a `LtEq` / `Lt` on the same column —
-/// legacy shape, kept for direct `PredicateBuilder` users that still build
-/// the conjunction explicitly.
+/// Native `Between` leaves and explicit comparisons participate together.
+/// Float comparisons intentionally follow the residual filter's bit-preserving
+/// total order, not bitmap dictionary ordering (which canonicalizes all NaNs).
+/// Bitmap floating ranges conservatively return all non-null candidates.
pub(crate) fn extract_between<'a>(
predicates: &[(PredicateOperator, &'a [Datum], &'a DataType)],
) -> ExtractBetweenResult<'a> {
- // Shape 1: native Between leaf — single tuple is enough.
- for (i, (op, literals, dt)) in predicates.iter().enumerate() {
- if matches!(op, PredicateOperator::Between) && literals.len() == 2 {
- let between = BetweenInfo {
- from: &literals[0],
- to: &literals[1],
- from_inclusive: true,
- to_inclusive: true,
- data_type: dt,
- };
- let remaining: Vec<_> = predicates
- .iter()
- .enumerate()
- .filter(|(j, _)| *j != i)
- .map(|(_, p)| *p)
- .collect();
- return (Some(between), remaining);
- }
- }
-
- if predicates.len() < 2 {
+ let Some((_, _, data_type)) = predicates.first() else {
+ return (None, Vec::new());
+ };
+ if predicates.len() == 1 && predicates[0].0 != PredicateOperator::Between {
return (None, predicates.to_vec());
}
-
- let mut lower: Option<(usize, bool)> = None; // (index, inclusive)
- let mut upper: Option<(usize, bool)> = None;
-
- for (i, (op, literals, _)) in predicates.iter().enumerate() {
- if literals.len() != 1 {
+ let cmp = crate::btree::make_key_comparator(data_type);
+ let mut lower: Option<(&Datum, Vec<u8>, bool)> = None;
+ let mut upper: Option<(&Datum, Vec<u8>, bool)> = None;
+ let mut remaining = Vec::new();
+ for &(op, literals, ty) in predicates {
+ if ty != *data_type {
+ remaining.push((op, literals, ty));
continue;
}
- match op {
- PredicateOperator::GtEq if lower.is_none() => lower = Some((i,
true)),
- PredicateOperator::Gt if lower.is_none() => lower = Some((i,
false)),
- PredicateOperator::LtEq if upper.is_none() => upper = Some((i,
true)),
- PredicateOperator::Lt if upper.is_none() => upper = Some((i,
false)),
- _ => {}
+ let (from, to) = match (op, literals) {
+ (PredicateOperator::GtEq, [value]) => (Some((value, true)), None),
+ (PredicateOperator::Gt, [value]) => (Some((value, false)), None),
+ (PredicateOperator::LtEq, [value]) => (None, Some((value, true))),
+ (PredicateOperator::Lt, [value]) => (None, Some((value, false))),
+ (PredicateOperator::Between, [from, to]) => (Some((from, true)),
Some((to, true))),
+ _ => {
+ remaining.push((op, literals, ty));
+ continue;
+ }
+ };
+ for (candidate, bound, tighter) in [
+ (from, &mut lower, Ordering::Greater),
+ (to, &mut upper, Ordering::Less),
+ ] {
+ if let Some((value, inclusive)) = candidate {
+ let key = serialize_datum(value, data_type);
+ match bound {
+ Some((_, existing, current_inclusive)) => match cmp(&key,
existing) {
+ Ordering::Equal => *current_inclusive &= inclusive,
+ order if order == tighter => *bound = Some((value,
key, inclusive)),
+ _ => {}
+ },
+ None => *bound = Some((value, key, inclusive)),
+ }
+ }
}
}
-
match (lower, upper) {
- (Some((li, from_inclusive)), Some((ui, to_inclusive))) => {
- let between = BetweenInfo {
- from: &predicates[li].1[0],
- to: &predicates[ui].1[0],
+ (Some((from, _, from_inclusive)), Some((to, _, to_inclusive))) => (
+ Some(BetweenInfo {
+ from,
+ to,
from_inclusive,
to_inclusive,
- data_type: predicates[li].2,
- };
- let remaining: Vec<_> = predicates
- .iter()
- .enumerate()
- .filter(|(i, _)| *i != li && *i != ui)
- .map(|(_, p)| *p)
- .collect();
- (Some(between), remaining)
- }
+ data_type,
+ }),
+ remaining,
+ ),
_ => (None, predicates.to_vec()),
}
}
diff --git a/crates/paimon/src/btree/reader.rs
b/crates/paimon/src/btree/reader.rs
index 7f8f1ba1..1debf239 100644
--- a/crates/paimon/src/btree/reader.rs
+++ b/crates/paimon/src/btree/reader.rs
@@ -27,13 +27,13 @@ use crate::btree::block::{BlockHandle, BlockReader};
use crate::btree::bloom_filter::BloomFilter;
use crate::btree::footer::{BTreeFileFooter, BloomFilterHandle,
BTREE_FOOTER_ENCODED_LENGTH};
use crate::btree::meta::BTreeIndexMeta;
+use crate::btree::posting_list;
use crate::btree::sst_file::{read_block_from_bytes, SstFileReader};
-use crate::btree::var_len::{decode_var_int, decode_var_long};
use crate::io::FileRead;
use crate::spec::murmur_hash::hash_bytes;
use roaring::RoaringTreemap;
use std::cmp::Ordering;
-use std::io::{self, Cursor};
+use std::io;
use tokio::sync::OnceCell;
struct LazyBloomFilter {
@@ -50,6 +50,7 @@ pub struct BTreeIndexReader<F: Fn(&[u8], &[u8]) -> Ordering> {
max_key: Option<Vec<u8>>,
key_comparator: F,
bloom_filter: Option<LazyBloomFilter>,
+ file_version: u32,
}
impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
@@ -105,6 +106,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
max_key: meta.last_key.clone(),
key_comparator,
bloom_filter,
+ file_version: footer.version,
})
}
@@ -149,7 +151,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
let (key, value, next_offset) = block.read_entry_at(offset);
offset = next_offset;
if predicate(key) {
- insert_row_ids_into(value, &mut result)?;
+ posting_list::add_to(value, self.file_version, &mut
result)?;
}
}
}
@@ -169,6 +171,12 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
let cmp = &self.key_comparator;
let mut result = RoaringTreemap::new();
+ match cmp(from, to) {
+ Ordering::Greater => return Ok(result),
+ Ordering::Equal if !from_inclusive || !to_inclusive => return
Ok(result),
+ _ => {}
+ }
+
// Seek in index block to find the first data block that may contain
`from`
let index_block = self.sst_reader.index_block();
let (_, mut index_iter) = index_block.seek_and_iter(from, cmp);
@@ -246,7 +254,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
return Ok(true);
}
- insert_row_ids_into(value, result)?;
+ posting_list::add_to(value, self.file_version, result)?;
}
Ok(false)
}
@@ -334,7 +342,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
let (found, mut entry_iter) = block.seek_and_iter(key, cmp);
let mut result = RoaringTreemap::new();
if let (true, Some((_entry_key, value))) = (found, entry_iter.next()) {
- insert_row_ids_into(value, &mut result)?;
+ posting_list::add_to(value, self.file_version, &mut result)?;
}
Ok(result)
}
@@ -457,7 +465,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
for key in block_keys {
let (found, mut entry_iter) = block.seek_and_iter(key, cmp);
if let (true, Some((_entry_key, value))) = (found,
entry_iter.next()) {
- insert_row_ids_into(value, &mut result)?;
+ posting_list::add_to(value, self.file_version, &mut
result)?;
}
}
}
@@ -511,19 +519,3 @@ fn verify_null_bitmap_crc(bitmap_bytes: &[u8], crc_bytes:
&[u8]) -> io::Result<(
}
Ok(())
}
-
-/// Deserialize row ids from value bytes and insert directly into bitmap.
-fn insert_row_ids_into(data: &[u8], bitmap: &mut RoaringTreemap) ->
io::Result<()> {
- let mut cursor = Cursor::new(data);
- let count = decode_var_int(&mut cursor)?;
- if count < 0 {
- return Err(io::Error::new(
- io::ErrorKind::InvalidData,
- format!("Invalid row id count: {count}"),
- ));
- }
- for _ in 0..count {
- bitmap.insert(decode_var_long(&mut cursor)? as u64);
- }
- Ok(())
-}
diff --git a/crates/paimon/src/btree/tests.rs b/crates/paimon/src/btree/tests.rs
index 89ac793b..9e9d6c92 100644
--- a/crates/paimon/src/btree/tests.rs
+++ b/crates/paimon/src/btree/tests.rs
@@ -94,10 +94,346 @@ async fn test_write_read_roundtrip() {
assert_eq!(all.len(), 100);
}
+#[tokio::test]
+async fn test_tightest_bounds_across_key_types_and_versions() {
+ use crate::btree::query::extract_between;
+ use crate::btree::{make_key_comparator, serialize_datum};
+ use crate::spec::*;
+ let cases = vec![
+ (
+ DataType::TinyInt(TinyIntType::new()),
+ (0..6).map(Datum::TinyInt).collect::<Vec<_>>(),
+ ),
+ (
+ DataType::SmallInt(SmallIntType::new()),
+ (0..6).map(Datum::SmallInt).collect(),
+ ),
+ (
+ DataType::Int(IntType::new()),
+ (0..6).map(Datum::Int).collect(),
+ ),
+ (
+ DataType::BigInt(BigIntType::new()),
+ (0..6).map(Datum::Long).collect(),
+ ),
+ (
+ DataType::Float(FloatType::new()),
+ (0..6).map(|v| Datum::Float(v as f32)).collect(),
+ ),
+ (
+ DataType::Double(DoubleType::new()),
+ (0..6).map(|v| Datum::Double(v as f64)).collect(),
+ ),
+ (
+ DataType::VarChar(VarCharType::string_type()),
+ (0..6).map(|v| Datum::String(v.to_string())).collect(),
+ ),
+ (
+ DataType::Date(DateType::new()),
+ (0..6).map(Datum::Date).collect(),
+ ),
+ (
+ DataType::Time(TimeType::new(3).unwrap()),
+ (0..6).map(Datum::Time).collect(),
+ ),
+ (
+ DataType::Timestamp(TimestampType::new(9).unwrap()),
+ (0..6)
+ .map(|v| Datum::Timestamp {
+ millis: 0,
+ nanos: v,
+ })
+ .collect(),
+ ),
+ (
+ DataType::Decimal(DecimalType::new(10, 0).unwrap()),
+ (0..6)
+ .map(|v| Datum::Decimal {
+ unscaled: v,
+ precision: 10,
+ scale: 0,
+ })
+ .collect(),
+ ),
+ (
+ DataType::Decimal(DecimalType::new(30, 0).unwrap()),
+ (0..6)
+ .map(|v| Datum::Decimal {
+ unscaled: v,
+ precision: 30,
+ scale: 0,
+ })
+ .collect(),
+ ),
+ ];
+ for (ty, values) in cases {
+ for version in [1, 2] {
+ let buf = VecFileWrite::new();
+ let mut writer =
+ BTreeIndexWriter::new(Box::new(buf.clone()), 64,
BlockCompressionType::None)
+ .with_file_version(version)
+ .unwrap();
+ writer.write(None, 6).await.unwrap();
+ for (row, value) in values.iter().enumerate() {
+ writer
+ .write(Some(&serialize_datum(value, &ty)), row as i64)
+ .await
+ .unwrap();
+ }
+ let result = writer.finish().await.unwrap();
+ let reader = write_and_open(&buf, &result,
make_key_comparator(&ty)).await;
+ for reverse in [false, true] {
+ // A native Between plus multiple pairs, with strictness ties.
+ let native = vec![values[0].clone(), values[5].clone()];
+ let mut predicates = vec![
+ (PredicateOperator::Between, native.as_slice(), &ty),
+ (PredicateOperator::GtEq, &values[1..2], &ty),
+ (PredicateOperator::LtEq, &values[4..5], &ty),
+ (PredicateOperator::Gt, &values[1..2], &ty),
+ (PredicateOperator::Lt, &values[4..5], &ty),
+ (PredicateOperator::IsNotNull, &values[0..0], &ty),
+ ];
+ if reverse {
+ predicates.reverse();
+ }
+ let (range, remaining) = extract_between(&predicates);
+ assert_eq!(remaining.len(), 1);
+ assert_eq!(remaining[0].0, PredicateOperator::IsNotNull);
+ let range = range.unwrap();
+ assert!(!range.is_empty());
+ let actual = reader
+ .range_query(
+ &serialize_datum(range.from, &ty),
+ &serialize_datum(range.to, &ty),
+ range.from_inclusive,
+ range.to_inclusive,
+ )
+ .await
+ .unwrap();
+ assert_eq!(
+ actual.iter().collect::<Vec<_>>(),
+ vec![2, 3],
+ "{ty:?}, V{version}"
+ );
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn test_empty_range_does_not_read_data_blocks() {
+ let buf = VecFileWrite::new();
+ let mut writer = BTreeIndexWriter::new(Box::new(buf.clone()), 64,
BlockCompressionType::None);
+ for row in 0..100 {
+ writer.write(Some(&int_key(row)), row as i64).await.unwrap();
+ }
+ let result = writer.finish().await.unwrap();
+ let data = Bytes::from(buf.to_vec());
+ let reads = Arc::new(Mutex::new(Vec::new()));
+ let reader = BTreeIndexReader::open(
+ Box::new(RecordingFileRead {
+ data: data.clone(),
+ ranges: reads.clone(),
+ }),
+ data.len() as u64,
+ &result.meta,
+ int_cmp,
+ )
+ .await
+ .unwrap();
+ reads.lock().unwrap().clear();
+ for (from, to, lower, upper) in [
+ (50, 40, true, true),
+ (50, 50, false, true),
+ (50, 50, true, false),
+ (50, 50, false, false),
+ ] {
+ assert!(reader
+ .range_query(&int_key(from), &int_key(to), lower, upper)
+ .await
+ .unwrap()
+ .is_empty());
+ }
+ assert!(reads.lock().unwrap().is_empty());
+}
+
fn read_footer(data: &[u8]) -> BTreeFileFooter {
BTreeFileFooter::read_footer(&data[data.len() -
BTREE_FOOTER_ENCODED_LENGTH..]).unwrap()
}
+#[tokio::test]
+async fn test_java_v2_posting_encodings() {
+ // Existing low-level tests use arbitrary big-endian keys. Java's INT key
+ // serializer uses little-endian bytes, so use the production codec here.
+ let data_type = DataType::Int(crate::spec::IntType::new());
+ let int_key = |v| crate::btree::serialize_datum(&Datum::Int(v),
&data_type);
+ for fixture in [
+
include_bytes!("../../testdata/btree/btree_v2_java_none.bin").as_slice(),
+
include_bytes!("../../testdata/btree/btree_v2_java_lz4.bin").as_slice(),
+ ] {
+ let reader = BTreeIndexReader::open(
+ Box::new(BytesFileRead(Bytes::copy_from_slice(fixture))),
+ fixture.len() as u64,
+ &BTreeIndexMeta::new(Some(int_key(0)), Some(int_key(5)), true),
+ crate::btree::make_key_comparator(&data_type),
+ )
+ .await
+ .unwrap();
+ let expected = [
+ [7].into_iter().collect::<roaring::RoaringTreemap>(),
+ [2, 130, 65538, (1u64 << 32) + 5].into_iter().collect(),
+ (60000..140000).collect(),
+ ((1u64 << 32) - 10..(1u64 << 32) + 10000).collect(),
+ [i64::MAX as u64].into_iter().collect(),
+ (0..4096).map(|id| (1u64 << 40) + id * 128).collect(),
+ ];
+ for (key, rows) in expected.iter().enumerate() {
+ assert_eq!(
+ reader.query_equal(&int_key(key as i32)).await.unwrap(),
+ *rows
+ );
+ }
+ assert_eq!(
+ reader.null_bitmap().iter().collect::<Vec<_>>(),
+ vec![4, (1u64 << 32) + 20000]
+ );
+ assert_eq!(
+ reader.query_in(&[&int_key(1), &int_key(3)]).await.unwrap(),
+ &expected[1] | &expected[3]
+ );
+ assert_eq!(
+ reader
+ .range_query(&int_key(1), &int_key(3), false, false)
+ .await
+ .unwrap(),
+ expected[2]
+ );
+ assert_eq!(
+ reader.scan_entries(|key| key == int_key(2)).await.unwrap(),
+ expected[2]
+ );
+ }
+}
+
+#[tokio::test]
+async fn test_v2_writer_matches_java_fixture() {
+ let data_type = DataType::Int(crate::spec::IntType::new());
+ let int_key = |v| crate::btree::serialize_datum(&Datum::Int(v),
&data_type);
+ let buf = VecFileWrite::new();
+ let mut writer = BTreeIndexWriter::with_comparator(
+ Box::new(buf.clone()),
+ 64,
+ BlockCompressionType::None,
+ crate::btree::make_key_comparator(&data_type),
+ )
+ .with_file_version(2)
+ .unwrap();
+ writer.write(Some(&int_key(0)), 7).await.unwrap();
+ for row in [2, 130, 65538, (1i64 << 32) + 5] {
+ writer.write(Some(&int_key(1)), row).await.unwrap();
+ }
+ for row in 60000..140000 {
+ writer.write(Some(&int_key(2)), row).await.unwrap();
+ }
+ for row in (1i64 << 32) - 10..(1i64 << 32) + 10000 {
+ writer.write(Some(&int_key(3)), row).await.unwrap();
+ }
+ writer.write(Some(&int_key(4)), i64::MAX).await.unwrap();
+ for id in 0..4096 {
+ writer
+ .write(Some(&int_key(5)), (1i64 << 40) + id * 128)
+ .await
+ .unwrap();
+ }
+ writer.write(None, 4).await.unwrap();
+ writer.write(None, (1i64 << 32) + 20000).await.unwrap();
+ writer.finish().await.unwrap();
+ assert_eq!(
+ buf.to_vec(),
+ include_bytes!("../../testdata/btree/btree_v2_java_none.bin")
+ );
+ assert!(
+ include_bytes!("../../testdata/btree/btree_v2_java_lz4.bin").len()
+ <
include_bytes!("../../testdata/btree/btree_v2_java_none.bin").len()
+ );
+}
+
+#[test]
+fn test_unsupported_version_is_rejected() {
+ for version in [0u32, 3, u32::MAX] {
+ let mut bytes =
include_bytes!("../../testdata/btree/btree_v2_java_none.bin").to_vec();
+ let pos = bytes.len() - 8;
+ bytes[pos..pos + 4].copy_from_slice(&version.to_le_bytes());
+ assert!(BTreeFileFooter::read_footer(&bytes[bytes.len() -
52..]).is_err());
+ assert!(BTreeIndexWriter::new(
+ Box::new(VecFileWrite::new()),
+ 64,
+ BlockCompressionType::None
+ )
+ .with_file_version(version)
+ .is_err());
+ }
+}
+
+#[tokio::test]
+#[ignore = "manual release benchmark; no wall-clock assertions"]
+async fn benchmark_btree_v1_v2_postings() {
+ use std::hint::black_box;
+ use std::time::Instant;
+ let count = 1_200_000;
+ let mut readers = Vec::new();
+ for version in [1, 2] {
+ let buf = VecFileWrite::new();
+ let mut writer =
+ BTreeIndexWriter::new(Box::new(buf.clone()), 64 * 1024,
BlockCompressionType::None)
+ .with_file_version(version)
+ .unwrap();
+ for key in [0, 1] {
+ for id in 0..count {
+ if (id % 10 != 0) as i32 == key {
+ writer.write(Some(&int_key(key)), id).await.unwrap();
+ }
+ }
+ }
+ writer.write(Some(&int_key(2)), count).await.unwrap();
+ let result = writer.finish().await.unwrap();
+ let reader = write_and_open(&buf, &result, int_cmp).await;
+ println!(
+ "BTREE_BENCH version={version} file_bytes={}",
+ buf.to_vec().len()
+ );
+ readers.push(reader);
+ }
+ for (key, expected) in [(0, count / 10), (1, count * 9 / 10), (2, 1)] {
+ for reader in &readers {
+ for _ in 0..3 {
+ assert_eq!(
+ reader.query_equal(&int_key(key)).await.unwrap().len(),
+ expected as u64
+ );
+ }
+ }
+ let mut times = [Vec::new(), Vec::new()];
+ for round in 0..7 {
+ for version in [round % 2, 1 - round % 2] {
+ let start = Instant::now();
+ let rows =
readers[version].query_equal(&int_key(key)).await.unwrap();
+ let elapsed = start.elapsed().as_secs_f64() * 1000.0;
+ assert_eq!(rows.len(), expected as u64);
+ black_box(rows);
+ times[version].push(elapsed);
+ }
+ }
+ for samples in &mut times {
+ samples.sort_by(f64::total_cmp);
+ }
+ println!(
+ "BTREE_BENCH key={key} rows={expected} v1_ms={:.3} v2_ms={:.3}",
+ times[0][3], times[1][3]
+ );
+ }
+}
+
#[tokio::test]
async fn test_bloom_filter_is_disabled_by_default() {
let buf = VecFileWrite::new();
diff --git a/crates/paimon/src/btree/writer.rs
b/crates/paimon/src/btree/writer.rs
index 6743174f..0ded1392 100644
--- a/crates/paimon/src/btree/writer.rs
+++ b/crates/paimon/src/btree/writer.rs
@@ -24,8 +24,8 @@
use crate::btree::block::{BlockCompressionType, BlockHandle};
use crate::btree::footer::BTreeFileFooter;
use crate::btree::meta::BTreeIndexMeta;
+use crate::btree::posting_list;
use crate::btree::sst_file::SstFileWriter;
-use crate::btree::var_len::{encode_var_int, encode_var_long};
use crate::io::FileWrite;
use roaring::RoaringTreemap;
use std::cmp::Ordering;
@@ -46,6 +46,7 @@ pub struct BTreeIndexWriter<F: Fn(&[u8], &[u8]) -> Ordering> {
null_bitmap: Option<RoaringTreemap>,
row_count: u64,
key_comparator: F,
+ file_version: u32,
}
/// Result of finishing a BTree index write.
@@ -84,6 +85,7 @@ impl BTreeIndexWriter<fn(&[u8], &[u8]) -> Ordering> {
null_bitmap: None,
row_count: 0,
key_comparator: |a, b| a.cmp(b),
+ file_version: 1,
}
}
}
@@ -139,13 +141,34 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexWriter<F>
{
null_bitmap: None,
row_count: 0,
key_comparator: cmp,
+ file_version: 1,
}
}
+ /// Select the on-disk version before writing any rows. V1 remains the
default;
+ /// V2 requires readers that support adaptive postings.
+ pub fn with_file_version(mut self, version: u32) -> io::Result<Self> {
+ if !matches!(version, 1 | 2) || self.row_count != 0 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ "BTree file version must be 1 or 2 and set before writing",
+ ));
+ }
+ self.file_version = version;
+ Ok(self)
+ }
+
/// Write a key and its associated row id.
/// If key is None, the row id is added to the null bitmap.
/// Keys must be written in sorted order; entries with the same key are
combined.
+ /// In V2, row IDs for the same key must also be strictly increasing.
pub async fn write(&mut self, key: Option<&[u8]>, row_id: i64) ->
io::Result<()> {
+ if row_id < 0 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ "BTree row ID must be non-negative",
+ ));
+ }
self.row_count += 1;
match key {
@@ -177,12 +200,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexWriter<F> {
return Ok(());
}
- // Serialize row id list: var_len_int(count) + var_len_long(id) * count
- let mut value_buf = Vec::with_capacity(self.current_row_ids.len() * 9
+ 5);
- encode_var_int(&mut value_buf, self.current_row_ids.len() as i32)?;
- for &row_id in &self.current_row_ids {
- encode_var_long(&mut value_buf, row_id)?;
- }
+ let value_buf = posting_list::serialize(&self.current_row_ids,
self.file_version)?;
self.current_row_ids.clear();
if let Some(ref key) = self.last_key {
@@ -237,8 +255,9 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexWriter<F> {
let index_block_handle = self.sst_writer.write_index_block().await?;
// Write footer
- let footer =
+ let mut footer =
BTreeFileFooter::new(bloom_filter_handle, index_block_handle,
null_bitmap_handle);
+ footer.version = self.file_version;
let footer_bytes = footer.write_footer();
self.sst_writer.write_raw(&footer_bytes).await?;
diff --git a/crates/paimon/src/table/global_index_scanner.rs
b/crates/paimon/src/table/global_index_scanner.rs
index c61f3825..5fbeffa6 100644
--- a/crates/paimon/src/table/global_index_scanner.rs
+++ b/crates/paimon/src/table/global_index_scanner.rs
@@ -20,6 +20,7 @@
//!
//! Reference:
[org.apache.paimon.index.GlobalIndexScanner](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexScanner.java)
+mod all_match;
mod deletion_vectors;
mod entry;
mod evaluator;
@@ -62,6 +63,8 @@ const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS";
struct QueryIoProbe {
active: TestAtomicUsize,
peak: TestAtomicUsize,
+ predicate_queries: TestAtomicUsize,
+ range_queries: TestAtomicUsize,
}
#[cfg(test)]
@@ -262,6 +265,7 @@ impl GlobalIndexScanner {
file_name: entry.index_file.file_name.clone(),
index_type: kind,
file_size: entry.index_file.file_size,
+ row_count: entry.index_file.row_count,
row_range_start: global_meta.row_range_start,
row_range_end: global_meta.row_range_end,
external_path: entry.index_file.external_path.clone(),
diff --git a/crates/paimon/src/table/global_index_scanner/all_match.rs
b/crates/paimon/src/table/global_index_scanner/all_match.rs
new file mode 100644
index 00000000..57da267b
--- /dev/null
+++ b/crates/paimon/src/table/global_index_scanner/all_match.rs
@@ -0,0 +1,103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Prove complete scalar BTree domains match without opening index files.
+
+use super::entry::{sorted_entry_meta, GlobalIndexEntry, GlobalIndexFileKind};
+use crate::btree::{make_key_comparator, serialize_datum};
+use crate::spec::{DataType, Datum, PredicateOperator};
+use std::cmp::Ordering::{Equal, Greater, Less};
+use std::collections::HashMap;
+
+pub(super) fn all_matching_entries(
+ entries: &[&GlobalIndexEntry],
+ predicates: &[(PredicateOperator, &[Datum], &DataType)],
+) -> Vec<Vec<bool>> {
+ let mut groups: HashMap<(i64, i64), Vec<usize>> = HashMap::new();
+ for (index, entry) in entries.iter().enumerate() {
+ if entry.index_type == GlobalIndexFileKind::BTree {
+ groups
+ .entry((entry.row_range_start, entry.row_range_end))
+ .or_default()
+ .push(index);
+ }
+ }
+ let comparisons: Vec<_> = predicates
+ .iter()
+ .map(|(op, literals, ty)| {
+ (
+ *op,
+ literals
+ .iter()
+ .map(|literal| serialize_datum(literal, ty))
+ .collect::<Vec<_>>(),
+ make_key_comparator(ty),
+ )
+ })
+ .collect();
+ let mut result = vec![vec![false; entries.len()]; predicates.len()];
+ for ((start, end), files) in groups {
+ let Some(expected_count) = end
+ .checked_sub(start)
+ .and_then(|len| len.checked_add(1))
+ .filter(|len| *len > 0 && start >= 0)
+ else {
+ continue;
+ };
+ // Multiple files can partition keys within the SAME source row domain.
+ // A single file's min/max never proves that it contains every source
row.
+ let count = files.iter().try_fold(0i64, |count, &index| {
+ let rows = entries[index].row_count;
+ (rows > 0).then(|| count.checked_add(rows)).flatten()
+ });
+ if count != Some(expected_count) {
+ continue;
+ }
+ for (predicate_index, (op, values, cmp)) in
comparisons.iter().enumerate() {
+ let all_match = files.iter().all(|&index| {
+ let meta = sorted_entry_meta(entries[index]);
+ let (Some(first), Some(last)) = (&meta.first_key,
&meta.last_key) else {
+ return false;
+ };
+ !meta.has_nulls && {
+ if cmp(first, last) == Greater {
+ return false;
+ }
+ match (op, values.as_slice()) {
+ (PredicateOperator::Eq, [value]) => {
+ cmp(first, value) == Equal && cmp(last, value) ==
Equal
+ }
+ (PredicateOperator::Lt, [value]) => cmp(last, value)
== Less,
+ (PredicateOperator::LtEq, [value]) => cmp(last, value)
!= Greater,
+ (PredicateOperator::Gt, [value]) => cmp(first, value)
== Greater,
+ (PredicateOperator::GtEq, [value]) => cmp(first,
value) != Less,
+ (PredicateOperator::Between, [from, to]) => {
+ cmp(first, from) != Less && cmp(last, to) !=
Greater
+ }
+ _ => false,
+ }
+ }
+ });
+ if all_match {
+ for &index in &files {
+ result[predicate_index][index] = true;
+ }
+ }
+ }
+ }
+ result
+}
diff --git a/crates/paimon/src/table/global_index_scanner/entry.rs
b/crates/paimon/src/table/global_index_scanner/entry.rs
index 8fb3fc8e..d402f69b 100644
--- a/crates/paimon/src/table/global_index_scanner/entry.rs
+++ b/crates/paimon/src/table/global_index_scanner/entry.rs
@@ -30,6 +30,7 @@ pub(super) struct GlobalIndexEntry {
pub(super) file_name: String,
pub(super) index_type: GlobalIndexFileKind,
pub(super) file_size: i64,
+ pub(super) row_count: i64,
pub(super) row_range_start: i64,
pub(super) row_range_end: i64,
pub(super) external_path: Option<String>,
diff --git a/crates/paimon/src/table/global_index_scanner/evaluator.rs
b/crates/paimon/src/table/global_index_scanner/evaluator.rs
index 9b2daea1..b2ef50e5 100644
--- a/crates/paimon/src/table/global_index_scanner/evaluator.rs
+++ b/crates/paimon/src/table/global_index_scanner/evaluator.rs
@@ -17,6 +17,7 @@
//! Predicate orchestration and index-query planning.
+use super::all_match::all_matching_entries;
use super::entry::{
bitmap_meta_may_match, bitmap_meta_may_match_between,
multivalue_meta_may_match,
sorted_entry_meta, GlobalIndexEntry, GlobalIndexFileKind,
@@ -43,7 +44,6 @@ use crate::table::bitmap_global_index_format::{
use crate::table::RowRange;
use crate::{Error, Result};
use futures::{StreamExt, TryStreamExt};
-use roaring::RoaringTreemap;
use std::collections::HashSet;
use std::future::Future;
@@ -299,15 +299,60 @@ impl GlobalIndexScanner {
})
.collect::<Result<Vec<_>>>()?;
let predicates = normalized_predicates.as_slice();
-
// Try to detect between pattern and split into (between, remaining)
let (between, remaining) = extract_between(predicates);
+ let coverage = || {
+ crate::table::merge_row_ranges(
+ entries
+ .iter()
+ .map(|entry| RowRange::new(entry.row_range_start,
entry.row_range_end))
+ .collect(),
+ )
+ };
+ if between.as_ref().is_some_and(|range| range.is_empty()) {
+ return Ok(Some((Vec::new(), coverage())));
+ }
+
let effective_predicates = if between.is_some() {
&remaining
} else {
predicates
};
+ // A redundant conjunct is an identity independently of the other
+ // predicates on this field. Do not decode its entire posting list.
+ let predicate_all_matches = all_matching_entries(entries,
effective_predicates);
+ let between_all_matches = between
+ .as_ref()
+ .map(|range| {
+ let endpoints = [
+ (
+ if range.from_inclusive {
+ PredicateOperator::GtEq
+ } else {
+ PredicateOperator::Gt
+ },
+ std::slice::from_ref(range.from),
+ range.data_type,
+ ),
+ (
+ if range.to_inclusive {
+ PredicateOperator::LtEq
+ } else {
+ PredicateOperator::Lt
+ },
+ std::slice::from_ref(range.to),
+ range.data_type,
+ ),
+ ];
+ let matches = all_matching_entries(entries, &endpoints);
+ matches[0]
+ .iter()
+ .zip(&matches[1])
+ .map(|(lower, upper)| *lower && *upper)
+ .collect::<Vec<_>>()
+ })
+ .unwrap_or_else(|| vec![false; entries.len()]);
// Pre-compute comparators and serialized keys for file-level pruning
per predicate
let pruning_info: Vec<_> = effective_predicates
@@ -367,8 +412,16 @@ impl GlobalIndexScanner {
.iter()
.enumerate()
.map(|(i, (op, _, _))| {
- requires_fallback_scan(*op)
- .then(|| self.fallback_scan_plan(entries,
&predicate_matches[i]))
+ requires_fallback_scan(*op).then(|| {
+ self.fallback_scan_plan(
+ entries,
+ &predicate_matches[i]
+ .iter()
+ .zip(&predicate_all_matches[i])
+ .map(|(matches, all)| *matches && !all)
+ .collect::<Vec<_>>(),
+ )
+ })
})
.collect();
@@ -399,19 +452,36 @@ impl GlobalIndexScanner {
}
None => Vec::new(),
};
- let between_fallback_plan = between
- .as_ref()
- .map(|_| self.fallback_scan_plan(entries,
&between_matches_by_entry));
+ let between_fallback_plan = between.as_ref().map(|_| {
+ self.fallback_scan_plan(
+ entries,
+ &between_matches_by_entry
+ .iter()
+ .zip(&between_all_matches)
+ .map(|(matches, all)| *matches && !all)
+ .collect::<Vec<_>>(),
+ )
+ });
let mut query_plans = Vec::with_capacity(entries.len());
+ let mut all_matching_ranges = Vec::new();
for (entry_idx, entry) in entries.iter().enumerate() {
+ if (between.is_none() || between_all_matches[entry_idx])
+ && predicate_all_matches
+ .iter()
+ .all(|matches| matches[entry_idx])
+ {
+ all_matching_ranges.push(RowRange::new(entry.row_range_start,
entry.row_range_end));
+ continue;
+ }
// Also check if between range may match
let between_matches = between
.as_ref()
.is_some_and(|_| between_matches_by_entry[entry_idx]);
- let between_evaluated_for_entry =
between_fallback_plan.is_some_and(|plan| {
- fallback_plan_evaluates_entry(plan, entry.index_type,
between_matches)
- });
+ let between_evaluated_for_entry = between_all_matches[entry_idx]
+ || between_fallback_plan.is_some_and(|plan| {
+ fallback_plan_evaluates_entry(plan, entry.index_type,
between_matches)
+ });
// When a Between conjunct exists but the file does not overlap its
// range, the whole AND cannot match — drop the file regardless of
@@ -430,6 +500,10 @@ impl GlobalIndexScanner {
between_matches && !between_evaluated_for_entry &&
between_fallback_plan.is_some();
let matching_predicates: Vec<usize> =
(0..effective_predicates.len())
.filter(|&i| {
+ if predicate_all_matches[i][entry_idx] {
+ file_evaluated = true;
+ return false;
+ }
let predicate_matches_entry =
predicate_matches[i][entry_idx];
let predicate_evaluated_for_entry =
predicate_fallback_plans[i].is_none_or(|plan| {
@@ -461,10 +535,18 @@ impl GlobalIndexScanner {
continue;
}
+ let query_between = between_evaluated_for_entry &&
!between_all_matches[entry_idx];
+ if !query_between && matching_predicates.is_empty() {
+ // Only a proven identity remains (other conjuncts may have
+ // declined their budget). Preserve its indexed domain and
+ // leave the residual filter to the ordinary table reader.
+ all_matching_ranges.push(RowRange::new(entry.row_range_start,
entry.row_range_end));
+ continue;
+ }
query_plans.push(EntryQueryPlan {
entry_idx,
between_matches,
- between_evaluated: between_evaluated_for_entry,
+ between_evaluated: query_between,
matching_predicates,
});
}
@@ -495,21 +577,30 @@ impl GlobalIndexScanner {
let result = self
.query_entry(entry, data_type, between, &plan,
effective_predicates)
.await?;
- Ok((entry.row_range_start, result))
+ let ranges = result.bitmap.as_ref().map(|bitmap| {
+ bitmap_to_ranges(bitmap)
+ .into_iter()
+ .map(|range| {
+ RowRange::new(
+ range.from() + entry.row_range_start,
+ range.to() + entry.row_range_start,
+ )
+ })
+ .collect::<Vec<_>>()
+ });
+ Ok((ranges, result.declined))
});
- let (all_row_ids, declined) = try_fold_bounded(
+ let (all_ranges, declined) = try_fold_bounded(
futures,
self.global_index_thread_num,
- (RoaringTreemap::new(), false),
- |(all_row_ids, declined), (row_range_start, file_result)| {
- if file_result.declined {
+ (all_matching_ranges, false),
+ |(all_ranges, declined), (ranges, file_declined)| {
+ if file_declined {
*declined = true;
return;
}
- if let Some(bitmap) = file_result.bitmap {
- for row_id in bitmap.iter() {
- all_row_ids.insert(row_id + row_range_start as u64);
- }
+ if let Some(ranges) = ranges {
+ all_ranges.extend(ranges);
}
},
)
@@ -519,13 +610,10 @@ impl GlobalIndexScanner {
return Ok(None);
}
- let coverage = crate::table::merge_row_ranges(
- entries
- .iter()
- .map(|entry| RowRange::new(entry.row_range_start,
entry.row_range_end))
- .collect(),
- );
- Ok(Some((bitmap_to_ranges(&all_row_ids), coverage)))
+ Ok(Some((
+ crate::table::merge_row_ranges(all_ranges),
+ coverage(),
+ )))
}
fn find_field_id_by_name(&self, column: &str) -> Result<Option<i32>> {
diff --git a/crates/paimon/src/table/global_index_scanner/reader.rs
b/crates/paimon/src/table/global_index_scanner/reader.rs
index b74b4cde..246eeccd 100644
--- a/crates/paimon/src/table/global_index_scanner/reader.rs
+++ b/crates/paimon/src/table/global_index_scanner/reader.rs
@@ -113,6 +113,12 @@ impl GlobalIndexScanner {
let mut file_result = None;
if plan.between_matches && plan.between_evaluated {
+ #[cfg(test)]
+ if let Some(probe) = &self.query_io_probe {
+ probe
+ .range_queries
+ .fetch_add(1, super::TestOrdering::SeqCst);
+ }
let between = between.expect("evaluated between query is present");
let serialize_key = match entry.index_type {
GlobalIndexFileKind::BTree => serialize_datum,
@@ -139,6 +145,12 @@ impl GlobalIndexScanner {
}
for &idx in &plan.matching_predicates {
+ #[cfg(test)]
+ if let Some(probe) = &self.query_io_probe {
+ probe
+ .predicate_queries
+ .fetch_add(1, super::TestOrdering::SeqCst);
+ }
let (op, literals, data_type) = &effective_predicates[idx];
let Some(bitmap) = reader
.as_ref()
diff --git a/crates/paimon/src/table/global_index_scanner/row_ranges.rs
b/crates/paimon/src/table/global_index_scanner/row_ranges.rs
index abfe5df7..a306d88a 100644
--- a/crates/paimon/src/table/global_index_scanner/row_ranges.rs
+++ b/crates/paimon/src/table/global_index_scanner/row_ranges.rs
@@ -24,32 +24,30 @@ use std::collections::{HashMap, HashSet};
/// Convert a RoaringTreemap to merged RowRanges (already sorted and
deduplicated).
pub(super) fn bitmap_to_ranges(bitmap: &RoaringTreemap) -> Vec<RowRange> {
- if bitmap.is_empty() {
- return Vec::new();
- }
- let mut ranges = Vec::new();
- let mut iter = bitmap.iter();
- let first = iter.next().unwrap();
- let mut start = first as i64;
- let mut end = start;
-
- for id in iter {
- let id = id as i64;
- if id == end + 1 {
- end = id;
- } else {
+ let mut ranges: Vec<RowRange> = Vec::new();
+ // Walk containers/runs instead of every set bit. Adjacent high-32
partitions
+ // must still be joined to preserve the canonical range layout.
+ for (high, low) in bitmap.bitmaps() {
+ let base = u64::from(high) << 32;
+ let mut iter = low.iter();
+ while let Some(range) = iter.next_range() {
+ let start = (base | u64::from(*range.start())) as i64;
+ let end = (base | u64::from(*range.end())) as i64;
+ if let Some(last) = ranges.last_mut() {
+ if last.to().checked_add(1) == Some(start) {
+ *last = RowRange::new(last.from(), end);
+ continue;
+ }
+ }
ranges.push(RowRange::new(start, end));
- start = id;
- end = id;
}
}
- ranges.push(RowRange::new(start, end));
ranges
}
/// Intersect two sorted range lists using RowRangeIndex for efficient binary
search.
pub(super) fn intersect_sorted_ranges(a: &[RowRange], b: &[RowRange]) ->
Vec<RowRange> {
- let idx = RowRangeIndex::create(a.to_vec());
+ let idx = RowRangeIndex::from_sorted_ranges(a.to_vec());
let mut result = Vec::new();
for r in b {
result.extend(idx.intersected_ranges(r.from(), r.to()));
@@ -187,6 +185,11 @@ impl RowRangeIndex {
/// Ranges are sorted and merged to eliminate overlaps.
pub fn create(ranges: Vec<RowRange>) -> Self {
let ranges = merge_row_ranges(ranges);
+ Self::from_sorted_ranges(ranges)
+ }
+
+ // Internal callers with canonical ranges must not sort and merge them
again.
+ fn from_sorted_ranges(ranges: Vec<RowRange>) -> Self {
let starts: Vec<i64> = ranges.iter().map(|r| r.from()).collect();
let ends: Vec<i64> = ranges.iter().map(|r| r.to()).collect();
Self {
diff --git a/crates/paimon/src/table/global_index_scanner/tests.rs
b/crates/paimon/src/table/global_index_scanner/tests.rs
index 6c3d12f8..4bd731ab 100644
--- a/crates/paimon/src/table/global_index_scanner/tests.rs
+++ b/crates/paimon/src/table/global_index_scanner/tests.rs
@@ -95,6 +95,115 @@ fn test_intersect_sorted_ranges() {
assert_eq!(result, vec![RowRange::new(5, 10), RowRange::new(20, 25)]);
}
+#[test]
+fn test_bitmap_ranges_across_containers_and_high_keys() {
+ for run in [1u64, 9, 31, 64, 256, 65536] {
+ let start = (1u64 << 32) - 10000;
+ let expected: Vec<_> = (0..100)
+ .map(|i| {
+ let from = start + i * (run + 1);
+ RowRange::new(from as i64, (from + run - 1) as i64)
+ })
+ .collect();
+ let mut bitmap = RoaringTreemap::new();
+ for range in &expected {
+ bitmap.insert_range(range.from() as u64..=range.to() as u64);
+ }
+ assert_eq!(bitmap_to_ranges(&bitmap), expected);
+ bitmap.optimize();
+ assert_eq!(bitmap_to_ranges(&bitmap), expected);
+ }
+ let mut bitmap = RoaringTreemap::new();
+ bitmap.insert_range((1u64 << 32) - 10..=(1u64 << 32) + 10);
+ bitmap.insert(i64::MAX as u64);
+ assert_eq!(
+ bitmap_to_ranges(&bitmap),
+ vec![
+ RowRange::new((1i64 << 32) - 10, (1i64 << 32) + 10),
+ RowRange::new(i64::MAX, i64::MAX),
+ ]
+ );
+}
+
+#[test]
+#[ignore = "manual release benchmark; no wall-clock assertions"]
+fn benchmark_bitmap_range_transport() {
+ use std::hint::black_box;
+ use std::time::Instant;
+ fn baseline(bitmap: &RoaringTreemap, offset: u64) -> Vec<RowRange> {
+ let mut shifted = RoaringTreemap::new();
+ for id in bitmap {
+ shifted.insert(id + offset);
+ }
+ let mut ranges = Vec::new();
+ let mut current: Option<(i64, i64)> = None;
+ for id in shifted {
+ let id = id as i64;
+ match current {
+ Some((start, end)) if id == end + 1 => current = Some((start,
id)),
+ Some((start, end)) => {
+ ranges.push(RowRange::new(start, end));
+ current = Some((id, id));
+ }
+ None => current = Some((id, id)),
+ }
+ }
+ if let Some((start, end)) = current {
+ ranges.push(RowRange::new(start, end));
+ }
+ ranges
+ }
+ fn optimized(bitmap: &RoaringTreemap, offset: u64) -> Vec<RowRange> {
+ crate::table::merge_row_ranges(
+ bitmap_to_ranges(bitmap)
+ .into_iter()
+ .map(|r| RowRange::new(r.from() + offset as i64, r.to() +
offset as i64))
+ .collect(),
+ )
+ }
+ for (name, count, run) in [
+ ("continuous", 12_000_000u64, 12_000_000),
+ ("nine-hit-one-gap", 1_200_000, 9),
+ ("singletons", 1_200_000, 1),
+ ] {
+ let mut bitmap = RoaringTreemap::new();
+ let step = if run == 1 { 10 } else { run + 1 };
+ for start in (0..count).step_by(step as usize) {
+ bitmap.insert_range(start..(start + run).min(count));
+ }
+ bitmap.optimize();
+ let offset = (1u64 << 32) - 10000;
+ let expected = baseline(&bitmap, offset);
+ assert_eq!(optimized(&bitmap, offset), expected);
+ for _ in 0..3 {
+ black_box(baseline(&bitmap, offset));
+ black_box(optimized(&bitmap, offset));
+ }
+ let mut times = [Vec::new(), Vec::new()];
+ for round in 0..7 {
+ for index in [round % 2, 1 - round % 2] {
+ let started = Instant::now();
+ black_box(if index == 0 {
+ baseline(&bitmap, offset)
+ } else {
+ optimized(&bitmap, offset)
+ });
+ times[index].push(started.elapsed().as_secs_f64() * 1000.0);
+ }
+ }
+ for samples in &mut times {
+ samples.sort_by(f64::total_cmp);
+ }
+ println!(
+ "RANGE_BENCH {name} rows={} ranges={} before_ms={:.3}
after_ms={:.3}",
+ bitmap.len(),
+ expected.len(),
+ times[0][3],
+ times[1][3]
+ );
+ }
+}
+
#[test]
fn test_intersect_no_overlap() {
let a = vec![RowRange::new(0, 5)];
@@ -496,6 +605,7 @@ fn test_mixed_fm_and_btree_select_compatible_index_family()
{
file_name: "name.btree".to_string(),
index_type: GlobalIndexFileKind::BTree,
file_size: 1,
+ row_count: 10,
row_range_start: 0,
row_range_end: 9,
external_path: None,
@@ -505,6 +615,7 @@ fn test_mixed_fm_and_btree_select_compatible_index_family()
{
file_name: "name.fm".to_string(),
index_type: GlobalIndexFileKind::FM,
file_size: 1,
+ row_count: 10,
row_range_start: 0,
row_range_end: 9,
external_path: None,
@@ -615,6 +726,535 @@ fn int_eq(column: &str, index: usize, value: i32) ->
Predicate {
}
}
+#[tokio::test]
+async fn test_btree_all_match_without_index_io() {
+ let io = crate::io::FileIOBuilder::new("memory").build().unwrap();
+ let fields = int_schema_fields();
+ let meta = BTreeIndexMeta::new(
+ Some(7i32.to_le_bytes().to_vec()),
+ Some(7i32.to_le_bytes().to_vec()),
+ false,
+ );
+ let mut entry = make_global_index_entry("not-created.index", 1, 100, 109,
&meta);
+ entry.index_file.row_count = 10;
+ entry.index_file.file_size = i64::MAX;
+ for (op, values) in [
+ (PredicateOperator::Eq, vec![7]),
+ (PredicateOperator::Lt, vec![8]),
+ (PredicateOperator::LtEq, vec![7]),
+ (PredicateOperator::Gt, vec![6]),
+ (PredicateOperator::GtEq, vec![7]),
+ (PredicateOperator::Between, vec![7, 7]),
+ ] {
+ let predicate = Predicate::Leaf {
+ column: "id".into(),
+ index: 0,
+ data_type: fields[0].data_type().clone(),
+ op,
+ literals: values.into_iter().map(Datum::Int).collect(),
+ };
+ let ranges = evaluate_global_index_fast_with_fallback_size(
+ &io,
+ "memory:/absent",
+ std::slice::from_ref(&entry),
+ &[predicate],
+ &fields,
+ 0,
+ 0,
+ )
+ .await
+ .unwrap();
+ assert_eq!(ranges, Some(vec![RowRange::new(100, 109)]), "{op}");
+ }
+ for ops in [
+ vec![PredicateOperator::Lt],
+ vec![PredicateOperator::Gt],
+ vec![PredicateOperator::Eq, PredicateOperator::Gt],
+ ] {
+ let predicates = ops
+ .into_iter()
+ .map(|op| Predicate::Leaf {
+ column: "id".into(),
+ index: 0,
+ data_type: fields[0].data_type().clone(),
+ op,
+ literals: vec![Datum::Int(7)],
+ })
+ .collect();
+ assert_eq!(
+ evaluate_global_index_fast(
+ &io,
+ "memory:/absent",
+ std::slice::from_ref(&entry),
+ &[Predicate::and(predicates)],
+ &fields,
+ )
+ .await
+ .unwrap(),
+ Some(vec![])
+ );
+ }
+ // Matching value bounds alone must not invent rows for incomplete/unknown
coverage.
+ for count in [-1, 0, 9, 11] {
+ entry.index_file.row_count = count;
+ assert!(evaluate_global_index_fast(
+ &io,
+ "memory:/absent",
+ std::slice::from_ref(&entry),
+ &[int_eq("id", 0, 7)],
+ &fields
+ )
+ .await
+ .is_err());
+ }
+}
+
+#[tokio::test]
+async fn test_all_match_does_not_hide_partial_shard_over_budget() {
+ let (io, path, file, tmp) =
setup_testdata_table("btree_int_100_no_compress.bin");
+ let fields = int_schema_fields();
+ let mut all = make_global_index_entry(
+ "absent.index",
+ 1,
+ 100,
+ 199,
+ &BTreeIndexMeta::new(Some(le_int_key(7)), Some(le_int_key(7)), false),
+ );
+ all.index_file.row_count = 100;
+ all.index_file.file_size = i64::MAX;
+ let mut partial = make_global_index_entry(
+ &file,
+ 1,
+ 300,
+ 399,
+ &BTreeIndexMeta::new(Some(le_int_key(0)), Some(le_int_key(198)),
false),
+ );
+ let size = std::fs::metadata(tmp.path().join("index").join(&file))
+ .unwrap()
+ .len() as i64;
+ partial.index_file.file_size = size;
+ partial.index_file.row_count = 100;
+ let entries = vec![all, partial];
+ for between in [false, true] {
+ let mut predicates = vec![Predicate::Leaf {
+ column: "id".into(),
+ index: 0,
+ data_type: fields[0].data_type().clone(),
+ op: PredicateOperator::Lt,
+ literals: vec![Datum::Int(11)],
+ }];
+ if between {
+ predicates.push(Predicate::Leaf {
+ column: "id".into(),
+ index: 0,
+ data_type: fields[0].data_type().clone(),
+ op: PredicateOperator::Gt,
+ literals: vec![Datum::Int(2)],
+ });
+ }
+ for budget in [size, size - 1] {
+ let actual = evaluate_global_index_fast_with_fallback_size(
+ &io,
+ &path,
+ &entries,
+ &[Predicate::and(predicates.clone())],
+ &fields,
+ budget,
+ i64::MAX,
+ )
+ .await
+ .unwrap();
+ let expected = (budget == size).then(|| {
+ vec![
+ RowRange::new(100, 199),
+ RowRange::new(if between { 302 } else { 300 }, 305),
+ ]
+ });
+ assert_eq!(actual, expected, "between={between}, budget={budget}");
+ }
+ }
+}
+
+#[tokio::test]
+async fn test_all_match_requires_every_key_shard() {
+ let io = crate::io::FileIOBuilder::new("memory").build().unwrap();
+ let path = "memory:/all_match_key_shards";
+ let mut entries = Vec::new();
+ for (key, rows) in [(7, 0..40), (9, 40..100)] {
+ let name = format!("key-{key}.index");
+ let buf = VecFileWrite::new();
+ let mut writer =
+ BTreeIndexWriter::new(Box::new(buf.clone()), 64,
BlockCompressionType::None);
+ for id in rows {
+ writer.write(Some(&le_int_key(key)), id).await.unwrap();
+ }
+ let result = writer.finish().await.unwrap();
+ io.new_output(&format!("{path}/index/{name}"))
+ .unwrap()
+ .write(bytes::Bytes::from(buf.to_vec()))
+ .await
+ .unwrap();
+ let mut entry = make_global_index_entry(&name, 1, 100, 199,
&result.meta);
+ entry.index_file.row_count = result.row_count as i64;
+ entries.push(entry);
+ }
+ assert_eq!(
+ evaluate_global_index_fast(
+ &io,
+ path,
+ &entries,
+ &[int_eq("id", 0, 7)],
+ &int_schema_fields()
+ )
+ .await
+ .unwrap(),
+ Some(vec![RowRange::new(100, 139)])
+ );
+}
+
+#[tokio::test]
+async fn test_btree_all_match_coverage_nulls_and_boolean_siblings() {
+ let (io, path, file, _tmp) =
setup_testdata_table("btree_int_100_no_compress.bin");
+ let meta = BTreeIndexMeta::new(Some(le_int_key(7)), Some(le_int_key(7)),
false);
+ let mut all = make_global_index_entry("absent.index", 1, 100, 199, &meta);
+ all.index_file.row_count = 100;
+ let selective = make_global_index_entry(
+ &file,
+ 2,
+ 100,
+ 199,
+ &BTreeIndexMeta::new(Some(le_int_key(0)), Some(le_int_key(198)),
false),
+ );
+ let fields = two_field_schema_fields();
+ let predicates = vec![int_eq("id", 0, 7), int_eq("value", 1, 18)];
+ let entries = vec![all.clone(), selective];
+ assert_eq!(
+ evaluate_global_index_fast(
+ &io,
+ &path,
+ &entries,
+ &[Predicate::and(predicates.clone())],
+ &fields
+ )
+ .await
+ .unwrap(),
+ Some(vec![RowRange::new(109, 109)])
+ );
+ assert_eq!(
+ evaluate_global_index_fast(&io, &path, &entries,
&[Predicate::or(predicates)], &fields)
+ .await
+ .unwrap(),
+ Some(vec![RowRange::new(100, 199)])
+ );
+
+ for (mode, expected) in [
+ (GlobalIndexSearchMode::Fast, vec![RowRange::new(100, 199)]),
+ (GlobalIndexSearchMode::Full, vec![RowRange::new(0, 219)]),
+ (GlobalIndexSearchMode::Detail, vec![RowRange::new(100, 209)]),
+ ] {
+ let actual = evaluate_global_index(GlobalIndexEvaluation {
+ file_io: &io,
+ table_path: &path,
+ index_entries: std::slice::from_ref(&all),
+ predicates: &[int_eq("id", 0, 7)],
+ schema_fields: &fields,
+ search_mode: mode,
+ global_index_thread_num: 2,
+ btree_fallback_scan_max_size: 0,
+ bitmap_fallback_scan_max_size: 0,
+ fm_read_options: FMReadOptions::default(),
+ next_row_id: Some(220),
+ data_ranges: &[RowRange::new(100, 209)],
+ })
+ .await
+ .unwrap();
+ assert_eq!(actual, Some(expected));
+ }
+
+ // Two key-partitioned files share a source domain; neither alone is
complete.
+ let mut first = all.clone();
+ first.index_file.row_count = 40;
+ let mut second = all.clone();
+ second.index_file.file_name = "also-absent.index".into();
+ second.index_file.row_count = 60;
+ assert_eq!(
+ evaluate_global_index_fast(
+ &io,
+ &path,
+ &[first.clone(), second.clone()],
+ &[int_eq("id", 0, 7)],
+ &fields
+ )
+ .await
+ .unwrap(),
+ Some(vec![RowRange::new(100, 199)])
+ );
+ assert!(
+ evaluate_global_index_fast(&io, &path, &[first], &[int_eq("id", 0,
7)], &fields)
+ .await
+ .is_err()
+ );
+ let nullable = BTreeIndexMeta::new(Some(le_int_key(7)),
Some(le_int_key(7)), true);
+ all.index_file
+ .global_index_meta
+ .as_mut()
+ .unwrap()
+ .index_meta = Some(nullable.serialize());
+ assert!(
+ evaluate_global_index_fast(&io, &path, &[all], &[int_eq("id", 0, 7)],
&fields)
+ .await
+ .is_err()
+ );
+}
+
+#[tokio::test]
+async fn test_bounded_range_all_endpoints_and_predicate_orders() {
+ let (io, path, file, _tmp) =
setup_testdata_table("btree_int_100_no_compress.bin");
+ let fields = int_schema_fields();
+ let entries = vec![make_global_index_entry(
+ &file,
+ 1,
+ 100,
+ 199,
+ &BTreeIndexMeta::new(Some(le_int_key(0)), Some(le_int_key(198)),
false),
+ )];
+ for (low, low_inclusive) in [
+ (PredicateOperator::Gt, false),
+ (PredicateOperator::GtEq, true),
+ ] {
+ for (high, high_inclusive) in [
+ (PredicateOperator::Lt, false),
+ (PredicateOperator::LtEq, true),
+ ] {
+ for (lower, upper) in [(30, 40), (30, 30), (40, 30)] {
+ for reverse in [false, true] {
+ let mut predicates: Vec<_> = [(low, lower), (high, upper)]
+ .into_iter()
+ .map(|(op, v)| Predicate::Leaf {
+ column: "id".into(),
+ index: 0,
+ data_type: fields[0].data_type().clone(),
+ op,
+ literals: vec![Datum::Int(v)],
+ })
+ .collect();
+ if reverse {
+ predicates.reverse();
+ }
+ let actual = evaluate_global_index_fast(
+ &io,
+ &path,
+ &entries,
+ &[Predicate::and(predicates)],
+ &fields,
+ )
+ .await
+ .unwrap()
+ .unwrap();
+ let expected: Vec<_> = (0..100)
+ .filter(|id| {
+ let key = *id * 2;
+ (key > lower || low_inclusive && key == lower)
+ && (key < upper || high_inclusive && key ==
upper)
+ })
+ .map(|id| id as i64 + 100)
+ .collect();
+ let actual: Vec<_> = actual.iter().flat_map(|r|
r.from()..=r.to()).collect();
+ assert_eq!(actual, expected);
+ }
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn test_scalar_optimization_query_counts() {
+ let (io, path, file, _tmp) =
setup_testdata_table("btree_int_100_no_compress.bin");
+ let fields = int_schema_fields();
+ let mut entry = make_global_index_entry(
+ &file,
+ 1,
+ 100,
+ 199,
+ &BTreeIndexMeta::new(Some(le_int_key(0)), Some(le_int_key(198)),
false),
+ );
+ entry.index_file.row_count = 100;
+ let leaf = |op, value| Predicate::Leaf {
+ column: "id".into(),
+ index: 0,
+ data_type: fields[0].data_type().clone(),
+ op,
+ literals: vec![Datum::Int(value)],
+ };
+ for (conditions, expected, query_counts) in [
+ (
+ vec![
+ leaf(PredicateOperator::GtEq, 0),
+ leaf(PredicateOperator::Eq, 50),
+ ],
+ vec![RowRange::new(125, 125)],
+ (1, 0),
+ ),
+ (
+ vec![
+ leaf(PredicateOperator::GtEq, 10),
+ leaf(PredicateOperator::LtEq, 180),
+ leaf(PredicateOperator::Gt, 30),
+ leaf(PredicateOperator::Lt, 40),
+ ],
+ vec![RowRange::new(116, 119)],
+ (0, 1),
+ ),
+ (
+ vec![
+ leaf(PredicateOperator::Gt, 40),
+ leaf(PredicateOperator::Lt, 30),
+ ],
+ vec![],
+ (0, 0),
+ ),
+ (
+ vec![
+ leaf(PredicateOperator::Gt, 30),
+ leaf(PredicateOperator::LtEq, 30),
+ ],
+ vec![],
+ (0, 0),
+ ),
+ ] {
+ for reverse in [false, true] {
+ let mut conditions = conditions.clone();
+ if reverse {
+ conditions.reverse();
+ }
+ let mut scanner = GlobalIndexScanner::create(
+ &io,
+ &path,
+ 2,
+ i64::MAX,
+ i64::MAX,
+ std::slice::from_ref(&entry),
+ &fields,
+ )
+ .unwrap()
+ .unwrap();
+ let probe = Arc::new(QueryIoProbe::default());
+ scanner.query_io_probe = Some(probe.clone());
+ let result = scanner
+ .evaluate(&Predicate::and(conditions))
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(result.row_ranges, expected);
+ assert_eq!(
+ (
+ probe.predicate_queries.load(TestOrdering::SeqCst),
+ probe.range_queries.load(TestOrdering::SeqCst)
+ ),
+ query_counts
+ );
+ if query_counts == (0, 0) {
+ assert_eq!(probe.peak(), 0);
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn test_scalar_optimization_empty_range_never_opens_index() {
+ let io = crate::io::FileIOBuilder::new("memory").build().unwrap();
+ let fields = int_schema_fields();
+ let entry = make_global_index_entry(
+ "must-not-open.index",
+ 1,
+ 100,
+ 199,
+ &BTreeIndexMeta::new(Some(le_int_key(0)), Some(le_int_key(198)),
false),
+ );
+ for (from, to) in [(40, 30), (30, 30)] {
+ let predicates: Vec<_> = [(PredicateOperator::Gt, from),
(PredicateOperator::LtEq, to)]
+ .into_iter()
+ .map(|(op, v)| Predicate::Leaf {
+ column: "id".into(),
+ index: 0,
+ data_type: fields[0].data_type().clone(),
+ op,
+ literals: vec![Datum::Int(v)],
+ })
+ .collect();
+ for (mode, expected) in [
+ (GlobalIndexSearchMode::Fast, vec![]),
+ (
+ GlobalIndexSearchMode::Full,
+ vec![RowRange::new(0, 99), RowRange::new(200, 219)],
+ ),
+ (GlobalIndexSearchMode::Detail, vec![RowRange::new(200, 209)]),
+ ] {
+ let actual = evaluate_global_index(GlobalIndexEvaluation {
+ file_io: &io,
+ table_path: "memory:/absent",
+ index_entries: std::slice::from_ref(&entry),
+ predicates: &[Predicate::and(predicates.clone())],
+ schema_fields: &fields,
+ search_mode: mode,
+ global_index_thread_num: 2,
+ btree_fallback_scan_max_size: 0,
+ bitmap_fallback_scan_max_size: 0,
+ fm_read_options: FMReadOptions::default(),
+ next_row_id: Some(220),
+ data_ranges: &[RowRange::new(100, 209)],
+ })
+ .await
+ .unwrap();
+ assert_eq!(actual, Some(expected));
+ }
+ }
+}
+
+#[tokio::test]
+async fn
test_all_matching_conjunct_keeps_domain_when_other_conjunct_declines() {
+ let io = crate::io::FileIOBuilder::new("memory").build().unwrap();
+ let fields = string_schema_fields();
+ let mut entry = make_global_index_entry(
+ "must-not-open.index",
+ 1,
+ 100,
+ 109,
+ &BTreeIndexMeta::new(Some(b"hit".to_vec()), Some(b"miss".to_vec()),
false),
+ );
+ entry.index_file.row_count = 10;
+ entry.index_file.file_size = 100;
+ let predicates = [
+ (PredicateOperator::Lt, "z"),
+ (PredicateOperator::Contains, "hit"),
+ ]
+ .into_iter()
+ .map(|(op, value)| Predicate::Leaf {
+ column: "name".into(),
+ index: 0,
+ data_type: fields[0].data_type().clone(),
+ op,
+ literals: vec![Datum::String(value.into())],
+ })
+ .collect();
+ let actual = evaluate_global_index_fast_with_fallback_size(
+ &io,
+ "memory:/absent",
+ &[entry],
+ &[Predicate::and(predicates)],
+ &fields,
+ 0,
+ 0,
+ )
+ .await
+ .unwrap();
+ assert_eq!(
+ actual,
+ Some(vec![RowRange::new(100, 109)]),
+ "retain candidates for residual filtering, not empty"
+ );
+}
+
#[test]
fn test_unindexed_ranges_fast_mode_empty() {
let file_io = crate::io::FileIOBuilder::new("memory").build().unwrap();
@@ -1191,6 +1831,99 @@ async fn
test_bitmap_nan_equality_uses_direct_lookup_with_fallback_scan_disabled
.await;
}
+#[tokio::test]
+async fn test_bitmap_floating_bound_fusion_preserves_candidates_and_fallback()
{
+ for (ty, low, high, zero, minus_one) in [
+ (
+ DataType::Float(crate::spec::FloatType::new()),
+ Datum::Float(f32::from_bits(0xffc0_0001)),
+ Datum::Float(f32::from_bits(0x7fc0_0001)),
+ Datum::Float(0.0),
+ Datum::Float(-1.0),
+ ),
+ (
+ DataType::Double(crate::spec::DoubleType::new()),
+ Datum::Double(f64::from_bits(0xfff8_0000_0000_0001)),
+ Datum::Double(f64::from_bits(0x7ff8_0000_0000_0001)),
+ Datum::Double(0.0),
+ Datum::Double(-1.0),
+ ),
+ ] {
+ let output = VecFileWrite::new();
+ let mut writer = BitmapGlobalIndexWriter::new(
+ Box::new(output.clone()),
+ 1,
+ BlockCompressionType::None,
+ make_bitmap_key_comparator(&ty),
+ );
+ writer
+ .write(Some(&serialize_bitmap_datum(&zero, &ty)), 0)
+ .unwrap();
+ writer.write(None, 1).unwrap();
+ let result = writer.finish().await.unwrap();
+ let bytes = output.to_vec();
+ let io = crate::io::FileIOBuilder::new("memory").build().unwrap();
+ let path = "memory:/bitmap_bound_fusion";
+ io.new_output(&format!("{path}/index/bitmap.index"))
+ .unwrap()
+ .write(bytes::Bytes::from(bytes.clone()))
+ .await
+ .unwrap();
+ let mut entry = make_global_index_entry_with_type(
+ BITMAP_GLOBAL_INDEX_TYPE,
+ "bitmap.index",
+ 1,
+ 100,
+ 101,
+ &result.meta,
+ );
+ entry.index_file.file_size = bytes.len() as i64;
+ let fields = vec![DataField::new(1, "id".into(), ty.clone())];
+ let leaf = |op, value: Datum| Predicate::Leaf {
+ column: "id".into(),
+ index: 0,
+ data_type: ty.clone(),
+ op,
+ literals: vec![value],
+ };
+ for extra_bound in [false, true] {
+ for reverse in [false, true] {
+ let mut conditions = vec![
+ leaf(PredicateOperator::Gt, low.clone()),
+ leaf(PredicateOperator::Lt, high.clone()),
+ ];
+ if extra_bound {
+ conditions.push(leaf(PredicateOperator::Gt,
minus_one.clone()));
+ }
+ if reverse {
+ conditions.reverse();
+ }
+ for budget in [i64::MAX, 0] {
+ let actual = evaluate_global_index_fast_with_fallback_size(
+ &io,
+ path,
+ std::slice::from_ref(&entry),
+ &[Predicate::and(conditions.clone())],
+ &fields,
+ 0,
+ budget,
+ )
+ .await
+ .unwrap();
+ assert_eq!(
+ actual,
+ if budget == 0 {
+ None
+ } else {
+ Some(vec![RowRange::new(100, 100)])
+ }
+ );
+ }
+ }
+ }
+ }
+}
+
fn legacy_floating_comparator(data_type: &DataType) -> BoxedCmp {
match data_type {
DataType::Float(_) => Box::new(|left, right| {
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
index fff014d5..cace6bec 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
@@ -657,6 +657,22 @@ async fn scan_ids(table: &Table, predicate: Predicate) ->
Vec<i32> {
.try_collect::<Vec<_>>()
.await
.unwrap();
+ // The column-group fixture must merge payloads by RowID, not just return
+ // the right number of rows from the indexed key/flag group.
+ for batch in &batches {
+ if let Some(payload) = batch.column_by_name("payload") {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let payload =
payload.as_any().downcast_ref::<Int64Array>().unwrap();
+ assert_eq!(payload.null_count(), 0);
+ for row in 0..batch.num_rows() {
+ assert_eq!(payload.value(row), i64::from(ids.value(row)) * 17);
+ }
+ }
+ }
let mut ids = batches
.iter()
.flat_map(|batch| {
@@ -676,8 +692,15 @@ async fn scan_ids(table: &Table, predicate: Predicate) ->
Vec<i32> {
#[tokio::test]
async fn test_execute_writes_btree_index_manifest_and_file() {
+ for version in [1, 2] {
+ assert_btree_build_version(version).await;
+ }
+}
+
+async fn assert_btree_build_version(version: u32) {
let table_path = "memory:/test_btree_global_index_builder_e2e";
let mut options = table_options("10");
+ options.insert("btree-index.file-version".into(), version.to_string());
options.insert(
"btree-index.bloom-filter.enabled".to_string(),
"true".to_string(),
@@ -742,6 +765,7 @@ async fn
test_execute_writes_btree_index_manifest_and_file() {
.bloom_filter_handle
.expect("enabled BTree build must write a Bloom filter");
assert_eq!(bloom_handle.expected_entries, 2);
+ assert_eq!(footer.version, version);
let global_meta = index_file
.global_index_meta
@@ -805,6 +829,230 @@ async fn
test_execute_writes_btree_index_manifest_and_file() {
plan.splits()[0].row_ranges(),
Some(&[RowRange::new(0, 0), RowRange::new(2, 2)][..])
);
+ let predicate = PredicateBuilder::new(scan_table.schema().fields())
+ .equal("name", crate::spec::Datum::String("alice".to_string()))
+ .unwrap();
+ assert_eq!(scan_ids(&scan_table, predicate).await, vec![1, 3]);
+}
+
+async fn column_group_btree_table(rows: i32, version: u32) -> Table {
+ use crate::table::data_evolution_writer::DataEvolutionPartialWriter;
+ let mut options = table_options("2000000");
+ options.insert("btree-index.file-version".into(), version.to_string());
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::string_type()))
+ .column("payload", DataType::BigInt(crate::spec::BigIntType::new()))
+ .options(options)
+ .build()
+ .unwrap();
+ let table = Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "column_groups"),
+ format!("memory:/btree_column_groups_v{version}"),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ setup_dirs(&table).await;
+ let batch = data_batch(
+ (0..rows).collect(),
+ (0..rows)
+ .map(|id| if id % 10 == 0 { "miss" } else { "hit" })
+ .collect(),
+ );
+ let mut write =
+ DataEvolutionPartialWriter::new(&table, vec!["id".into(),
"name".into()]).unwrap();
+ write
+ .write_partial_batch(
+ crate::spec::EMPTY_BINARY_ROW.to_serialized_bytes(),
+ 0,
+ 0,
+ 0,
+ batch,
+ )
+ .await
+ .unwrap();
+ let mut messages = write.prepare_commit().await.unwrap();
+ // Seed an append column group: normal commit allocates its RowIDs.
+ // Subsequent partial-column writes reuse that committed RowID domain.
+ for message in &mut messages {
+ message.check_from_snapshot = None;
+ for file in &mut message.new_files {
+ file.first_row_id = None;
+ }
+ }
+ TableCommit::new(table.clone(), "bench".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![ArrowField::new(
+ "payload",
+ ArrowDataType::Int64,
+ true,
+ )])),
+ vec![Arc::new(Int64Array::from(
+ (0..rows).map(|id| i64::from(id) * 17).collect::<Vec<_>>(),
+ ))],
+ )
+ .unwrap();
+ let mut payload = DataEvolutionPartialWriter::new(&table,
vec!["payload".into()]).unwrap();
+ payload
+ .write_partial_batch(
+ crate::spec::EMPTY_BINARY_ROW.to_serialized_bytes(),
+ 0,
+ 0,
+ 1,
+ batch,
+ )
+ .await
+ .unwrap();
+ TableCommit::new(table.clone(), "payload".into())
+ .commit(payload.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ let plan = table.new_read_builder().new_scan().plan().await.unwrap();
+ let files: Vec<_> = plan
+ .splits()
+ .iter()
+ .flat_map(|split| split.data_files())
+ .collect();
+ assert_eq!(files.len(), 2);
+ for file in &files {
+ assert!(file.file_name.ends_with(".parquet"));
+ assert_eq!(file.first_row_id, Some(0));
+ assert_eq!(file.row_count, i64::from(rows));
+ }
+ assert!(files
+ .iter()
+ .any(|file| file.write_cols.as_deref() == Some(&["id".into(),
"name".into()])));
+ assert!(files
+ .iter()
+ .any(|file| file.write_cols.as_deref() == Some(&["payload".into()])));
+ for column in ["id", "name"] {
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_column(column)
+ .execute()
+ .await
+ .unwrap();
+ }
+ table
+}
+
+#[tokio::test]
+async fn test_scalar_column_group_reads() {
+ for version in [1, 2] {
+ let table = column_group_btree_table(4096, version).await;
+ let builder = PredicateBuilder::new(table.schema().fields());
+ for (predicate, expected) in [
+ (
+ builder.equal("name", Datum::String("miss".into())).unwrap(),
+ (0..4096).step_by(10).collect(),
+ ),
+ (
+ Predicate::and(vec![
+ builder.greater_or_equal("id", Datum::Int(0)).unwrap(),
+ builder.equal("id", Datum::Int(1)).unwrap(),
+ ]),
+ vec![1],
+ ),
+ ] {
+ assert_eq!(scan_ids(&table, predicate).await, expected);
+ }
+ }
+}
+
+#[tokio::test]
+#[ignore = "manual release benchmark; default Parquet, column groups, planning
and full reads"]
+async fn benchmark_btree_data_evolution_read() {
+ use crate::spec::Datum;
+ use std::time::Instant;
+ let rows = 1_200_000;
+ for version in [1, 2] {
+ let table = column_group_btree_table(rows, version).await;
+ let mut options = table.schema().options().clone();
+ options.insert("global-index.enabled".into(), "false".into());
+ let no_index = Table::new(
+ table.file_io().clone(),
+ table.identifier().clone(),
+ table.location().to_string(),
+ table.schema().copy_with_replaced_options(options),
+ None,
+ );
+ let builder = PredicateBuilder::new(table.schema().fields());
+ let point = builder.equal("id", Datum::Int(1)).unwrap();
+ let dense = builder.equal("name",
Datum::String("hit".into())).unwrap();
+ let sparse = builder.equal("name",
Datum::String("miss".into())).unwrap();
+ let all = builder
+ .less_than("name", Datum::String("z".into()))
+ .unwrap();
+ let bounded = Predicate::and(vec![
+ builder.greater_than("id", Datum::Int(rows / 2)).unwrap(),
+ builder.less_than("id", Datum::Int(rows / 2 + 100)).unwrap(),
+ ]);
+ let point_with_full_bound = Predicate::and(vec![
+ builder.greater_or_equal("id", Datum::Int(0)).unwrap(),
+ point.clone(),
+ ]);
+ let multiple_bounds = Predicate::and(vec![
+ builder.greater_or_equal("id", Datum::Int(0)).unwrap(),
+ builder.less_than("id", Datum::Int(rows)).unwrap(),
+ bounded.clone(),
+ ]);
+ for (name, predicate, expected) in [
+ (
+ "90pct",
+ dense.clone(),
+ (0..rows).filter(|id| id % 10 != 0).collect::<Vec<_>>(),
+ ),
+ (
+ "10pct",
+ sparse,
+ (0..rows).filter(|id| id % 10 == 0).collect(),
+ ),
+ ("point", point.clone(), vec![1]),
+ ("point_and_full_same_field", point_with_full_bound, vec![1]),
+ (
+ "multiple_bounds",
+ multiple_bounds,
+ (rows / 2 + 1..rows / 2 + 100).collect(),
+ ),
+ (
+ "point_and_90pct",
+ Predicate::and(vec![point, dense]),
+ vec![1],
+ ),
+ ("all_match", all, (0..rows).collect()),
+ ("bounded", bounded, (rows / 2 + 1..rows / 2 + 100).collect()),
+ ] {
+ for source in [&table, &no_index] {
+ for _ in 0..3 {
+ assert_eq!(scan_ids(source, predicate.clone()).await,
expected);
+ }
+ }
+ let sources = [&table, &no_index];
+ let mut samples = [Vec::new(), Vec::new()];
+ for round in 0..7 {
+ for i in [round % 2, 1 - round % 2] {
+ let start = Instant::now();
+ let actual = scan_ids(sources[i], predicate.clone()).await;
+ let elapsed = start.elapsed().as_secs_f64() * 1000.0;
+ assert_eq!(actual, expected);
+ samples[i].push(elapsed);
+ }
+ }
+ for times in &mut samples {
+ times.sort_by(f64::total_cmp);
+ }
+ println!(
+ "TABLE_BENCH version={version} case={name} rows={}
index_ms={:.3} scan_ms={:.3}",
+ expected.len(),
+ samples[0][3],
+ samples[1][3]
+ );
+ }
+ }
}
#[tokio::test]
diff --git
a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
index 1b7996ae..aa71e501 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
@@ -112,7 +112,12 @@ impl SortedGlobalIndexBuildBuilder<'_> {
write_options.compression_level,
write_options.bloom_filter_enabled,
cmp,
- );
+ )
+ .with_file_version(write_options.btree_file_version)
+ .map_err(|error| Error::DataInvalid {
+ message: "Invalid BTree file version".to_string(),
+ source: Some(Box::new(error)),
+ })?;
for (key, local_row_id) in &rows {
writer
.write(key.as_deref(), *local_row_id)
diff --git a/crates/paimon/src/table/sorted_global_index_options.rs
b/crates/paimon/src/table/sorted_global_index_options.rs
index e7a16422..bea4a269 100644
--- a/crates/paimon/src/table/sorted_global_index_options.rs
+++ b/crates/paimon/src/table/sorted_global_index_options.rs
@@ -24,6 +24,7 @@ use crate::{Error, Result};
use std::collections::HashMap;
pub(crate) const BTREE_BLOCK_SIZE_OPTION: &str = "btree-index.block-size";
+pub(crate) const BTREE_FILE_VERSION_OPTION: &str = "btree-index.file-version";
pub(crate) const BTREE_BLOOM_FILTER_ENABLED_OPTION: &str =
"btree-index.bloom-filter.enabled";
pub(crate) const BTREE_COMPRESSION_OPTION: &str = "btree-index.compression";
pub(crate) const BTREE_COMPRESSION_LEVEL_OPTION: &str =
"btree-index.compression-level";
@@ -45,6 +46,7 @@ pub(crate) struct SortedIndexWriteOptions {
pub(crate) compression_type: BlockCompressionType,
pub(crate) compression_level: i32,
pub(crate) bloom_filter_enabled: bool,
+ pub(crate) btree_file_version: u32,
}
impl SortedIndexWriteOptions {
@@ -76,6 +78,26 @@ impl SortedIndexWriteOptions {
};
Ok(Self {
+ btree_file_version: if index_type == BTREE_GLOBAL_INDEX_TYPE {
+ match options
+ .get(BTREE_FILE_VERSION_OPTION)
+ .map(|s| s.trim())
+ .unwrap_or("1")
+ {
+ "1" => 1,
+ "2" => 2,
+ raw => {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Option '{BTREE_FILE_VERSION_OPTION}' must be
1 or 2, got: {raw}"
+ ),
+ source: None,
+ })
+ }
+ }
+ } else {
+ 1
+ },
block_size: parse_block_size(options, block_size_option,
default_block_size)?,
compression_type: parse_compression(options, compression_option)?,
compression_level: parse_compression_level(
@@ -185,6 +207,7 @@ mod tests {
compression_type: BlockCompressionType::None,
compression_level: 1,
bloom_filter_enabled: false,
+ btree_file_version: 1,
}
);
for index_type in [BITMAP_GLOBAL_INDEX_TYPE,
MULTIVALUE_GLOBAL_INDEX_TYPE] {
@@ -195,6 +218,7 @@ mod tests {
compression_type: BlockCompressionType::None,
compression_level: 1,
bloom_filter_enabled: false,
+ btree_file_version: 1,
}
);
}
@@ -234,6 +258,7 @@ mod tests {
compression_type: BlockCompressionType::Lz4,
compression_level: 7,
bloom_filter_enabled: false,
+ btree_file_version: 1,
}
);
}
@@ -253,6 +278,13 @@ mod tests {
fn test_invalid_options() {
for (index_type, option, value) in [
(BTREE_GLOBAL_INDEX_TYPE, BTREE_BLOCK_SIZE_OPTION, "0"),
+ (BTREE_GLOBAL_INDEX_TYPE, BTREE_FILE_VERSION_OPTION, "0"),
+ (BTREE_GLOBAL_INDEX_TYPE, BTREE_FILE_VERSION_OPTION, "3"),
+ (
+ BTREE_GLOBAL_INDEX_TYPE,
+ BTREE_FILE_VERSION_OPTION,
+ "invalid",
+ ),
(
BITMAP_GLOBAL_INDEX_TYPE,
BITMAP_DICTIONARY_BLOCK_SIZE_OPTION,
diff --git a/crates/paimon/testdata/btree/GenerateBTreeV2.java
b/crates/paimon/testdata/btree/GenerateBTreeV2.java
new file mode 100644
index 00000000..dd250704
--- /dev/null
+++ b/crates/paimon/testdata/btree/GenerateBTreeV2.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.paimon.globalindex.btree;
+
+import org.apache.paimon.compression.BlockCompressionFactory;
+import org.apache.paimon.compression.BlockCompressionType;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.types.IntType;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+
+/** Generate compatibility fixtures with the Java production BTree V2 writer.
*/
+public class GenerateBTreeV2 {
+ public static void main(String[] args) throws Exception {
+ for (BlockCompressionType codec : new BlockCompressionType[] {
+ BlockCompressionType.NONE, BlockCompressionType.LZ4}) {
+ final ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ GlobalIndexFileWriter files = new GlobalIndexFileWriter() {
+ public String newFileName(String prefix) { return "fixture"; }
+ public PositionOutputStream newOutputStream(String name) {
+ return new PositionOutputStream() {
+ public long getPos() { return bytes.size(); }
+ public void write(int b) { bytes.write(b); }
+ public void write(byte[] b) { bytes.write(b, 0,
b.length); }
+ public void write(byte[] b, int off, int len) {
bytes.write(b, off, len); }
+ public void flush() {}
+ public void close() {}
+ };
+ }
+ };
+ BTreeIndexWriter writer = new BTreeIndexWriter(
+ files, KeySerializer.create(new IntType()), 64, null,
+ BlockCompressionFactory.create(codec), 2);
+ writer.write(0, 7);
+ for (long id : new long[] {2, 130, 65538, (1L << 32) + 5}) {
writer.write(1, id); }
+ for (long id = 60000; id < 140000; id++) { writer.write(2, id); }
+ for (long id = (1L << 32) - 10; id < (1L << 32) + 10000; id++) {
writer.write(3, id); }
+ writer.write(4, Long.MAX_VALUE);
+ // Repeated deltas make this block genuinely compressible with LZ4.
+ for (long id = 0; id < 4096; id++) { writer.write(5, (1L << 40) +
id * 128); }
+ writer.write(null, 4);
+ writer.write(null, (1L << 32) + 20000);
+ writer.finish();
+ Files.write(Paths.get(args[0], "btree_v2_java_" +
codec.name().toLowerCase() + ".bin"),
+ bytes.toByteArray());
+ }
+ }
+}
diff --git a/crates/paimon/testdata/btree/README.md
b/crates/paimon/testdata/btree/README.md
new file mode 100644
index 00000000..fbdcbaba
--- /dev/null
+++ b/crates/paimon/testdata/btree/README.md
@@ -0,0 +1,44 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# BTree V2 interoperability fixtures
+
+`btree_v2_java_none.bin` and `btree_v2_java_lz4.bin` are generated by the Java
+production `BTreeIndexWriter`, using `GenerateBTreeV2.java` in its package to
+select file version 2. They were generated against apache/paimon `b5bcf9babce`.
+
+Integer keys 0–5 exercise a singleton, a sparse delta list, dense Roaring runs,
+a run crossing the high-32-bit boundary, `Long.MAX_VALUE`, and a compressible
+delta list. Two null rows exercise the unchanged null-bitmap format. Blocks
+have a target size of 64 bytes. The LZ4 fixture contains both compressed and
+uncompressed blocks: the Java SST writer compresses only when it saves space.
+
+To regenerate, compile the helper with the Java `paimon-common` / `paimon-api`
+classes and their runtime dependencies on the classpath, then run:
+
+```sh
+javac -cp "$PAIMON_CLASSPATH" -d "$CLASSES_DIR" GenerateBTreeV2.java
+java -cp "$CLASSES_DIR:$PAIMON_CLASSPATH" \
+ org.apache.paimon.globalindex.btree.GenerateBTreeV2 .
+```
+
+Rust tests read the Java files and compare the Rust V2 writer's uncompressed
+output byte-for-byte with the Java fixture. A writer/reader round-trip alone
+would not catch incompatible posting tags, portable bitmap layout, or footer
+versions.
diff --git a/crates/paimon/testdata/btree/btree_v2_java_lz4.bin
b/crates/paimon/testdata/btree/btree_v2_java_lz4.bin
new file mode 100644
index 00000000..2c7d92e5
Binary files /dev/null and b/crates/paimon/testdata/btree/btree_v2_java_lz4.bin
differ
diff --git a/crates/paimon/testdata/btree/btree_v2_java_none.bin
b/crates/paimon/testdata/btree/btree_v2_java_none.bin
new file mode 100644
index 00000000..d665b357
Binary files /dev/null and
b/crates/paimon/testdata/btree/btree_v2_java_none.bin differ
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 28c292bb..7598ee53 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1143,6 +1143,23 @@ options. BTree additionally accepts
`btree-index.bloom-filter.enabled` (default
`false`) to accelerate equality and `IN` lookups. Per-call options override
table
options. Bitmap and multivalue global indexes use Java-compatible bitmap files.
+BTree reads both Java-compatible V1 and V2 index files. Writing defaults to V1
+for compatibility with older readers. Set `btree-index.file-version=2`
explicitly
+to use adaptive single-row, delta-list, and Roaring postings; every reader of
+the table must support V2 before enabling it. Existing index files do not need
+to be rebuilt for the reader to handle a mix of versions.
+
+BTree min/max metadata can also avoid index-file reads when every key matches,
+there are no nulls, and the file set's row counts prove complete coverage of
+the same source row-ID range. These all-matching files do not consume the
+fallback index-scan byte budget; partially matching files still do. The proof
is
+applied separately to each conjunct, so an all-matching bound does not force a
+full posting-list scan alongside a selective equality on the same column.
+Multiple lower/upper bounds (including `BETWEEN`) on one column are tightened
+into one range query; contradictory bounds produce no indexed candidates
+without opening index files. Unindexed-tail handling and residual filtering
+continue to follow the selected global-index search mode.
+
FM global indexes support character-string columns and exact byte-substring
`contains`, `IS NULL`, and `IS NOT NULL` predicates. They use the
Java-compatible partitioned V1 format. `sorted-index.records-per-range` bounds
@@ -2436,6 +2453,7 @@ deletion vectors enabled.
| `global-index.row-count-per-shard` | `100000` | Maximum row count per vector
global-index shard. |
| `sorted-index.records-per-range` | `100000` | Maximum row count per BTree,
bitmap, multivalue, or FM global-index file range; falls back to legacy
`btree-index.records-per-range`. |
| `btree-index.block-size` | `64kb` | Target BTree data-block size. |
+| `btree-index.file-version` | `1` | BTree file version to write: `1` or `2`.
V2 uses adaptive postings and requires V2-capable readers. |
| `btree-index.bloom-filter.enabled` | `false` | Writes a Bloom filter used to
avoid BTree data-block reads for missing equality and `IN` keys. |
| `btree-index.compression` | `none` | BTree block compression: `none`,
`zstd`, `lz4`, or `lzo`. |
| `btree-index.compression-level` | `1` | BTree compression level (used by
codecs that support levels). |
@@ -2453,7 +2471,7 @@ deletion vectors enabled.
| `fm-index.read-cache-size` | `64mb` | Scan-scoped decoded FM block cache
size. |
| `fm-index.demand-page-size` | `512kb` | Target compressed-block read-ahead
size for FM demand paging. |
| `fm-index.locate-cost-ratio` | `0.001` | Maximum estimated locate work as a
fraction of indexed text; denser matches fall back to the source scan. |
-| `btree-index.fallback-scan-max-size` | `256mb` | Maximum total size of
selected BTree global-index files for fallback scans used by range/between and
suffix/contains/complex LIKE predicates; `0` disables BTree fallback index
scans. |
+| `btree-index.fallback-scan-max-size` | `256mb` | Maximum total size of BTree
global-index files requiring fallback scans for range/between and
suffix/contains/complex LIKE predicates; files proven to match completely by
metadata need no scan and are excluded. `0` disables BTree fallback index
scans. |
| `bitmap-index.fallback-scan-max-size` | `256mb` | Maximum total size of
selected bitmap global-index files for fallback scans used by range/between and
suffix/contains/complex LIKE predicates; `0` disables bitmap fallback index
scans. |
| `global-index.search-mode` | `fast` | Global index coverage mode for reads:
`fast`, `full`, or `detail`. |
| `global-index.thread-num` | `32` | Number of concurrent global-index search
tasks; must be greater than 0 and must not exceed the runtime's task limit.
This does not limit Vindex file range reads. |
