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 5c02f0a Support Java-compatible bitmap global index (#478)
5c02f0a is described below
commit 5c02f0a196a006014d7e764234416d207e3c3198
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 8 09:01:25 2026 +0800
Support Java-compatible bitmap global index (#478)
---
crates/integrations/datafusion/src/procedures.rs | 19 +-
crates/integrations/datafusion/tests/procedures.rs | 114 ++-
crates/paimon/src/btree/key_serde.rs | 133 ++-
crates/paimon/src/btree/meta.rs | 57 +-
crates/paimon/src/btree/mod.rs | 2 +-
crates/paimon/src/btree/query.rs | 70 +-
crates/paimon/src/btree/reader.rs | 30 +
crates/paimon/src/btree/tests.rs | 55 ++
crates/paimon/src/spec/core_options.rs | 97 +++
crates/paimon/src/spec/mod.rs | 1 +
crates/paimon/src/spec/predicate.rs | 2 +-
.../paimon/src/table/bitmap_global_index_reader.rs | 892 +++++++++++++++++++++
.../src/table/btree_global_index_build_builder.rs | 219 ++++-
.../src/table/btree_global_index_drop_builder.rs | 34 +-
crates/paimon/src/table/global_index_scanner.rs | 866 ++++++++++++++++++--
crates/paimon/src/table/global_index_types.rs | 29 +
crates/paimon/src/table/mod.rs | 2 +
crates/paimon/src/table/table_scan.rs | 6 +
.../testdata/bitmap/bitmap_varchar_java.index | Bin 0 -> 455 bytes
.../testdata/bitmap/bitmap_varchar_java.index.meta | Bin 0 -> 22 bytes
docs/src/sql.md | 20 +-
21 files changed, 2505 insertions(+), 143 deletions(-)
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index 0778839..76f3660 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -24,8 +24,10 @@
//! - `CALL sys.rollback_to_timestamp(table => '...', timestamp => ...)`
//! - `CALL sys.create_tag_from_timestamp(table => '...', tag => '...',
timestamp => ...)`
//! - `CALL sys.create_global_index(table => '...', index_column => '...',
index_type => 'btree')`
+//! - `CALL sys.create_global_index(table => '...', index_column => '...',
index_type => 'bitmap')`
//! - `CALL sys.create_global_index(table => '...', index_column => '...',
index_type => 'ivf-pq')`
//! - `CALL sys.drop_global_index(table => '...', index_column => '...',
index_type => 'btree')`
+//! - `CALL sys.drop_global_index(table => '...', index_column => '...',
index_type => 'bitmap')`
//! - `CALL sys.create_lumina_index(table => '...', index_column => '...')`
use std::collections::HashMap;
@@ -545,15 +547,17 @@ async fn proc_create_global_index(
.get("index_type")
.map(String::as_str)
.unwrap_or("btree");
- if index_type.eq_ignore_ascii_case("btree") {
+ if is_sorted_global_index_type(index_type) {
if args.contains_key("options") {
return Err(DataFusionError::NotImplemented(
- "create_global_index options are not supported for btree
yet".to_string(),
+ "create_global_index options are not supported for sorted
global indexes yet"
+ .to_string(),
));
}
let mut builder = table.new_btree_global_index_build_builder();
builder.with_index_column(index_column);
+ builder.with_index_type(index_type);
builder.execute().await.map_err(to_datafusion_error)?;
} else if is_vindex_index_type(index_type) {
let mut builder = table.new_vindex_index_build_builder(index_type);
@@ -564,7 +568,7 @@ async fn proc_create_global_index(
builder.execute().await.map_err(to_datafusion_error)?;
} else {
return Err(DataFusionError::NotImplemented(format!(
- "create_global_index only supports index_type => 'btree' or vindex
types \
+ "create_global_index only supports index_type => 'btree',
'bitmap', or vindex types \
('ivf-flat', 'ivf-pq', 'ivf-hnsw-flat', 'ivf-hnsw-sq'), got
'{index_type}'"
)));
}
@@ -583,9 +587,9 @@ async fn proc_drop_global_index(
.get("index_type")
.map(String::as_str)
.unwrap_or("btree");
- if !index_type.eq_ignore_ascii_case("btree") {
+ if !is_sorted_global_index_type(index_type) {
return Err(DataFusionError::NotImplemented(format!(
- "drop_global_index only supports index_type => 'btree', got
'{index_type}'"
+ "drop_global_index only supports index_type => 'btree' or
'bitmap', got '{index_type}'"
)));
}
if args.contains_key("partitions") {
@@ -601,10 +605,15 @@ async fn proc_drop_global_index(
let mut builder = table.new_btree_global_index_drop_builder();
builder.with_index_column(index_column);
+ builder.with_index_type(index_type);
builder.execute().await.map_err(to_datafusion_error)?;
ok_result(ctx)
}
+fn is_sorted_global_index_type(index_type: &str) -> bool {
+ index_type.eq_ignore_ascii_case("btree") ||
index_type.eq_ignore_ascii_case("bitmap")
+}
+
fn parse_key_value_options(options: &str) -> DFResult<HashMap<String, String>>
{
let mut parsed = HashMap::new();
for entry in options.split(',').map(str::trim).filter(|s| !s.is_empty()) {
diff --git a/crates/integrations/datafusion/tests/procedures.rs
b/crates/integrations/datafusion/tests/procedures.rs
index 9e6a256..15ac8d1 100644
--- a/crates/integrations/datafusion/tests/procedures.rs
+++ b/crates/integrations/datafusion/tests/procedures.rs
@@ -153,13 +153,13 @@ async fn test_create_global_index_requires_index_column()
{
}
#[tokio::test]
-async fn test_create_global_index_rejects_non_btree() {
- let (_tmp, sql_context) =
setup_btree_global_index_table("btree_bad_type").await;
+async fn test_create_global_index_rejects_unsupported_index_type() {
+ let (_tmp, sql_context) =
setup_btree_global_index_table("global_index_bad_type").await;
assert_sql_error(
&sql_context,
- "CALL sys.create_global_index(table => 'test_db.btree_bad_type',
index_column => 'id', index_type => 'bitmap')",
- "only supports index_type => 'btree'",
+ "CALL sys.create_global_index(table =>
'test_db.global_index_bad_type', index_column => 'id', index_type =>
'full-text')",
+ "only supports index_type => 'btree', 'bitmap'",
)
.await;
}
@@ -208,6 +208,75 @@ async fn
test_create_global_index_builds_btree_and_filter_reads() {
assert_eq!(rows, vec![(2, "bob".to_string())]);
}
+#[tokio::test]
+async fn test_create_global_index_btree_string_fallback_scan_reads() {
+ let (_tmp, sql_context) =
setup_btree_global_index_table("btree_string_fallback").await;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.btree_string_fallback (id, name) VALUES \
+ (1, 'alice'), (2, 'alpine'), (3, 'bob'), (4, 'carol'), (5, 'malice')",
+ )
+ .await;
+
+ exec(
+ &sql_context,
+ "CALL sys.create_global_index(table =>
'test_db.btree_string_fallback', index_column => 'name', index_type =>
'btree')",
+ )
+ .await;
+
+ let rows = collect_id_name(
+ &sql_context,
+ "SELECT id, name FROM paimon.test_db.btree_string_fallback WHERE name
LIKE 'a%c_'",
+ )
+ .await;
+ assert_eq!(rows, vec![(1, "alice".to_string())]);
+}
+
+#[tokio::test]
+async fn test_create_global_index_builds_bitmap_with_java_format() {
+ let (_tmp, sql_context) =
setup_btree_global_index_table("bitmap_build").await;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.bitmap_build (id, name) VALUES (1,
'alice'), (2, 'bob'), (3, 'alice')",
+ )
+ .await;
+
+ exec(
+ &sql_context,
+ "CALL sys.create_global_index(table => 'test_db.bitmap_build',
index_column => 'name', index_type => 'bitmap')",
+ )
+ .await;
+
+ let index_count = row_count(
+ &sql_context,
+ "SELECT * FROM paimon.test_db.`bitmap_build$table_indexes` \
+ WHERE index_type = 'bitmap' AND row_range_start = 0 AND row_range_end
= 2 \
+ AND index_field_name = 'name'",
+ )
+ .await;
+ assert_eq!(index_count, 1);
+
+ let rows = collect_id_name(
+ &sql_context,
+ "SELECT id, name FROM paimon.test_db.bitmap_build WHERE name =
'alice'",
+ )
+ .await;
+ assert_eq!(
+ rows,
+ vec![(1, "alice".to_string()), (3, "alice".to_string())]
+ );
+
+ let contains_rows = collect_id_name(
+ &sql_context,
+ "SELECT id, name FROM paimon.test_db.bitmap_build WHERE name LIKE
'%lic%'",
+ )
+ .await;
+ assert_eq!(
+ contains_rows,
+ vec![(1, "alice".to_string()), (3, "alice".to_string())]
+ );
+}
+
#[tokio::test]
async fn test_drop_global_index_removes_btree_and_reads_fallback() {
let (_tmp, sql_context) =
setup_btree_global_index_table("btree_drop").await;
@@ -256,6 +325,35 @@ async fn
test_drop_global_index_removes_btree_and_reads_fallback() {
assert_eq!(rows, vec![(2, "bob".to_string()), (3, "carol".to_string())]);
}
+#[tokio::test]
+async fn test_drop_global_index_removes_bitmap() {
+ let (_tmp, sql_context) =
setup_btree_global_index_table("bitmap_drop").await;
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.bitmap_drop (id, name) VALUES (1,
'alice'), (2, 'bob')",
+ )
+ .await;
+ exec(
+ &sql_context,
+ "CALL sys.create_global_index(table => 'test_db.bitmap_drop',
index_column => 'name', index_type => 'bitmap')",
+ )
+ .await;
+
+ exec(
+ &sql_context,
+ "CALL sys.drop_global_index(table => 'test_db.bitmap_drop',
index_column => 'name', index_type => 'bitmap')",
+ )
+ .await;
+
+ let index_count = row_count(
+ &sql_context,
+ "SELECT * FROM paimon.test_db.`bitmap_drop$table_indexes` \
+ WHERE index_type = 'bitmap' AND index_field_name = 'name'",
+ )
+ .await;
+ assert_eq!(index_count, 0);
+}
+
#[tokio::test]
async fn test_drop_global_index_requires_index_column() {
let (_tmp, sql_context) =
setup_btree_global_index_table("btree_drop_missing_col").await;
@@ -269,13 +367,13 @@ async fn test_drop_global_index_requires_index_column() {
}
#[tokio::test]
-async fn test_drop_global_index_rejects_non_btree() {
- let (_tmp, sql_context) =
setup_btree_global_index_table("btree_drop_bad_type").await;
+async fn test_drop_global_index_rejects_unsupported_index_type() {
+ let (_tmp, sql_context) =
setup_btree_global_index_table("global_index_drop_bad_type").await;
assert_sql_error(
&sql_context,
- "CALL sys.drop_global_index(table => 'test_db.btree_drop_bad_type',
index_column => 'id', index_type => 'bitmap')",
- "only supports index_type => 'btree'",
+ "CALL sys.drop_global_index(table =>
'test_db.global_index_drop_bad_type', index_column => 'id', index_type =>
'full-text')",
+ "only supports index_type => 'btree' or 'bitmap'",
)
.await;
}
diff --git a/crates/paimon/src/btree/key_serde.rs
b/crates/paimon/src/btree/key_serde.rs
index 22d7254..d1b7834 100644
--- a/crates/paimon/src/btree/key_serde.rs
+++ b/crates/paimon/src/btree/key_serde.rs
@@ -84,15 +84,9 @@ pub fn make_key_comparator(data_type: &DataType) ->
KeyComparator {
})
}
DataType::Decimal(d) if d.precision() > DECIMAL_COMPACT_PRECISION => {
- // Non-compact: raw unscaled bytes, compare as big-endian signed
+ // Non-compact Decimal keys use Java BigInteger.toByteArray()
bytes.
Box::new(|a: &[u8], b: &[u8]| {
- let a_neg = !a.is_empty() && (a[0] & 0x80) != 0;
- let b_neg = !b.is_empty() && (b[0] & 0x80) != 0;
- match (a_neg, b_neg) {
- (true, false) => Ordering::Less,
- (false, true) => Ordering::Greater,
- _ => a.cmp(b),
- }
+
decode_java_big_integer_i128(a).cmp(&decode_java_big_integer_i128(b))
})
}
// Compact Timestamp/LocalZonedTimestamp (precision <= 3): millis as
i64 LE
@@ -142,10 +136,14 @@ pub fn serialize_datum(datum: &Datum, data_type:
&DataType) -> Vec<u8> {
precision,
..
} => {
- if *precision <= DECIMAL_COMPACT_PRECISION {
+ let key_precision = match data_type {
+ DataType::Decimal(decimal_type) => decimal_type.precision(),
+ _ => *precision,
+ };
+ if key_precision <= DECIMAL_COMPACT_PRECISION {
(*unscaled as i64).to_le_bytes().to_vec()
} else {
- unscaled.to_be_bytes().to_vec()
+ encode_java_big_integer_i128(*unscaled)
}
}
Datum::Bytes(v) => v.clone(),
@@ -160,3 +158,118 @@ pub fn serialize_datum(datum: &Datum, data_type:
&DataType) -> Vec<u8> {
}
}
}
+
+fn encode_java_big_integer_i128(value: i128) -> Vec<u8> {
+ let bytes = value.to_be_bytes();
+ let mut start = 0;
+ while start < bytes.len() - 1 {
+ let current = bytes[start];
+ let next = bytes[start + 1];
+ if (current == 0x00 && next & 0x80 == 0) || (current == 0xff && next &
0x80 != 0) {
+ start += 1;
+ } else {
+ break;
+ }
+ }
+ bytes[start..].to_vec()
+}
+
+fn decode_java_big_integer_i128(bytes: &[u8]) -> i128 {
+ if bytes.is_empty() {
+ return 0;
+ }
+ let negative = bytes[0] & 0x80 != 0;
+ let mut value = if negative { -1 } else { 0 };
+ for &byte in bytes {
+ value = (value << 8) | i128::from(byte);
+ }
+ value
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::{DecimalType, IntType};
+
+ #[test]
+ fn test_serialize_non_compact_decimal_matches_java_big_integer_bytes() {
+ let data_type = DataType::Decimal(DecimalType::new(20, 0).unwrap());
+
+ assert_eq!(
+ serialize_datum(
+ &Datum::Decimal {
+ unscaled: 127,
+ precision: 20,
+ scale: 0,
+ },
+ &data_type,
+ ),
+ vec![0x7f]
+ );
+ assert_eq!(
+ serialize_datum(
+ &Datum::Decimal {
+ unscaled: 128,
+ precision: 20,
+ scale: 0,
+ },
+ &data_type,
+ ),
+ vec![0x00, 0x80]
+ );
+ assert_eq!(
+ serialize_datum(
+ &Datum::Decimal {
+ unscaled: -129,
+ precision: 20,
+ scale: 0,
+ },
+ &data_type,
+ ),
+ vec![0xff, 0x7f]
+ );
+ }
+
+ #[test]
+ fn
test_decimal_key_serialization_uses_column_precision_not_literal_precision() {
+ let compact_column = DataType::Decimal(DecimalType::new(10,
0).unwrap());
+ let non_compact_column = DataType::Decimal(DecimalType::new(20,
0).unwrap());
+
+ let compact_column_key = serialize_datum(
+ &Datum::Decimal {
+ unscaled: 128,
+ precision: 20,
+ scale: 0,
+ },
+ &compact_column,
+ );
+ assert_eq!(compact_column_key, 128i64.to_le_bytes());
+
+ let non_compact_column_key = serialize_datum(
+ &Datum::Decimal {
+ unscaled: 128,
+ precision: 10,
+ scale: 0,
+ },
+ &non_compact_column,
+ );
+ assert_eq!(non_compact_column_key, vec![0x00, 0x80]);
+ }
+
+ #[test]
+ fn test_compare_non_compact_decimal_uses_numeric_order() {
+ let cmp = make_key_comparator(&DataType::Decimal(DecimalType::new(20,
0).unwrap()));
+ let key_127 = encode_java_big_integer_i128(127);
+ let key_128 = encode_java_big_integer_i128(128);
+ let key_minus_129 = encode_java_big_integer_i128(-129);
+
+ assert_eq!(cmp(&key_127, &key_128), Ordering::Less);
+ assert_eq!(cmp(&key_minus_129, &key_127), Ordering::Less);
+ }
+
+ #[test]
+ fn test_compact_numeric_still_uses_little_endian() {
+ let key = serialize_datum(&Datum::Int(42),
&DataType::Int(IntType::new()));
+ assert_eq!(key, 42i32.to_le_bytes());
+ }
+}
diff --git a/crates/paimon/src/btree/meta.rs b/crates/paimon/src/btree/meta.rs
index a7ae174..ea511ea 100644
--- a/crates/paimon/src/btree/meta.rs
+++ b/crates/paimon/src/btree/meta.rs
@@ -19,14 +19,19 @@
//!
//! Serialization format (little-endian):
//! ```text
-//! | first_key_length (4) | first_key_bytes | last_key_length (4) |
last_key_bytes | has_nulls (1) |
+//! | first_key_length (4) | first_key_bytes | last_key_length (4) |
last_key_bytes |
+//! | has_nulls (1) | format_version (1) | null_key_flags (1) |
//! ```
-//! If first_key or last_key is null, their length is written as 0.
+//! Null key flags distinguish empty serialized keys from absent keys.
use crate::spec::PredicateOperator;
use std::cmp::Ordering;
use std::io;
+const FORMAT_VERSION_WITH_NULL_FLAGS: u8 = 1;
+const FIRST_KEY_IS_NULL: u8 = 1;
+const LAST_KEY_IS_NULL: u8 = 1 << 1;
+
/// Index meta for each BTree index file.
#[derive(Debug, Clone)]
pub struct BTreeIndexMeta {
@@ -111,11 +116,12 @@ impl BTreeIndexMeta {
cmp(first_key, to_key) != Ordering::Greater && cmp(last_key, from_key)
!= Ordering::Less
}
- /// Serialize to bytes (compatible with Java BTreeIndexMeta.serialize()).
+ /// Serialize to bytes (compatible with Java
SortedIndexFileMeta.serialize()).
pub fn serialize(&self) -> Vec<u8> {
let fk_len = self.first_key.as_ref().map_or(0, |k| k.len());
let lk_len = self.last_key.as_ref().map_or(0, |k| k.len());
- let mut buf = Vec::with_capacity(fk_len + lk_len + 9);
+ let mut buf = Vec::with_capacity(fk_len + lk_len + 11);
+ let mut null_key_flags = 0u8;
// first key
match &self.first_key {
@@ -125,6 +131,7 @@ impl BTreeIndexMeta {
}
None => {
buf.extend_from_slice(&0i32.to_le_bytes());
+ null_key_flags |= FIRST_KEY_IS_NULL;
}
}
@@ -136,16 +143,19 @@ impl BTreeIndexMeta {
}
None => {
buf.extend_from_slice(&0i32.to_le_bytes());
+ null_key_flags |= LAST_KEY_IS_NULL;
}
}
// has_nulls
buf.push(if self.has_nulls { 1 } else { 0 });
+ buf.push(FORMAT_VERSION_WITH_NULL_FLAGS);
+ buf.push(null_key_flags);
buf
}
- /// Deserialize from bytes (compatible with Java
BTreeIndexMeta.deserialize()).
+ /// Deserialize from bytes (compatible with Java
SortedIndexFileMeta.deserialize()).
pub fn deserialize(data: &[u8]) -> io::Result<Self> {
if data.len() < 9 {
return Err(io::Error::new(
@@ -158,9 +168,7 @@ impl BTreeIndexMeta {
let fk_len = i32::from_le_bytes(data[pos..pos +
4].try_into().unwrap()) as usize;
pos += 4;
- let first_key = if fk_len == 0 {
- None
- } else {
+ let mut first_key = {
let key = data[pos..pos + fk_len].to_vec();
pos += fk_len;
Some(key)
@@ -168,15 +176,31 @@ impl BTreeIndexMeta {
let lk_len = i32::from_le_bytes(data[pos..pos +
4].try_into().unwrap()) as usize;
pos += 4;
- let last_key = if lk_len == 0 {
- None
- } else {
+ let mut last_key = {
let key = data[pos..pos + lk_len].to_vec();
pos += lk_len;
Some(key)
};
let has_nulls = data[pos] == 1;
+ pos += 1;
+
+ if data.len().saturating_sub(pos) >= 2 {
+ let format_version = data[pos];
+ pos += 1;
+ if format_version == FORMAT_VERSION_WITH_NULL_FLAGS {
+ let null_key_flags = data[pos];
+ if null_key_flags & FIRST_KEY_IS_NULL != 0 {
+ first_key = None;
+ }
+ if null_key_flags & LAST_KEY_IS_NULL != 0 {
+ last_key = None;
+ }
+ }
+ } else if fk_len == 0 && lk_len == 0 && has_nulls {
+ first_key = None;
+ last_key = None;
+ }
Ok(Self {
first_key,
@@ -210,6 +234,17 @@ mod tests {
assert!(decoded.has_nulls);
}
+ #[test]
+ fn test_meta_empty_string_keys_are_not_null() {
+ let meta = BTreeIndexMeta::new(Some(Vec::new()), Some(Vec::new()),
false);
+ let encoded = meta.serialize();
+ let decoded = BTreeIndexMeta::deserialize(&encoded).unwrap();
+ assert_eq!(decoded.first_key, Some(Vec::new()));
+ assert_eq!(decoded.last_key, Some(Vec::new()));
+ assert!(!decoded.only_nulls());
+ assert!(!decoded.has_nulls);
+ }
+
#[test]
fn test_meta_no_nulls() {
let meta = BTreeIndexMeta::new(Some(b"key1".to_vec()),
Some(b"key2".to_vec()), false);
diff --git a/crates/paimon/src/btree/mod.rs b/crates/paimon/src/btree/mod.rs
index ca68faa..0371013 100644
--- a/crates/paimon/src/btree/mod.rs
+++ b/crates/paimon/src/btree/mod.rs
@@ -43,7 +43,7 @@ mod meta;
pub(crate) mod query;
mod reader;
mod sst_file;
-mod var_len;
+pub(crate) mod var_len;
mod writer;
pub use block::BlockCompressionType;
diff --git a/crates/paimon/src/btree/query.rs b/crates/paimon/src/btree/query.rs
index 0aed9d4..df4c275 100644
--- a/crates/paimon/src/btree/query.rs
+++ b/crates/paimon/src/btree/query.rs
@@ -22,7 +22,7 @@
use crate::btree::key_serde::serialize_datum;
use crate::btree::reader::BTreeIndexReader;
-use crate::spec::{DataType, Datum, PredicateOperator};
+use crate::spec::{like_match, DataType, Datum, PredicateOperator};
use roaring::RoaringTreemap;
use std::cmp::Ordering;
use std::io;
@@ -109,17 +109,71 @@ where
all_non_null -= inside;
Ok(all_non_null)
}
- PredicateOperator::StartsWith
- | PredicateOperator::EndsWith
- | PredicateOperator::Contains
- | PredicateOperator::Like => Err(io::Error::new(
- io::ErrorKind::Unsupported,
- format!("BTree index does not support op: {op}"),
- )),
+ PredicateOperator::StartsWith => {
+ let key = serialize_datum(&literals[0], data_type);
+ self.query_prefix(&key).await
+ }
+ PredicateOperator::EndsWith => {
+ ensure_character_string(data_type, op)?;
+ let suffix = serialize_datum(&literals[0], data_type);
+ if suffix.is_empty() {
+ return self.all_non_null_rows().await;
+ }
+ self.scan_entries(move |key| key.ends_with(&suffix)).await
+ }
+ PredicateOperator::Contains => {
+ ensure_character_string(data_type, op)?;
+ let needle = serialize_datum(&literals[0], data_type);
+ if needle.is_empty() {
+ return self.all_non_null_rows().await;
+ }
+ self.scan_entries(move |key| contains_bytes(key, &needle))
+ .await
+ }
+ PredicateOperator::Like => {
+ ensure_character_string(data_type, op)?;
+ let pattern = string_literal(literals, op)?.to_string();
+ self.scan_entries(move |key| {
+ std::str::from_utf8(key).is_ok_and(|value|
like_match(value, &pattern))
+ })
+ .await
+ }
}
}
}
+fn ensure_character_string(data_type: &DataType, op: PredicateOperator) ->
io::Result<()> {
+ if matches!(data_type, DataType::Char(_) | DataType::VarChar(_)) {
+ Ok(())
+ } else {
+ Err(io::Error::new(
+ io::ErrorKind::Unsupported,
+ format!("BTree index {op} only supports string columns"),
+ ))
+ }
+}
+
+fn string_literal(literals: &[Datum], op: PredicateOperator) ->
io::Result<&str> {
+ match literals.first() {
+ Some(Datum::String(value)) => Ok(value),
+ Some(other) => Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ format!("BTree index {op} requires a string literal, got {other}"),
+ )),
+ None => Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ format!("BTree index {op} requires one literal"),
+ )),
+ }
+}
+
+fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
+ !needle.is_empty()
+ && haystack
+ .windows(needle.len())
+ .any(|window| window == needle)
+}
+
/// Detected between pattern from predicate pairs.
pub(crate) struct BetweenInfo<'a> {
pub from: &'a Datum,
diff --git a/crates/paimon/src/btree/reader.rs
b/crates/paimon/src/btree/reader.rs
index 6de0c3a..f983633 100644
--- a/crates/paimon/src/btree/reader.rs
+++ b/crates/paimon/src/btree/reader.rs
@@ -113,6 +113,36 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
.await
}
+ /// Scan all non-null entries and collect row ids for keys matching
`predicate`.
+ pub async fn scan_entries(
+ &self,
+ predicate: impl Fn(&[u8]) -> bool + Send + Sync,
+ ) -> io::Result<RoaringTreemap> {
+ let Some(min_key) = self.min_key.as_deref() else {
+ return Ok(RoaringTreemap::new());
+ };
+
+ let cmp = &self.key_comparator;
+ let index_block = self.sst_reader.index_block();
+ let (_, mut index_iter) = index_block.seek_and_iter(min_key, cmp);
+ let mut result = RoaringTreemap::new();
+
+ while let Some((_key, handle_bytes)) = index_iter.next() {
+ let handle = BlockHandle::decode(handle_bytes)?;
+ let block = self.read_data_block(&handle).await?;
+ let mut offset = 0;
+ while offset < block.data.len() {
+ let (key, value, next_offset) = block.read_entry_at(offset);
+ offset = next_offset;
+ if predicate(key) {
+ insert_row_ids_into(value, &mut result)?;
+ }
+ }
+ }
+
+ Ok(result)
+ }
+
/// Range query: returns a bitmap of all row ids whose keys fall in [from,
to]
/// with configurable inclusivity. Reads data blocks on demand.
pub async fn range_query(
diff --git a/crates/paimon/src/btree/tests.rs b/crates/paimon/src/btree/tests.rs
index 0ef9687..9641508 100644
--- a/crates/paimon/src/btree/tests.rs
+++ b/crates/paimon/src/btree/tests.rs
@@ -17,9 +17,11 @@
use crate::btree::block::BlockCompressionType;
use crate::btree::meta::BTreeIndexMeta;
+use crate::btree::query::IndexQuery;
use crate::btree::reader::BTreeIndexReader;
use crate::btree::test_util::{BytesFileRead, VecFileWrite};
use crate::btree::writer::BTreeIndexWriter;
+use crate::spec::{DataType, Datum, PredicateOperator, VarCharType};
use bytes::Bytes;
fn int_key(v: i32) -> Vec<u8> {
@@ -217,6 +219,59 @@ async fn test_prefix_query() {
assert_eq!(bm.len(), keys.len() as u64);
}
+#[tokio::test]
+async fn test_string_fallback_scan_query() {
+ let buf = VecFileWrite::new();
+ let mut writer = BTreeIndexWriter::new(Box::new(buf.clone()), 32,
BlockCompressionType::None);
+
+ writer.write(None, 99).await.unwrap();
+ let keys = [
+ ("alice", 0),
+ ("alpine", 1),
+ ("bob", 2),
+ ("carol", 3),
+ ("malice", 4),
+ ("office", 5),
+ ];
+ for (key, row_id) in keys {
+ writer.write(Some(key.as_bytes()), row_id).await.unwrap();
+ }
+
+ let result = writer.finish().await.unwrap();
+ let reader = write_and_open(&buf, &result, |a: &[u8], b: &[u8]|
a.cmp(b)).await;
+ let data_type = DataType::VarChar(VarCharType::string_type());
+
+ let ends_with = reader
+ .query(
+ PredicateOperator::EndsWith,
+ &[Datum::String("ice".to_string())],
+ &data_type,
+ )
+ .await
+ .unwrap();
+ assert_eq!(ends_with.iter().collect::<Vec<_>>(), vec![0, 4, 5]);
+
+ let contains = reader
+ .query(
+ PredicateOperator::Contains,
+ &[Datum::String("o".to_string())],
+ &data_type,
+ )
+ .await
+ .unwrap();
+ assert_eq!(contains.iter().collect::<Vec<_>>(), vec![2, 3, 5]);
+
+ let like = reader
+ .query(
+ PredicateOperator::Like,
+ &[Datum::String("a%c_".to_string())],
+ &data_type,
+ )
+ .await
+ .unwrap();
+ assert_eq!(like.iter().collect::<Vec<_>>(), vec![0]);
+}
+
#[tokio::test]
async fn test_not_equal_query() {
let buf = VecFileWrite::new();
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index e7d0fa6..3d5d79b 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -25,6 +25,8 @@ const GLOBAL_INDEX_SEARCH_MODE_OPTION: &str =
"global-index.search-mode";
const GLOBAL_INDEX_ROW_COUNT_PER_SHARD_OPTION: &str =
"global-index.row-count-per-shard";
const GLOBAL_INDEX_COLUMN_UPDATE_ACTION_OPTION: &str =
"global-index.column-update-action";
const SORTED_INDEX_RECORDS_PER_RANGE_OPTION: &str =
"sorted-index.records-per-range";
+const BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION: &str =
"btree-index.fallback-scan-max-size";
+const BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION: &str =
"bitmap-index.fallback-scan-max-size";
const SOURCE_SPLIT_TARGET_SIZE_OPTION: &str = "source.split.target-size";
const SOURCE_SPLIT_OPEN_FILE_COST_OPTION: &str = "source.split.open-file-cost";
const PARTITION_DEFAULT_NAME_OPTION: &str = "partition.default-name";
@@ -87,6 +89,7 @@ const DEFAULT_WRITE_PARQUET_BUFFER_SIZE: i64 = 256 * 1024 *
1024;
const DYNAMIC_BUCKET_TARGET_ROW_NUM_OPTION: &str =
"dynamic-bucket.target-row-num";
const DEFAULT_DYNAMIC_BUCKET_TARGET_ROW_NUM: i64 = 200_000;
const DEFAULT_GLOBAL_INDEX_ROW_COUNT_PER_SHARD: i64 = 100_000;
+const DEFAULT_GLOBAL_INDEX_FALLBACK_SCAN_MAX_SIZE: i64 = 256 * 1024 * 1024;
const BLOB_AS_DESCRIPTOR_OPTION: &str = "blob-as-descriptor";
const BLOB_DESCRIPTOR_FIELD_OPTION: &str = "blob-descriptor-field";
@@ -434,6 +437,37 @@ impl<'a> CoreOptions<'a> {
Ok(value)
}
+ pub fn btree_index_fallback_scan_max_size(&self) -> crate::Result<i64> {
+ self.fallback_scan_max_size(BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION)
+ }
+
+ pub fn bitmap_index_fallback_scan_max_size(&self) -> crate::Result<i64> {
+ self.fallback_scan_max_size(BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION)
+ }
+
+ fn fallback_scan_max_size(&self, option_name: &'static str) ->
crate::Result<i64> {
+ let value = match self.options.get(option_name) {
+ Some(raw) => parse_memory_size(raw).ok_or_else(||
crate::Error::DataInvalid {
+ message: format!(
+ "Option '{}' must be a valid memory size, got: {}",
+ option_name, raw
+ ),
+ source: None,
+ })?,
+ None => DEFAULT_GLOBAL_INDEX_FALLBACK_SCAN_MAX_SIZE,
+ };
+ if value < 0 {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Option '{}' must be greater than or equal to 0, got: {}",
+ option_name, value
+ ),
+ source: None,
+ });
+ }
+ Ok(value)
+ }
+
pub fn global_index_column_update_action(
&self,
) -> crate::Result<GlobalIndexColumnUpdateAction> {
@@ -836,6 +870,14 @@ mod tests {
core_options.sorted_index_records_per_range().unwrap(),
100_000
);
+ assert_eq!(
+ core_options.btree_index_fallback_scan_max_size().unwrap(),
+ 256 * 1024 * 1024
+ );
+ assert_eq!(
+ core_options.bitmap_index_fallback_scan_max_size().unwrap(),
+ 256 * 1024 * 1024
+ );
assert_eq!(
core_options.global_index_column_update_action().unwrap(),
GlobalIndexColumnUpdateAction::ThrowError
@@ -865,6 +907,14 @@ mod tests {
SORTED_INDEX_RECORDS_PER_RANGE_OPTION.to_string(),
"4096".to_string(),
),
+ (
+ BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION.to_string(),
+ "4 mb".to_string(),
+ ),
+ (
+ BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION.to_string(),
+ "8 mb".to_string(),
+ ),
(
GLOBAL_INDEX_COLUMN_UPDATE_ACTION_OPTION.to_string(),
"DROP_PARTITION_INDEX".to_string(),
@@ -883,6 +933,14 @@ mod tests {
2048
);
assert_eq!(core_options.sorted_index_records_per_range().unwrap(),
4096);
+ assert_eq!(
+ core_options.btree_index_fallback_scan_max_size().unwrap(),
+ 4 * 1024 * 1024
+ );
+ assert_eq!(
+ core_options.bitmap_index_fallback_scan_max_size().unwrap(),
+ 8 * 1024 * 1024
+ );
assert_eq!(
core_options.global_index_column_update_action().unwrap(),
GlobalIndexColumnUpdateAction::DropPartitionIndex
@@ -955,6 +1013,45 @@ mod tests {
}
}
+ #[test]
+ fn test_global_index_fallback_scan_max_size_values() {
+ for option_name in [
+ BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION,
+ BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION,
+ ] {
+ let options = HashMap::from([(option_name.to_string(),
"0".to_string())]);
+ let core = CoreOptions::new(&options);
+ let value = match option_name {
+ BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION => {
+ core.btree_index_fallback_scan_max_size()
+ }
+ BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION => {
+ core.bitmap_index_fallback_scan_max_size()
+ }
+ _ => unreachable!(),
+ };
+ assert_eq!(value.unwrap(), 0);
+
+ for value in ["-1", "abc"] {
+ let options = HashMap::from([(option_name.to_string(),
value.to_string())]);
+ let core = CoreOptions::new(&options);
+
+ let err = match option_name {
+ BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION => {
+ core.btree_index_fallback_scan_max_size()
+ }
+ BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION => {
+ core.bitmap_index_fallback_scan_max_size()
+ }
+ _ => unreachable!(),
+ }
+ .expect_err("invalid fallback scan max size should fail");
+ assert!(matches!(err, crate::Error::DataInvalid { message, .. }
+ if message.contains(option_name)));
+ }
+ }
+ }
+
#[test]
fn test_parse_memory_size() {
assert_eq!(parse_memory_size("1024"), Some(1024));
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index 12a0cc6..eb4dc54 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -94,6 +94,7 @@ pub(crate) use predicate::datum_cmp;
pub(crate) use predicate::eval_row;
pub(crate) use predicate::extract_datum;
pub(crate) use predicate::java_bytes_cmp;
+pub(crate) use predicate::like_match;
pub use predicate::{
field_idx_to_partition_idx, Datum, Predicate, PredicateBuilder,
PredicateOperator,
};
diff --git a/crates/paimon/src/spec/predicate.rs
b/crates/paimon/src/spec/predicate.rs
index d44463e..badabb8 100644
--- a/crates/paimon/src/spec/predicate.rs
+++ b/crates/paimon/src/spec/predicate.rs
@@ -1165,7 +1165,7 @@ fn optimize_like_pattern(pattern: &str) -> LikeShape {
/// * `_` matches exactly one character,
/// * `\X` matches the literal `X` for any character `X` (the backslash is
/// consumed); a trailing `\` matches a literal backslash.
-fn like_match(value: &str, pattern: &str) -> bool {
+pub(crate) fn like_match(value: &str, pattern: &str) -> bool {
let value: Vec<char> = value.chars().collect();
let pattern: Vec<char> = pattern.chars().collect();
like_match_chars(&value, 0, &pattern, 0)
diff --git a/crates/paimon/src/table/bitmap_global_index_reader.rs
b/crates/paimon/src/table/bitmap_global_index_reader.rs
new file mode 100644
index 0000000..0c6172b
--- /dev/null
+++ b/crates/paimon/src/table/bitmap_global_index_reader.rs
@@ -0,0 +1,892 @@
+// 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.
+
+//! Reader for Java Paimon's bitmap global index file format.
+//!
+//! Reference: `org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexFormat`.
+
+use crate::btree::var_len::{decode_var_int, decode_var_long, encode_var_int,
encode_var_long};
+use crate::btree::BTreeIndexMeta;
+use crate::btree::{make_key_comparator, serialize_datum, BlockCompressionType};
+use crate::io::{FileRead, FileWrite};
+use crate::spec::{like_match, DataType, Datum, PredicateOperator};
+use bytes::Bytes;
+use roaring::RoaringTreemap;
+use std::cmp::Ordering;
+use std::collections::BTreeMap;
+use std::io::Write;
+use std::io::{self, Cursor, Read};
+
+const MAGIC: i32 = 0x4247_4958;
+const VERSION: i32 = 1;
+const FOOTER_LENGTH: usize = 48;
+const BLOCK_TRAILER_LENGTH: usize = 5;
+
+#[derive(Clone, Copy)]
+struct BlockInfo {
+ offset: u64,
+ length: usize,
+}
+
+#[derive(Clone)]
+struct DictionaryBlockMeta {
+ first_key: Vec<u8>,
+ block: BlockInfo,
+}
+
+struct DictionaryEntry {
+ key: Vec<u8>,
+ bitmap_block: BlockInfo,
+}
+
+struct Footer {
+ null_rows_block: BlockInfo,
+ non_null_rows_block: BlockInfo,
+ index_block: BlockInfo,
+}
+
+/// Result of finishing a Java-compatible bitmap global index write.
+pub(crate) struct BitmapWriteResult {
+ pub(crate) meta: BTreeIndexMeta,
+ pub(crate) row_count: u64,
+}
+
+/// Writer for Java Paimon's `BitmapGlobalIndexFormat`.
+pub(crate) struct BitmapGlobalIndexWriter<F: Fn(&[u8], &[u8]) -> Ordering> {
+ writer: Box<dyn FileWrite>,
+ dictionary_block_size: usize,
+ compression_type: BlockCompressionType,
+ key_comparator: F,
+ bitmaps: BTreeMap<Vec<u8>, RoaringTreemap>,
+ null_rows: RoaringTreemap,
+ non_null_rows: RoaringTreemap,
+ first_key: Option<Vec<u8>>,
+ last_key: Option<Vec<u8>>,
+ row_count: u64,
+}
+
+impl<F: Fn(&[u8], &[u8]) -> Ordering> BitmapGlobalIndexWriter<F> {
+ pub(crate) fn new(
+ writer: Box<dyn FileWrite>,
+ dictionary_block_size: usize,
+ compression_type: BlockCompressionType,
+ key_comparator: F,
+ ) -> Self {
+ Self {
+ writer,
+ dictionary_block_size,
+ compression_type,
+ key_comparator,
+ bitmaps: BTreeMap::new(),
+ null_rows: RoaringTreemap::new(),
+ non_null_rows: RoaringTreemap::new(),
+ first_key: None,
+ last_key: None,
+ row_count: 0,
+ }
+ }
+
+ pub(crate) fn write(&mut self, key: Option<&[u8]>, relative_row_id: i64)
-> io::Result<()> {
+ if relative_row_id < 0 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ format!("Bitmap global index row id must be non-negative:
{relative_row_id}"),
+ ));
+ }
+
+ self.row_count += 1;
+ match key {
+ Some(key) => {
+ let row_id = relative_row_id as u64;
+ self.non_null_rows.insert(row_id);
+ self.bitmaps.entry(key.to_vec()).or_default().insert(row_id);
+ self.update_min_max(key);
+ }
+ None => {
+ self.null_rows.insert(relative_row_id as u64);
+ }
+ }
+ Ok(())
+ }
+
+ pub(crate) async fn finish(mut self) -> io::Result<BitmapWriteResult> {
+ let mut bytes = Vec::new();
+ write_bitmap_index_bytes(
+ &mut bytes,
+ &self.null_rows,
+ &self.non_null_rows,
+ &self.bitmaps,
+ self.dictionary_block_size,
+ self.compression_type,
+ )?;
+ self.writer
+ .write(Bytes::from(bytes))
+ .await
+ .map_err(|e| io::Error::other(e.to_string()))?;
+ self.writer
+ .close()
+ .await
+ .map_err(|e| io::Error::other(e.to_string()))?;
+
+ Ok(BitmapWriteResult {
+ meta: BTreeIndexMeta::new(self.first_key, self.last_key,
!self.null_rows.is_empty()),
+ row_count: self.row_count,
+ })
+ }
+
+ fn update_min_max(&mut self, key: &[u8]) {
+ if self
+ .first_key
+ .as_ref()
+ .is_none_or(|existing| (self.key_comparator)(key,
existing).is_lt())
+ {
+ self.first_key = Some(key.to_vec());
+ }
+ if self
+ .last_key
+ .as_ref()
+ .is_none_or(|existing| (self.key_comparator)(key,
existing).is_gt())
+ {
+ self.last_key = Some(key.to_vec());
+ }
+ }
+}
+
+pub(crate) struct BitmapGlobalIndexReader {
+ reader: Box<dyn FileRead>,
+ footer: Footer,
+ dictionary_blocks: Vec<DictionaryBlockMeta>,
+}
+
+impl BitmapGlobalIndexReader {
+ pub(crate) async fn open(reader: Box<dyn FileRead>, file_size: u64) ->
io::Result<Self> {
+ let footer = read_footer(reader.as_ref(), file_size).await?;
+ let index_block = read_compressible_block(reader.as_ref(),
footer.index_block).await?;
+ let dictionary_blocks = read_index_block(&index_block)?;
+ Ok(Self {
+ reader,
+ footer,
+ dictionary_blocks,
+ })
+ }
+
+ pub(crate) async fn query(
+ &self,
+ op: PredicateOperator,
+ literals: &[Datum],
+ data_type: &DataType,
+ ) -> io::Result<RoaringTreemap> {
+ match op {
+ PredicateOperator::Eq => {
+ let key = serialize_datum(&literals[0], data_type);
+ self.equal(&key).await
+ }
+ PredicateOperator::NotEq => {
+ let mut result = self.is_not_null().await?;
+ let key = serialize_datum(&literals[0], data_type);
+ result -= self.equal(&key).await?;
+ Ok(result)
+ }
+ PredicateOperator::In => {
+ let keys = literals
+ .iter()
+ .map(|literal| serialize_datum(literal, data_type))
+ .collect::<Vec<_>>();
+ self.in_keys(&keys).await
+ }
+ PredicateOperator::NotIn => {
+ let mut result = self.is_not_null().await?;
+ let keys = literals
+ .iter()
+ .map(|literal| serialize_datum(literal, data_type))
+ .collect::<Vec<_>>();
+ result -= self.in_keys(&keys).await?;
+ Ok(result)
+ }
+ PredicateOperator::IsNull => self.is_null().await,
+ PredicateOperator::IsNotNull => self.is_not_null().await,
+ PredicateOperator::Lt => {
+ let key = serialize_datum(&literals[0], data_type);
+ self.scan_dictionary(data_type, |candidate, cmp|
cmp(candidate, &key).is_lt())
+ .await
+ }
+ PredicateOperator::LtEq => {
+ let key = serialize_datum(&literals[0], data_type);
+ self.scan_dictionary(data_type, |candidate, cmp|
!cmp(candidate, &key).is_gt())
+ .await
+ }
+ PredicateOperator::Gt => {
+ let key = serialize_datum(&literals[0], data_type);
+ self.scan_dictionary(data_type, |candidate, cmp|
cmp(candidate, &key).is_gt())
+ .await
+ }
+ PredicateOperator::GtEq => {
+ let key = serialize_datum(&literals[0], data_type);
+ self.scan_dictionary(data_type, |candidate, cmp|
!cmp(candidate, &key).is_lt())
+ .await
+ }
+ PredicateOperator::Between => {
+ let from = serialize_datum(&literals[0], data_type);
+ let to = serialize_datum(&literals[1], data_type);
+ self.scan_dictionary(data_type, |candidate, cmp| {
+ !cmp(candidate, &from).is_lt() && !cmp(candidate,
&to).is_gt()
+ })
+ .await
+ }
+ PredicateOperator::NotBetween => {
+ let mut result = self.is_not_null().await?;
+ let from = serialize_datum(&literals[0], data_type);
+ let to = serialize_datum(&literals[1], data_type);
+ let inside = self
+ .scan_dictionary(data_type, |candidate, cmp| {
+ !cmp(candidate, &from).is_lt() && !cmp(candidate,
&to).is_gt()
+ })
+ .await?;
+ result -= inside;
+ Ok(result)
+ }
+ PredicateOperator::StartsWith => {
+ if !is_character_string(data_type) {
+ return Err(io::Error::new(
+ io::ErrorKind::Unsupported,
+ "Bitmap global index starts_with only supports string
columns",
+ ));
+ }
+ let prefix = serialize_datum(&literals[0], data_type);
+ if prefix.is_empty() {
+ return self.is_not_null().await;
+ }
+ self.scan_serialized_dictionary(|candidate|
candidate.starts_with(&prefix))
+ .await
+ }
+ PredicateOperator::EndsWith => {
+ if !is_character_string(data_type) {
+ return Err(io::Error::new(
+ io::ErrorKind::Unsupported,
+ "Bitmap global index ends_with only supports string
columns",
+ ));
+ }
+ let suffix = serialize_datum(&literals[0], data_type);
+ if suffix.is_empty() {
+ return self.is_not_null().await;
+ }
+ self.scan_serialized_dictionary(|candidate|
candidate.ends_with(&suffix))
+ .await
+ }
+ PredicateOperator::Contains => {
+ if !is_character_string(data_type) {
+ return Err(io::Error::new(
+ io::ErrorKind::Unsupported,
+ "Bitmap global index contains only supports string
columns",
+ ));
+ }
+ let needle = serialize_datum(&literals[0], data_type);
+ if needle.is_empty() {
+ return self.is_not_null().await;
+ }
+ self.scan_serialized_dictionary(|candidate|
contains_bytes(candidate, &needle))
+ .await
+ }
+ PredicateOperator::Like => {
+ if !is_character_string(data_type) {
+ return Err(io::Error::new(
+ io::ErrorKind::Unsupported,
+ "Bitmap global index like only supports string
columns",
+ ));
+ }
+ let pattern = string_literal(literals, op)?.to_string();
+ self.scan_serialized_dictionary(|candidate| {
+ std::str::from_utf8(candidate).is_ok_and(|value|
like_match(value, &pattern))
+ })
+ .await
+ }
+ }
+ }
+
+ pub(crate) async fn range_query(
+ &self,
+ from: &[u8],
+ to: &[u8],
+ data_type: &DataType,
+ from_inclusive: bool,
+ to_inclusive: bool,
+ ) -> io::Result<RoaringTreemap> {
+ self.scan_dictionary(data_type, |candidate, cmp| {
+ let from_cmp = cmp(candidate, from);
+ let to_cmp = cmp(candidate, to);
+ (from_cmp.is_gt() || (from_inclusive && from_cmp.is_eq()))
+ && (to_cmp.is_lt() || (to_inclusive && to_cmp.is_eq()))
+ })
+ .await
+ }
+
+ async fn is_null(&self) -> io::Result<RoaringTreemap> {
+ self.read_bitmap(self.footer.null_rows_block).await
+ }
+
+ async fn is_not_null(&self) -> io::Result<RoaringTreemap> {
+ self.read_bitmap(self.footer.non_null_rows_block).await
+ }
+
+ async fn equal(&self, key: &[u8]) -> io::Result<RoaringTreemap> {
+ match self.find_bitmap_block(key).await? {
+ Some(block) => self.read_bitmap(block).await,
+ None => Ok(RoaringTreemap::new()),
+ }
+ }
+
+ async fn in_keys(&self, keys: &[Vec<u8>]) -> io::Result<RoaringTreemap> {
+ let mut sorted_keys = keys.to_vec();
+ sorted_keys.sort();
+ sorted_keys.dedup();
+
+ let mut result = RoaringTreemap::new();
+ for key in sorted_keys {
+ if let Some(block) = self.find_bitmap_block(&key).await? {
+ result |= self.read_bitmap(block).await?;
+ }
+ }
+ Ok(result)
+ }
+
+ async fn scan_dictionary(
+ &self,
+ data_type: &DataType,
+ predicate: impl Fn(&[u8], &dyn Fn(&[u8], &[u8]) -> Ordering) -> bool,
+ ) -> io::Result<RoaringTreemap> {
+ let cmp = make_key_comparator(data_type);
+ self.scan_serialized_dictionary(|candidate| predicate(candidate,
cmp.as_ref()))
+ .await
+ }
+
+ async fn scan_serialized_dictionary(
+ &self,
+ predicate: impl Fn(&[u8]) -> bool,
+ ) -> io::Result<RoaringTreemap> {
+ let mut result = RoaringTreemap::new();
+ for block_meta in &self.dictionary_blocks {
+ for entry in self.read_dictionary_block(block_meta.block).await? {
+ if predicate(&entry.key) {
+ result |= self.read_bitmap(entry.bitmap_block).await?;
+ }
+ }
+ }
+ Ok(result)
+ }
+
+ async fn find_bitmap_block(&self, key: &[u8]) ->
io::Result<Option<BlockInfo>> {
+ let Some(block_meta) = self.find_dictionary_block_meta(key) else {
+ return Ok(None);
+ };
+
+ for entry in self.read_dictionary_block(block_meta.block).await? {
+ match compare_unsigned(&entry.key, key) {
+ Ordering::Equal => return Ok(Some(entry.bitmap_block)),
+ Ordering::Greater => return Ok(None),
+ Ordering::Less => {}
+ }
+ }
+ Ok(None)
+ }
+
+ fn find_dictionary_block_meta(&self, key: &[u8]) ->
Option<&DictionaryBlockMeta> {
+ if self.dictionary_blocks.is_empty() {
+ return None;
+ }
+ let mut low = 0usize;
+ let mut high = self.dictionary_blocks.len();
+ while low < high {
+ let mid = (low + high) / 2;
+ if compare_unsigned(&self.dictionary_blocks[mid].first_key, key)
!= Ordering::Greater {
+ low = mid + 1;
+ } else {
+ high = mid;
+ }
+ }
+ low.checked_sub(1)
+ .and_then(|index| self.dictionary_blocks.get(index))
+ }
+
+ async fn read_dictionary_block(&self, block: BlockInfo) ->
io::Result<Vec<DictionaryEntry>> {
+ let bytes = read_compressible_block(self.reader.as_ref(),
block).await?;
+ let mut cursor = Cursor::new(bytes.as_slice());
+ let entry_count = decode_var_int(&mut cursor)?;
+ if entry_count < 0 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!("Invalid bitmap dictionary entry count:
{entry_count}"),
+ ));
+ }
+ let mut entries = Vec::with_capacity(entry_count as usize);
+ for _ in 0..entry_count {
+ let key = read_key(&mut cursor)?;
+ let offset = decode_var_long(&mut cursor)?;
+ let length = decode_var_int(&mut cursor)?;
+ entries.push(DictionaryEntry {
+ key,
+ bitmap_block: block_info(offset, length)?,
+ });
+ }
+ Ok(entries)
+ }
+
+ async fn read_bitmap(&self, block: BlockInfo) ->
io::Result<RoaringTreemap> {
+ let bytes = self
+ .reader
+ .read(block.offset..block.offset + block.length as u64)
+ .await
+ .map_err(|e| io::Error::other(e.to_string()))?;
+ RoaringTreemap::deserialize_from(bytes.as_ref())
+ .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))
+ }
+}
+
+async fn read_footer(reader: &dyn FileRead, file_size: u64) ->
io::Result<Footer> {
+ if file_size < FOOTER_LENGTH as u64 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ "Invalid bitmap global index file size",
+ ));
+ }
+ let bytes = reader
+ .read(file_size - FOOTER_LENGTH as u64..file_size)
+ .await
+ .map_err(|e| io::Error::other(e.to_string()))?;
+
+ let null_rows_block = block_info(read_i64_be(&bytes, 0)?,
read_i32_be(&bytes, 8)?)?;
+ let non_null_rows_block = block_info(read_i64_be(&bytes, 12)?,
read_i32_be(&bytes, 20)?)?;
+ let index_block = block_info(read_i64_be(&bytes, 24)?, read_i32_be(&bytes,
32)?)?;
+ let value_count = read_i32_be(&bytes, 36)?;
+ let version = read_i32_be(&bytes, 40)?;
+ let magic = read_i32_be(&bytes, 44)?;
+
+ if magic != MAGIC {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ "File is not a bitmap global index file",
+ ));
+ }
+ if version != VERSION {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!("Unsupported bitmap global index file version: {version}"),
+ ));
+ }
+ if value_count < 0 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ "Invalid bitmap value count",
+ ));
+ }
+
+ Ok(Footer {
+ null_rows_block,
+ non_null_rows_block,
+ index_block,
+ })
+}
+
+fn read_index_block(bytes: &[u8]) -> io::Result<Vec<DictionaryBlockMeta>> {
+ let mut cursor = Cursor::new(bytes);
+ let block_count = decode_var_int(&mut cursor)?;
+ if block_count < 0 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!("Invalid bitmap dictionary block count: {block_count}"),
+ ));
+ }
+ let mut blocks = Vec::with_capacity(block_count as usize);
+ for _ in 0..block_count {
+ let first_key = read_key(&mut cursor)?;
+ let offset = decode_var_long(&mut cursor)?;
+ let length = decode_var_int(&mut cursor)?;
+ blocks.push(DictionaryBlockMeta {
+ first_key,
+ block: block_info(offset, length)?,
+ });
+ }
+ blocks.sort_by(|left, right| compare_unsigned(&left.first_key,
&right.first_key));
+ Ok(blocks)
+}
+
+async fn read_compressible_block(reader: &dyn FileRead, block: BlockInfo) ->
io::Result<Vec<u8>> {
+ if block.length > usize::MAX - BLOCK_TRAILER_LENGTH {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ "Bitmap block is too large",
+ ));
+ }
+ let bytes = reader
+ .read(block.offset..block.offset + (block.length +
BLOCK_TRAILER_LENGTH) as u64)
+ .await
+ .map_err(|e| io::Error::other(e.to_string()))?;
+ let block_bytes = &bytes[..block.length];
+ let trailer = &bytes[block.length..block.length + BLOCK_TRAILER_LENGTH];
+ let compression_type =
BlockCompressionType::from_persistent_id(trailer[0])?;
+ let expected_crc = u32::from_le_bytes([trailer[1], trailer[2], trailer[3],
trailer[4]]);
+ let actual_crc = compute_crc32(block_bytes, compression_type);
+ if expected_crc != actual_crc {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!(
+ "Bitmap block CRC mismatch: expected 0x{expected_crc:08X}, got
0x{actual_crc:08X}"
+ ),
+ ));
+ }
+
+ match compression_type {
+ BlockCompressionType::None => Ok(block_bytes.to_vec()),
+ BlockCompressionType::Zstd => {
+ let mut cursor = Cursor::new(block_bytes);
+ let uncompressed_size = decode_var_int(&mut cursor)? as usize;
+ let compressed_start = cursor.position() as usize;
+ let compressed_data = &block_bytes[compressed_start..];
+ let mut decompressed = vec![0u8; uncompressed_size];
+ let actual = zstd::bulk::decompress_to_buffer(compressed_data,
&mut decompressed)
+ .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
+ if actual != uncompressed_size {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!(
+ "Bitmap block decompressed size mismatch: expected
{uncompressed_size}, got {actual}"
+ ),
+ ));
+ }
+ Ok(decompressed)
+ }
+ _ => Err(io::Error::new(
+ io::ErrorKind::Unsupported,
+ format!(
+ "Bitmap global index compression type {:?} is not supported",
+ compression_type
+ ),
+ )),
+ }
+}
+
+fn compute_crc32(data: &[u8], compression_type: BlockCompressionType) -> u32 {
+ let mut hasher = crc32fast::Hasher::new();
+ hasher.update(data);
+ hasher.update(&[compression_type as u8]);
+ hasher.finalize()
+}
+
+fn read_key(input: &mut impl Read) -> io::Result<Vec<u8>> {
+ let key_length = decode_var_int(input)?;
+ if key_length < 0 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!("Invalid bitmap key length: {key_length}"),
+ ));
+ }
+ let mut key = vec![0; key_length as usize];
+ input.read_exact(&mut key)?;
+ Ok(key)
+}
+
+fn block_info(offset: i64, length: i32) -> io::Result<BlockInfo> {
+ if offset < 0 || length < 0 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!("Invalid bitmap block info: offset={offset},
length={length}"),
+ ));
+ }
+ Ok(BlockInfo {
+ offset: offset as u64,
+ length: length as usize,
+ })
+}
+
+fn read_i64_be(bytes: &[u8], offset: usize) -> io::Result<i64> {
+ let end = offset + 8;
+ let value = bytes
+ .get(offset..end)
+ .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Bitmap
footer is truncated"))?;
+ Ok(i64::from_be_bytes(value.try_into().unwrap()))
+}
+
+fn read_i32_be(bytes: &[u8], offset: usize) -> io::Result<i32> {
+ let end = offset + 4;
+ let value = bytes
+ .get(offset..end)
+ .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Bitmap
footer is truncated"))?;
+ Ok(i32::from_be_bytes(value.try_into().unwrap()))
+}
+
+fn compare_unsigned(left: &[u8], right: &[u8]) -> Ordering {
+ for (left, right) in left.iter().zip(right.iter()) {
+ match left.cmp(right) {
+ Ordering::Equal => {}
+ non_eq => return non_eq,
+ }
+ }
+ left.len().cmp(&right.len())
+}
+
+fn is_character_string(data_type: &DataType) -> bool {
+ matches!(data_type, DataType::Char(_) | DataType::VarChar(_))
+}
+
+fn string_literal(literals: &[Datum], op: PredicateOperator) ->
io::Result<&str> {
+ match literals.first() {
+ Some(Datum::String(value)) => Ok(value),
+ Some(other) => Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ format!("Bitmap global index {op} requires a string literal, got
{other}"),
+ )),
+ None => Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ format!("Bitmap global index {op} requires one literal"),
+ )),
+ }
+}
+
+fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
+ !needle.is_empty()
+ && haystack
+ .windows(needle.len())
+ .any(|window| window == needle)
+}
+
+fn write_bitmap_index_bytes(
+ out: &mut Vec<u8>,
+ null_rows: &RoaringTreemap,
+ non_null_rows: &RoaringTreemap,
+ bitmaps: &BTreeMap<Vec<u8>, RoaringTreemap>,
+ dictionary_block_size: usize,
+ compression_type: BlockCompressionType,
+) -> io::Result<()> {
+ if dictionary_block_size == 0 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ "Bitmap dictionary block size must be greater than 0",
+ ));
+ }
+
+ let null_rows_block = write_bitmap_block(out, null_rows)?;
+ let non_null_rows_block = write_bitmap_block(out, non_null_rows)?;
+ let (dictionary_blocks, value_count) =
+ write_dictionary_and_bitmap_blocks(out, bitmaps,
dictionary_block_size, compression_type)?;
+ let index_block = write_index_block(out, &dictionary_blocks,
compression_type)?;
+
+ out.extend_from_slice(&u64_to_i64(null_rows_block.offset)?.to_be_bytes());
+
out.extend_from_slice(&usize_to_i32(null_rows_block.length)?.to_be_bytes());
+
out.extend_from_slice(&u64_to_i64(non_null_rows_block.offset)?.to_be_bytes());
+
out.extend_from_slice(&usize_to_i32(non_null_rows_block.length)?.to_be_bytes());
+ out.extend_from_slice(&u64_to_i64(index_block.offset)?.to_be_bytes());
+ out.extend_from_slice(&usize_to_i32(index_block.length)?.to_be_bytes());
+ out.extend_from_slice(&usize_to_i32(value_count)?.to_be_bytes());
+ out.extend_from_slice(&VERSION.to_be_bytes());
+ out.extend_from_slice(&MAGIC.to_be_bytes());
+ Ok(())
+}
+
+fn write_bitmap_block(out: &mut Vec<u8>, bitmap: &RoaringTreemap) ->
io::Result<BlockInfo> {
+ let offset = out.len() as u64;
+ bitmap.serialize_into(&mut *out)?;
+ Ok(BlockInfo {
+ offset,
+ length: out.len() - offset as usize,
+ })
+}
+
+fn write_dictionary_and_bitmap_blocks(
+ out: &mut Vec<u8>,
+ bitmaps: &BTreeMap<Vec<u8>, RoaringTreemap>,
+ dictionary_block_size: usize,
+ compression_type: BlockCompressionType,
+) -> io::Result<(Vec<DictionaryBlockMeta>, usize)> {
+ let mut block_metas = Vec::new();
+ let mut current = DictionaryBlockBuilder::default();
+ let mut value_count = 0usize;
+
+ for (key, bitmap) in bitmaps {
+ let bitmap_block = write_bitmap_block(out, bitmap)?;
+ let entry = DictionaryEntry {
+ key: key.clone(),
+ bitmap_block,
+ };
+ if current.has_entries() && current.estimated_size_after(&entry) >
dictionary_block_size {
+ block_metas.push(write_dictionary_block(
+ out,
+ ¤t.entries,
+ compression_type,
+ )?);
+ current = DictionaryBlockBuilder::default();
+ }
+ current.add(entry);
+ value_count += 1;
+ }
+
+ if current.has_entries() {
+ block_metas.push(write_dictionary_block(
+ out,
+ ¤t.entries,
+ compression_type,
+ )?);
+ }
+ Ok((block_metas, value_count))
+}
+
+fn write_dictionary_block(
+ out: &mut Vec<u8>,
+ entries: &[DictionaryEntry],
+ compression_type: BlockCompressionType,
+) -> io::Result<DictionaryBlockMeta> {
+ let mut bytes = Vec::new();
+ encode_var_int(&mut bytes, usize_to_i32(entries.len())?)?;
+ for entry in entries {
+ encode_var_int(&mut bytes, usize_to_i32(entry.key.len())?)?;
+ bytes.extend_from_slice(&entry.key);
+ encode_var_long(&mut bytes, u64_to_i64(entry.bitmap_block.offset)?)?;
+ encode_var_int(&mut bytes, usize_to_i32(entry.bitmap_block.length)?)?;
+ }
+ let block = write_compressible_block(out, &bytes, compression_type)?;
+ Ok(DictionaryBlockMeta {
+ first_key: entries[0].key.clone(),
+ block,
+ })
+}
+
+fn write_index_block(
+ out: &mut Vec<u8>,
+ blocks: &[DictionaryBlockMeta],
+ compression_type: BlockCompressionType,
+) -> io::Result<BlockInfo> {
+ let mut bytes = Vec::new();
+ encode_var_int(&mut bytes, usize_to_i32(blocks.len())?)?;
+ for block in blocks {
+ encode_var_int(&mut bytes, usize_to_i32(block.first_key.len())?)?;
+ bytes.extend_from_slice(&block.first_key);
+ encode_var_long(&mut bytes, u64_to_i64(block.block.offset)?)?;
+ encode_var_int(&mut bytes, usize_to_i32(block.block.length)?)?;
+ }
+ write_compressible_block(out, &bytes, compression_type)
+}
+
+fn write_compressible_block(
+ out: &mut Vec<u8>,
+ bytes: &[u8],
+ compression_type: BlockCompressionType,
+) -> io::Result<BlockInfo> {
+ let (block_bytes, actual_compression_type) = encode_block(bytes,
compression_type)?;
+ let offset = out.len() as u64;
+ out.write_all(&block_bytes)?;
+ let crc = compute_crc32(&block_bytes, actual_compression_type);
+ out.write_all(&[actual_compression_type as u8])?;
+ out.write_all(&crc.to_le_bytes())?;
+ Ok(BlockInfo {
+ offset,
+ length: block_bytes.len(),
+ })
+}
+
+fn encode_block(
+ bytes: &[u8],
+ compression_type: BlockCompressionType,
+) -> io::Result<(Vec<u8>, BlockCompressionType)> {
+ match compression_type {
+ BlockCompressionType::None => Ok((bytes.to_vec(),
BlockCompressionType::None)),
+ BlockCompressionType::Zstd => {
+ let compressed = zstd::bulk::compress(bytes, 3)
+ .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
+ let mut encoded = Vec::with_capacity(5 + compressed.len());
+ encode_var_int(&mut encoded, usize_to_i32(bytes.len())?)?;
+ encoded.extend_from_slice(&compressed);
+ if encoded.len() < bytes.len() - (bytes.len() / 8) {
+ Ok((encoded, BlockCompressionType::Zstd))
+ } else {
+ Ok((bytes.to_vec(), BlockCompressionType::None))
+ }
+ }
+ _ => Err(io::Error::new(
+ io::ErrorKind::Unsupported,
+ format!(
+ "Bitmap global index compression type {:?} is not supported",
+ compression_type
+ ),
+ )),
+ }
+}
+
+#[derive(Default)]
+struct DictionaryBlockBuilder {
+ entries: Vec<DictionaryEntry>,
+ entries_size: usize,
+}
+
+impl DictionaryBlockBuilder {
+ fn has_entries(&self) -> bool {
+ !self.entries.is_empty()
+ }
+
+ fn estimated_size_after(&self, entry: &DictionaryEntry) -> usize {
+ estimated_var_len_int_size(self.entries.len() + 1)
+ + self.entries_size
+ + entry.estimated_size()
+ }
+
+ fn add(&mut self, entry: DictionaryEntry) {
+ self.entries_size += entry.estimated_size();
+ self.entries.push(entry);
+ }
+}
+
+impl DictionaryEntry {
+ fn estimated_size(&self) -> usize {
+ estimated_var_len_int_size(self.key.len())
+ + self.key.len()
+ + estimated_var_len_long_size(self.bitmap_block.offset)
+ + estimated_var_len_int_size(self.bitmap_block.length)
+ }
+}
+
+fn estimated_var_len_int_size(mut value: usize) -> usize {
+ let mut size = 1;
+ while (value & !0x7f) != 0 {
+ value >>= 7;
+ size += 1;
+ }
+ size
+}
+
+fn estimated_var_len_long_size(mut value: u64) -> usize {
+ let mut size = 1;
+ while (value & !0x7f) != 0 {
+ value >>= 7;
+ size += 1;
+ }
+ size
+}
+
+fn usize_to_i32(value: usize) -> io::Result<i32> {
+ i32::try_from(value).map_err(|_| {
+ io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!("Bitmap global index value is too large: {value}"),
+ )
+ })
+}
+
+fn u64_to_i64(value: u64) -> io::Result<i64> {
+ i64::try_from(value).map_err(|_| {
+ io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!("Bitmap global index offset is too large: {value}"),
+ )
+ })
+}
diff --git a/crates/paimon/src/table/btree_global_index_build_builder.rs
b/crates/paimon/src/table/btree_global_index_build_builder.rs
index 57b2ff2..2f242ca 100644
--- a/crates/paimon/src/table/btree_global_index_build_builder.rs
+++ b/crates/paimon/src/table/btree_global_index_build_builder.rs
@@ -15,6 +15,10 @@
// specific language governing permissions and limitations
// under the License.
+use super::bitmap_global_index_reader::{BitmapGlobalIndexWriter,
BitmapWriteResult};
+use super::global_index_types::{
+ normalize_sorted_global_index_type, BITMAP_GLOBAL_INDEX_TYPE,
BTREE_GLOBAL_INDEX_TYPE,
+};
use crate::btree::{make_key_comparator, serialize_datum, BTreeIndexWriter,
BlockCompressionType};
use crate::spec::{
bucket_dir_name, extract_datum_from_arrow, BinaryRow, CoreOptions,
DataField, DataFileMeta,
@@ -31,15 +35,16 @@ use futures::TryStreamExt;
use std::cmp::Ordering;
use std::collections::HashMap;
-const BTREE_INDEX_TYPE: &str = "btree";
const INDEX_DIR: &str = "index";
const BTREE_BLOCK_SIZE: usize = 4 * 1024;
+const BITMAP_DICTIONARY_BLOCK_SIZE: usize = 16 * 1024;
type BTreeKeyRow = (Option<Vec<u8>>, i64);
pub struct BTreeGlobalIndexBuildBuilder<'a> {
table: &'a Table,
index_column: Option<String>,
+ index_type: String,
}
impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
@@ -47,6 +52,7 @@ impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
Self {
table,
index_column: None,
+ index_type: BTREE_GLOBAL_INDEX_TYPE.to_string(),
}
}
@@ -55,12 +61,25 @@ impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
self
}
+ pub fn with_index_type(&mut self, index_type: &str) -> &mut Self {
+ self.index_type = index_type.to_string();
+ self
+ }
+
pub async fn execute(&self) -> Result<usize> {
+ let index_type =
normalize_sorted_global_index_type(&self.index_type).ok_or_else(|| {
+ Error::Unsupported {
+ message: format!(
+ "Sorted global index build only supports index_type =>
'btree' or 'bitmap', got '{}'",
+ self.index_type
+ ),
+ }
+ })?;
let index_column = self
.index_column
.as_deref()
.ok_or_else(|| Error::DataInvalid {
- message: "BTree global index column is required".to_string(),
+ message: "Sorted global index column is required".to_string(),
source: None,
})?;
@@ -79,7 +98,7 @@ impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
.get_latest_snapshot()
.await?
.ok_or_else(|| Error::DataInvalid {
- message: "Cannot build BTree global index without a
snapshot".to_string(),
+ message: "Cannot build sorted global index without a
snapshot".to_string(),
source: None,
})?;
@@ -127,7 +146,7 @@ impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
self.table.clone(),
format!(
"global-index-{}-create-{}",
- BTREE_INDEX_TYPE,
+ index_type,
uuid::Uuid::new_v4()
),
)
@@ -143,6 +162,14 @@ impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
index_field: &DataField,
index_column: &str,
) -> Result<IndexFileMeta> {
+ let index_type =
normalize_sorted_global_index_type(&self.index_type).ok_or_else(|| {
+ Error::Unsupported {
+ message: format!(
+ "Sorted global index build only supports index_type =>
'btree' or 'bitmap', got '{}'",
+ self.index_type
+ ),
+ }
+ })?;
let row_count = checked_row_count(shard.row_range_start,
shard.row_range_end)?;
let mut rows = extract_index_rows(self.table, shard, index_column,
index_field).await?;
let cmp = make_key_comparator(index_field.data_type());
@@ -155,7 +182,7 @@ impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
self.table.location().trim_end_matches('/')
))
.await?;
- let file_name = format!("btree-global-index-{}.index",
uuid::Uuid::new_v4());
+ let file_name = format!("{index_type}-global-index-{}.index",
uuid::Uuid::new_v4());
let index_path = format!(
"{}/{INDEX_DIR}/{}",
self.table.location().trim_end_matches('/'),
@@ -163,31 +190,64 @@ impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
);
let output = self.table.file_io().new_output(&index_path)?;
let writer = output.writer().await?;
- let mut writer = BTreeIndexWriter::with_comparator(
- writer,
- BTREE_BLOCK_SIZE,
- BlockCompressionType::None,
- cmp,
- );
- for (key, local_row_id) in &rows {
- writer
- .write(key.as_deref(), *local_row_id)
- .await
- .map_err(|e| Error::DataInvalid {
- message: format!("Failed to write BTree global index file
'{file_name}'"),
+ let (written_row_count, index_meta) = match index_type {
+ BTREE_GLOBAL_INDEX_TYPE => {
+ let mut writer = BTreeIndexWriter::with_comparator(
+ writer,
+ BTREE_BLOCK_SIZE,
+ BlockCompressionType::None,
+ cmp,
+ );
+ for (key, local_row_id) in &rows {
+ writer
+ .write(key.as_deref(), *local_row_id)
+ .await
+ .map_err(|e| Error::DataInvalid {
+ message: format!(
+ "Failed to write BTree global index file
'{file_name}'"
+ ),
+ source: Some(Box::new(e)),
+ })?;
+ }
+ let write_result = writer.finish().await.map_err(|e|
Error::DataInvalid {
+ message: format!("Failed to finish BTree global index file
'{file_name}'"),
source: Some(Box::new(e)),
})?;
- }
- let write_result = writer.finish().await.map_err(|e|
Error::DataInvalid {
- message: format!("Failed to finish BTree global index file
'{file_name}'"),
- source: Some(Box::new(e)),
- })?;
+ (write_result.row_count, write_result.meta)
+ }
+ BITMAP_GLOBAL_INDEX_TYPE => {
+ let cmp = make_key_comparator(index_field.data_type());
+ let mut writer = BitmapGlobalIndexWriter::new(
+ writer,
+ BITMAP_DICTIONARY_BLOCK_SIZE,
+ BlockCompressionType::None,
+ cmp,
+ );
+ for (key, local_row_id) in &rows {
+ writer.write(key.as_deref(), *local_row_id).map_err(|e| {
+ Error::DataInvalid {
+ message: format!(
+ "Failed to write bitmap global index file
'{file_name}'"
+ ),
+ source: Some(Box::new(e)),
+ }
+ })?;
+ }
+ let BitmapWriteResult { row_count, meta } =
+ writer.finish().await.map_err(|e| Error::DataInvalid {
+ message: format!("Failed to finish bitmap global index
file '{file_name}'"),
+ source: Some(Box::new(e)),
+ })?;
+ (row_count, meta)
+ }
+ _ => unreachable!("normalized sorted global index type"),
+ };
- if write_result.row_count != u64::try_from(row_count).unwrap() {
+ if written_row_count != u64::try_from(row_count).unwrap() {
return Err(Error::DataInvalid {
message: format!(
- "BTree global index expected {} rows, wrote {}",
- row_count, write_result.row_count
+ "Sorted global index expected {} rows, wrote {}",
+ row_count, written_row_count
),
source: None,
});
@@ -195,7 +255,7 @@ impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
let status = self.table.file_io().get_status(&index_path).await?;
Ok(IndexFileMeta {
- index_type: BTREE_INDEX_TYPE.to_string(),
+ index_type: index_type.to_string(),
file_name,
file_size: checked_i32(
status.size,
@@ -208,7 +268,7 @@ impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
row_range_end: shard.row_range_end,
index_field_id: index_field.id(),
extra_field_ids: None,
- index_meta: Some(write_result.meta.serialize()),
+ index_meta: Some(index_meta.serialize()),
}),
})
}
@@ -1159,7 +1219,7 @@ mod tests {
assert_eq!(index_entries.len(), 1);
let index_file = &index_entries[0].index_file;
- assert_eq!(index_file.index_type, BTREE_INDEX_TYPE);
+ assert_eq!(index_file.index_type, BTREE_GLOBAL_INDEX_TYPE);
assert!(index_file.file_name.starts_with("btree-global-index-"));
assert_eq!(index_file.row_count, 3);
assert!(index_file.file_size > 0);
@@ -1188,6 +1248,109 @@ mod tests {
predicates: &[predicate],
schema_fields: table.schema().fields(),
search_mode: GlobalIndexSearchMode::Fast,
+ btree_fallback_scan_max_size: i64::MAX,
+ bitmap_fallback_scan_max_size: i64::MAX,
+ next_row_id: snapshot.next_row_id(),
+ data_ranges: &[],
+ })
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(row_ranges, vec![RowRange::new(0, 0), RowRange::new(2, 2)]);
+ }
+
+ #[tokio::test]
+ async fn test_execute_writes_bitmap_index_manifest_and_java_file() {
+ let table_path = "memory:/test_bitmap_global_index_builder_e2e";
+ let table = test_table_with_path(table_path, table_options("10"));
+ setup_dirs(&table).await;
+
+ let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ table_write
+ .write_arrow_batch(&data_batch(vec![1, 2, 3], vec!["alice", "bob",
"alice"]))
+ .await
+ .unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ let shard_count = table
+ .new_btree_global_index_build_builder()
+ .with_index_column("name")
+ .with_index_type(BITMAP_GLOBAL_INDEX_TYPE)
+ .execute()
+ .await
+ .unwrap();
+ assert_eq!(shard_count, 1);
+
+ let snapshot_manager =
+ SnapshotManager::new(table.file_io().clone(),
table.location().to_string());
+ let snapshot = snapshot_manager
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let index_manifest = snapshot.index_manifest().expect("index
manifest");
+ let index_entries = IndexManifest::read(
+ table.file_io(),
+ &format!("{table_path}/manifest/{index_manifest}"),
+ )
+ .await
+ .unwrap();
+ assert_eq!(index_entries.len(), 1);
+
+ let index_file = &index_entries[0].index_file;
+ assert_eq!(index_file.index_type, BITMAP_GLOBAL_INDEX_TYPE);
+ assert!(index_file.file_name.starts_with("bitmap-global-index-"));
+ assert_eq!(index_file.row_count, 3);
+ assert!(index_file.file_size > 0);
+
+ let global_meta = index_file
+ .global_index_meta
+ .as_ref()
+ .expect("global index meta");
+ let bitmap_meta =
+
crate::btree::BTreeIndexMeta::deserialize(global_meta.index_meta.as_ref().unwrap())
+ .unwrap();
+ assert_eq!(bitmap_meta.first_key, Some(b"alice".to_vec()));
+ assert_eq!(bitmap_meta.last_key, Some(b"bob".to_vec()));
+ assert!(!bitmap_meta.has_nulls);
+
+ let index_path = format!("{table_path}/index/{}",
index_file.file_name);
+ let input = table.file_io().new_input(&index_path).unwrap();
+ let file_size = input.metadata().await.unwrap().size;
+ let reader = input.reader().await.unwrap();
+ let bitmap_reader =
+
crate::table::bitmap_global_index_reader::BitmapGlobalIndexReader::open(
+ Box::new(reader),
+ file_size,
+ )
+ .await
+ .unwrap();
+ let bitmap = bitmap_reader
+ .query(
+ crate::spec::PredicateOperator::Eq,
+ &[crate::spec::Datum::String("alice".to_string())],
+ table.schema().fields()[1].data_type(),
+ )
+ .await
+ .unwrap();
+ assert_eq!(bitmap.iter().collect::<Vec<_>>(), vec![0, 2]);
+
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .equal("name", crate::spec::Datum::String("alice".to_string()))
+ .unwrap();
+ let row_ranges = evaluate_global_index(GlobalIndexEvaluation {
+ file_io: table.file_io(),
+ table_path: table.location(),
+ index_entries: &index_entries,
+ predicates: &[predicate],
+ schema_fields: table.schema().fields(),
+ search_mode: GlobalIndexSearchMode::Fast,
+ btree_fallback_scan_max_size: i64::MAX,
+ bitmap_fallback_scan_max_size: i64::MAX,
next_row_id: snapshot.next_row_id(),
data_ranges: &[],
})
diff --git a/crates/paimon/src/table/btree_global_index_drop_builder.rs
b/crates/paimon/src/table/btree_global_index_drop_builder.rs
index 50f0e30..a1c497e 100644
--- a/crates/paimon/src/table/btree_global_index_drop_builder.rs
+++ b/crates/paimon/src/table/btree_global_index_drop_builder.rs
@@ -15,16 +15,16 @@
// specific language governing permissions and limitations
// under the License.
+use super::global_index_types::{normalize_sorted_global_index_type,
BTREE_GLOBAL_INDEX_TYPE};
use crate::spec::{DataField, FileKind, IndexFileMeta, IndexManifest};
use crate::table::{CommitMessage, SnapshotManager, Table, TableCommit};
use crate::{Error, Result};
use std::collections::HashMap;
-const BTREE_INDEX_TYPE: &str = "btree";
-
pub struct BTreeGlobalIndexDropBuilder<'a> {
table: &'a Table,
index_column: Option<String>,
+ index_type: String,
}
impl<'a> BTreeGlobalIndexDropBuilder<'a> {
@@ -32,6 +32,7 @@ impl<'a> BTreeGlobalIndexDropBuilder<'a> {
Self {
table,
index_column: None,
+ index_type: BTREE_GLOBAL_INDEX_TYPE.to_string(),
}
}
@@ -40,12 +41,25 @@ impl<'a> BTreeGlobalIndexDropBuilder<'a> {
self
}
+ pub fn with_index_type(&mut self, index_type: &str) -> &mut Self {
+ self.index_type = index_type.to_string();
+ self
+ }
+
pub async fn execute(&self) -> Result<usize> {
+ let index_type =
normalize_sorted_global_index_type(&self.index_type).ok_or_else(|| {
+ Error::Unsupported {
+ message: format!(
+ "Sorted global index drop only supports index_type =>
'btree' or 'bitmap', got '{}'",
+ self.index_type
+ ),
+ }
+ })?;
let index_column = self
.index_column
.as_deref()
.ok_or_else(|| Error::DataInvalid {
- message: "BTree global index column is required".to_string(),
+ message: "Sorted global index column is required".to_string(),
source: None,
})?;
let index_field = find_index_field(self.table, index_column)?;
@@ -70,7 +84,7 @@ impl<'a> BTreeGlobalIndexDropBuilder<'a> {
HashMap::new();
let mut dropped = 0;
for entry in index_entries {
- if entry.kind != FileKind::Add || entry.index_file.index_type !=
BTREE_INDEX_TYPE {
+ if entry.kind != FileKind::Add || entry.index_file.index_type !=
index_type {
continue;
}
let Some(global_meta) =
entry.index_file.global_index_meta.as_ref() else {
@@ -110,11 +124,7 @@ impl<'a> BTreeGlobalIndexDropBuilder<'a> {
TableCommit::new(
self.table.clone(),
- format!(
- "global-index-{}-drop-{}",
- BTREE_INDEX_TYPE,
- uuid::Uuid::new_v4()
- ),
+ format!("global-index-{}-drop-{}", index_type,
uuid::Uuid::new_v4()),
)
.commit_if_latest_snapshot(messages, snapshot.id())
.await?;
@@ -288,8 +298,8 @@ mod tests {
vec![data_file("data-0.parquet")],
);
message.new_index_files = vec![
- global_index_file(BTREE_INDEX_TYPE, "btree-id.index", 0, 0, 9),
- global_index_file(BTREE_INDEX_TYPE, "btree-name.index", 1, 0, 9),
+ global_index_file(BTREE_GLOBAL_INDEX_TYPE, "btree-id.index", 0, 0,
9),
+ global_index_file(BTREE_GLOBAL_INDEX_TYPE, "btree-name.index", 1,
0, 9),
global_index_file("full-text", "fulltext-id.index", 0, 100, 109),
hash_index_file("hash.index"),
deletion_vector_index_file("dv.index"),
@@ -331,7 +341,7 @@ mod tests {
let mut message = CommitMessage::new(vec![], 0,
vec![data_file("data-0.parquet")]);
message.new_index_files = vec
+use super::bitmap_global_index_reader::BitmapGlobalIndexReader;
+use super::global_index_types::{
+ normalize_sorted_global_index_type, BITMAP_GLOBAL_INDEX_TYPE,
BTREE_GLOBAL_INDEX_TYPE,
+};
use crate::btree::query::{extract_between, IndexQuery};
use crate::btree::{make_key_comparator, serialize_datum, BTreeIndexMeta,
BTreeIndexReader};
use crate::deletion_vector::DeletionVectorFactory;
@@ -43,7 +47,6 @@ type EvaluateFuture<'a> = std::pin::Pin<
type PredicateTuple<'a> = (PredicateOperator, &'a [Datum], &'a DataType);
-const BTREE_INDEX_TYPE: &str = "btree";
const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS";
const INDEX_DIR: &str = "index";
@@ -56,11 +59,13 @@ struct GlobalIndexScanResult {
///
/// The scanner filters index manifest entries for global index files,
/// uses BTreeIndexMeta for file-level pruning, then reads matching
-/// BTree files to evaluate predicates and collect row IDs.
+/// BTree or bitmap files to evaluate predicates and collect row IDs.
/// Opened BTreeIndexReaders are cached for reuse across evaluations.
pub(crate) struct GlobalIndexScanner {
file_io: FileIO,
table_path: String,
+ btree_fallback_scan_max_size: i64,
+ bitmap_fallback_scan_max_size: i64,
/// Global index entries grouped by field_id.
entries_by_field: Vec<(i32, Vec<GlobalIndexEntry>)>,
/// Indexed row-id coverage grouped by field_id.
@@ -74,16 +79,84 @@ pub(crate) struct GlobalIndexScanner {
/// A resolved global index entry with parsed metadata.
struct GlobalIndexEntry {
file_name: String,
+ index_type: GlobalIndexFileKind,
+ file_size: i64,
row_range_start: i64,
meta: BTreeIndexMeta,
}
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum GlobalIndexFileKind {
+ BTree,
+ Bitmap,
+}
+
+enum OpenedGlobalIndexReader {
+ BTree(BTreeIndexReader<BoxedCmp>),
+ Bitmap(BitmapGlobalIndexReader),
+}
+
+#[derive(Clone, Copy, Default)]
+struct FallbackScanPlan {
+ selected_btree: usize,
+ selected_bitmap: usize,
+ allow_btree: bool,
+ allow_bitmap: bool,
+}
+
+impl FallbackScanPlan {
+ fn allowed(self, kind: GlobalIndexFileKind) -> bool {
+ match kind {
+ GlobalIndexFileKind::BTree => self.allow_btree,
+ GlobalIndexFileKind::Bitmap => self.allow_bitmap,
+ }
+ }
+}
+
+impl OpenedGlobalIndexReader {
+ async fn query(
+ &self,
+ op: PredicateOperator,
+ literals: &[Datum],
+ data_type: &DataType,
+ ) -> std::io::Result<RoaringTreemap> {
+ match self {
+ Self::BTree(reader) => reader.query(op, literals, data_type).await,
+ Self::Bitmap(reader) => reader.query(op, literals,
data_type).await,
+ }
+ }
+
+ async fn range_query(
+ &self,
+ from: &[u8],
+ to: &[u8],
+ data_type: &DataType,
+ from_inclusive: bool,
+ to_inclusive: bool,
+ ) -> std::io::Result<RoaringTreemap> {
+ match self {
+ Self::BTree(reader) => {
+ reader
+ .range_query(from, to, from_inclusive, to_inclusive)
+ .await
+ }
+ Self::Bitmap(reader) => {
+ reader
+ .range_query(from, to, data_type, from_inclusive,
to_inclusive)
+ .await
+ }
+ }
+ }
+}
+
impl GlobalIndexScanner {
/// Create a scanner from index manifest entries.
/// Returns `None` if there are no global index entries.
pub(crate) fn create(
file_io: &FileIO,
table_path: &str,
+ btree_fallback_scan_max_size: i64,
+ bitmap_fallback_scan_max_size: i64,
index_entries: &[IndexManifestEntry],
schema_fields: &[DataField],
) -> Option<Self> {
@@ -95,15 +168,16 @@ impl GlobalIndexScanner {
if entry.kind != FileKind::Add {
continue;
}
- if entry.index_file.index_type != BTREE_INDEX_TYPE {
+ let Some(index_type) =
normalize_sorted_global_index_type(&entry.index_file.index_type)
+ else {
continue;
- }
+ };
let global_meta = match &entry.index_file.global_index_meta {
Some(m) => m,
None => continue,
};
- let btree_meta = global_meta
+ let sorted_meta = global_meta
.index_meta
.as_ref()
.and_then(|bytes| BTreeIndexMeta::deserialize(bytes).ok())
@@ -111,8 +185,14 @@ impl GlobalIndexScanner {
let resolved = GlobalIndexEntry {
file_name: entry.index_file.file_name.clone(),
+ index_type: match index_type {
+ BTREE_GLOBAL_INDEX_TYPE => GlobalIndexFileKind::BTree,
+ BITMAP_GLOBAL_INDEX_TYPE => GlobalIndexFileKind::Bitmap,
+ _ => unreachable!("normalized sorted global index type"),
+ },
+ file_size: i64::from(entry.index_file.file_size),
row_range_start: global_meta.row_range_start,
- meta: btree_meta,
+ meta: sorted_meta,
};
let row_range = RowRange::new(global_meta.row_range_start,
global_meta.row_range_end);
@@ -142,6 +222,8 @@ impl GlobalIndexScanner {
Some(Self {
file_io: file_io.clone(),
table_path: table_path.trim_end_matches('/').to_string(),
+ btree_fallback_scan_max_size,
+ bitmap_fallback_scan_max_size,
entries_by_field: entries_by_field.into_iter().collect(),
coverage_by_field,
schema_fields: schema_fields.to_vec(),
@@ -161,7 +243,7 @@ impl GlobalIndexScanner {
data_type,
..
} => {
- if !is_btree_supported_op(*op) {
+ if !is_sorted_global_index_supported_op(*op) {
return Ok(None);
}
let field_id = self.find_field_id_by_name(column)?;
@@ -197,7 +279,7 @@ impl GlobalIndexScanner {
..
} = child
{
- if is_btree_supported_op(*op) {
+ if is_sorted_global_index_supported_op(*op) {
if let Some(field_id) =
self.find_field_id_by_name(column)? {
if
self.entries_for_field(field_id).is_some() {
leaf_groups.entry(field_id).or_default().push((
@@ -306,22 +388,47 @@ impl GlobalIndexScanner {
})
.collect();
- for entry in entries {
- // Check if any predicate may match this file (use
effective_predicates for pruning)
- let matching_predicates: Vec<usize> =
(0..effective_predicates.len())
- .filter(|&i| {
- entry
- .meta
- .may_match(pruning_info[i].0, &pruning_info[i].2,
&pruning_info[i].1)
- })
- .collect();
+ let predicate_matches: Vec<Vec<bool>> = pruning_info
+ .iter()
+ .map(|(op, cmp, serialized)| {
+ entries
+ .iter()
+ .map(|entry| entry.meta.may_match(*op, serialized, cmp))
+ .collect()
+ })
+ .collect();
+ let predicate_fallback_plans: Vec<Option<FallbackScanPlan>> =
effective_predicates
+ .iter()
+ .enumerate()
+ .map(|(i, (op, _, _))| {
+ requires_fallback_scan(*op)
+ .then(|| self.fallback_scan_plan(entries,
&predicate_matches[i]))
+ })
+ .collect();
- // Also check if between range may match
- let between_matches = between.as_ref().is_some_and(|b| {
+ let between_matches_by_entry: Vec<bool> = match between.as_ref() {
+ Some(b) => {
let cmp = make_key_comparator(b.data_type);
let from_key = serialize_datum(b.from, b.data_type);
let to_key = serialize_datum(b.to, b.data_type);
- entry.meta.may_match_between(&from_key, &to_key, &cmp)
+ entries
+ .iter()
+ .map(|entry| entry.meta.may_match_between(&from_key,
&to_key, &cmp))
+ .collect()
+ }
+ None => Vec::new(),
+ };
+ let between_fallback_plan = between
+ .as_ref()
+ .map(|_| self.fallback_scan_plan(entries,
&between_matches_by_entry));
+
+ for (entry_idx, entry) in entries.iter().enumerate() {
+ // 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)
});
// When a Between conjunct exists but the file does not overlap its
@@ -331,11 +438,44 @@ impl GlobalIndexScanner {
// (e.g. `BETWEEN 10 AND 20 AND id >= 0` on a file [30, 40]) would
// be retained because `file_result` is initialized from the
// remaining bitmap, silently dropping the Between conjunct.
- if between.is_some() && !between_matches {
+ if between_evaluated_for_entry && !between_matches {
continue;
}
- if matching_predicates.is_empty() && !between_matches {
+ let mut file_evaluated = between_evaluated_for_entry;
+ let mut file_cannot_match = false;
+ let mut file_has_unsupported_match =
+ between_matches && !between_evaluated_for_entry &&
between_fallback_plan.is_some();
+ let matching_predicates: Vec<usize> =
(0..effective_predicates.len())
+ .filter(|&i| {
+ let predicate_matches_entry =
predicate_matches[i][entry_idx];
+ let predicate_evaluated_for_entry =
+ predicate_fallback_plans[i].is_none_or(|plan| {
+ fallback_plan_evaluates_entry(
+ plan,
+ entry.index_type,
+ predicate_matches_entry,
+ )
+ });
+ if !predicate_evaluated_for_entry {
+ file_has_unsupported_match |= predicate_matches_entry;
+ return false;
+ }
+ file_evaluated = true;
+ if !predicate_matches_entry {
+ file_cannot_match = true;
+ return false;
+ }
+ true
+ })
+ .collect();
+ if file_cannot_match {
+ continue;
+ }
+ if !file_evaluated {
+ if file_has_unsupported_match {
+ return Ok(None);
+ }
continue;
}
@@ -344,22 +484,34 @@ impl GlobalIndexScanner {
.map(|b| b.data_type)
.or_else(|| effective_predicates.first().map(|p| p.2))
.unwrap_or(predicates[0].2);
- let reader = self
- .get_or_open_reader(&entry.file_name, &entry.meta, data_type)
- .await?;
+ let mut reader = if (between_matches &&
between_evaluated_for_entry)
+ || !matching_predicates.is_empty()
+ {
+ Some(self.open_reader_for_entry(entry, data_type).await?)
+ } else {
+ None
+ };
- let mut file_result: Option<RoaringTreemap> = None;
+ let mut file_result = None;
// Execute between query first if applicable
- if between_matches {
+ if between_matches && between_evaluated_for_entry {
if let Some(b) = &between {
let from_key = serialize_datum(b.from, b.data_type);
let to_key = serialize_datum(b.to, b.data_type);
let bitmap = reader
- .range_query(&from_key, &to_key, b.from_inclusive,
b.to_inclusive)
+ .as_ref()
+ .expect("reader is opened when between matches")
+ .range_query(
+ &from_key,
+ &to_key,
+ b.data_type,
+ b.from_inclusive,
+ b.to_inclusive,
+ )
.await
.map_err(|e| crate::Error::DataInvalid {
- message: "BTree query failed".to_string(),
+ message: "Global index query failed".to_string(),
source: Some(Box::new(e)),
})?;
file_result = Some(bitmap);
@@ -369,12 +521,15 @@ impl GlobalIndexScanner {
// Evaluate remaining predicates
for &idx in &matching_predicates {
let (op, literals, dt) = &effective_predicates[idx];
- let bitmap = reader.query(*op, literals, dt).await.map_err(|e|
{
- crate::Error::DataInvalid {
- message: "BTree query failed".to_string(),
+ let bitmap = reader
+ .as_ref()
+ .expect("reader is opened when predicates match")
+ .query(*op, literals, dt)
+ .await
+ .map_err(|e| crate::Error::DataInvalid {
+ message: "Global index query failed".to_string(),
source: Some(Box::new(e)),
- }
- })?;
+ })?;
file_result = Some(match file_result {
None => bitmap,
Some(mut existing) => {
@@ -384,8 +539,11 @@ impl GlobalIndexScanner {
});
}
- // Return reader to cache
- self.return_reader(entry.file_name.clone(), reader);
+ // Return BTree readers to cache. Bitmap readers are cheap wrappers
+ // around one opened file and are not cached yet.
+ if let Some(OpenedGlobalIndexReader::BTree(reader)) =
reader.take() {
+ self.return_reader(entry.file_name.clone(), reader);
+ }
if let Some(bitmap) = file_result {
for rid in bitmap.iter() {
@@ -403,12 +561,12 @@ impl GlobalIndexScanner {
file_name: &str,
meta: &BTreeIndexMeta,
data_type: &DataType,
- ) -> Result<BTreeIndexReader<BoxedCmp>> {
+ ) -> Result<OpenedGlobalIndexReader> {
// Try to take from cache
{
let mut cache = self.reader_cache.lock().unwrap();
if let Some(reader) = cache.remove(file_name) {
- return Ok(reader);
+ return Ok(OpenedGlobalIndexReader::BTree(reader));
}
}
@@ -421,12 +579,96 @@ impl GlobalIndexScanner {
let cmp = make_key_comparator(data_type);
BTreeIndexReader::open(Box::new(file_reader), file_size, meta, cmp)
.await
+ .map(OpenedGlobalIndexReader::BTree)
.map_err(|e| crate::Error::DataInvalid {
message: format!("Failed to open BTree index file:
{file_name}"),
source: Some(Box::new(e)),
})
}
+ async fn open_reader_for_entry(
+ &self,
+ entry: &GlobalIndexEntry,
+ data_type: &DataType,
+ ) -> Result<OpenedGlobalIndexReader> {
+ match entry.index_type {
+ GlobalIndexFileKind::BTree => {
+ self.get_or_open_reader(&entry.file_name, &entry.meta,
data_type)
+ .await
+ }
+ GlobalIndexFileKind::Bitmap => self
+ .open_bitmap_reader(&entry.file_name)
+ .await
+ .map(OpenedGlobalIndexReader::Bitmap)
+ .map_err(|e| crate::Error::DataInvalid {
+ message: format!(
+ "Failed to open bitmap global index file: {}",
+ entry.file_name
+ ),
+ source: Some(Box::new(e)),
+ }),
+ }
+ }
+
+ async fn open_bitmap_reader(
+ &self,
+ file_name: &str,
+ ) -> std::io::Result<BitmapGlobalIndexReader> {
+ let path = format!("{}/{INDEX_DIR}/{}", self.table_path, file_name);
+ let input = self
+ .file_io
+ .new_input(&path)
+ .map_err(|e| std::io::Error::other(e.to_string()))?;
+ let file_size = input
+ .metadata()
+ .await
+ .map_err(|e| std::io::Error::other(e.to_string()))?
+ .size;
+ let file_reader = input
+ .reader()
+ .await
+ .map_err(|e| std::io::Error::other(e.to_string()))?;
+ BitmapGlobalIndexReader::open(Box::new(file_reader), file_size).await
+ }
+
+ fn fallback_scan_plan(
+ &self,
+ entries: &[GlobalIndexEntry],
+ selected: &[bool],
+ ) -> FallbackScanPlan {
+ let mut plan = FallbackScanPlan::default();
+ let mut btree_total = 0i64;
+ let mut bitmap_total = 0i64;
+ let mut btree_valid = true;
+ let mut bitmap_valid = true;
+
+ for (entry, selected) in entries.iter().zip(selected) {
+ if !selected {
+ continue;
+ }
+ match entry.index_type {
+ GlobalIndexFileKind::BTree => {
+ plan.selected_btree += 1;
+ btree_valid &= add_file_size(&mut btree_total,
entry.file_size);
+ }
+ GlobalIndexFileKind::Bitmap => {
+ plan.selected_bitmap += 1;
+ bitmap_valid &= add_file_size(&mut bitmap_total,
entry.file_size);
+ }
+ }
+ }
+
+ plan.allow_btree = plan.selected_btree > 0
+ && btree_valid
+ && self.btree_fallback_scan_max_size > 0
+ && btree_total <= self.btree_fallback_scan_max_size;
+ plan.allow_bitmap = plan.selected_bitmap > 0
+ && bitmap_valid
+ && self.bitmap_fallback_scan_max_size > 0
+ && bitmap_total <= self.bitmap_fallback_scan_max_size;
+ plan
+ }
+
/// Return a reader to the cache for future reuse.
fn return_reader(&self, file_name: String, reader:
BTreeIndexReader<BoxedCmp>) {
let mut cache = self.reader_cache.lock().unwrap();
@@ -512,10 +754,10 @@ impl GlobalIndexScanner {
}
}
-/// Whether the b-tree global index can evaluate this operator directly.
+/// Whether the sorted global index can evaluate this operator directly.
/// Operators that fall outside this set bypass the index and are evaluated
/// later in the read pipeline (stats prune + parquet row filter).
-fn is_btree_supported_op(op: PredicateOperator) -> bool {
+fn is_sorted_global_index_supported_op(op: PredicateOperator) -> bool {
matches!(
op,
PredicateOperator::Eq
@@ -530,9 +772,49 @@ fn is_btree_supported_op(op: PredicateOperator) -> bool {
| PredicateOperator::IsNotNull
| PredicateOperator::Between
| PredicateOperator::NotBetween
+ | PredicateOperator::StartsWith
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains
+ | PredicateOperator::Like
)
}
+fn requires_fallback_scan(op: PredicateOperator) -> bool {
+ matches!(
+ op,
+ PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq
+ | PredicateOperator::Between
+ | PredicateOperator::NotBetween
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains
+ | PredicateOperator::Like
+ )
+}
+
+fn fallback_plan_evaluates_entry(
+ plan: FallbackScanPlan,
+ kind: GlobalIndexFileKind,
+ selected: bool,
+) -> bool {
+ !selected || plan.allowed(kind)
+}
+
+fn add_file_size(total: &mut i64, file_size: i64) -> bool {
+ if file_size < 0 {
+ return false;
+ }
+ match total.checked_add(file_size) {
+ Some(next) => {
+ *total = next;
+ true
+ }
+ None => false,
+ }
+}
+
/// Convert a RoaringTreemap to merged RowRanges (already sorted and
deduplicated).
fn bitmap_to_ranges(bitmap: &RoaringTreemap) -> Vec<RowRange> {
if bitmap.is_empty() {
@@ -938,6 +1220,8 @@ pub(crate) struct GlobalIndexEvaluation<'a> {
pub(crate) predicates: &'a [Predicate],
pub(crate) schema_fields: &'a [DataField],
pub(crate) search_mode: GlobalIndexSearchMode,
+ pub(crate) btree_fallback_scan_max_size: i64,
+ pub(crate) bitmap_fallback_scan_max_size: i64,
pub(crate) next_row_id: Option<i64>,
pub(crate) data_ranges: &'a [RowRange],
}
@@ -948,6 +1232,8 @@ pub(crate) async fn evaluate_global_index(
let scanner = match GlobalIndexScanner::create(
evaluation.file_io,
evaluation.table_path,
+ evaluation.btree_fallback_scan_max_size,
+ evaluation.bitmap_fallback_scan_max_size,
evaluation.index_entries,
evaluation.schema_fields,
) {
@@ -1128,12 +1414,50 @@ mod tests {
(file_io, table_path, testdata_name.to_string(), tmp)
}
+ type JavaBitmapTestdataTable = (FileIO, String, String, BTreeIndexMeta,
tempfile::TempDir);
+
+ fn setup_java_bitmap_testdata_table() -> JavaBitmapTestdataTable {
+ const FILE_NAME: &str = "bitmap_varchar_java.index";
+ let src = format!("{}/testdata/bitmap/{FILE_NAME}",
env!("CARGO_MANIFEST_DIR"));
+ let meta_src = format!(
+ "{}/testdata/bitmap/{FILE_NAME}.meta",
+ env!("CARGO_MANIFEST_DIR")
+ );
+ let tmp = tempfile::tempdir().unwrap();
+ let index_dir = tmp.path().join("index");
+ std::fs::create_dir_all(&index_dir).unwrap();
+ std::fs::copy(&src, index_dir.join(FILE_NAME)).unwrap();
+ let meta =
BTreeIndexMeta::deserialize(&std::fs::read(meta_src).unwrap()).unwrap();
+
+ let table_path = format!("file://{}", tmp.path().display());
+ let file_io = crate::io::FileIOBuilder::new("file").build().unwrap();
+ (file_io, table_path, FILE_NAME.to_string(), meta, tmp)
+ }
+
fn make_global_index_entry(
file_name: &str,
field_id: i32,
row_range_start: i64,
row_range_end: i64,
meta: &BTreeIndexMeta,
+ ) -> crate::spec::IndexManifestEntry {
+ make_global_index_entry_with_type(
+ BTREE_GLOBAL_INDEX_TYPE,
+ file_name,
+ field_id,
+ row_range_start,
+ row_range_end,
+ meta,
+ )
+ }
+
+ fn make_global_index_entry_with_type(
+ index_type: &str,
+ file_name: &str,
+ field_id: i32,
+ row_range_start: i64,
+ row_range_end: i64,
+ meta: &BTreeIndexMeta,
) -> crate::spec::IndexManifestEntry {
use crate::spec::{GlobalIndexMeta, IndexFileMeta};
IndexManifestEntry {
@@ -1142,7 +1466,7 @@ mod tests {
partition: vec![],
bucket: 0,
index_file: IndexFileMeta {
- index_type: BTREE_INDEX_TYPE.to_string(),
+ index_type: index_type.to_string(),
file_name: file_name.to_string(),
file_size: 0,
row_count: 0,
@@ -1166,12 +1490,41 @@ mod tests {
)]
}
+ fn string_schema_fields() -> Vec<DataField> {
+ vec![DataField::new(
+ 1,
+ "name".to_string(),
+ DataType::VarChar(crate::spec::VarCharType::string_type()),
+ )]
+ }
+
async fn evaluate_global_index_fast(
file_io: &FileIO,
table_path: &str,
entries: &[IndexManifestEntry],
predicates: &[Predicate],
fields: &[DataField],
+ ) -> Result<Option<Vec<RowRange>>> {
+ evaluate_global_index_fast_with_fallback_size(
+ file_io,
+ table_path,
+ entries,
+ predicates,
+ fields,
+ i64::MAX,
+ i64::MAX,
+ )
+ .await
+ }
+
+ async fn evaluate_global_index_fast_with_fallback_size(
+ file_io: &FileIO,
+ table_path: &str,
+ entries: &[IndexManifestEntry],
+ predicates: &[Predicate],
+ fields: &[DataField],
+ btree_fallback_scan_max_size: i64,
+ bitmap_fallback_scan_max_size: i64,
) -> Result<Option<Vec<RowRange>>> {
super::evaluate_global_index(super::GlobalIndexEvaluation {
file_io,
@@ -1180,6 +1533,8 @@ mod tests {
predicates,
schema_fields: fields,
search_mode: GlobalIndexSearchMode::Fast,
+ btree_fallback_scan_max_size,
+ bitmap_fallback_scan_max_size,
next_row_id: None,
data_ranges: &[],
})
@@ -1217,8 +1572,15 @@ mod tests {
let meta = BTreeIndexMeta::new(None, None, false);
let entries = vec![make_global_index_entry("idx", 1, 0, 49, &meta)];
let fields = int_schema_fields();
- let scanner =
- GlobalIndexScanner::create(&file_io, "memory:/t", &entries,
&fields).expect("scanner");
+ let scanner = GlobalIndexScanner::create(
+ &file_io,
+ "memory:/t",
+ i64::MAX,
+ i64::MAX,
+ &entries,
+ &fields,
+ )
+ .expect("scanner");
let ranges = scanner
.unindexed_ranges(
@@ -1237,8 +1599,15 @@ mod tests {
let meta = BTreeIndexMeta::new(None, None, false);
let entries = vec![make_global_index_entry("idx", 1, 0, 49, &meta)];
let fields = int_schema_fields();
- let scanner =
- GlobalIndexScanner::create(&file_io, "memory:/t", &entries,
&fields).expect("scanner");
+ let scanner = GlobalIndexScanner::create(
+ &file_io,
+ "memory:/t",
+ i64::MAX,
+ i64::MAX,
+ &entries,
+ &fields,
+ )
+ .expect("scanner");
let ranges = scanner
.unindexed_ranges(
@@ -1257,8 +1626,15 @@ mod tests {
let meta = BTreeIndexMeta::new(None, None, false);
let entries = vec![make_global_index_entry("idx", 1, 0, 49, &meta)];
let fields = int_schema_fields();
- let scanner =
- GlobalIndexScanner::create(&file_io, "memory:/t", &entries,
&fields).expect("scanner");
+ let scanner = GlobalIndexScanner::create(
+ &file_io,
+ "memory:/t",
+ i64::MAX,
+ i64::MAX,
+ &entries,
+ &fields,
+ )
+ .expect("scanner");
let ranges = scanner
.unindexed_ranges(
@@ -1284,8 +1660,15 @@ mod tests {
make_global_index_entry("idx_value", 2, 0, 99, &meta),
];
let fields = two_field_schema_fields();
- let scanner =
- GlobalIndexScanner::create(&file_io, "memory:/t", &entries,
&fields).expect("scanner");
+ let scanner = GlobalIndexScanner::create(
+ &file_io,
+ "memory:/t",
+ i64::MAX,
+ i64::MAX,
+ &entries,
+ &fields,
+ )
+ .expect("scanner");
let predicate = Predicate::and(vec![int_eq("id", 0, 7),
int_eq("value", 1, 8)]);
let ranges = scanner
@@ -1300,8 +1683,15 @@ mod tests {
let meta = BTreeIndexMeta::new(None, None, false);
let entries = vec![make_global_index_entry("idx_id", 1, 0, 49, &meta)];
let fields = two_field_schema_fields();
- let scanner =
- GlobalIndexScanner::create(&file_io, "memory:/t", &entries,
&fields).expect("scanner");
+ let scanner = GlobalIndexScanner::create(
+ &file_io,
+ "memory:/t",
+ i64::MAX,
+ i64::MAX,
+ &entries,
+ &fields,
+ )
+ .expect("scanner");
let predicate = Predicate::and(vec![int_eq("id", 0, 7),
int_eq("value", 1, 8)]);
let ranges = scanner
@@ -1322,8 +1712,15 @@ mod tests {
.unwrap()
.extra_field_ids = Some(vec![2]);
let fields = two_field_schema_fields();
- let scanner =
- GlobalIndexScanner::create(&file_io, "memory:/t", &[entry],
&fields).expect("scanner");
+ let scanner = GlobalIndexScanner::create(
+ &file_io,
+ "memory:/t",
+ i64::MAX,
+ i64::MAX,
+ &[entry],
+ &fields,
+ )
+ .expect("scanner");
let ranges = scanner
.unindexed_ranges(
@@ -1386,6 +1783,343 @@ mod tests {
assert_eq!(ranges, vec![RowRange::new(25, 25)]);
}
+ #[tokio::test]
+ async fn test_evaluate_java_bitmap_golden_index_eq_and_null() {
+ let data_type =
DataType::VarChar(crate::spec::VarCharType::string_type());
+ let (file_io, table_path, file_name, meta, _tmp) =
setup_java_bitmap_testdata_table();
+ let entries = vec![make_global_index_entry_with_type(
+ BITMAP_GLOBAL_INDEX_TYPE,
+ &file_name,
+ 1,
+ 100,
+ 109,
+ &meta,
+ )];
+ let fields = string_schema_fields();
+ assert_eq!(meta.first_key, Some(b"alpha".to_vec()));
+ assert_eq!(meta.last_key, Some(b"office".to_vec()));
+ assert!(meta.has_nulls);
+
+ let eq_predicates = vec![Predicate::Leaf {
+ column: "name".to_string(),
+ index: 0,
+ data_type: data_type.clone(),
+ op: PredicateOperator::Eq,
+ literals: vec![Datum::String("k2".to_string())],
+ }];
+ let eq_result =
+ evaluate_global_index_fast(&file_io, &table_path, &entries,
&eq_predicates, &fields)
+ .await
+ .unwrap();
+ assert_eq!(eq_result.unwrap(), vec![RowRange::new(105, 106)]);
+
+ let null_predicates = vec![Predicate::Leaf {
+ column: "name".to_string(),
+ index: 0,
+ data_type,
+ op: PredicateOperator::IsNull,
+ literals: vec![],
+ }];
+ let null_result =
+ evaluate_global_index_fast(&file_io, &table_path, &entries,
&null_predicates, &fields)
+ .await
+ .unwrap();
+ assert_eq!(null_result.unwrap(), vec![RowRange::new(104, 104)]);
+ }
+
+ #[tokio::test]
+ async fn test_evaluate_java_bitmap_golden_index_string_fallback_scan() {
+ let data_type =
DataType::VarChar(crate::spec::VarCharType::string_type());
+ let (file_io, table_path, file_name, meta, _tmp) =
setup_java_bitmap_testdata_table();
+ let entries = vec![make_global_index_entry_with_type(
+ BITMAP_GLOBAL_INDEX_TYPE,
+ &file_name,
+ 1,
+ 100,
+ 109,
+ &meta,
+ )];
+ let fields = string_schema_fields();
+
+ let ends_with_predicates = vec![Predicate::Leaf {
+ column: "name".to_string(),
+ index: 0,
+ data_type: data_type.clone(),
+ op: PredicateOperator::EndsWith,
+ literals: vec![Datum::String("ta".to_string())],
+ }];
+ let ends_with_result = evaluate_global_index_fast(
+ &file_io,
+ &table_path,
+ &entries,
+ &ends_with_predicates,
+ &fields,
+ )
+ .await
+ .unwrap();
+ assert_eq!(
+ ends_with_result.unwrap(),
+ vec![RowRange::new(101, 101), RowRange::new(103, 103)]
+ );
+
+ let contains_predicates = vec![Predicate::Leaf {
+ column: "name".to_string(),
+ index: 0,
+ data_type: data_type.clone(),
+ op: PredicateOperator::Contains,
+ literals: vec![Datum::String("ph".to_string())],
+ }];
+ let contains_result = evaluate_global_index_fast(
+ &file_io,
+ &table_path,
+ &entries,
+ &contains_predicates,
+ &fields,
+ )
+ .await
+ .unwrap();
+ assert_eq!(
+ contains_result.unwrap(),
+ vec![RowRange::new(100, 100), RowRange::new(102, 102)]
+ );
+
+ let like_predicates = vec![Predicate::Leaf {
+ column: "name".to_string(),
+ index: 0,
+ data_type: data_type.clone(),
+ op: PredicateOperator::Like,
+ literals: vec![Datum::String("%ha%".to_string())],
+ }];
+ let like_result =
+ evaluate_global_index_fast(&file_io, &table_path, &entries,
&like_predicates, &fields)
+ .await
+ .unwrap();
+ assert_eq!(
+ like_result.unwrap(),
+ vec![RowRange::new(100, 100), RowRange::new(102, 102)]
+ );
+
+ let less_than_predicates = vec![Predicate::Leaf {
+ column: "name".to_string(),
+ index: 0,
+ data_type: data_type.clone(),
+ op: PredicateOperator::Lt,
+ literals: vec![Datum::String("delta".to_string())],
+ }];
+ let less_than_result = evaluate_global_index_fast(
+ &file_io,
+ &table_path,
+ &entries,
+ &less_than_predicates,
+ &fields,
+ )
+ .await
+ .unwrap();
+ assert_eq!(less_than_result.unwrap(), vec![RowRange::new(100, 102)]);
+
+ let mut over_limit_entries = vec![make_global_index_entry_with_type(
+ BITMAP_GLOBAL_INDEX_TYPE,
+ &file_name,
+ 1,
+ 100,
+ 109,
+ &meta,
+ )];
+ over_limit_entries[0].index_file.file_size = 2;
+ let over_limit_less_than =
evaluate_global_index_fast_with_fallback_size(
+ &file_io,
+ &table_path,
+ &over_limit_entries,
+ &less_than_predicates,
+ &fields,
+ i64::MAX,
+ 1,
+ )
+ .await
+ .unwrap();
+ assert!(
+ over_limit_less_than.is_none(),
+ "range predicates require fallback dictionary scans and should be
unsupported over budget"
+ );
+
+ let no_match_contains = vec![Predicate::Leaf {
+ column: "name".to_string(),
+ index: 0,
+ data_type: data_type.clone(),
+ op: PredicateOperator::Contains,
+ literals: vec![Datum::String("zz".to_string())],
+ }];
+ let over_limit_result = evaluate_global_index_fast_with_fallback_size(
+ &file_io,
+ &table_path,
+ &over_limit_entries,
+ &no_match_contains,
+ &fields,
+ i64::MAX,
+ 1,
+ )
+ .await
+ .unwrap();
+ assert!(
+ over_limit_result.is_none(),
+ "fallback scans over budget should be unsupported instead of
returning full coverage"
+ );
+
+ let direct_with_over_limit_fallback = vec![Predicate::and(vec![
+ Predicate::Leaf {
+ column: "name".to_string(),
+ index: 0,
+ data_type: data_type.clone(),
+ op: PredicateOperator::Eq,
+ literals: vec![Datum::String("k2".to_string())],
+ },
+ Predicate::Leaf {
+ column: "name".to_string(),
+ index: 0,
+ data_type,
+ op: PredicateOperator::Contains,
+ literals: vec![Datum::String("zz".to_string())],
+ },
+ ])];
+ let direct_result = evaluate_global_index_fast_with_fallback_size(
+ &file_io,
+ &table_path,
+ &over_limit_entries,
+ &direct_with_over_limit_fallback,
+ &fields,
+ i64::MAX,
+ 1,
+ )
+ .await
+ .unwrap();
+ assert_eq!(direct_result.unwrap(), vec![RowRange::new(105, 106)]);
+ }
+
+ #[tokio::test]
+ async fn test_btree_fallback_scan_over_limit_is_unsupported() {
+ let (file_io, table_path, file_name, tmp) =
+ setup_testdata_table("btree_varchar_100_no_compress.bin");
+ let meta = BTreeIndexMeta::new(Some(b"a".to_vec()),
Some(b"yyyy".to_vec()), false);
+ let fields = string_schema_fields();
+ let data_type =
DataType::VarChar(crate::spec::VarCharType::string_type());
+ let predicates = vec![Predicate::Leaf {
+ column: "name".to_string(),
+ index: 0,
+ data_type,
+ op: PredicateOperator::Contains,
+ literals: vec![Datum::String("not-present".to_string())],
+ }];
+
+ let entries = vec![make_global_index_entry(&file_name, 1, 0, 99,
&meta)];
+ let exact_result = evaluate_global_index_fast_with_fallback_size(
+ &file_io,
+ &table_path,
+ &entries,
+ &predicates,
+ &fields,
+ i64::MAX,
+ i64::MAX,
+ )
+ .await
+ .unwrap();
+ assert_eq!(exact_result.unwrap(), Vec::<RowRange>::new());
+
+ let mut over_limit_entries = vec![make_global_index_entry(&file_name,
1, 0, 99, &meta)];
+ over_limit_entries[0].index_file.file_size = 2;
+ let over_limit_result = evaluate_global_index_fast_with_fallback_size(
+ &file_io,
+ &table_path,
+ &over_limit_entries,
+ &predicates,
+ &fields,
+ 1,
+ i64::MAX,
+ )
+ .await
+ .unwrap();
+ assert!(
+ over_limit_result.is_none(),
+ "fallback scans over budget should be unsupported instead of
returning full coverage"
+ );
+
+ let second_file_name = "btree_varchar_100_no_compress_2.bin";
+ std::fs::copy(
+ tmp.path().join("index").join(&file_name),
+ tmp.path().join("index").join(second_file_name),
+ )
+ .unwrap();
+ let mut first = make_global_index_entry(&file_name, 1, 0, 99, &meta);
+ first.index_file.file_size = 1;
+ let mut second = make_global_index_entry(second_file_name, 1, 100,
199, &meta);
+ second.index_file.file_size = 1;
+ let total_over_limit_result =
evaluate_global_index_fast_with_fallback_size(
+ &file_io,
+ &table_path,
+ &[first, second],
+ &predicates,
+ &fields,
+ 1,
+ i64::MAX,
+ )
+ .await
+ .unwrap();
+ assert!(
+ total_over_limit_result.is_none(),
+ "fallback budget should use selected files' total size, not
per-file size"
+ );
+ }
+
+ #[tokio::test]
+ async fn
test_fallback_scan_over_limit_with_mixed_index_kinds_is_unsupported() {
+ let (file_io, table_path, file_name, _tmp) =
+ setup_testdata_table("btree_varchar_100_no_compress.bin");
+ let btree_meta = BTreeIndexMeta::new(Some(b"a".to_vec()),
Some(b"yyyy".to_vec()), false);
+ let bitmap_meta = BTreeIndexMeta::new(Some(b"m".to_vec()),
Some(b"z".to_vec()), false);
+ let fields = string_schema_fields();
+ let predicates = vec![Predicate::Leaf {
+ column: "name".to_string(),
+ index: 0,
+ data_type:
DataType::VarChar(crate::spec::VarCharType::string_type()),
+ op: PredicateOperator::Lt,
+ literals: vec![Datum::String("delta".to_string())],
+ }];
+
+ let mut btree = make_global_index_entry_with_type(
+ BTREE_GLOBAL_INDEX_TYPE,
+ &file_name,
+ 1,
+ 0,
+ 99,
+ &btree_meta,
+ );
+ btree.index_file.file_size = 2;
+ let mut bitmap = make_global_index_entry_with_type(
+ BITMAP_GLOBAL_INDEX_TYPE,
+ "bitmap-no-match.index",
+ 1,
+ 100,
+ 199,
+ &bitmap_meta,
+ );
+ bitmap.index_file.file_size = 1;
+
+ let result = evaluate_global_index_fast_with_fallback_size(
+ &file_io,
+ &table_path,
+ &[btree, bitmap],
+ &predicates,
+ &fields,
+ 1,
+ i64::MAX,
+ )
+ .await
+ .unwrap();
+ assert!(
+ result.is_none(),
+ "an over-budget selected BTree file must stay unsupported even if
bitmap files are pruned by metadata"
+ );
+ }
+
#[tokio::test]
async fn test_evaluate_global_index_full_mode_includes_unindexed_tail() {
let (file_io, table_path, file_name, _tmp) =
@@ -1402,6 +2136,8 @@ mod tests {
predicates: &predicates,
schema_fields: &fields,
search_mode: GlobalIndexSearchMode::Full,
+ btree_fallback_scan_max_size: i64::MAX,
+ bitmap_fallback_scan_max_size: i64::MAX,
next_row_id: Some(150),
data_ranges: &[],
})
@@ -1452,6 +2188,8 @@ mod tests {
predicates: &predicates,
schema_fields: &fields,
search_mode: GlobalIndexSearchMode::Full,
+ btree_fallback_scan_max_size: i64::MAX,
+ bitmap_fallback_scan_max_size: i64::MAX,
next_row_id: Some(100),
data_ranges: &[],
})
@@ -1483,6 +2221,8 @@ mod tests {
predicates: &predicates,
schema_fields: &fields,
search_mode: GlobalIndexSearchMode::Detail,
+ btree_fallback_scan_max_size: i64::MAX,
+ bitmap_fallback_scan_max_size: i64::MAX,
next_row_id: Some(150),
data_ranges: &data_ranges,
})
@@ -1531,6 +2271,24 @@ mod tests {
.unwrap();
let ranges = result.unwrap();
assert_eq!(ranges, vec![RowRange::new(5, 10)]);
+
+ let mut over_limit_entries = vec![make_global_index_entry(&file_name,
1, 0, 99, &meta)];
+ over_limit_entries[0].index_file.file_size = 2;
+ let over_limit_result = evaluate_global_index_fast_with_fallback_size(
+ &file_io,
+ &table_path,
+ &over_limit_entries,
+ &predicates,
+ &fields,
+ 1,
+ i64::MAX,
+ )
+ .await
+ .unwrap();
+ assert!(
+ over_limit_result.is_none(),
+ "between/range predicates require fallback scans and should be
unsupported over budget"
+ );
}
#[tokio::test]
diff --git a/crates/paimon/src/table/global_index_types.rs
b/crates/paimon/src/table/global_index_types.rs
new file mode 100644
index 0000000..bfa4430
--- /dev/null
+++ b/crates/paimon/src/table/global_index_types.rs
@@ -0,0 +1,29 @@
+// 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.
+
+pub(crate) const BTREE_GLOBAL_INDEX_TYPE: &str = "btree";
+pub(crate) const BITMAP_GLOBAL_INDEX_TYPE: &str = "bitmap";
+
+pub(crate) fn normalize_sorted_global_index_type(index_type: &str) ->
Option<&'static str> {
+ if index_type.eq_ignore_ascii_case(BTREE_GLOBAL_INDEX_TYPE) {
+ Some(BTREE_GLOBAL_INDEX_TYPE)
+ } else if index_type.eq_ignore_ascii_case(BITMAP_GLOBAL_INDEX_TYPE) {
+ Some(BITMAP_GLOBAL_INDEX_TYPE)
+ } else {
+ None
+ }
+}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 9e6b623..373efa5 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -19,6 +19,7 @@
pub(crate) mod aggregator;
pub(crate) mod bin_pack;
+mod bitmap_global_index_reader;
mod blob_file_writer;
mod branch_manager;
mod btree_global_index_build_builder;
@@ -39,6 +40,7 @@ mod data_file_writer;
#[cfg(feature = "fulltext")]
mod full_text_search_builder;
pub(crate) mod global_index_scanner;
+mod global_index_types;
mod hybrid_search_builder;
mod kv_file_reader;
mod kv_file_writer;
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 115fb01..67184a2 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -1065,6 +1065,10 @@ impl<'a> TableScan<'a> {
} else {
Vec::new()
};
+ let btree_index_fallback_scan_max_size =
+ core_options.btree_index_fallback_scan_max_size()?;
+ let bitmap_index_fallback_scan_max_size =
+ core_options.bitmap_index_fallback_scan_max_size()?;
let snapshot_id = snapshot.id();
let base_path = table_path.trim_end_matches('/');
@@ -1119,6 +1123,8 @@ impl<'a> TableScan<'a> {
predicates: &self.data_predicates,
schema_fields: self.table.schema().fields(),
search_mode,
+ btree_fallback_scan_max_size:
btree_index_fallback_scan_max_size,
+ bitmap_fallback_scan_max_size:
bitmap_index_fallback_scan_max_size,
next_row_id: snapshot.next_row_id(),
data_ranges: &global_index_detail_data_ranges,
},
diff --git a/crates/paimon/testdata/bitmap/bitmap_varchar_java.index
b/crates/paimon/testdata/bitmap/bitmap_varchar_java.index
new file mode 100644
index 0000000..028fc46
Binary files /dev/null and
b/crates/paimon/testdata/bitmap/bitmap_varchar_java.index differ
diff --git a/crates/paimon/testdata/bitmap/bitmap_varchar_java.index.meta
b/crates/paimon/testdata/bitmap/bitmap_varchar_java.index.meta
new file mode 100644
index 0000000..5571240
Binary files /dev/null and
b/crates/paimon/testdata/bitmap/bitmap_varchar_java.index.meta differ
diff --git a/docs/src/sql.md b/docs/src/sql.md
index b97a654..934eb27 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -648,10 +648,18 @@ CALL sys.create_global_index(
index_column => 'id',
index_type => 'btree'
);
+
+CALL sys.create_global_index(
+ table => 'paimon.my_db.my_table',
+ index_column => 'tag',
+ index_type => 'bitmap'
+);
```
-`index_type` defaults to `btree`. BTree indexes support scalar columns and do
-not accept the `options` argument yet.
+`index_type` defaults to `btree`. BTree and bitmap global indexes support
+scalar columns and do not accept the `options` argument yet. Bitmap global
+indexes use the same on-disk file format as Java Paimon's
+`BitmapGlobalIndexFormat`.
The current global-index builders require a row-tracking data-evolution table
with global indexes enabled. They do not support primary-key tables or tables
@@ -726,7 +734,7 @@ FROM paimon.my_db.items$table_indexes;
### drop_global_index
-Drop a committed BTree global index:
+Drop a committed sorted global index:
```sql
CALL sys.drop_global_index(
@@ -736,7 +744,7 @@ CALL sys.drop_global_index(
);
```
-Only BTree indexes can be dropped through this procedure currently.
+BTree and bitmap global indexes can be dropped through this procedure
currently.
### create_lumina_index
@@ -1384,7 +1392,7 @@ Columns:
|---|---|---|
| `partition` | STRING | Partition spec for the indexed data, or `NULL` for
unpartitioned tables |
| `bucket` | INT | Bucket id covered by the index file |
-| `index_type` | STRING | Index type, such as `btree`, `ivf-flat`, `lumina`,
or `DELETION_VECTORS` |
+| `index_type` | STRING | Index type, such as `btree`, `bitmap`, `ivf-flat`,
`lumina`, or `DELETION_VECTORS` |
| `file_name` | STRING | Index file name under the table index directory |
| `file_size` | BIGINT | Index file size in bytes |
| `row_count` | BIGINT | Number of rows covered by the index file |
@@ -1516,6 +1524,8 @@ deletion vectors enabled.
| `global-index.enabled` | `false` | Enables global index metadata and
global-index-aware reads. |
| `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
range. |
+| `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. |
+| `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`. |
### Variant Shredding Options
