You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/10/03 12:34:25 UTC
(paimon-rust) branch main updated: feat(btree): support composite global indexes (#1015)
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 c41ce209 feat(btree): support composite global indexes (#1015)
c41ce209 is described below
commit c41ce2099fae18101e9faaf7501311c5c2340679
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Sat Oct 3 20:34:20 2026 +0800
feat(btree): support composite global indexes (#1015)
---
crates/paimon/src/btree/composite_key.rs | 268 +++++++
crates/paimon/src/btree/composite_predicate.rs | 457 +++++++++++
crates/paimon/src/btree/composite_tests.rs | 664 ++++++++++++++++
crates/paimon/src/btree/key_serde.rs | 52 ++
crates/paimon/src/btree/mod.rs | 8 +
crates/paimon/src/btree/reader.rs | 52 ++
crates/paimon/src/spec/mod.rs | 1 +
crates/paimon/src/spec/predicate.rs | 2 +-
.../paimon/src/table/global_index_build_common.rs | 68 +-
.../paimon/src/table/global_index_drop_builder.rs | 47 +-
crates/paimon/src/table/global_index_scanner.rs | 28 +-
.../src/table/global_index_scanner/composite.rs | 227 ++++++
.../src/table/global_index_scanner/evaluator.rs | 104 ++-
.../src/table/global_index_scanner/reader.rs | 6 +-
.../paimon/src/table/global_index_scanner/tests.rs | 60 +-
.../src/table/sorted_global_index_build_builder.rs | 38 +-
.../extraction.rs | 95 ++-
.../sorted_global_index_build_builder/tests.rs | 875 +++++++++++++++++++++
.../validation.rs | 1 +
.../sorted_global_index_build_builder/writer.rs | 35 +-
.../testdata/btree/GenerateCompositeBTree.java | 96 +++
crates/paimon/testdata/btree/README.md | 35 +
.../testdata/btree/btree_composite_java_scalar.key | Bin 0 -> 52 bytes
.../testdata/btree/btree_composite_v1_java_lz4.bin | Bin 0 -> 873 bytes
.../btree/btree_composite_v1_java_lz4.meta | Bin 0 -> 23 bytes
.../btree/btree_composite_v1_java_none.bin | Bin 0 -> 1144 bytes
.../btree/btree_composite_v1_java_none.meta | Bin 0 -> 23 bytes
.../testdata/btree/btree_composite_v2_java_lz4.bin | Bin 0 -> 799 bytes
.../btree/btree_composite_v2_java_lz4.meta | Bin 0 -> 23 bytes
.../btree/btree_composite_v2_java_none.bin | Bin 0 -> 1249 bytes
.../btree/btree_composite_v2_java_none.meta | Bin 0 -> 23 bytes
docs/src/sql.md | 46 ++
32 files changed, 3199 insertions(+), 66 deletions(-)
diff --git a/crates/paimon/src/btree/composite_key.rs b/crates/paimon/src/btree/composite_key.rs
new file mode 100644
index 00000000..71584529
--- /dev/null
+++ b/crates/paimon/src/btree/composite_key.rs
@@ -0,0 +1,268 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Java `CompositeKeySerializer` / `RowCompactedSerializer` tuple encoding.
+
+use super::key_serde::{fixed_key_bytes, make_key_comparator, serialize_datum, KeyComparator};
+use super::var_len::{encode_var_int, try_decode_var_int_from_slice};
+use crate::spec::{DataField, DataType, Datum};
+use crate::{Error, Result};
+use std::cmp::Ordering;
+
+pub(crate) struct CompositeKeyCodec {
+ types: Vec<DataType>,
+ comparators: Vec<KeyComparator>,
+}
+
+/// Read fields directly from a compacted row slice, without materializing a row
+/// or allocating a component list. The cursor is local to each comparison.
+pub(crate) struct CompositeKeyCursor<'codec, 'key> {
+ codec: &'codec CompositeKeyCodec,
+ key: &'key [u8],
+ position: usize,
+ offset: usize,
+}
+
+fn invalid(message: &str) -> Error {
+ Error::DataInvalid {
+ message: format!("Invalid composite BTree key: {message}"),
+ source: None,
+ }
+}
+
+/// Serialize a tuple in index-column order, including NULL components. A tuple
+/// containing only NULL components is still a key, not a scalar NULL posting.
+pub fn serialize_composite_key(values: &[Option<Datum>], fields: &[DataField]) -> Result<Vec<u8>> {
+ CompositeKeyCodec::new(fields).serialize(values)
+}
+
+impl CompositeKeyCodec {
+ pub(crate) fn new(fields: &[DataField]) -> Self {
+ Self {
+ types: fields.iter().map(|f| f.data_type().clone()).collect(),
+ comparators: fields
+ .iter()
+ .map(|f| make_key_comparator(f.data_type()))
+ .collect(),
+ }
+ }
+
+ pub(crate) fn serialize(&self, values: &[Option<Datum>]) -> Result<Vec<u8>> {
+ if values.len() != self.types.len() || values.is_empty() {
+ return Err(invalid("tuple arity differs from index fields"));
+ }
+ // One RowKind byte (INSERT = 0), then packed NULL bits.
+ let mut bytes = vec![0; 1 + values.len().div_ceil(8)];
+ for (i, (value, ty)) in values.iter().zip(&self.types).enumerate() {
+ if !supported_type(ty) {
+ return Err(invalid("unsupported non-scalar component type"));
+ }
+ match value {
+ None => bytes[1 + i / 8] |= 1 << (i % 8),
+ Some(value) => {
+ crate::spec::validate_datum_matches_type(value, ty)?;
+ let component = self.serialize_component(i, value);
+ if variable_width(ty) {
+ let len = i32::try_from(component.len())
+ .map_err(|_| invalid("component too large"))?;
+ encode_var_int(&mut bytes, len).expect("writing to Vec");
+ }
+ bytes.extend(component);
+ }
+ }
+ }
+ // Also validate the input types and complete layout before writing it.
+ self.validate_key(&bytes)?;
+ Ok(bytes)
+ }
+
+ pub(crate) fn serialize_component(&self, i: usize, value: &Datum) -> Vec<u8> {
+ // Java Float/Double.compare treats every NaN as equal. Bloom hashing
+ // therefore uses the canonical positive NaN payload for each component.
+ match value {
+ Datum::Float(v) if v.is_nan() => 0x7fc00000u32.to_le_bytes().to_vec(),
+ Datum::Double(v) if v.is_nan() => 0x7ff8000000000000u64.to_le_bytes().to_vec(),
+ _ => serialize_datum(value, &self.types[i]),
+ }
+ }
+
+ pub(crate) fn cursor<'a>(&self, key: &'a [u8]) -> Result<CompositeKeyCursor<'_, 'a>> {
+ let offset = 1 + self.types.len().div_ceil(8);
+ if key.len() < offset {
+ return Err(invalid("truncated header"));
+ }
+ Ok(CompositeKeyCursor {
+ codec: self,
+ key,
+ position: 0,
+ offset,
+ })
+ }
+
+ pub(crate) fn validate_key(&self, key: &[u8]) -> Result<()> {
+ self.cursor(key)?.finish()
+ }
+
+ pub(crate) fn compare_component(
+ &self,
+ i: usize,
+ a: Option<&[u8]>,
+ b: Option<&[u8]>,
+ ) -> Result<Ordering> {
+ match (a, b) {
+ (None, None) => Ok(Ordering::Equal),
+ (None, Some(_)) => Ok(Ordering::Less),
+ (Some(_), None) => Ok(Ordering::Greater),
+ (Some(a), Some(b)) => match &self.types[i] {
+ DataType::Float(_) => {
+ let a = f32::from_le_bytes(fixed_key_bytes::<4>(a, "FLOAT")?);
+ let b = f32::from_le_bytes(fixed_key_bytes::<4>(b, "FLOAT")?);
+ Ok(
+ (if a.is_nan() { f32::NAN } else { a }).total_cmp(&if b.is_nan() {
+ f32::NAN
+ } else {
+ b
+ }),
+ )
+ }
+ DataType::Double(_) => {
+ let a = f64::from_le_bytes(fixed_key_bytes::<8>(a, "DOUBLE")?);
+ let b = f64::from_le_bytes(fixed_key_bytes::<8>(b, "DOUBLE")?);
+ Ok(
+ (if a.is_nan() { f64::NAN } else { a }).total_cmp(&if b.is_nan() {
+ f64::NAN
+ } else {
+ b
+ }),
+ )
+ }
+ _ => (self.comparators[i])(a, b),
+ },
+ }
+ }
+
+ pub(crate) fn compare_keys(&self, a: &[u8], b: &[u8]) -> Result<Ordering> {
+ let mut a = self.cursor(a)?;
+ let mut b = self.cursor(b)?;
+ let mut order = Ordering::Equal;
+ for i in 0..self.types.len() {
+ order = self.compare_component(i, a.next_component()?, b.next_component()?)?;
+ if order != Ordering::Equal {
+ break;
+ }
+ }
+ a.finish()?;
+ b.finish()?;
+ Ok(order)
+ }
+}
+
+impl<'key> CompositeKeyCursor<'_, 'key> {
+ pub(crate) fn next_component(&mut self) -> Result<Option<&'key [u8]>> {
+ let i = self.position;
+ let ty = self
+ .codec
+ .types
+ .get(i)
+ .ok_or_else(|| invalid("read past tuple arity"))?;
+ self.position += 1;
+ if self.key[1 + i / 8] & (1 << (i % 8)) != 0 {
+ return Ok(None);
+ }
+ let len = if variable_width(ty) {
+ let (len, consumed) = try_decode_var_int_from_slice(self.key, self.offset)?;
+ self.offset += consumed;
+ usize::try_from(len).map_err(|_| invalid("negative component length"))?
+ } else {
+ match ty {
+ DataType::Boolean(_) | DataType::TinyInt(_) => 1,
+ DataType::SmallInt(_) => 2,
+ DataType::Int(_) | DataType::Date(_) | DataType::Time(_) | DataType::Float(_) => 4,
+ DataType::BigInt(_) | DataType::Double(_) | DataType::Decimal(_) => 8,
+ DataType::Timestamp(t) if t.precision() > 3 => {
+ timestamp_len(self.key, self.offset)?
+ }
+ DataType::LocalZonedTimestamp(t) if t.precision() > 3 => {
+ timestamp_len(self.key, self.offset)?
+ }
+ DataType::Timestamp(_) | DataType::LocalZonedTimestamp(_) => 8,
+ _ => return Err(invalid("unsupported non-scalar component type")),
+ }
+ };
+ let end = self
+ .offset
+ .checked_add(len)
+ .ok_or_else(|| invalid("component length overflow"))?;
+ let component = self
+ .key
+ .get(self.offset..end)
+ .ok_or_else(|| invalid("truncated component"))?;
+ self.codec
+ .compare_component(i, Some(component), Some(component))?;
+ self.offset = end;
+ Ok(Some(component))
+ }
+
+ pub(crate) fn finish(mut self) -> Result<()> {
+ // Preserve validation of suffix fields even when an earlier field has
+ // already decided the order, so incompatible keys still trigger fallback.
+ while self.position < self.codec.types.len() {
+ self.next_component()?;
+ }
+ if self.offset != self.key.len() {
+ return Err(invalid("trailing bytes or incompatible schema"));
+ }
+ Ok(())
+ }
+}
+
+fn supported_type(ty: &DataType) -> bool {
+ variable_width(ty)
+ || matches!(
+ ty,
+ DataType::Boolean(_)
+ | DataType::TinyInt(_)
+ | DataType::SmallInt(_)
+ | DataType::Int(_)
+ | DataType::BigInt(_)
+ | DataType::Float(_)
+ | DataType::Double(_)
+ | DataType::Date(_)
+ | DataType::Time(_)
+ | DataType::Decimal(_)
+ | DataType::Timestamp(_)
+ | DataType::LocalZonedTimestamp(_)
+ )
+}
+
+fn variable_width(ty: &DataType) -> bool {
+ matches!(
+ ty,
+ DataType::Char(_) | DataType::VarChar(_) | DataType::Binary(_) | DataType::VarBinary(_)
+ ) || matches!(ty, DataType::Decimal(d) if d.precision() > 18)
+}
+
+fn timestamp_len(key: &[u8], offset: usize) -> Result<usize> {
+ let start = offset
+ .checked_add(8)
+ .ok_or_else(|| invalid("timestamp overflow"))?;
+ let (nanos, consumed) = try_decode_var_int_from_slice(key, start)?;
+ if !(0..1_000_000).contains(&nanos) {
+ return Err(invalid("invalid timestamp nanoseconds"));
+ }
+ Ok(8 + consumed)
+}
diff --git a/crates/paimon/src/btree/composite_predicate.rs b/crates/paimon/src/btree/composite_predicate.rs
new file mode 100644
index 00000000..cb96f520
--- /dev/null
+++ b/crates/paimon/src/btree/composite_predicate.rs
@@ -0,0 +1,457 @@
+// 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.
+
+//! Left-prefix tuple intervals matching Java `CompositeBTreePredicate`.
+
+use super::key_serde::normalize_key_literal;
+use super::CompositeKeyCodec;
+use crate::spec::{DataField, Datum, Predicate, PredicateOperator};
+use crate::Result;
+use std::cmp::Ordering;
+
+pub(crate) const MAX_INTERVALS: usize = 256;
+type Component = Option<Vec<u8>>;
+
+pub(crate) struct CompositeBound {
+ values: Vec<Component>,
+ after: bool,
+}
+
+impl CompositeBound {
+ fn new(values: Vec<Component>, after: bool) -> Self {
+ Self { values, after }
+ }
+
+ pub(crate) fn compare_key(&self, key: &[u8], codec: &CompositeKeyCodec) -> Result<Ordering> {
+ let mut cursor = codec.cursor(key)?;
+ let mut order = Ordering::Equal;
+ for (i, value) in self.values.iter().enumerate() {
+ order = codec.compare_component(i, cursor.next_component()?, value.as_deref())?;
+ if order != Ordering::Equal {
+ break;
+ }
+ }
+ cursor.finish()?;
+ if order != Ordering::Equal {
+ return Ok(order);
+ }
+ Ok(if self.after {
+ Ordering::Less
+ } else {
+ Ordering::Greater
+ })
+ }
+
+ fn compare(&self, other: &Self, codec: &CompositeKeyCodec) -> Result<Ordering> {
+ for i in 0..self.values.len().min(other.values.len()) {
+ let order = codec.compare_component(
+ i,
+ self.values[i].as_deref(),
+ other.values[i].as_deref(),
+ )?;
+ if order != Ordering::Equal {
+ return Ok(order);
+ }
+ }
+ Ok(match self.values.len().cmp(&other.values.len()) {
+ Ordering::Less => {
+ if self.after {
+ Ordering::Greater
+ } else {
+ Ordering::Less
+ }
+ }
+ Ordering::Greater => {
+ if other.after {
+ Ordering::Less
+ } else {
+ Ordering::Greater
+ }
+ }
+ Ordering::Equal => self.after.cmp(&other.after),
+ })
+ }
+}
+
+pub(crate) struct CompositeInterval {
+ pub(crate) lower: CompositeBound,
+ pub(crate) upper: CompositeBound,
+ pub(crate) point_key: Option<Vec<u8>>,
+}
+
+pub(crate) struct CompositePlan {
+ pub(crate) fields: Vec<DataField>,
+ pub(crate) codec: CompositeKeyCodec,
+ pub(crate) intervals: Vec<CompositeInterval>,
+ pub(crate) bound_columns: usize,
+ pub(crate) equal_columns: usize,
+}
+
+impl CompositePlan {
+ pub(crate) fn is_point_lookup(&self) -> bool {
+ self.equal_columns == self.fields.len()
+ }
+
+ pub(crate) fn plan(fields: &[DataField], predicate: &Predicate) -> Result<Option<Self>> {
+ let codec = CompositeKeyCodec::new(fields);
+ let leaves = predicate
+ .clone()
+ .split_and()
+ .into_iter()
+ .filter_map(as_leaf)
+ .collect::<Vec<_>>();
+ let mut prefixes: Vec<Vec<Option<Datum>>> = vec![Vec::new()];
+ let mut equal_columns = 0;
+ let mut range = Vec::new();
+ for (i, field) in fields.iter().enumerate() {
+ let mut column = leaves
+ .iter()
+ .filter(|leaf| {
+ let Predicate::Leaf {
+ column,
+ data_type,
+ op,
+ ..
+ } = leaf
+ else {
+ unreachable!()
+ };
+ column == field.name()
+ && data_type.equals_ignore_nullable(field.data_type())
+ && supported(*op)
+ })
+ .cloned()
+ .collect::<Vec<_>>();
+ for leaf in &mut column {
+ let Predicate::Leaf { literals, .. } = leaf else {
+ unreachable!()
+ };
+ for literal in literals {
+ if !normalize_key_literal(literal, field.data_type()) {
+ return Ok(None);
+ }
+ }
+ }
+ let selected = column
+ .iter()
+ .filter(|leaf| point(operator(leaf)))
+ .min_by_key(|leaf| match leaf {
+ Predicate::Leaf { literals, op, .. } => {
+ if *op == PredicateOperator::IsNull {
+ 0
+ } else {
+ literals.len()
+ }
+ }
+ _ => unreachable!(),
+ });
+ let Some(selected) = selected else {
+ range = column;
+ break;
+ };
+ let Predicate::Leaf { op, literals, .. } = selected else {
+ unreachable!()
+ };
+ let candidates = if *op == PredicateOperator::IsNull {
+ if field.data_type().is_nullable() {
+ vec![None]
+ } else {
+ Vec::new()
+ }
+ } else {
+ literals.iter().cloned().map(Some).collect()
+ };
+ let mut domain: Vec<Option<Datum>> = Vec::new();
+ for candidate in candidates {
+ let encoded = candidate.as_ref().map(|v| codec.serialize_component(i, v));
+ let mut matches = true;
+ for leaf in &column {
+ if !matches_leaf(&codec, i, encoded.as_deref(), leaf)? {
+ matches = false;
+ break;
+ }
+ }
+ if matches {
+ let mut duplicate = false;
+ for value in &domain {
+ let value = value.as_ref().map(|v| codec.serialize_component(i, v));
+ if codec.compare_component(i, encoded.as_deref(), value.as_deref())?
+ == Ordering::Equal
+ {
+ duplicate = true;
+ break;
+ }
+ }
+ if !duplicate {
+ domain.push(candidate);
+ }
+ if domain.len() > MAX_INTERVALS {
+ break;
+ }
+ }
+ }
+ equal_columns += 1;
+ if domain.is_empty() {
+ return Ok(Some(Self {
+ fields: fields.to_vec(),
+ codec,
+ intervals: Vec::new(),
+ bound_columns: equal_columns,
+ equal_columns,
+ }));
+ }
+ if prefixes.len() * domain.len() > MAX_INTERVALS {
+ return Ok(None);
+ }
+ let mut expanded = Vec::new();
+ for prefix in prefixes {
+ for value in &domain {
+ let mut next = prefix.clone();
+ next.push(value.clone());
+ expanded.push(next);
+ }
+ }
+ prefixes = expanded;
+ }
+ let bound_columns = equal_columns + usize::from(!range.is_empty());
+ if bound_columns == 0 {
+ return Ok(None);
+ }
+ let mut intervals = Vec::new();
+ for prefix in prefixes {
+ let values = prefix
+ .iter()
+ .enumerate()
+ .map(|(i, v)| v.as_ref().map(|v| codec.serialize_component(i, v)))
+ .collect::<Vec<_>>();
+ let mut lower = CompositeBound::new(values.clone(), false);
+ let mut upper = CompositeBound::new(values.clone(), true);
+ if !range.is_empty() {
+ let mut null_prefix = values.clone();
+ null_prefix.push(None);
+ lower = CompositeBound::new(null_prefix, true);
+ for leaf in &range {
+ let Predicate::Leaf { op, literals, .. } = leaf else {
+ unreachable!()
+ };
+ if *op == PredicateOperator::IsNotNull {
+ continue;
+ }
+ for (value, after, is_lower) in match op {
+ PredicateOperator::Gt | PredicateOperator::GtEq => {
+ vec![(&literals[0], *op == PredicateOperator::Gt, true)]
+ }
+ PredicateOperator::Lt | PredicateOperator::LtEq => {
+ vec![(&literals[0], *op == PredicateOperator::LtEq, false)]
+ }
+ PredicateOperator::Between => {
+ vec![(&literals[0], false, true), (&literals[1], true, false)]
+ }
+ _ => unreachable!(),
+ } {
+ let mut next = values.clone();
+ next.push(Some(codec.serialize_component(equal_columns, value)));
+ let next = CompositeBound::new(next, after);
+ if is_lower && next.compare(&lower, &codec)? == Ordering::Greater {
+ lower = next;
+ } else if !is_lower && next.compare(&upper, &codec)? == Ordering::Less {
+ upper = next;
+ }
+ }
+ }
+ }
+ if lower.compare(&upper, &codec)? == Ordering::Less {
+ let point_key = if equal_columns == fields.len() {
+ Some(codec.serialize(&prefix)?)
+ } else {
+ None
+ };
+ intervals.push(CompositeInterval {
+ lower,
+ upper,
+ point_key,
+ });
+ }
+ }
+ Ok(Some(Self {
+ fields: fields.to_vec(),
+ codec,
+ intervals,
+ bound_columns,
+ equal_columns,
+ }))
+ }
+
+ pub(crate) fn may_match(&self, first: Option<&[u8]>, last: Option<&[u8]>) -> Result<bool> {
+ if self.intervals.is_empty() {
+ return Ok(false);
+ }
+ let (Some(first), Some(last)) = (first, last) else {
+ return Ok(true);
+ };
+ for interval in &self.intervals {
+ if interval.lower.compare_key(last, &self.codec)? == Ordering::Greater
+ && interval.upper.compare_key(first, &self.codec)? == Ordering::Less
+ {
+ return Ok(true);
+ }
+ }
+ Ok(false)
+ }
+}
+
+fn operator(leaf: &Predicate) -> PredicateOperator {
+ let Predicate::Leaf { op, .. } = leaf else {
+ unreachable!()
+ };
+ *op
+}
+
+fn point(op: PredicateOperator) -> bool {
+ matches!(
+ op,
+ PredicateOperator::Eq | PredicateOperator::In | PredicateOperator::IsNull
+ )
+}
+fn supported(op: PredicateOperator) -> bool {
+ point(op)
+ || matches!(
+ op,
+ PredicateOperator::Gt
+ | PredicateOperator::GtEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Between
+ | PredicateOperator::IsNotNull
+ )
+}
+
+fn as_leaf(predicate: Predicate) -> Option<Predicate> {
+ match predicate {
+ Predicate::Leaf {
+ op,
+ ref literals,
+ ref data_type,
+ ..
+ } if supported(op) => {
+ let valid = match op {
+ PredicateOperator::Eq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq => literals.len() == 1,
+ PredicateOperator::Between => literals.len() == 2,
+ _ => true,
+ };
+ (valid
+ && literals.iter().all(|literal| {
+ crate::spec::validate_datum_matches_type(literal, data_type).is_ok()
+ }))
+ .then_some(predicate)
+ }
+ Predicate::Or(children) => {
+ let mut first: Option<Predicate> = None;
+ let mut values = Vec::new();
+ for child in children {
+ let leaf = as_leaf(child)?;
+ let Predicate::Leaf {
+ ref column,
+ ref data_type,
+ ref literals,
+ op,
+ ..
+ } = leaf
+ else {
+ return None;
+ };
+ if !matches!(op, PredicateOperator::Eq | PredicateOperator::In) {
+ return None;
+ }
+ if let Some(Predicate::Leaf {
+ column: name,
+ data_type: ty,
+ ..
+ }) = &first
+ {
+ if name != column || ty != data_type {
+ return None;
+ }
+ }
+ values.extend(literals.iter().cloned());
+ first.get_or_insert(leaf);
+ }
+ let Predicate::Leaf {
+ column,
+ index,
+ data_type,
+ ..
+ } = first?
+ else {
+ return None;
+ };
+ Some(Predicate::Leaf {
+ column,
+ index,
+ data_type,
+ op: PredicateOperator::In,
+ literals: values,
+ })
+ }
+ _ => None,
+ }
+}
+
+fn matches_leaf(
+ codec: &CompositeKeyCodec,
+ i: usize,
+ value: Option<&[u8]>,
+ leaf: &Predicate,
+) -> Result<bool> {
+ let Predicate::Leaf { op, literals, .. } = leaf else {
+ unreachable!()
+ };
+ if *op == PredicateOperator::IsNull {
+ return Ok(value.is_none());
+ }
+ if value.is_none() {
+ return Ok(false);
+ }
+ if *op == PredicateOperator::IsNotNull {
+ return Ok(true);
+ }
+ let compare = |literal: &Datum| {
+ codec.compare_component(i, value, Some(&codec.serialize_component(i, literal)))
+ };
+ Ok(match op {
+ PredicateOperator::Eq => compare(&literals[0])? == Ordering::Equal,
+ PredicateOperator::In => {
+ let mut found = false;
+ for literal in literals {
+ found |= compare(literal)? == Ordering::Equal;
+ }
+ found
+ }
+ PredicateOperator::Gt => compare(&literals[0])? == Ordering::Greater,
+ PredicateOperator::GtEq => compare(&literals[0])? != Ordering::Less,
+ PredicateOperator::Lt => compare(&literals[0])? == Ordering::Less,
+ PredicateOperator::LtEq => compare(&literals[0])? != Ordering::Greater,
+ PredicateOperator::Between => {
+ compare(&literals[0])? != Ordering::Less && compare(&literals[1])? != Ordering::Greater
+ }
+ _ => unreachable!(),
+ })
+}
diff --git a/crates/paimon/src/btree/composite_tests.rs b/crates/paimon/src/btree/composite_tests.rs
new file mode 100644
index 00000000..ffc090af
--- /dev/null
+++ b/crates/paimon/src/btree/composite_tests.rs
@@ -0,0 +1,664 @@
+// 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.
+
+use super::test_util::{BytesFileRead, VecFileWrite};
+use super::*;
+use crate::spec::{
+ BinaryRow, DataField, DataType, Datum, IntType, Predicate, PredicateBuilder, RowType,
+ VarCharType,
+};
+use bytes::Bytes;
+use std::cmp::Ordering;
+
+fn fields() -> Vec<DataField> {
+ vec![
+ DataField::new(
+ 10,
+ "category".into(),
+ DataType::VarChar(VarCharType::string_type()),
+ ),
+ DataField::new(20, "item".into(), DataType::Int(IntType::new())),
+ DataField::new(
+ 30,
+ "tag".into(),
+ DataType::VarChar(VarCharType::string_type()),
+ ),
+ ]
+}
+fn tuples() -> Vec<Vec<Option<Datum>>> {
+ let mut rows = Vec::new();
+ for category in [None, Some("a"), Some("b")] {
+ for item in [None, Some(-2), Some(0), Some(1), Some(2)] {
+ for tag in [None, Some(""), Some("z")] {
+ rows.push(vec![
+ category.map(|v| Datum::String(v.into())),
+ item.map(Datum::Int),
+ tag.map(|v| Datum::String(v.into())),
+ ]);
+ }
+ }
+ }
+ rows
+}
+
+#[test]
+fn composite_compacted_encoding_and_comparison() {
+ let fields = fields();
+ let codec = CompositeKeyCodec::new(&fields);
+ assert_eq!(codec.serialize(&[None, None, None]).unwrap(), vec![0, 7]);
+ assert_eq!(
+ codec
+ .serialize(&[Some(Datum::String("a".into())), Some(Datum::Int(-2)), None])
+ .unwrap(),
+ vec![0, 4, 1, b'a', 254, 255, 255, 255]
+ );
+ let encoded = tuples()
+ .iter()
+ .map(|tuple| codec.serialize(tuple).unwrap())
+ .collect::<Vec<_>>();
+ for pair in encoded.windows(2) {
+ assert_eq!(
+ codec.compare_keys(&pair[0], &pair[1]).unwrap(),
+ Ordering::Less
+ );
+ }
+ for malformed in [
+ vec![],
+ vec![0],
+ vec![0, 0, 128],
+ vec![0, 0, 127],
+ vec![0, 7, 0],
+ ] {
+ assert!(codec.validate_key(&malformed).is_err());
+ }
+}
+
+#[test]
+fn composite_query_declines_decimal_overflow_and_timestamp_precision_loss() {
+ use crate::spec::{DecimalType, LocalZonedTimestampType, TimestampType};
+
+ let fields = vec![
+ DataField::new(
+ 0,
+ "d".into(),
+ DataType::Decimal(DecimalType::new(18, 2).unwrap()),
+ ),
+ DataField::new(1, "item".into(), DataType::Int(IntType::new())),
+ ];
+ let b = PredicateBuilder::new(&fields);
+ for (unscaled, scale) in [
+ (i128::MAX, 0), // Exact rescaling would overflow i128.
+ (i128::from(i64::MAX) + 1, 2), // Compact encoding would wrap the integer.
+ (1, 3), // Dropping the final digit would change the bound.
+ ] {
+ let predicate = b
+ .less_than(
+ "d",
+ Datum::Decimal {
+ unscaled,
+ precision: 38,
+ scale,
+ },
+ )
+ .unwrap();
+ assert!(CompositePlan::plan(&fields, &predicate).unwrap().is_none());
+ }
+ // A zero with an arbitrary literal scale is exactly representable.
+ let predicate = b
+ .equal(
+ "d",
+ Datum::Decimal {
+ unscaled: 0,
+ precision: 38,
+ scale: 38,
+ },
+ )
+ .unwrap();
+ let plan = CompositePlan::plan(&fields, &predicate).unwrap().unwrap();
+ let codec = CompositeKeyCodec::new(&fields);
+ let zero = codec
+ .serialize(&[
+ Some(Datum::Decimal {
+ unscaled: 0,
+ precision: 18,
+ scale: 2,
+ }),
+ Some(Datum::Int(1)),
+ ])
+ .unwrap();
+ assert!(plan.may_match(Some(&zero), Some(&zero)).unwrap());
+ for (ty, literal) in [
+ (
+ DataType::Timestamp(TimestampType::new(3).unwrap()),
+ Datum::Timestamp {
+ millis: 1000,
+ nanos: 1,
+ },
+ ),
+ (
+ DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(3).unwrap()),
+ Datum::LocalZonedTimestamp {
+ millis: 1000,
+ nanos: 1,
+ },
+ ),
+ ] {
+ let fields = vec![
+ DataField::new(0, "ts".into(), ty),
+ DataField::new(1, "item".into(), DataType::Int(IntType::new())),
+ ];
+ let predicate = PredicateBuilder::new(&fields)
+ .less_than("ts", literal)
+ .unwrap();
+ assert!(CompositePlan::plan(&fields, &predicate).unwrap().is_none());
+ }
+}
+
+#[test]
+fn composite_slice_ordering_handles_offsets_and_typed_fields() {
+ use crate::spec::VarBinaryType;
+
+ let fields = vec![
+ DataField::new(0, "number".into(), DataType::Int(IntType::new())),
+ DataField::new(
+ 1,
+ "text".into(),
+ DataType::VarChar(VarCharType::string_type()),
+ ),
+ DataField::new(
+ 2,
+ "bytes".into(),
+ DataType::VarBinary(VarBinaryType::new(130).unwrap()),
+ ),
+ ];
+ // Tuple order differs from raw byte order for little-endian integers,
+ // length-prefixed strings, and the NULL header. Binary bytes are unsigned.
+ let ordered = [
+ (None, None, None),
+ (Some(1), None, Some(vec![0xff])),
+ (Some(1), Some(String::new()), None),
+ (Some(1), Some("a".repeat(128)), Some(vec![0xff])),
+ (Some(1), Some("b".into()), None),
+ (Some(1), Some("b".into()), Some(vec![1, 2, 3])),
+ (Some(1), Some("b".into()), Some(vec![1, 2, 4])),
+ (Some(1), Some("b".into()), Some(vec![0xff])),
+ (Some(256), None, None),
+ ];
+ let mut storage = vec![0xee; 17];
+ let mut ranges = Vec::new();
+ for (number, text, bytes) in ordered {
+ let key = serialize_composite_key(
+ &[
+ number.map(Datum::Int),
+ text.map(Datum::String),
+ bytes.map(Datum::Bytes),
+ ],
+ &fields,
+ )
+ .unwrap();
+ let start = storage.len();
+ storage.extend(key);
+ ranges.push(start..storage.len());
+ storage.extend([0xff; 3]);
+ }
+ let compare = make_key_comparator(&DataType::Row(RowType::new(fields)));
+ for (i, left) in ranges.iter().enumerate() {
+ for (j, right) in ranges.iter().enumerate() {
+ assert_eq!(
+ compare(&storage[left.clone()], &storage[right.clone()]).unwrap(),
+ i.cmp(&j),
+ "tuple {i} versus {j}"
+ );
+ }
+ }
+}
+
+#[test]
+fn composite_slice_comparison_rejects_invalid_suffixes() {
+ let fields = fields();
+ let codec = CompositeKeyCodec::new(&fields);
+ let key = |category: &str| {
+ codec
+ .serialize(&[
+ Some(Datum::String(category.into())),
+ Some(Datum::Int(1)),
+ Some(Datum::String("z".into())),
+ ])
+ .unwrap()
+ };
+ let valid = key("a");
+ let mut truncated = key("b");
+ truncated.pop();
+ let mut trailing = key("b");
+ trailing.push(0);
+ let predicate = PredicateBuilder::new(&fields)
+ .equal("category", Datum::String("a".into()))
+ .unwrap();
+ let plan = CompositePlan::plan(&fields, &predicate).unwrap().unwrap();
+ for malformed in [truncated, trailing] {
+ // An earlier column decides the order, but the incompatible suffix must
+ // still decline the index rather than allowing metadata to prune rows.
+ assert!(codec.compare_keys(&valid, &malformed).is_err());
+ assert!(codec.compare_keys(&malformed, &valid).is_err());
+ assert!(plan.intervals[0]
+ .lower
+ .compare_key(&malformed, &plan.codec)
+ .is_err());
+ assert!(plan.intervals[0]
+ .upper
+ .compare_key(&malformed, &plan.codec)
+ .is_err());
+ }
+}
+
+#[test]
+fn composite_nan_keys_are_canonical_and_null_header_spans_bytes() {
+ use crate::spec::{DoubleType, FloatType};
+ let fields = vec![
+ DataField::new(0, "f".into(), DataType::Float(FloatType::new())),
+ DataField::new(1, "d".into(), DataType::Double(DoubleType::new())),
+ ];
+ let codec = CompositeKeyCodec::new(&fields);
+ let key = codec
+ .serialize(&[Some(Datum::Float(f32::NAN)), Some(Datum::Double(f64::NAN))])
+ .unwrap();
+ let alternate = codec
+ .serialize(&[
+ Some(Datum::Float(f32::from_bits(0xffc00001))),
+ Some(Datum::Double(f64::from_bits(0xfff8000000000001))),
+ ])
+ .unwrap();
+ assert_eq!(key, alternate);
+ assert_eq!(
+ codec.compare_keys(&key, &alternate).unwrap(),
+ Ordering::Equal
+ );
+ let inf = codec
+ .serialize(&[
+ Some(Datum::Float(f32::INFINITY)),
+ Some(Datum::Double(f64::INFINITY)),
+ ])
+ .unwrap();
+ assert_eq!(codec.compare_keys(&inf, &key).unwrap(), Ordering::Less);
+ let fields = (0..9)
+ .map(|i| DataField::new(i, format!("k{i}"), DataType::Int(IntType::new())))
+ .collect::<Vec<_>>();
+ assert_eq!(
+ serialize_composite_key(&[const { None }; 9], &fields).unwrap(),
+ vec![0, 255, 1]
+ );
+}
+
+#[test]
+fn composite_interval_expansion_and_residual_columns() {
+ let fields = fields();
+ let b = PredicateBuilder::new(&fields);
+ let duplicate = b
+ .is_in("category", vec![Datum::String("a".into()); 1000])
+ .unwrap();
+ assert_eq!(
+ CompositePlan::plan(&fields, &duplicate)
+ .unwrap()
+ .unwrap()
+ .intervals
+ .len(),
+ 1
+ );
+ let too_many = b.is_in("item", (0..257).map(Datum::Int).collect()).unwrap();
+ let predicate = Predicate::and(vec![
+ b.equal("category", Datum::String("a".into())).unwrap(),
+ too_many,
+ ]);
+ assert!(CompositePlan::plan(&fields, &predicate).unwrap().is_none());
+ let predicate = Predicate::and(vec![
+ b.greater_than("category", Datum::String("a".into()))
+ .unwrap(),
+ b.equal("item", Datum::Int(1)).unwrap(),
+ ]);
+ let plan = CompositePlan::plan(&fields, &predicate).unwrap().unwrap();
+ assert_eq!(plan.bound_columns, 1);
+ assert_eq!(plan.equal_columns, 0);
+ assert!(
+ CompositePlan::plan(&fields, &b.equal("item", Datum::Int(1)).unwrap())
+ .unwrap()
+ .is_none()
+ );
+ let contradiction = Predicate::and(vec![
+ b.equal("category", Datum::String("a".into())).unwrap(),
+ b.equal("category", Datum::String("b".into())).unwrap(),
+ ]);
+ assert!(CompositePlan::plan(&fields, &contradiction)
+ .unwrap()
+ .unwrap()
+ .intervals
+ .is_empty());
+}
+
+fn predicates(fields: &[DataField]) -> Vec<Predicate> {
+ let b = PredicateBuilder::new(fields);
+ let mut predicates = vec![
+ b.equal("category", Datum::String("a".into())).unwrap(),
+ b.is_null("category").unwrap(),
+ b.is_not_null("category").unwrap(),
+ ];
+ for prefix in [
+ b.equal("category", Datum::String("a".into())).unwrap(),
+ b.is_null("category").unwrap(),
+ b.is_in(
+ "category",
+ vec![
+ Datum::String("a".into()),
+ Datum::String("b".into()),
+ Datum::String("a".into()),
+ ],
+ )
+ .unwrap(),
+ ] {
+ for range in [
+ b.greater_than("item", Datum::Int(0)).unwrap(),
+ b.greater_or_equal("item", Datum::Int(0)).unwrap(),
+ b.less_than("item", Datum::Int(1)).unwrap(),
+ b.less_or_equal("item", Datum::Int(1)).unwrap(),
+ b.between("item", Datum::Int(-2), Datum::Int(1)).unwrap(),
+ b.is_null("item").unwrap(),
+ b.is_not_null("item").unwrap(),
+ b.is_in("item", vec![Datum::Int(-2), Datum::Int(2)])
+ .unwrap(),
+ Predicate::and(vec![
+ b.greater_than("item", Datum::Int(1)).unwrap(),
+ b.less_than("item", Datum::Int(1)).unwrap(),
+ ]),
+ ] {
+ predicates.push(Predicate::and(vec![prefix.clone(), range]));
+ }
+ }
+ predicates.push(Predicate::and(vec![
+ b.is_null("category").unwrap(),
+ b.is_null("item").unwrap(),
+ b.is_null("tag").unwrap(),
+ ]));
+ predicates.push(Predicate::and(vec![
+ b.equal("tag", Datum::String("z".into())).unwrap(),
+ b.equal("item", Datum::Int(1)).unwrap(),
+ b.equal("category", Datum::String("a".into())).unwrap(),
+ ]));
+ predicates
+}
+
+/// Exercise range endpoints at every key position, including NULL prefixes,
+/// nested disjunctions, duplicate IN values, and contradictory bounds. The
+/// row predicate evaluator below is independent of the tuple interval planner.
+fn predicate_matrix(fields: &[DataField]) -> Vec<Predicate> {
+ let b = PredicateBuilder::new(fields);
+ let prefixes = vec![
+ b.is_null("category").unwrap(),
+ b.equal("category", Datum::String("a".into())).unwrap(),
+ b.equal("category", Datum::String("b".into())).unwrap(),
+ b.is_in(
+ "category",
+ vec![
+ Datum::String("b".into()),
+ Datum::String("a".into()),
+ Datum::String("b".into()),
+ ],
+ )
+ .unwrap(),
+ Predicate::Or(vec![
+ b.equal("category", Datum::String("a".into())).unwrap(),
+ Predicate::Or(vec![
+ b.equal("category", Datum::String("b".into())).unwrap(),
+ b.equal("category", Datum::String("a".into())).unwrap(),
+ ]),
+ ]),
+ ];
+ let mut items = vec![b.is_null("item").unwrap(), b.is_not_null("item").unwrap()];
+ for value in [-3, -2, -1, 0, 1, 2, 3, 256] {
+ items.extend([
+ b.equal("item", Datum::Int(value)).unwrap(),
+ b.greater_than("item", Datum::Int(value)).unwrap(),
+ b.greater_or_equal("item", Datum::Int(value)).unwrap(),
+ b.less_than("item", Datum::Int(value)).unwrap(),
+ b.less_or_equal("item", Datum::Int(value)).unwrap(),
+ ]);
+ }
+ items.push(
+ b.is_in("item", vec![Datum::Int(2), Datum::Int(-2), Datum::Int(2)])
+ .unwrap(),
+ );
+ for lower in [-3, -2, 0, 1, 2, 3] {
+ for upper in [-3, -2, 0, 1, 2, 3] {
+ // Use raw AND so contradictions reach the planner rather than a
+ // builder short-circuit to AlwaysFalse.
+ items.push(Predicate::And(vec![
+ b.greater_or_equal("item", Datum::Int(lower)).unwrap(),
+ b.less_than("item", Datum::Int(upper)).unwrap(),
+ ]));
+ }
+ }
+ let mut result = Vec::new();
+ for prefix in prefixes {
+ for item in &items {
+ result.push(Predicate::And(vec![prefix.clone(), item.clone()]));
+ }
+ for item in [
+ b.is_null("item").unwrap(),
+ b.equal("item", Datum::Int(0)).unwrap(),
+ b.is_in("item", vec![Datum::Int(2), Datum::Int(-2)])
+ .unwrap(),
+ ] {
+ for value in ["", "a", "z", "zz", "联合"] {
+ for tag in [
+ b.equal("tag", Datum::String(value.into())).unwrap(),
+ b.greater_than("tag", Datum::String(value.into())).unwrap(),
+ b.less_or_equal("tag", Datum::String(value.into())).unwrap(),
+ ] {
+ result.push(Predicate::And(vec![prefix.clone(), item.clone(), tag]));
+ }
+ }
+ for tag in [b.is_null("tag").unwrap(), b.is_not_null("tag").unwrap()] {
+ result.push(Predicate::And(vec![prefix.clone(), item.clone(), tag]));
+ }
+ }
+ }
+ result
+}
+
+#[tokio::test]
+async fn composite_java_files_and_rust_writer_match() {
+ // Larger LZ4 blocks must actually compress, rather than exercising only
+ // the uncompressed fallback of a writer configured for LZ4.
+ assert!(
+ include_bytes!("../../testdata/btree/btree_composite_v1_java_lz4.bin").len()
+ < include_bytes!("../../testdata/btree/btree_composite_v1_java_none.bin").len()
+ );
+ assert!(
+ include_bytes!("../../testdata/btree/btree_composite_v2_java_lz4.bin").len()
+ < include_bytes!("../../testdata/btree/btree_composite_v2_java_none.bin").len()
+ );
+ let fields = fields();
+ let codec = CompositeKeyCodec::new(&fields);
+ let key_type = DataType::Row(RowType::new(fields.clone()));
+ for (version, compression, fixture, metadata) in [
+ (
+ 1,
+ BlockCompressionType::None,
+ include_bytes!("../../testdata/btree/btree_composite_v1_java_none.bin").as_slice(),
+ include_bytes!("../../testdata/btree/btree_composite_v1_java_none.meta").as_slice(),
+ ),
+ (
+ 1,
+ BlockCompressionType::Lz4,
+ include_bytes!("../../testdata/btree/btree_composite_v1_java_lz4.bin").as_slice(),
+ include_bytes!("../../testdata/btree/btree_composite_v1_java_lz4.meta").as_slice(),
+ ),
+ (
+ 2,
+ BlockCompressionType::None,
+ include_bytes!("../../testdata/btree/btree_composite_v2_java_none.bin").as_slice(),
+ include_bytes!("../../testdata/btree/btree_composite_v2_java_none.meta").as_slice(),
+ ),
+ (
+ 2,
+ BlockCompressionType::Lz4,
+ include_bytes!("../../testdata/btree/btree_composite_v2_java_lz4.bin").as_slice(),
+ include_bytes!("../../testdata/btree/btree_composite_v2_java_lz4.meta").as_slice(),
+ ),
+ ] {
+ let meta = BTreeIndexMeta::deserialize(metadata).unwrap();
+ let reader = BTreeIndexReader::open(
+ Box::new(BytesFileRead(Bytes::copy_from_slice(fixture))),
+ fixture.len() as u64,
+ &meta,
+ make_key_comparator(&key_type),
+ )
+ .await
+ .unwrap();
+ let tuples = tuples();
+ for predicate in predicates(&fields)
+ .into_iter()
+ .chain(predicate_matrix(&fields))
+ {
+ let plan = CompositePlan::plan(&fields, &predicate).unwrap().unwrap();
+ let mut expected = roaring::RoaringTreemap::new();
+ for (id, tuple) in tuples.iter().enumerate() {
+ let row = BinaryRow::from_datums(
+ &tuple
+ .iter()
+ .zip(&fields)
+ .map(|(v, field)| (v.as_ref(), field.data_type()))
+ .collect::<Vec<_>>(),
+ );
+ if crate::spec::eval_row(&predicate, &row).unwrap() {
+ expected.insert(id as u64);
+ expected.insert(id as u64 + 1000);
+ }
+ }
+ assert_eq!(
+ reader.query_composite(&plan).await.unwrap(),
+ expected,
+ "{predicate:?}, version={version}"
+ );
+ }
+ assert!(reader.null_bitmap().await.unwrap().is_empty());
+ let buf = VecFileWrite::new();
+ let mut writer = BTreeIndexWriter::with_comparator_and_options(
+ Box::new(buf.clone()),
+ if compression == BlockCompressionType::None {
+ 64
+ } else {
+ 512
+ },
+ compression,
+ 1,
+ false,
+ make_key_comparator(&key_type),
+ )
+ .with_file_version(version)
+ .unwrap();
+ for (id, tuple) in tuples.iter().enumerate() {
+ let key = codec.serialize(tuple).unwrap();
+ writer.write(Some(&key), id as i64).await.unwrap();
+ writer.write(Some(&key), id as i64 + 1000).await.unwrap();
+ }
+ let result = writer.finish().await.unwrap();
+ assert_eq!(result.meta.serialize(), metadata);
+ if compression == BlockCompressionType::None {
+ assert_eq!(buf.to_vec(), fixture);
+ }
+ }
+}
+
+#[test]
+fn composite_java_scalar_types_match_compacted_encoding() {
+ use crate::spec::{
+ BigIntType, BooleanType, DecimalType, DoubleType, FloatType, SmallIntType, TimestampType,
+ TinyIntType,
+ };
+ let types = vec![
+ DataType::Boolean(BooleanType::new()),
+ DataType::TinyInt(TinyIntType::new()),
+ DataType::SmallInt(SmallIntType::new()),
+ DataType::Int(IntType::new()),
+ DataType::BigInt(BigIntType::new()),
+ DataType::Float(FloatType::new()),
+ DataType::Double(DoubleType::new()),
+ DataType::Decimal(DecimalType::new(20, 2).unwrap()),
+ DataType::Timestamp(TimestampType::new(6).unwrap()),
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::Int(IntType::new()),
+ ];
+ let fields = types
+ .into_iter()
+ .enumerate()
+ .map(|(i, ty)| DataField::new(i as i32, format!("k{i}"), ty))
+ .collect::<Vec<_>>();
+ let codec = CompositeKeyCodec::new(&fields);
+ let values = vec![
+ Some(Datum::Bool(true)),
+ Some(Datum::TinyInt(-3)),
+ Some(Datum::SmallInt(-257)),
+ Some(Datum::Int(-12345)),
+ Some(Datum::Long(1234567890123)),
+ Some(Datum::Float(f32::from_bits(0xffc00001))),
+ Some(Datum::Double(f64::from_bits(0xfff8000000000001))),
+ Some(Datum::Decimal {
+ unscaled: -129,
+ precision: 20,
+ scale: 2,
+ }),
+ Some(Datum::Timestamp {
+ millis: -123000,
+ nanos: 999999,
+ }),
+ Some(Datum::String("联合".into())),
+ None,
+ ];
+ let key = codec.serialize(&values).unwrap();
+ assert_eq!(
+ key,
+ include_bytes!("../../testdata/btree/btree_composite_java_scalar.key")
+ );
+ let mut larger = values.clone();
+ larger[7] = Some(Datum::Decimal {
+ unscaled: 128,
+ precision: 20,
+ scale: 2,
+ });
+ assert_eq!(
+ codec
+ .compare_keys(&key, &codec.serialize(&larger).unwrap())
+ .unwrap(),
+ Ordering::Less
+ );
+ larger = values.clone();
+ larger[8] = Some(Datum::Timestamp {
+ millis: -123000,
+ nanos: 0,
+ });
+ assert_eq!(
+ codec
+ .compare_keys(&key, &codec.serialize(&larger).unwrap())
+ .unwrap(),
+ Ordering::Greater
+ );
+ let mut wrong_type = values;
+ wrong_type[3] = Some(Datum::Float(0.0));
+ assert!(
+ codec.serialize(&wrong_type).is_err(),
+ "same-width mismatched types must be rejected"
+ );
+}
diff --git a/crates/paimon/src/btree/key_serde.rs b/crates/paimon/src/btree/key_serde.rs
index 3ef651dd..ddc12e87 100644
--- a/crates/paimon/src/btree/key_serde.rs
+++ b/crates/paimon/src/btree/key_serde.rs
@@ -28,6 +28,54 @@ const TIMESTAMP_COMPACT_PRECISION: u32 = 3;
/// Decimal precision <= 18 is compact (fits in i64).
const DECIMAL_COMPACT_PRECISION: u32 = 18;
+/// Slice comparison uses the column's physical layout. Preserve the literal's
+/// value when converting to that layout, or decline the index so the ordinary
+/// data predicate can compare the original value without losing precision.
+pub(crate) fn normalize_key_literal(literal: &mut Datum, data_type: &DataType) -> bool {
+ match (literal, data_type) {
+ (
+ Datum::Decimal {
+ unscaled,
+ precision,
+ scale,
+ },
+ DataType::Decimal(ty),
+ ) => {
+ if *unscaled != 0 && *scale != ty.scale() {
+ let Some(factor) = 10i128.checked_pow(scale.abs_diff(ty.scale())) else {
+ return false;
+ };
+ if *scale < ty.scale() {
+ let Some(value) = unscaled.checked_mul(factor) else {
+ return false;
+ };
+ *unscaled = value;
+ } else {
+ if *unscaled % factor != 0 {
+ return false;
+ }
+ *unscaled /= factor;
+ }
+ }
+ if ty.precision() <= DECIMAL_COMPACT_PRECISION && i64::try_from(*unscaled).is_err() {
+ return false;
+ }
+ *precision = ty.precision();
+ *scale = ty.scale();
+ true
+ }
+ (Datum::Timestamp { nanos, .. }, DataType::Timestamp(ty)) => {
+ (0..1_000_000).contains(nanos)
+ && (ty.precision() > TIMESTAMP_COMPACT_PRECISION || *nanos == 0)
+ }
+ (Datum::LocalZonedTimestamp { nanos, .. }, DataType::LocalZonedTimestamp(ty)) => {
+ (0..1_000_000).contains(nanos)
+ && (ty.precision() > TIMESTAMP_COMPACT_PRECISION || *nanos == 0)
+ }
+ _ => true,
+ }
+}
+
/// Key comparator type alias.
///
/// Fallible because the bytes come from an index file that may have been written when
@@ -122,6 +170,10 @@ pub(crate) fn is_key_comparison_failure(error: &std::io::Error) -> bool {
/// For variable-length types (string, bytes), uses lexicographic byte comparison.
pub fn make_key_comparator(data_type: &DataType) -> KeyComparator {
match data_type {
+ DataType::Row(row) => {
+ let codec = super::CompositeKeyCodec::new(row.fields());
+ Box::new(move |a, b| codec.compare_keys(a, b))
+ }
DataType::TinyInt(_) => Box::new(|a: &[u8], b: &[u8]| {
let av = i8::from_le_bytes(fixed_key_bytes::<1>(a, "TINYINT")?);
let bv = i8::from_le_bytes(fixed_key_bytes::<1>(b, "TINYINT")?);
diff --git a/crates/paimon/src/btree/mod.rs b/crates/paimon/src/btree/mod.rs
index 9baa8662..64488333 100644
--- a/crates/paimon/src/btree/mod.rs
+++ b/crates/paimon/src/btree/mod.rs
@@ -38,6 +38,8 @@
mod block;
mod bloom_filter;
+mod composite_key;
+mod composite_predicate;
mod data_block_cache;
mod footer;
pub(crate) mod key_serde;
@@ -53,6 +55,9 @@ pub use block::BlockCompressionType;
pub(crate) use block::{
compress_block, compress_codec_block, compute_crc32, decompress_block, decompress_codec_block,
};
+pub use composite_key::serialize_composite_key;
+pub(crate) use composite_key::CompositeKeyCodec;
+pub(crate) use composite_predicate::CompositePlan;
pub(crate) use data_block_cache::BTreeDataBlockCache;
pub use footer::BTreeFileFooter;
pub use key_serde::{make_key_comparator, serialize_datum};
@@ -65,3 +70,6 @@ pub use writer::BTreeIndexWriter;
pub(crate) mod test_util;
#[cfg(test)]
mod tests;
+
+#[cfg(test)]
+mod composite_tests;
diff --git a/crates/paimon/src/btree/reader.rs b/crates/paimon/src/btree/reader.rs
index 579190c1..44fe15b6 100644
--- a/crates/paimon/src/btree/reader.rs
+++ b/crates/paimon/src/btree/reader.rs
@@ -220,6 +220,58 @@ impl<F: Fn(&[u8], &[u8]) -> crate::Result<Ordering>> BTreeIndexReader<F> {
Ok(result)
}
+ /// Seek to each virtual tuple boundary; NULLs in individual components are
+ /// ordinary tuple keys and never use the scalar NULL bitmap.
+ pub(crate) async fn query_composite(
+ &self,
+ plan: &super::CompositePlan,
+ ) -> io::Result<RoaringTreemap> {
+ let mut result = RoaringTreemap::new();
+ for interval in &plan.intervals {
+ if let Some(key) = &interval.point_key {
+ result |= self.query_equal(key).await?;
+ continue;
+ }
+ let cmp = |key: &[u8], _: &[u8]| {
+ interval
+ .lower
+ .compare_key(key, &plan.codec)
+ .map_err(key_comparison_io_error)
+ };
+ let (_, mut index_iter) = self.sst_reader.index_block().seek_and_iter(&[], &cmp)?;
+ let mut first = true;
+ while let Some((_, handle_bytes)) = index_iter.next() {
+ let handle = BlockHandle::decode(handle_bytes)?;
+ let block = self.read_data_block(&handle).await?;
+ let mut offset = if first {
+ first = false;
+ block.seek_and_iter(&[], &cmp)?.1.offset
+ } else {
+ 0
+ };
+ let mut done = false;
+ while offset < block.data.len() {
+ let (key, value, next) = block.read_entry_at(offset);
+ offset = next;
+ if interval
+ .upper
+ .compare_key(key, &plan.codec)
+ .map_err(key_comparison_io_error)?
+ != Ordering::Less
+ {
+ done = true;
+ break;
+ }
+ posting_list::add_to(value, self.file_version, &mut result)?;
+ }
+ if done {
+ break;
+ }
+ }
+ }
+ 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/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index 5fe8faa1..611579e8 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -114,6 +114,7 @@ pub(crate) use predicate::extract_datum;
pub(crate) use predicate::like_match;
#[cfg(test)]
pub(crate) use predicate::row_id_leaf;
+pub(crate) use predicate::validate_datum_matches_type;
pub use predicate::{
field_idx_to_partition_idx, Datum, Predicate, PredicateBuilder, PredicateOperator, Transform,
TransformInput,
diff --git a/crates/paimon/src/spec/predicate.rs b/crates/paimon/src/spec/predicate.rs
index dd437fdf..2abe34eb 100644
--- a/crates/paimon/src/spec/predicate.rs
+++ b/crates/paimon/src/spec/predicate.rs
@@ -1541,7 +1541,7 @@ impl PredicateBuilder {
///
/// This is a fail-fast check at predicate construction time, preventing
/// type mismatches from propagating to evaluators.
-fn validate_datum_matches_type(datum: &Datum, data_type: &DataType) -> Result<()> {
+pub(crate) fn validate_datum_matches_type(datum: &Datum, data_type: &DataType) -> Result<()> {
let ok = matches!(
(datum, data_type),
(Datum::Bool(_), DataType::Boolean(_))
diff --git a/crates/paimon/src/table/global_index_build_common.rs b/crates/paimon/src/table/global_index_build_common.rs
index 3b255a25..ecc23878 100644
--- a/crates/paimon/src/table/global_index_build_common.rs
+++ b/crates/paimon/src/table/global_index_build_common.rs
@@ -24,12 +24,78 @@
//! `index_type` string, so it lives here once rather than being copied into
//! each builder.
-use crate::spec::{FileKind, IndexManifest};
+use crate::spec::{DataField, FileKind, IndexManifest};
use crate::table::{merge_row_ranges, RowRange, Table};
use crate::{Error, Result};
pub(crate) mod vector;
+pub(crate) enum IndexColumns {
+ Column(String),
+ Columns(Vec<String>),
+}
+
+/// Resolve an ordered definition, shared by build and drop. The leading field
+/// and ordered extra IDs together identify a composite definition.
+pub(crate) fn resolve_index_fields<'a>(
+ table: &'a Table,
+ columns: &IndexColumns,
+ index_type: &str,
+) -> Result<Vec<&'a DataField>> {
+ let columns = match columns {
+ IndexColumns::Column(column) => {
+ if table
+ .schema()
+ .fields()
+ .iter()
+ .any(|field| field.name() == column)
+ {
+ vec![column.as_str()]
+ } else {
+ column.split(',').map(str::trim).collect()
+ }
+ }
+ IndexColumns::Columns(columns) => columns.iter().map(String::as_str).collect(),
+ };
+ if columns.is_empty() {
+ return Err(Error::DataInvalid {
+ message: "Global index column list is empty".into(),
+ source: None,
+ });
+ }
+ let mut fields = Vec::new();
+ for column in columns {
+ if column.is_empty() {
+ return Err(Error::DataInvalid {
+ message: "Global index column list contains an empty column".into(),
+ source: None,
+ });
+ }
+ let field = table
+ .schema()
+ .fields()
+ .iter()
+ .find(|field| field.name() == column)
+ .ok_or_else(|| Error::ColumnNotExist {
+ full_name: table.identifier().full_name(),
+ column: column.to_string(),
+ })?;
+ if fields.contains(&field) {
+ return Err(Error::DataInvalid {
+ message: format!("Duplicate global index column '{column}'"),
+ source: None,
+ });
+ }
+ fields.push(field);
+ }
+ if fields.len() > 1 && index_type != super::global_index_types::BTREE_GLOBAL_INDEX_TYPE {
+ return Err(Error::Unsupported {
+ message: "Only BTree global indexes support multiple index columns".into(),
+ });
+ }
+ Ok(fields)
+}
+
/// Java `sameExtraFieldIds`: null/empty are equal; otherwise exact ordered equality.
pub(crate) fn same_extra_field_ids(a: Option<&[i32]>, b: Option<&[i32]>) -> bool {
let a = a.unwrap_or(&[]);
diff --git a/crates/paimon/src/table/global_index_drop_builder.rs b/crates/paimon/src/table/global_index_drop_builder.rs
index 170e7097..bbee8b1d 100644
--- a/crates/paimon/src/table/global_index_drop_builder.rs
+++ b/crates/paimon/src/table/global_index_drop_builder.rs
@@ -15,18 +15,19 @@
// specific language governing permissions and limitations
// under the License.
+use super::global_index_build_common::IndexColumns;
use super::global_index_types::{
normalize_global_index_type_for_drop, BTREE_GLOBAL_INDEX_TYPE,
SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP,
};
-use crate::spec::{DataField, FileKind, IndexFileMeta, IndexManifest};
+use crate::spec::{FileKind, IndexFileMeta, IndexManifest};
use crate::table::{CommitMessage, SnapshotManager, Table, TableCommit};
use crate::{Error, Result};
use std::collections::HashMap;
pub struct GlobalIndexDropBuilder<'a> {
table: &'a Table,
- index_column: Option<String>,
+ index_column: Option<IndexColumns>,
index_type: String,
}
@@ -40,7 +41,15 @@ impl<'a> GlobalIndexDropBuilder<'a> {
}
pub fn with_index_column(&mut self, column: &str) -> &mut Self {
- self.index_column = Some(column.to_string());
+ self.index_column = Some(IndexColumns::Column(column.to_string()));
+ self
+ }
+
+ /// Drop exactly the definition with these ordered BTree columns.
+ pub fn with_index_columns(&mut self, columns: &[&str]) -> &mut Self {
+ self.index_column = Some(IndexColumns::Columns(
+ columns.iter().map(|column| (*column).to_string()).collect(),
+ ));
self
}
@@ -66,12 +75,21 @@ impl<'a> GlobalIndexDropBuilder<'a> {
})?;
let index_column = self
.index_column
- .as_deref()
+ .as_ref()
.ok_or_else(|| Error::DataInvalid {
message: "Global index column is required".to_string(),
source: None,
})?;
- let index_field = find_index_field(self.table, index_column)?;
+ let index_fields = super::global_index_build_common::resolve_index_fields(
+ self.table,
+ index_column,
+ index_type,
+ )?;
+ let extra_field_ids = index_fields
+ .iter()
+ .skip(1)
+ .map(|field| field.id())
+ .collect::<Vec<_>>();
let snapshot_manager = SnapshotManager::new(
self.table.file_io().clone(),
@@ -104,7 +122,12 @@ impl<'a> GlobalIndexDropBuilder<'a> {
let Some(global_meta) = entry.index_file.global_index_meta.as_ref() else {
continue;
};
- if global_meta.index_field_id != index_field.id() {
+ if global_meta.index_field_id != index_fields[0].id()
+ || !super::global_index_build_common::same_extra_field_ids(
+ global_meta.extra_field_ids.as_deref(),
+ Some(&extra_field_ids),
+ )
+ {
continue;
}
dropped += 1;
@@ -147,18 +170,6 @@ impl<'a> GlobalIndexDropBuilder<'a> {
}
}
-fn find_index_field<'a>(table: &'a Table, column: &str) -> Result<&'a DataField> {
- table
- .schema()
- .fields()
- .iter()
- .find(|field| field.name() == column)
- .ok_or_else(|| Error::ColumnNotExist {
- full_name: table.identifier().full_name(),
- column: column.to_string(),
- })
-}
-
#[cfg(test)]
mod tests {
use super::*;
diff --git a/crates/paimon/src/table/global_index_scanner.rs b/crates/paimon/src/table/global_index_scanner.rs
index 09f838c3..008b323a 100644
--- a/crates/paimon/src/table/global_index_scanner.rs
+++ b/crates/paimon/src/table/global_index_scanner.rs
@@ -21,6 +21,7 @@
//! Reference: [org.apache.paimon.index.GlobalIndexScanner](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexScanner.java)
mod all_match;
+mod composite;
mod deletion_vectors;
mod entry;
mod evaluator;
@@ -124,6 +125,9 @@ pub(crate) struct GlobalIndexScanner {
fm_read_context: Arc<FMReadContext>,
/// Global index entries grouped by field_id.
entries_by_field: Vec<(i32, Vec<GlobalIndexEntry>)>,
+ /// Composite BTree files grouped by the full ordered definition. These
+ /// must never be passed to a scalar reader for their leading column.
+ composite_entries: Vec<(Vec<i32>, Vec<GlobalIndexEntry>)>,
/// Indexed row-id coverage grouped by field_id.
#[cfg(test)]
coverage_by_field: HashMap<i32, Vec<RowRange>>,
@@ -192,6 +196,7 @@ impl GlobalIndexScanner {
}
let mut entries_by_field: std::collections::HashMap<i32, Vec<GlobalIndexEntry>> =
std::collections::HashMap::new();
+ let mut composite_entries: HashMap<Vec<i32>, Vec<GlobalIndexEntry>> = HashMap::new();
#[cfg(test)]
let mut coverage_by_field: HashMap<i32, Vec<RowRange>> = HashMap::new();
@@ -305,15 +310,26 @@ impl GlobalIndexScanner {
}
}
- entries_by_field
- .entry(global_meta.index_field_id)
- .or_default()
- .push(resolved);
+ if kind == GlobalIndexFileKind::BTree
+ && global_meta
+ .extra_field_ids
+ .as_ref()
+ .is_some_and(|ids| !ids.is_empty())
+ {
+ let mut ids = vec![global_meta.index_field_id];
+ ids.extend(global_meta.extra_field_ids.as_ref().unwrap());
+ composite_entries.entry(ids).or_default().push(resolved);
+ } else {
+ entries_by_field
+ .entry(global_meta.index_field_id)
+ .or_default()
+ .push(resolved);
+ }
}
validate_fm_file_sets(&entries_by_field)?;
- if entries_by_field.is_empty() {
+ if entries_by_field.is_empty() && composite_entries.is_empty() {
return Ok(None);
}
@@ -328,6 +344,7 @@ impl GlobalIndexScanner {
fm_read_options,
fm_read_context: Arc::new(FMReadContext::new(fm_read_options.cache_size)),
entries_by_field: entries_by_field.into_iter().collect(),
+ composite_entries: composite_entries.into_iter().collect(),
#[cfg(test)]
coverage_by_field,
schema_fields: schema_fields.to_vec(),
@@ -386,7 +403,6 @@ pub(crate) async fn evaluate_global_index(
Some(s) => s,
None => return Ok(None),
};
-
let combined = Predicate::and(evaluation.predicates.to_vec());
let scan_result = match scanner.evaluate(&combined).await? {
diff --git a/crates/paimon/src/table/global_index_scanner/composite.rs b/crates/paimon/src/table/global_index_scanner/composite.rs
new file mode 100644
index 00000000..f61d0e4f
--- /dev/null
+++ b/crates/paimon/src/table/global_index_scanner/composite.rs
@@ -0,0 +1,227 @@
+// 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.
+
+//! Composite definition selection, pruning, and bounded shard queries.
+
+use super::entry::{sorted_entry_meta, GlobalIndexEntry, GlobalIndexFileKind};
+use super::evaluator::{try_fold_bounded, GlobalIndexScanResult};
+use super::reader::OpenedGlobalIndexReader;
+use super::row_ranges::{bitmap_to_ranges, intersect_sorted_ranges};
+use super::GlobalIndexScanner;
+use crate::btree::key_serde::is_key_comparison_failure;
+use crate::btree::CompositePlan;
+use crate::spec::{DataType, Predicate, RowType};
+use crate::table::{merge_row_ranges, RowRange};
+use crate::{Error, Result};
+use std::cmp::Reverse;
+use std::collections::{HashMap, HashSet};
+
+fn coverage(entries: &[GlobalIndexEntry]) -> Vec<RowRange> {
+ merge_row_ranges(
+ entries
+ .iter()
+ .map(|entry| RowRange::new(entry.row_range_start, entry.row_range_end))
+ .collect(),
+ )
+}
+
+pub(super) struct CompositeIndexScanResult {
+ pub(super) result: GlobalIndexScanResult,
+ pub(super) key_field_ids: HashSet<i32>,
+}
+
+impl GlobalIndexScanner {
+ pub(super) async fn evaluate_composite(
+ &self,
+ predicate: &Predicate,
+ ) -> Result<Option<CompositeIndexScanResult>> {
+ let mut candidates = Vec::new();
+ for (ids, entries) in &self.composite_entries {
+ let fields = ids
+ .iter()
+ .map(|id| {
+ self.schema_fields
+ .iter()
+ .find(|field| field.id() == *id)
+ .cloned()
+ })
+ .collect::<Option<Vec<_>>>();
+ let Some(fields) = fields else {
+ continue;
+ };
+ if ids.iter().collect::<HashSet<_>>().len() != ids.len() {
+ continue;
+ }
+ // An unsupported type or literal layout declines this index.
+ let Ok(Some(plan)) = CompositePlan::plan(&fields, predicate) else {
+ continue;
+ };
+ let indexed_coverage = coverage(entries);
+ if plan.bound_columns == 1 && self.single_column_covers(ids[0], &indexed_coverage) {
+ continue;
+ }
+ let selected = entries
+ .iter()
+ .filter_map(|entry| {
+ let meta = sorted_entry_meta(entry);
+ match plan.may_match(meta.first_key.as_deref(), meta.last_key.as_deref()) {
+ Ok(true) => Some(Ok(entry)),
+ Ok(false) => None,
+ Err(error) => Some(Err(error)),
+ }
+ })
+ .collect::<Result<Vec<_>>>();
+ let Ok(selected) = selected else {
+ continue;
+ };
+ if !plan.is_point_lookup() && !selected.is_empty() {
+ let mut sizes = HashMap::new();
+ let valid = selected.iter().all(|entry| {
+ let size = sizes
+ .entry((entry.row_range_start, entry.row_range_end))
+ .or_insert(0i64);
+ if entry.file_size < 0 {
+ return false;
+ }
+ if let Some(total) = size.checked_add(entry.file_size) {
+ *size = total;
+ true
+ } else {
+ false
+ }
+ });
+ if self.btree_fallback_scan_max_size <= 0
+ || !valid
+ || sizes
+ .values()
+ .any(|size| *size > self.btree_fallback_scan_max_size)
+ {
+ continue;
+ }
+ }
+ candidates.push((plan, selected, indexed_coverage));
+ }
+ // Stable tie breaking avoids HashMap iteration deciding which coverage
+ // domain fast mode uses when equivalent definitions coexist.
+ candidates.sort_by_key(|(plan, _, _)| {
+ (
+ Reverse(plan.bound_columns),
+ Reverse(plan.equal_columns),
+ plan.fields.len(),
+ plan.fields
+ .iter()
+ .map(|field| field.id())
+ .collect::<Vec<_>>(),
+ )
+ });
+ for (plan, selected, indexed_coverage) in candidates {
+ let key_type = DataType::Row(RowType::new(plan.fields.clone()));
+ let mut queries = Vec::with_capacity(selected.len());
+ for entry in selected {
+ let plan = &plan;
+ let key_type = &key_type;
+ queries.push(async move {
+ let _guard = self.btree_file_lock(entry).lock_owned().await;
+ let _permit = self.query_semaphore.acquire().await.map_err(|error| {
+ Error::UnexpectedError {
+ message: "Global-index query budget was closed".into(),
+ source: Some(Box::new(error)),
+ }
+ })?;
+ let OpenedGlobalIndexReader::BTree(reader) = self
+ .get_or_open_reader(entry, sorted_entry_meta(entry), key_type)
+ .await?
+ else {
+ unreachable!()
+ };
+ let queried = reader.query_composite(plan).await;
+ self.return_reader(entry.resolved_path(&self.table_path), reader);
+ match queried {
+ Ok(bitmap) => Ok(Some(
+ bitmap_to_ranges(&bitmap)
+ .into_iter()
+ .map(|range| {
+ RowRange::new(
+ range.from() + entry.row_range_start,
+ range.to() + entry.row_range_start,
+ )
+ })
+ .collect::<Vec<_>>(),
+ )),
+ Err(error) if is_key_comparison_failure(&error) => Ok(None),
+ Err(error) => Err(Self::query_error(entry, error)),
+ }
+ });
+ }
+ let (ranges, declined) = try_fold_bounded(
+ queries,
+ self.global_index_thread_num,
+ (Vec::new(), false),
+ |(ranges, declined), result| {
+ if let Some(result) = result {
+ ranges.extend(result);
+ } else {
+ *declined = true;
+ }
+ },
+ )
+ .await?;
+ if declined {
+ continue;
+ }
+ return Ok(Some(CompositeIndexScanResult {
+ key_field_ids: plan.fields.iter().map(|field| field.id()).collect(),
+ result: GlobalIndexScanResult {
+ row_ranges: merge_row_ranges(ranges),
+ // Remaining predicates are always retained by the table reader.
+ evaluated_field_ids: plan
+ .fields
+ .iter()
+ .take(plan.bound_columns)
+ .map(|field| field.id())
+ .collect(),
+ indexed_coverage,
+ },
+ }));
+ }
+ Ok(None)
+ }
+
+ fn single_column_covers(&self, id: i32, composite_coverage: &[RowRange]) -> bool {
+ let Some((_, entries)) = self.entries_by_field.iter().find(|(field, _)| *field == id)
+ else {
+ return false;
+ };
+ let entries = entries
+ .iter()
+ .filter(|entry| {
+ matches!(
+ entry.index_type,
+ GlobalIndexFileKind::BTree | GlobalIndexFileKind::Bitmap
+ )
+ })
+ .collect::<Vec<_>>();
+ let single_coverage = merge_row_ranges(
+ entries
+ .iter()
+ .map(|entry| RowRange::new(entry.row_range_start, entry.row_range_end))
+ .collect(),
+ );
+ !entries.is_empty()
+ && intersect_sorted_ranges(&single_coverage, composite_coverage) == composite_coverage
+ }
+}
diff --git a/crates/paimon/src/table/global_index_scanner/evaluator.rs b/crates/paimon/src/table/global_index_scanner/evaluator.rs
index 6af6d379..dcaf3c26 100644
--- a/crates/paimon/src/table/global_index_scanner/evaluator.rs
+++ b/crates/paimon/src/table/global_index_scanner/evaluator.rs
@@ -33,6 +33,7 @@ use super::query_plan::{
use super::row_ranges::unindexed_ranges_for_coverage;
use super::row_ranges::{bitmap_to_ranges, intersect_sorted_ranges};
use super::GlobalIndexScanner;
+use crate::btree::key_serde::normalize_key_literal;
use crate::btree::query::extract_between;
use crate::btree::{make_key_comparator, serialize_datum};
#[cfg(test)]
@@ -44,6 +45,7 @@ use crate::table::bitmap_global_index_format::{
use crate::table::RowRange;
use crate::{Error, Result};
use futures::{StreamExt, TryStreamExt};
+use std::borrow::Cow;
use std::collections::HashSet;
use std::future::Future;
@@ -79,6 +81,31 @@ where
}
impl GlobalIndexScanner {
+ fn predicate_within_fields(
+ &self,
+ predicate: &Predicate,
+ fields: &HashSet<i32>,
+ ) -> Result<bool> {
+ match predicate {
+ Predicate::Leaf { column, .. } => Ok(self
+ .find_field_id_by_name(column)?
+ .is_some_and(|id| fields.contains(&id))),
+ Predicate::And(children) | Predicate::Or(children) => {
+ if children.is_empty() {
+ return Ok(false);
+ }
+ for child in children {
+ if !self.predicate_within_fields(child, fields)? {
+ return Ok(false);
+ }
+ }
+ Ok(true)
+ }
+ Predicate::Not(inner) => self.predicate_within_fields(inner, fields),
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => Ok(false),
+ }
+ }
+
/// Evaluate a predicate against the global indexes and return matching row ranges.
/// Returns `None` if the predicate cannot be evaluated by the global index.
pub(super) fn evaluate<'a>(&'a self, predicate: &'a Predicate) -> EvaluateFuture<'a> {
@@ -89,6 +116,11 @@ impl GlobalIndexScanner {
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}
Box::pin(async move {
+ if !matches!(predicate, Predicate::And(_)) {
+ if let Some(composite) = self.evaluate_composite(predicate).await? {
+ return Ok(Some(composite.result));
+ }
+ }
match predicate {
Predicate::Leaf {
column,
@@ -124,13 +156,51 @@ impl GlobalIndexScanner {
})
})
}
- Predicate::And(children) => {
+ Predicate::And(_) => {
+ let mut children = predicate.clone().split_and();
+ let mut row_ranges: Option<Vec<RowRange>> = None;
+ let mut indexed_coverage: Option<Vec<RowRange>> = None;
+ let mut evaluated_field_ids = HashSet::new();
+ loop {
+ let remaining_predicate = Predicate::And(children.clone());
+ let Some(composite) = self.evaluate_composite(&remaining_predicate).await?
+ else {
+ break;
+ };
+ let result = composite.result;
+ row_ranges = Some(match row_ranges {
+ None => result.row_ranges,
+ Some(existing) => {
+ intersect_sorted_ranges(&existing, &result.row_ranges)
+ }
+ });
+ indexed_coverage = Some(match indexed_coverage {
+ None => result.indexed_coverage,
+ Some(existing) => {
+ intersect_sorted_ranges(&existing, &result.indexed_coverage)
+ }
+ });
+ evaluated_field_ids.extend(result.evaluated_field_ids);
+ // Java keeps suffix constraints on this tuple as data
+ // filters, then continues planning independent indexes.
+ let previous_len = children.len();
+ let mut remaining = Vec::new();
+ for child in children {
+ if !self.predicate_within_fields(&child, &composite.key_field_ids)? {
+ remaining.push(child);
+ }
+ }
+ children = remaining;
+ if children.len() == previous_len || children.is_empty() {
+ break;
+ }
+ }
// Group leaf predicates by field_id to reuse readers
let mut leaf_groups: std::collections::HashMap<i32, Vec<PredicateTuple<'_>>> =
std::collections::HashMap::new();
let mut non_leaf_children = Vec::new();
- for child in children {
+ for child in &children {
if let Predicate::Leaf {
column,
op,
@@ -193,7 +263,7 @@ impl GlobalIndexScanner {
try_fold_bounded(
leaf_futures,
leaf_group_count.max(1),
- (None::<Vec<RowRange>>, None::<Vec<RowRange>>, HashSet::new()),
+ (row_ranges, indexed_coverage, evaluated_field_ids),
|(row_ranges, indexed_coverage, evaluated_field_ids),
(field_id, result)| {
if let Some((ranges, coverage)) = result {
@@ -289,7 +359,7 @@ impl GlobalIndexScanner {
entries: &[&GlobalIndexEntry],
predicates: &[(PredicateOperator, &[Datum], &DataType)],
) -> Result<Option<(Vec<RowRange>, Vec<RowRange>)>> {
- let normalized_predicates = predicates
+ let Some(normalized_predicates) = predicates
.iter()
.map(|(op, literals, data_type)| {
let key_type = if is_multivalue_predicate(*op) {
@@ -305,10 +375,30 @@ impl GlobalIndexScanner {
} else {
*data_type
};
- Ok((*op, *literals, key_type))
+ let mut literals = Cow::Borrowed(*literals);
+ if matches!(
+ key_type,
+ DataType::Decimal(_)
+ | DataType::Timestamp(_)
+ | DataType::LocalZonedTimestamp(_)
+ ) {
+ for literal in literals.to_mut() {
+ if !normalize_key_literal(literal, key_type) {
+ return Ok(None);
+ }
+ }
+ }
+ Ok(Some((*op, literals, key_type)))
})
- .collect::<Result<Vec<_>>>()?;
- let predicates = normalized_predicates.as_slice();
+ .collect::<Result<Option<Vec<_>>>>()?
+ else {
+ return Ok(None);
+ };
+ let predicates = normalized_predicates
+ .iter()
+ .map(|(op, literals, data_type)| (*op, literals.as_ref(), *data_type))
+ .collect::<Vec<_>>();
+ let predicates = predicates.as_slice();
// Try to detect between pattern and split into (between, remaining)
let (between, remaining) = extract_between(predicates);
diff --git a/crates/paimon/src/table/global_index_scanner/reader.rs b/crates/paimon/src/table/global_index_scanner/reader.rs
index dec996b1..f5024de5 100644
--- a/crates/paimon/src/table/global_index_scanner/reader.rs
+++ b/crates/paimon/src/table/global_index_scanner/reader.rs
@@ -205,7 +205,7 @@ impl GlobalIndexScanner {
})
}
- fn query_error(entry: &GlobalIndexEntry, error: std::io::Error) -> Error {
+ pub(super) fn query_error(entry: &GlobalIndexEntry, error: std::io::Error) -> Error {
Error::DataInvalid {
message: format!(
"Global index query failed for {} file '{}'",
@@ -220,7 +220,7 @@ impl GlobalIndexScanner {
/// cache is keyed by the resolved path (not the bare file name) so two entries
/// that share a file name but resolve to different locations — e.g. distinct
/// external paths — never reuse each other's reader.
- async fn get_or_open_reader(
+ pub(super) async fn get_or_open_reader(
&self,
entry: &GlobalIndexEntry,
meta: &BTreeIndexMeta,
@@ -421,7 +421,7 @@ impl GlobalIndexScanner {
}
/// Return a reader to the cache for future reuse.
- fn return_reader(&self, resolved_path: String, reader: BTreeIndexReader<BoxedCmp>) {
+ pub(super) fn return_reader(&self, resolved_path: String, reader: BTreeIndexReader<BoxedCmp>) {
let mut cache = self.reader_cache.lock().unwrap();
cache.insert(resolved_path, reader);
}
diff --git a/crates/paimon/src/table/global_index_scanner/tests.rs b/crates/paimon/src/table/global_index_scanner/tests.rs
index 2870bb1b..10f26742 100644
--- a/crates/paimon/src/table/global_index_scanner/tests.rs
+++ b/crates/paimon/src/table/global_index_scanner/tests.rs
@@ -1440,7 +1440,7 @@ fn test_unindexed_ranges_counts_extra_field_coverage() {
}
#[tokio::test]
-async fn test_evaluate_extra_field_only_without_composite_reader_falls_back() {
+async fn test_evaluate_composite_extra_field_only_without_left_prefix_falls_back() {
let (file_io, table_path, file_name, _tmp) =
setup_testdata_table("btree_int_100_no_compress.bin");
let meta = BTreeIndexMeta::new(Some(le_int_key(0)), Some(le_int_key(198)), false);
@@ -1459,7 +1459,7 @@ async fn test_evaluate_extra_field_only_without_composite_reader_falls_back() {
.unwrap();
assert!(
result.is_none(),
- "extra-field-only predicates must fall back until composite-key btree reads are supported"
+ "extra-field-only predicates must fall back without a constrained left prefix"
);
}
@@ -1783,6 +1783,52 @@ async fn test_evaluate_java_bitmap_golden_index_eq_and_null() {
assert_eq!(null_result.unwrap(), vec![RowRange::new(104, 104)]);
}
+#[tokio::test]
+async fn test_composite_prefers_fully_covering_leading_bitmap() {
+ let (file_io, table_path, file_name, meta, _tmp) = setup_java_bitmap_testdata_table();
+ let mut fields = string_schema_fields();
+ fields.push(DataField::new(
+ 2,
+ "id".into(),
+ DataType::Int(crate::spec::IntType::new()),
+ ));
+ let key = |name: &str, id| {
+ crate::btree::serialize_composite_key(
+ &[Some(Datum::String(name.into())), Some(Datum::Int(id))],
+ &fields,
+ )
+ .unwrap()
+ };
+ let composite_meta = BTreeIndexMeta::new(Some(key("alpha", 0)), Some(key("office", 9)), false);
+ let mut composite =
+ make_global_index_entry("missing_composite.btree", 1, 100, 109, &composite_meta);
+ composite
+ .index_file
+ .global_index_meta
+ .as_mut()
+ .unwrap()
+ .extra_field_ids = Some(vec![2]);
+ let bitmap =
+ make_global_index_entry_with_type(BITMAP_GLOBAL_INDEX_TYPE, &file_name, 1, 100, 109, &meta);
+ let predicates = [Predicate::Leaf {
+ column: "name".into(),
+ index: 0,
+ data_type: fields[0].data_type().clone(),
+ op: PredicateOperator::Eq,
+ literals: vec![Datum::String("k2".into())],
+ }];
+ let result = evaluate_global_index_fast(
+ &file_io,
+ &table_path,
+ &[composite, bitmap],
+ &predicates,
+ &fields,
+ )
+ .await
+ .unwrap();
+ assert_eq!(result.unwrap(), vec![RowRange::new(105, 106)]);
+}
+
async fn assert_bitmap_int_fixture(file_name: &str) {
let data_type = DataType::Int(crate::spec::IntType::new());
let (file_io, table_path, file_name, meta, _tmp) = setup_bitmap_testdata_table(file_name);
@@ -2571,15 +2617,19 @@ async fn test_evaluate_global_index_and_uses_evaluated_field_coverage_for_raw_fa
let file_io = crate::io::FileIOBuilder::new("file").build().unwrap();
let meta = BTreeIndexMeta::new(Some(le_int_key(0)), Some(le_int_key(198)), false);
- let mut first = make_global_index_entry("index_part1.bin", 1, 0, 49, &meta);
- first
+ let first = make_global_index_entry("index_part1.bin", 1, 0, 49, &meta);
+ // A separate tuple definition has narrower coverage and incompatible
+ // metadata. It must be declined without reading its keys as scalar INTs,
+ // and must not alter the coverage of the evaluated scalar definition.
+ let mut composite = make_global_index_entry("missing_composite.bin", 1, 0, 49, &meta);
+ composite
.index_file
.global_index_meta
.as_mut()
.unwrap()
.extra_field_ids = Some(vec![2]);
let second = make_global_index_entry("index_part2.bin", 1, 50, 99, &meta);
- let entries = vec![first, second];
+ let entries = vec![first, second, composite];
let fields = two_field_schema_fields();
let predicates = vec![Predicate::and(vec![
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder.rs b/crates/paimon/src/table/sorted_global_index_build_builder.rs
index b83761f6..d8619424 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder.rs
@@ -28,6 +28,7 @@ use planning::SortedGlobalIndexShard;
use validation::*;
use super::bitmap_global_index_format::{make_bitmap_key_comparator, serialize_bitmap_datum};
+use super::global_index_build_common::IndexColumns;
use super::global_index_types::{
normalize_queryable_global_index_type, BITMAP_GLOBAL_INDEX_TYPE, BTREE_GLOBAL_INDEX_TYPE,
FM_GLOBAL_INDEX_TYPE,
@@ -64,7 +65,7 @@ fn make_index_key_codec(index_type: &str, data_type: &DataType) -> (KeyComparato
pub struct SortedGlobalIndexBuildBuilder<'a> {
table: &'a Table,
- index_column: Option<String>,
+ index_column: Option<IndexColumns>,
index_type: String,
options: HashMap<String, String>,
}
@@ -84,7 +85,15 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
}
pub fn with_index_column(&mut self, column: &str) -> &mut Self {
- self.index_column = Some(column.to_string());
+ self.index_column = Some(IndexColumns::Column(column.to_string()));
+ self
+ }
+
+ /// Select BTree columns in physical tuple-key order.
+ pub fn with_index_columns(&mut self, columns: &[&str]) -> &mut Self {
+ self.index_column = Some(IndexColumns::Columns(
+ columns.iter().map(|column| (*column).to_string()).collect(),
+ ));
self
}
@@ -114,7 +123,7 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
})?;
let index_column = self
.index_column
- .as_deref()
+ .as_ref()
.ok_or_else(|| Error::DataInvalid {
message: "Sorted global index column is required".to_string(),
source: None,
@@ -134,8 +143,21 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
)?)
};
- let index_field = find_index_field(self.table, index_column)?;
- index_key_type(index_type, index_field)?;
+ let index_fields = super::global_index_build_common::resolve_index_fields(
+ self.table,
+ index_column,
+ index_type,
+ )?;
+ for field in &index_fields {
+ index_key_type(index_type, field)?;
+ }
+ let index_field = index_fields[0];
+ let extra_field_ids = index_fields
+ .iter()
+ .skip(1)
+ .map(|field| field.id())
+ .collect::<Vec<_>>();
+ let extra_field_ids = (!extra_field_ids.is_empty()).then_some(extra_field_ids);
let snapshot_manager = SnapshotManager::new(
self.table.file_io().clone(),
@@ -161,7 +183,7 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
snapshot.index_manifest(),
index_type,
index_field.id(),
- None, // single-column build; no extra fields today
+ extra_field_ids.as_deref(),
)
.await?;
@@ -184,7 +206,7 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
snapshot.index_manifest(),
index_type,
index_field.id(),
- None,
+ extra_field_ids.as_deref(),
&shards
.iter()
.map(|shard| RowRange::new(shard.row_range_start, shard.row_range_end))
@@ -204,7 +226,7 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
let mut messages = Vec::with_capacity(shard_count);
for shard in shards {
let index_file = match self
- .build_index_file(&shard, index_field, index_column, &write_options)
+ .build_index_file(&shard, &index_fields, &write_options)
.await
{
Ok(index_file) => index_file,
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs b/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
index c136f796..69f0fdae 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
@@ -35,18 +35,32 @@ use std::cmp::Ordering;
pub(super) async fn extract_index_rows(
table: &Table,
shard: &SortedGlobalIndexShard,
- index_column: &str,
- index_field: &DataField,
+ index_fields: &[&DataField],
index_type: &str,
serialize_key: SerializeKeyFn,
) -> Result<Vec<SortedIndexKeyRow>> {
let splits = build_read_splits_for_shard(shard)?;
let mut read_builder = table.new_read_builder();
- read_builder.with_projection(&[index_column, ROW_ID_FIELD_NAME])?;
+ let mut projection = index_fields
+ .iter()
+ .map(|field| field.name())
+ .collect::<Vec<_>>();
+ projection.push(ROW_ID_FIELD_NAME);
+ read_builder.with_projection(&projection)?;
let read = read_builder.new_read()?;
let batches = read.to_arrow(&splits)?.try_collect::<Vec<_>>().await?;
let expected_row_count = checked_row_count(shard.row_range_start, shard.row_range_end)?;
+ let index_field = index_fields[0];
+ let index_column = index_field.name();
+ if index_fields.len() > 1 {
+ return extract_composite_index_rows_from_batches(
+ &batches,
+ index_fields,
+ shard.row_range_start,
+ expected_row_count,
+ );
+ }
if index_type == MULTIVALUE_GLOBAL_INDEX_TYPE {
let DataType::Array(array_type) = index_field.data_type() else {
unreachable!("multivalue field was validated before extraction")
@@ -71,6 +85,81 @@ pub(super) async fn extract_index_rows(
}
}
+fn extract_composite_index_rows_from_batches(
+ batches: &[RecordBatch],
+ fields: &[&DataField],
+ row_range_start: i64,
+ expected_row_count: i64,
+) -> Result<Vec<SortedIndexKeyRow>> {
+ let fields = fields
+ .iter()
+ .map(|field| (*field).clone())
+ .collect::<Vec<_>>();
+ let codec = crate::btree::CompositeKeyCodec::new(&fields);
+ let mut expected_row_id = row_range_start;
+ let mut rows = Vec::new();
+ for batch in batches {
+ let indices = fields
+ .iter()
+ .map(|field| batch.schema().index_of(field.name()))
+ .collect::<std::result::Result<Vec<_>, _>>()
+ .map_err(|error| Error::DataInvalid {
+ message: format!("Composite index column missing: {error}"),
+ source: None,
+ })?;
+ let row_id_index = batch
+ .schema()
+ .index_of(ROW_ID_FIELD_NAME)
+ .map_err(|error| Error::DataInvalid {
+ message: format!("_ROW_ID column missing: {error}"),
+ source: None,
+ })?;
+ let row_ids = batch
+ .column(row_id_index)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Composite index requires Int64 _ROW_ID".into(),
+ source: None,
+ })?;
+ for row in 0..batch.num_rows() {
+ if row_ids.is_null(row) || row_ids.value(row) != expected_row_id {
+ return Err(Error::DataInvalid {
+ message: format!("Composite index expected non-null _ROW_ID {expected_row_id}"),
+ source: None,
+ });
+ }
+ let values = fields
+ .iter()
+ .zip(&indices)
+ .map(|(field, index)| {
+ extract_datum_from_arrow(batch, row, *index, field.data_type())
+ })
+ .collect::<Result<Vec<_>>>()?;
+ rows.push((
+ Some(codec.serialize(&values)?),
+ expected_row_id - row_range_start,
+ ));
+ expected_row_id = expected_row_id
+ .checked_add(1)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Composite index row ID overflow".into(),
+ source: None,
+ })?;
+ }
+ }
+ if expected_row_id - row_range_start != expected_row_count {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Composite index expected {expected_row_count} rows, got {}",
+ rows.len()
+ ),
+ source: None,
+ });
+ }
+ Ok(rows)
+}
+
pub(super) fn build_read_splits_for_shard(
shard: &SortedGlobalIndexShard,
) -> Result<Vec<DataSplit>> {
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
index b19d81ff..61122928 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
@@ -2609,3 +2609,878 @@ async fn incremental_build_splits_gap_around_mid_coverage_indexed_hole() {
);
}
}
+
+async fn composite_entries(
+ table: &Table,
+) -> (crate::spec::Snapshot, Vec<crate::spec::IndexManifestEntry>) {
+ let manager = SnapshotManager::new(table.file_io().clone(), table.location().into());
+ let snapshot = manager.get_latest_snapshot().await.unwrap().unwrap();
+ let entries = IndexManifest::read(
+ table.file_io(),
+ &manager.manifest_path(snapshot.index_manifest().unwrap()),
+ )
+ .await
+ .unwrap();
+ (snapshot, entries)
+}
+
+async fn composite_ranges(
+ table: &Table,
+ predicates: &[Predicate],
+ budget: i64,
+ mode: GlobalIndexSearchMode,
+) -> Option<Vec<RowRange>> {
+ let (snapshot, entries) = composite_entries(table).await;
+ evaluate_global_index(GlobalIndexEvaluation {
+ file_io: table.file_io(),
+ table_path: table.location(),
+ index_entries: &entries,
+ predicates,
+ schema_fields: table.schema().fields(),
+ search_mode: mode,
+ global_index_thread_num: 2,
+ btree_fallback_scan_max_size: budget,
+ btree_data_block_cache_size: 1024 * 1024,
+ bitmap_fallback_scan_max_size: budget,
+ fm_read_options: crate::fm_index::FMReadOptions::default(),
+ next_row_id: snapshot.next_row_id(),
+ data_ranges: &[RowRange::new(0, snapshot.next_row_id().unwrap() - 1)],
+ })
+ .await
+ .unwrap()
+}
+
+#[tokio::test]
+async fn test_composite_btree_build_query_coverage_and_drop() {
+ for version in [1, 2] {
+ let mut options = table_options("2");
+ options.insert("btree-index.file-version".into(), version.to_string());
+ options.insert("btree-index.bloom-filter.enabled".into(), "true".into());
+ let table =
+ test_table_with_path(&format!("memory:/composite_btree_e2e_v{version}"), options);
+ setup_dirs(&table).await;
+ let mut writer = TableWrite::new(&table, "composite-test".into()).unwrap();
+ writer
+ .write_arrow_batch(&data_batch(
+ vec![1, 2, 3, 1, 4, 5],
+ vec!["a", "a", "b", "a", "b", "c"],
+ ))
+ .await
+ .unwrap();
+ TableCommit::new(table.clone(), "composite-test".into())
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ assert_eq!(
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_columns(&["name", "id"])
+ .execute()
+ .await
+ .unwrap(),
+ 3
+ );
+ assert_eq!(
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_column(" name , id ")
+ .execute()
+ .await
+ .unwrap(),
+ 0
+ );
+ let b = PredicateBuilder::new(table.schema().fields());
+ let point = Predicate::and(vec![
+ b.equal("id", Datum::Int(1)).unwrap(),
+ b.equal("name", Datum::String("a".into())).unwrap(),
+ ]);
+ assert_eq!(
+ composite_ranges(
+ &table,
+ std::slice::from_ref(&point),
+ 0,
+ GlobalIndexSearchMode::Fast
+ )
+ .await
+ .unwrap(),
+ vec![RowRange::new(0, 0), RowRange::new(3, 3)]
+ );
+ assert_eq!(scan_ids(&table, point.clone()).await, vec![1, 1]);
+ let prefix = b.equal("name", Datum::String("a".into())).unwrap();
+ assert!(composite_ranges(
+ &table,
+ std::slice::from_ref(&prefix),
+ 0,
+ GlobalIndexSearchMode::Fast
+ )
+ .await
+ .is_none());
+ let range = Predicate::and(vec![
+ prefix.clone(),
+ b.greater_than("id", Datum::Int(1)).unwrap(),
+ ]);
+ assert_eq!(
+ composite_ranges(
+ &table,
+ std::slice::from_ref(&range),
+ i64::MAX,
+ GlobalIndexSearchMode::Fast
+ )
+ .await
+ .unwrap(),
+ vec![RowRange::new(1, 1)]
+ );
+ assert_eq!(scan_ids(&table, range.clone()).await, vec![2]);
+ let suffix = b.equal("id", Datum::Int(1)).unwrap();
+ assert!(
+ composite_ranges(&table, &[suffix], i64::MAX, GlobalIndexSearchMode::Fast)
+ .await
+ .is_none()
+ );
+ let disjunction = Predicate::or(vec![
+ point.clone(),
+ Predicate::and(vec![
+ b.equal("name", Datum::String("b".into())).unwrap(),
+ b.equal("id", Datum::Int(4)).unwrap(),
+ ]),
+ ]);
+ assert_eq!(
+ composite_ranges(&table, &[disjunction], 0, GlobalIndexSearchMode::Fast)
+ .await
+ .unwrap(),
+ vec![RowRange::new(0, 0), RowRange::new(3, 4)]
+ );
+ let (_, entries) = composite_entries(&table).await;
+ let max_size = entries
+ .iter()
+ .map(|entry| entry.index_file.file_size)
+ .max()
+ .unwrap();
+ assert!(
+ composite_ranges(
+ &table,
+ std::slice::from_ref(&prefix),
+ max_size,
+ GlobalIndexSearchMode::Fast
+ )
+ .await
+ .is_some(),
+ "budget is per row-range group, not summed across all shards"
+ );
+ assert!(entries.iter().all(|entry| entry
+ .index_file
+ .global_index_meta
+ .as_ref()
+ .unwrap()
+ .extra_field_ids
+ == Some(vec![0])));
+ // Append outside the composite coverage. FULL and DETAIL must retain
+ // those rows for residual data filtering; FAST uses indexed coverage.
+ let mut writer = TableWrite::new(&table, "append-test".into()).unwrap();
+ writer
+ .write_arrow_batch(&data_batch(vec![6], vec!["a"]))
+ .await
+ .unwrap();
+ TableCommit::new(table.clone(), "append-test".into())
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ for mode in [GlobalIndexSearchMode::Full, GlobalIndexSearchMode::Detail] {
+ assert_eq!(
+ composite_ranges(&table, std::slice::from_ref(&point), 0, mode)
+ .await
+ .unwrap(),
+ vec![
+ RowRange::new(0, 0),
+ RowRange::new(3, 3),
+ RowRange::new(6, 6)
+ ]
+ );
+ }
+ assert_eq!(scan_ids(&table, point).await, vec![1, 1]);
+ assert_eq!(
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_columns(&["name", "id"])
+ .execute()
+ .await
+ .unwrap(),
+ 1
+ );
+ // Independent scalar and reversed tuple definitions coexist.
+ assert!(
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_column("name")
+ .execute()
+ .await
+ .unwrap()
+ > 0
+ );
+ assert!(
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_columns(&["id", "name"])
+ .execute()
+ .await
+ .unwrap()
+ > 0
+ );
+ let (_, entries) = composite_entries(&table).await;
+ let target_count = entries
+ .iter()
+ .filter(|entry| {
+ let meta = entry.index_file.global_index_meta.as_ref().unwrap();
+ meta.index_field_id == 1 && meta.extra_field_ids == Some(vec![0])
+ })
+ .count();
+ assert_eq!(
+ table
+ .new_global_index_drop_builder()
+ .with_index_columns(&["name", "id"])
+ .execute()
+ .await
+ .unwrap(),
+ target_count
+ );
+ let (_, retained) = composite_entries(&table).await;
+ assert!(retained.iter().any(|entry| {
+ let meta = entry.index_file.global_index_meta.as_ref().unwrap();
+ meta.index_field_id == 1 && meta.extra_field_ids.is_none()
+ }));
+ assert!(retained.iter().any(|entry| entry
+ .index_file
+ .global_index_meta
+ .as_ref()
+ .unwrap()
+ .extra_field_ids
+ == Some(vec![1])));
+ assert_eq!(
+ table
+ .new_global_index_drop_builder()
+ .with_index_columns(&["name", "id"])
+ .execute()
+ .await
+ .unwrap(),
+ 0
+ );
+ }
+}
+
+#[tokio::test]
+async fn test_composite_btree_column_validation() {
+ let table = test_table(table_options("10"));
+ for columns in ["", "name,", "name,name", "name,missing"] {
+ assert!(table
+ .new_btree_global_index_build_builder()
+ .with_index_column(columns)
+ .execute()
+ .await
+ .is_err());
+ assert!(table
+ .new_global_index_drop_builder()
+ .with_index_column(columns)
+ .execute()
+ .await
+ .is_err());
+ }
+ for index_type in ["bitmap", "multivalue", "fm"] {
+ assert!(matches!(
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_columns(&["name", "id"])
+ .with_index_type(index_type)
+ .execute()
+ .await
+ .unwrap_err(),
+ Error::Unsupported { .. }
+ ));
+ }
+}
+
+#[tokio::test]
+async fn test_composite_btree_all_null_tuple_and_residual_filter() {
+ let table = test_table_with_path("memory:/composite_null_tuple", table_options("10"));
+ setup_dirs(&table).await;
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, true),
+ ArrowField::new("name", ArrowDataType::Utf8, true),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(vec![None, Some(1), Some(2)])) as ArrayRef,
+ Arc::new(StringArray::from(vec![None, Some("a"), None])) as ArrayRef,
+ ],
+ )
+ .unwrap();
+ let mut writer = TableWrite::new(&table, "null-test".into()).unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ TableCommit::new(table.clone(), "null-test".into())
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_columns(&["name", "id"])
+ .execute()
+ .await
+ .unwrap();
+ let b = PredicateBuilder::new(table.schema().fields());
+ let all_null = Predicate::and(vec![b.is_null("name").unwrap(), b.is_null("id").unwrap()]);
+ assert_eq!(
+ composite_ranges(&table, &[all_null], 0, GlobalIndexSearchMode::Fast)
+ .await
+ .unwrap(),
+ vec![RowRange::new(0, 0)]
+ );
+ let null_range = Predicate::and(vec![
+ b.is_null("name").unwrap(),
+ b.greater_than("id", Datum::Int(0)).unwrap(),
+ ]);
+ assert_eq!(
+ composite_ranges(&table, &[null_range], i64::MAX, GlobalIndexSearchMode::Fast)
+ .await
+ .unwrap(),
+ vec![RowRange::new(2, 2)]
+ );
+ // A constraint after the first range is a data filter, so the candidate
+ // includes row 1, while the ordinary reader rejects it.
+ let residual = Predicate::and(vec![
+ b.is_not_null("name").unwrap(),
+ b.equal("id", Datum::Int(2)).unwrap(),
+ ]);
+ assert_eq!(
+ composite_ranges(
+ &table,
+ std::slice::from_ref(&residual),
+ i64::MAX,
+ GlobalIndexSearchMode::Fast
+ )
+ .await
+ .unwrap(),
+ vec![RowRange::new(1, 1)]
+ );
+ assert!(scan_ids(&table, residual).await.is_empty());
+}
+
+#[tokio::test]
+async fn test_composite_btree_literal_precision_preserves_scan_results() {
+ use crate::spec::{DecimalType, TimestampType};
+ use arrow_array::{Decimal128Array, TimestampMillisecondArray};
+
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("d", DataType::Decimal(DecimalType::new(10, 2).unwrap()))
+ .column("ts", DataType::Timestamp(TimestampType::new(3).unwrap()))
+ .options(table_options("10"))
+ .build()
+ .unwrap();
+ let table = Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "literal_precision"),
+ "memory:/composite_literal_precision".into(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ setup_dirs(&table).await;
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("d", ArrowDataType::Decimal128(10, 2), false),
+ ArrowField::new(
+ "ts",
+ ArrowDataType::Timestamp(arrow_schema::TimeUnit::Millisecond, None),
+ false,
+ ),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef,
+ Arc::new(
+ Decimal128Array::from(vec![100, 200, 50])
+ .with_precision_and_scale(10, 2)
+ .unwrap(),
+ ),
+ Arc::new(TimestampMillisecondArray::from(vec![1000, 2000, 3000])),
+ ],
+ )
+ .unwrap();
+ let mut writer = TableWrite::new(&table, "precision-test".into()).unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ TableCommit::new(table.clone(), "precision-test".into())
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ let b = PredicateBuilder::new(table.schema().fields());
+ let decimal = |unscaled, scale| Datum::Decimal {
+ unscaled,
+ precision: 10,
+ scale,
+ };
+ let queries = vec![
+ (
+ Predicate::and(vec![
+ b.equal("d", decimal(10, 1)).unwrap(),
+ b.equal("id", Datum::Int(1)).unwrap(),
+ ]),
+ vec![1],
+ ),
+ (b.less_than("d", decimal(10, 1)).unwrap(), vec![3]),
+ (b.less_than("d", decimal(1005, 3)).unwrap(), vec![1, 3]),
+ (
+ Predicate::and(vec![
+ b.equal("d", decimal(500, 3)).unwrap(),
+ b.equal("id", Datum::Int(3)).unwrap(),
+ ]),
+ vec![3],
+ ),
+ (
+ b.less_than(
+ "ts",
+ Datum::Timestamp {
+ millis: 1000,
+ nanos: 1,
+ },
+ )
+ .unwrap(),
+ vec![1],
+ ),
+ (
+ b.is_in("d", vec![decimal(1, 0), decimal(500, 3), decimal(10, 1)])
+ .unwrap(),
+ vec![1, 3],
+ ),
+ (
+ b.between("d", decimal(5, 1), decimal(1000, 3)).unwrap(),
+ vec![1, 3],
+ ),
+ ];
+ for (predicate, expected) in &queries {
+ assert_eq!(scan_ids(&table, predicate.clone()).await, *expected);
+ }
+ for columns in [["d", "id"], ["ts", "id"]] {
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_columns(&columns)
+ .execute()
+ .await
+ .unwrap();
+ }
+ for (predicate, expected) in &queries {
+ assert_eq!(
+ scan_ids(&table, predicate.clone()).await,
+ *expected,
+ "{predicate:?}"
+ );
+ }
+ // Exact rescaling retains point lookups even with a zero range-scan budget.
+ assert_eq!(
+ composite_ranges(
+ &table,
+ &[queries[0].0.clone()],
+ 0,
+ GlobalIndexSearchMode::Fast
+ )
+ .await
+ .unwrap(),
+ vec![RowRange::new(0, 0)]
+ );
+ for index in [2, 4] {
+ assert!(composite_ranges(
+ &table,
+ &[queries[index].0.clone()],
+ i64::MAX,
+ GlobalIndexSearchMode::Fast
+ )
+ .await
+ .is_none());
+ }
+ // A fully covering scalar index is preferred for a leading-only query.
+ // That path must preserve the same literal values as the tuple query.
+ for index_type in ["btree", "bitmap"] {
+ for column in ["d", "ts"] {
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_column(column)
+ .with_index_type(index_type)
+ .execute()
+ .await
+ .unwrap();
+ }
+ for (predicate, expected) in &queries {
+ assert_eq!(
+ scan_ids(&table, predicate.clone()).await,
+ *expected,
+ "{index_type}: {predicate:?}"
+ );
+ }
+ for column in ["d", "ts"] {
+ table
+ .new_global_index_drop_builder()
+ .with_index_column(column)
+ .with_index_type(index_type)
+ .execute()
+ .await
+ .unwrap();
+ }
+ }
+}
+
+#[tokio::test]
+async fn test_composite_btree_intersects_independent_indexes() {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::string_type()))
+ .column("flag", DataType::Int(IntType::new()))
+ .options(table_options("10"))
+ .build()
+ .unwrap();
+ let table = Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "independent_indexes"),
+ "memory:/composite_independent_indexes".into(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ setup_dirs(&table).await;
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("name", ArrowDataType::Utf8, false),
+ ArrowField::new("flag", ArrowDataType::Int32, false),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as ArrayRef,
+ Arc::new(StringArray::from(vec!["a", "a", "b", "a"])),
+ Arc::new(Int32Array::from(vec![1, 0, 1, 1])),
+ ],
+ )
+ .unwrap();
+ let mut writer = TableWrite::new(&table, "intersection-test".into()).unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ TableCommit::new(table.clone(), "intersection-test".into())
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_type("bitmap")
+ .with_index_column("flag")
+ .execute()
+ .await
+ .unwrap();
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_columns(&["name", "id"])
+ .execute()
+ .await
+ .unwrap();
+ let b = PredicateBuilder::new(table.schema().fields());
+ let predicate = Predicate::And(vec![
+ b.equal("name", Datum::String("a".into())).unwrap(),
+ Predicate::And(vec![
+ b.greater_than("id", Datum::Int(1)).unwrap(),
+ b.equal("flag", Datum::Int(1)).unwrap(),
+ ]),
+ ]);
+ for _ in 0..2 {
+ assert_eq!(
+ composite_ranges(
+ &table,
+ std::slice::from_ref(&predicate),
+ i64::MAX,
+ GlobalIndexSearchMode::Fast
+ )
+ .await
+ .unwrap(),
+ vec![RowRange::new(3, 3)]
+ );
+ assert_eq!(scan_ids(&table, predicate.clone()).await, vec![4]);
+ // Repeat with a second composite definition instead of the scalar bitmap.
+ table
+ .new_global_index_drop_builder()
+ .with_index_column("flag")
+ .with_index_type("bitmap")
+ .execute()
+ .await
+ .unwrap();
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_columns(&["flag", "id"])
+ .execute()
+ .await
+ .unwrap();
+ }
+}
+
+#[tokio::test]
+async fn test_composite_btree_combines_different_index_coverage() {
+ for independent_type in ["bitmap", "composite"] {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::string_type()))
+ .column("flag", DataType::Int(IntType::new()))
+ .options(table_options("2"))
+ .build()
+ .unwrap();
+ let table = Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", independent_type),
+ format!("memory:/composite_partial_coverage_{independent_type}"),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ setup_dirs(&table).await;
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("name", ArrowDataType::Utf8, false),
+ ArrowField::new("flag", ArrowDataType::Int32, false),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 6])) as ArrayRef,
+ Arc::new(StringArray::from(vec!["a", "b", "a", "b", "a", "b"])),
+ Arc::new(Int32Array::from(vec![1, 0, 0, 1, 0, 1])),
+ ],
+ )
+ .unwrap();
+ let mut writer = TableWrite::new(&table, "coverage-test".into()).unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ TableCommit::new(table.clone(), "coverage-test".into())
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_columns(&["name", "id"])
+ .execute()
+ .await
+ .unwrap();
+ let mut independent = table.new_btree_global_index_build_builder();
+ if independent_type == "bitmap" {
+ independent
+ .with_index_type("bitmap")
+ .with_index_column("flag");
+ } else {
+ independent.with_index_columns(&["flag", "id"]);
+ }
+ independent.execute().await.unwrap();
+ // Keep real persisted files, with (name,id) covering [0,3] and the
+ // independent flag definition covering [2,5]. Neither covers the table.
+ let (snapshot, entries) = composite_entries(&table).await;
+ let mut deletions = HashMap::<_, Vec<IndexFileMeta>>::new();
+ for entry in entries {
+ let meta = entry.index_file.global_index_meta.as_ref().unwrap();
+ if (meta.index_field_id == 1 && meta.row_range_start == 4)
+ || (meta.index_field_id == 2 && meta.row_range_start == 0)
+ {
+ deletions
+ .entry((entry.partition, entry.bucket))
+ .or_default()
+ .push(entry.index_file);
+ }
+ }
+ assert_eq!(deletions.values().map(Vec::len).sum::<usize>(), 2);
+ let messages = deletions
+ .into_iter()
+ .map(|((partition, bucket), files)| {
+ let mut message = crate::table::CommitMessage::new(partition, bucket, vec![]);
+ message.deleted_index_files = files;
+ message
+ })
+ .collect();
+ TableCommit::new(table.clone(), "partial-coverage-test".into())
+ .commit_if_latest_snapshot(messages, snapshot.id())
+ .await
+ .unwrap();
+ let (_, entries) = composite_entries(&table).await;
+ for (field_id, expected) in [(1, RowRange::new(0, 3)), (2, RowRange::new(2, 5))] {
+ let ranges = entries
+ .iter()
+ .filter_map(|entry| {
+ let meta = entry.index_file.global_index_meta.as_ref().unwrap();
+ (meta.index_field_id == field_id)
+ .then(|| RowRange::new(meta.row_range_start, meta.row_range_end))
+ })
+ .collect();
+ assert_eq!(merge_row_ranges(ranges), vec![expected]);
+ }
+ let b = PredicateBuilder::new(table.schema().fields());
+ let tuple = Predicate::and(vec![
+ b.equal("name", Datum::String("a".into())).unwrap(),
+ b.greater_than("id", Datum::Int(0)).unwrap(),
+ ]);
+ let flag = b.equal("flag", Datum::Int(1)).unwrap();
+ let ordinary = table.copy_with_options(HashMap::from([(
+ "global-index.enabled".into(),
+ "false".into(),
+ )]));
+ for (predicate, fast_ranges, complete_ranges, fast_ids, complete_ids) in [
+ (
+ Predicate::and(vec![tuple.clone(), flag.clone()]),
+ vec![],
+ vec![RowRange::new(0, 1), RowRange::new(4, 5)],
+ vec![],
+ vec![1],
+ ),
+ (
+ Predicate::or(vec![tuple, flag]),
+ vec![
+ RowRange::new(0, 0),
+ RowRange::new(2, 3),
+ RowRange::new(5, 5),
+ ],
+ vec![RowRange::new(0, 5)],
+ vec![1, 3, 4, 6],
+ vec![1, 3, 4, 5, 6],
+ ),
+ ] {
+ assert_eq!(scan_ids(&ordinary, predicate.clone()).await, complete_ids);
+ for (mode, option) in [
+ (GlobalIndexSearchMode::Fast, "fast"),
+ (GlobalIndexSearchMode::Full, "full"),
+ (GlobalIndexSearchMode::Detail, "detail"),
+ ] {
+ let (ranges, ids) = if mode == GlobalIndexSearchMode::Fast {
+ (&fast_ranges, &fast_ids)
+ } else {
+ (&complete_ranges, &complete_ids)
+ };
+ assert_eq!(
+ composite_ranges(&table, std::slice::from_ref(&predicate), i64::MAX, mode)
+ .await
+ .unwrap(),
+ *ranges,
+ "{independent_type}, {option}, {predicate:?}"
+ );
+ let indexed = table.copy_with_options(HashMap::from([(
+ "global-index.search-mode".into(),
+ option.into(),
+ )]));
+ assert_eq!(
+ scan_ids(&indexed, predicate.clone()).await,
+ *ids,
+ "{independent_type}, {option}, {predicate:?}"
+ );
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn test_composite_btree_preserves_exact_column_names() {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("a,b", DataType::VarChar(VarCharType::string_type()))
+ .column(" c ", DataType::Int(IntType::new()))
+ .options(table_options("10"))
+ .build()
+ .unwrap();
+ let table = Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "exact_names"),
+ "memory:/composite_exact_names".into(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ setup_dirs(&table).await;
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("a,b", ArrowDataType::Utf8, false),
+ ArrowField::new(" c ", ArrowDataType::Int32, false),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef,
+ Arc::new(StringArray::from(vec!["x", "y"])),
+ Arc::new(Int32Array::from(vec![10, 20])),
+ ],
+ )
+ .unwrap();
+ let mut writer = TableWrite::new(&table, "name-test".into()).unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ TableCommit::new(table.clone(), "name-test".into())
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ assert_eq!(
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_columns(&["a,b", " c "])
+ .execute()
+ .await
+ .unwrap(),
+ 1
+ );
+ assert_eq!(
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_column("a,b")
+ .execute()
+ .await
+ .unwrap(),
+ 1
+ );
+ assert_eq!(
+ table
+ .new_btree_global_index_build_builder()
+ .with_index_column(" c ")
+ .execute()
+ .await
+ .unwrap(),
+ 1
+ );
+ let b = PredicateBuilder::new(table.schema().fields());
+ let point = Predicate::and(vec![
+ b.equal("a,b", Datum::String("x".into())).unwrap(),
+ b.equal(" c ", Datum::Int(10)).unwrap(),
+ ]);
+ assert_eq!(scan_ids(&table, point).await, vec![1]);
+ assert_eq!(
+ table
+ .new_global_index_drop_builder()
+ .with_index_columns(&["a,b", " c "])
+ .execute()
+ .await
+ .unwrap(),
+ 1
+ );
+ let (_, entries) = composite_entries(&table).await;
+ assert_eq!(entries.len(), 2);
+ assert!(entries.iter().all(|entry| entry
+ .index_file
+ .global_index_meta
+ .as_ref()
+ .unwrap()
+ .extra_field_ids
+ .is_none()));
+ assert_eq!(
+ table
+ .new_global_index_drop_builder()
+ .with_index_column("a,b")
+ .execute()
+ .await
+ .unwrap(),
+ 1
+ );
+ assert_eq!(
+ table
+ .new_global_index_drop_builder()
+ .with_index_column(" c ")
+ .execute()
+ .await
+ .unwrap(),
+ 1
+ );
+ assert!(table
+ .new_btree_global_index_build_builder()
+ .with_index_columns(&[])
+ .execute()
+ .await
+ .is_err());
+ assert!(table
+ .new_global_index_drop_builder()
+ .with_index_columns(&[])
+ .execute()
+ .await
+ .is_err());
+}
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/validation.rs b/crates/paimon/src/table/sorted_global_index_build_builder/validation.rs
index 59f9d001..c1e64c1f 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/validation.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/validation.rs
@@ -59,6 +59,7 @@ pub(super) fn validate_table_options(table: &Table, core_options: &CoreOptions)
Ok(())
}
+#[cfg(test)]
pub(super) fn find_index_field<'a>(table: &'a Table, column: &str) -> Result<&'a DataField> {
table
.schema()
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
index 8ef05b2d..fdacd5b7 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
@@ -28,7 +28,8 @@ use crate::btree::BTreeIndexWriter;
use crate::fm_index::{FMGlobalIndexWriter, FMWriteOptions};
use crate::io::FileWrite;
use crate::spec::{
- extract_datum_from_arrow, DataField, GlobalIndexMeta, IndexFileMeta, ROW_ID_FIELD_NAME,
+ extract_datum_from_arrow, DataField, DataType, GlobalIndexMeta, IndexFileMeta, RowType,
+ ROW_ID_FIELD_NAME,
};
use crate::table::bitmap_global_index_writer::{BitmapGlobalIndexWriter, BitmapWriteResult};
use crate::table::global_index_types::{
@@ -44,8 +45,7 @@ impl SortedGlobalIndexBuildBuilder<'_> {
pub(super) async fn build_index_file(
&self,
shard: &SortedGlobalIndexShard,
- index_field: &DataField,
- index_column: &str,
+ index_fields: &[&DataField],
write_options: &GlobalIndexWriteOptions,
) -> Result<IndexFileMeta> {
let index_type = normalize_queryable_global_index_type(&self.index_type).ok_or_else(|| {
@@ -57,7 +57,16 @@ impl SortedGlobalIndexBuildBuilder<'_> {
}
})?;
let row_count = checked_row_count(shard.row_range_start, shard.row_range_end)?;
- let key_type = index_key_type(index_type, index_field)?;
+ let index_field = index_fields[0];
+ let index_column = index_field.name();
+ let composite_type = DataType::Row(RowType::new(
+ index_fields.iter().map(|field| (*field).clone()).collect(),
+ ));
+ let key_type = if index_fields.len() > 1 {
+ &composite_type
+ } else {
+ index_key_type(index_type, index_field)?
+ };
let codec_type = if matches!(
index_type,
MULTIVALUE_GLOBAL_INDEX_TYPE | FM_GLOBAL_INDEX_TYPE
@@ -70,15 +79,7 @@ impl SortedGlobalIndexBuildBuilder<'_> {
let mut rows = if index_type == FM_GLOBAL_INDEX_TYPE {
Vec::new()
} else {
- extract_index_rows(
- self.table,
- shard,
- index_column,
- index_field,
- index_type,
- serialize_key,
- )
- .await?
+ extract_index_rows(self.table, shard, index_fields, index_type, serialize_key).await?
};
if !rows.is_empty() {
sort_index_rows(&mut rows, &cmp)?;
@@ -255,7 +256,13 @@ impl SortedGlobalIndexBuildBuilder<'_> {
row_range_start: shard.row_range_start,
row_range_end: shard.row_range_end,
index_field_id: index_field.id(),
- extra_field_ids: None,
+ extra_field_ids: (index_fields.len() > 1).then(|| {
+ index_fields
+ .iter()
+ .skip(1)
+ .map(|field| field.id())
+ .collect()
+ }),
source_meta: None,
index_meta: Some(index_meta),
}),
diff --git a/crates/paimon/testdata/btree/GenerateCompositeBTree.java b/crates/paimon/testdata/btree/GenerateCompositeBTree.java
new file mode 100644
index 00000000..81df04f0
--- /dev/null
+++ b/crates/paimon/testdata/btree/GenerateCompositeBTree.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.paimon.globalindex.btree;
+
+import org.apache.paimon.compression.BlockCompressionFactory;
+import org.apache.paimon.compression.BlockCompressionType;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.globalindex.CompositeKeySerializer;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+
+import java.io.ByteArrayOutputStream;
+import java.math.BigDecimal;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+
+/** Generate compatibility fixtures with the Java production composite BTree writer. */
+public class GenerateCompositeBTree {
+ public static void main(String[] args) throws Exception {
+ RowType scalarType = RowType.of(DataTypes.BOOLEAN(), DataTypes.TINYINT(),
+ DataTypes.SMALLINT(), DataTypes.INT(), DataTypes.BIGINT(), DataTypes.FLOAT(),
+ DataTypes.DOUBLE(), DataTypes.DECIMAL(20, 2), DataTypes.TIMESTAMP(6),
+ DataTypes.STRING(), DataTypes.INT());
+ GenericRow scalarKey = GenericRow.of(true, (byte) -3, (short) -257, -12345,
+ 1234567890123L, Float.intBitsToFloat(0xffc00001),
+ Double.longBitsToDouble(0xfff8000000000001L),
+ Decimal.fromBigDecimal(new BigDecimal("-1.29"), 20, 2), Timestamp.fromEpochMillis(-123000, 999999),
+ BinaryString.fromString("联合"), null);
+ Files.write(Paths.get(args[0], "btree_composite_java_scalar.key"),
+ new CompositeKeySerializer(scalarType).serialize(scalarKey));
+ for (int version : new int[] {1, 2}) {
+ for (BlockCompressionType codec : new BlockCompressionType[] {
+ BlockCompressionType.NONE, BlockCompressionType.LZ4}) {
+ final ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ GlobalIndexFileWriter files = new GlobalIndexFileWriter() {
+ public String newFileName(String prefix) { return "fixture"; }
+ public PositionOutputStream newOutputStream(String name) {
+ return new PositionOutputStream() {
+ public long getPos() { return bytes.size(); }
+ public void write(int b) { bytes.write(b); }
+ public void write(byte[] b) { bytes.write(b, 0, b.length); }
+ public void write(byte[] b, int off, int len) { bytes.write(b, off, len); }
+ public void flush() {}
+ public void close() {}
+ };
+ }
+ };
+ RowType type = RowType.of(DataTypes.STRING(), DataTypes.INT(), DataTypes.STRING());
+ BTreeIndexWriter writer = new BTreeIndexWriter(
+ files, new CompositeKeySerializer(type),
+ codec == BlockCompressionType.NONE ? 64 : 512, null,
+ BlockCompressionFactory.create(codec), version);
+ long id = 0;
+ for (String category : new String[] {null, "a", "b"}) {
+ for (Integer item : new Integer[] {null, -2, 0, 1, 2}) {
+ for (String tag : new String[] {null, "", "z"}) {
+ GenericRow key = GenericRow.of(
+ category == null ? null : BinaryString.fromString(category), item,
+ tag == null ? null : BinaryString.fromString(tag));
+ writer.write(key, id);
+ writer.write(key, id + 1000);
+ id++;
+ }
+ }
+ }
+ ResultEntry result = writer.finish().get(0);
+ String name = "btree_composite_v" + version + "_java_" + codec.name().toLowerCase();
+ Files.write(Paths.get(args[0], name + ".meta"), result.meta());
+ Files.write(Paths.get(args[0], name + ".bin"),
+ bytes.toByteArray());
+ }
+ }
+ }
+}
diff --git a/crates/paimon/testdata/btree/README.md b/crates/paimon/testdata/btree/README.md
index fbdcbaba..96cceed4 100644
--- a/crates/paimon/testdata/btree/README.md
+++ b/crates/paimon/testdata/btree/README.md
@@ -42,3 +42,38 @@ Rust tests read the Java files and compare the Rust V2 writer's uncompressed
output byte-for-byte with the Java fixture. A writer/reader round-trip alone
would not catch incompatible posting tags, portable bitmap layout, or footer
versions.
+
+## Composite BTree interoperability fixtures
+
+`btree_composite_v{1,2}_java_{none,lz4}.bin` and their `.meta` files were
+produced by `GenerateCompositeBTree.java` with the current Java
+`CompositeKeySerializer` and production `BTreeIndexWriter`, and verified by
+regenerating all files against the current Java sources. The verified serializer
+source SHA-256 is `f2cddf1eb63e81046dd25ad6a35a39f28a1e014cb603be0f44ecc603f14d547e`.
+The keys use `RowCompactedSerializer` bytes with a canonical INSERT header.
+
+Each fixture has 45 `(STRING, INT, STRING)` tuples: categories NULL/a/b,
+items NULL/-2/0/1/2, and tags NULL/empty/z. Tuple ordinal `i` has postings
+`i` and `i + 1000`; all-NULL tuples are ordinary keys. Uncompressed files use
+64-byte blocks to exercise seeks across blocks; LZ4 files use 512-byte blocks
+to ensure real compression. Rust tests compare queries with ordinary
+row filtering and compare uncompressed Rust output byte-for-byte with Java.
+
+Compile against the current Java classes and runtime dependencies (including
+RoaringBitmap, aircompressor, LZ4, and SLF4J). When Java target classes are stale,
+compile the relevant sources into the helper's output directory first:
+
+```sh
+javac -cp "$PAIMON_CLASSPATH" -d "$CLASSES_DIR" \
+ "$PAIMON_SOURCE/paimon-common/src/main/java/org/apache/paimon/data/serializer/RowCompactedSerializer.java" \
+ "$PAIMON_SOURCE/paimon-common/src/main/java/org/apache/paimon/globalindex/KeySerializer.java" \
+ "$PAIMON_SOURCE/paimon-common/src/main/java/org/apache/paimon/globalindex/CompositeKeySerializer.java" \
+ "$PAIMON_SOURCE/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java" \
+ GenerateCompositeBTree.java
+java -cp "$CLASSES_DIR:$PAIMON_CLASSPATH" \
+ org.apache.paimon.globalindex.btree.GenerateCompositeBTree .
+```
+
+`btree_composite_java_scalar.key` covers BOOLEAN, TINYINT, SMALLINT, INT,
+BIGINT, canonical FLOAT/DOUBLE NaNs, non-compact DECIMAL(20,2), TIMESTAMP(6),
+UTF-8 strings and a NULL suffix across multiple header bytes.
diff --git a/crates/paimon/testdata/btree/btree_composite_java_scalar.key b/crates/paimon/testdata/btree/btree_composite_java_scalar.key
new file mode 100644
index 00000000..360384e5
Binary files /dev/null and b/crates/paimon/testdata/btree/btree_composite_java_scalar.key differ
diff --git a/crates/paimon/testdata/btree/btree_composite_v1_java_lz4.bin b/crates/paimon/testdata/btree/btree_composite_v1_java_lz4.bin
new file mode 100644
index 00000000..a3e05730
Binary files /dev/null and b/crates/paimon/testdata/btree/btree_composite_v1_java_lz4.bin differ
diff --git a/crates/paimon/testdata/btree/btree_composite_v1_java_lz4.meta b/crates/paimon/testdata/btree/btree_composite_v1_java_lz4.meta
new file mode 100644
index 00000000..e1cf9ad1
Binary files /dev/null and b/crates/paimon/testdata/btree/btree_composite_v1_java_lz4.meta differ
diff --git a/crates/paimon/testdata/btree/btree_composite_v1_java_none.bin b/crates/paimon/testdata/btree/btree_composite_v1_java_none.bin
new file mode 100644
index 00000000..24cc494e
Binary files /dev/null and b/crates/paimon/testdata/btree/btree_composite_v1_java_none.bin differ
diff --git a/crates/paimon/testdata/btree/btree_composite_v1_java_none.meta b/crates/paimon/testdata/btree/btree_composite_v1_java_none.meta
new file mode 100644
index 00000000..e1cf9ad1
Binary files /dev/null and b/crates/paimon/testdata/btree/btree_composite_v1_java_none.meta differ
diff --git a/crates/paimon/testdata/btree/btree_composite_v2_java_lz4.bin b/crates/paimon/testdata/btree/btree_composite_v2_java_lz4.bin
new file mode 100644
index 00000000..25bd86bb
Binary files /dev/null and b/crates/paimon/testdata/btree/btree_composite_v2_java_lz4.bin differ
diff --git a/crates/paimon/testdata/btree/btree_composite_v2_java_lz4.meta b/crates/paimon/testdata/btree/btree_composite_v2_java_lz4.meta
new file mode 100644
index 00000000..e1cf9ad1
Binary files /dev/null and b/crates/paimon/testdata/btree/btree_composite_v2_java_lz4.meta differ
diff --git a/crates/paimon/testdata/btree/btree_composite_v2_java_none.bin b/crates/paimon/testdata/btree/btree_composite_v2_java_none.bin
new file mode 100644
index 00000000..af92beb3
Binary files /dev/null and b/crates/paimon/testdata/btree/btree_composite_v2_java_none.bin differ
diff --git a/crates/paimon/testdata/btree/btree_composite_v2_java_none.meta b/crates/paimon/testdata/btree/btree_composite_v2_java_none.meta
new file mode 100644
index 00000000..e1cf9ad1
Binary files /dev/null and b/crates/paimon/testdata/btree/btree_composite_v2_java_none.meta differ
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 259b2b2a..dcb3cd42 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1152,6 +1152,48 @@ to use adaptive single-row, delta-list, and Roaring postings; every reader of
the table must support V2 before enabling it. Existing index files do not need
to be rebuilt for the reader to handle a mix of versions.
+BTree also supports composite indexes. List scalar columns in tuple-key order:
+
+```sql
+CALL sys.create_global_index(
+ table => 'paimon.my_db.my_table',
+ index_column => 'category,item_number',
+ index_type => 'btree'
+);
+
+SELECT * FROM paimon.my_db.my_table
+WHERE category IN ('a', 'b') AND item_number >= 10;
+```
+
+Equality, `IN`, or `IS NULL` on every key column uses tuple point lookups.
+A leading prefix of those conditions may be followed by `>`, `>=`, `<`, `<=`,
+`BETWEEN`, or `IS NOT NULL` on the next column. Predicates after the first range
+column remain data filters; a predicate only on a later key column cannot use
+this index. Repeated `IN` values are deduplicated and intersected with other
+conditions on the same column before expansion, with a maximum of 256 tuple
+intervals. Larger expansions fall back to another index or an ordinary scan.
+Decimal literals are rescaled exactly to the column's scale. If a literal
+cannot be encoded without losing precision (including sub-millisecond bounds
+on compact timestamps), this definition is declined and data filtering retains
+the original literal. Independent indexes on other columns continue to be
+evaluated and intersected for `AND` predicates.
+
+Prefix and range scans honor `btree-index.fallback-scan-max-size` for selected
+files in each source row-ID range group. If any group exceeds the budget, the
+whole definition is declined; tuple point lookups bypass this budget. When
+multiple definitions match, selection prefers more bounded columns, then more
+point-constrained columns, then fewer total key columns. A dedicated BTree or
+bitmap index on the leading column is preferred for a leading-column-only predicate
+when it covers the composite definition's coverage. `full` and `detail` search
+modes retain unindexed rows for data filtering.
+
+The Rust builders accept `.with_index_columns(&["category", "item_number"])`
+or `.with_index_column("category,item_number")`. The list API preserves exact
+column names, including commas and whitespace. The string API first resolves
+an exact column name; otherwise it parses a comma-separated list. Rebuilding
+the same ordered definition fills missing row ranges. Composite, reversed-column, and scalar
+definitions can coexist. Bitmap, multivalue, and FM builds accept one column.
+
BTree min/max metadata can also avoid index-file reads when every key matches,
there are no nulls, and the file set's row counts prove complete coverage of
the same source row-ID range. These all-matching files do not consume the
@@ -1329,6 +1371,10 @@ CALL sys.drop_global_index(
`ivf-flat`, `ivf-pq`, `ivf-sq`, `ivf-rq`, and `diskann`. It defaults to `btree`, is
case-insensitive and surrounding whitespace is ignored.
+To drop a composite BTree definition, pass the same ordered `index_column`
+list used to build it, such as `'category,item_number'`. Scalar indexes and
+definitions with a different column order remain available.
+
### create_lumina_index
Build and commit a Lumina global vector index for a table column: