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 b2e53fa5 fix(global-index): stop reading index keys with a comparator
that can panic (#909)
b2e53fa5 is described below
commit b2e53fa552cf4ddfb979d15abccf9d9d5f494182
Author: jackylee <[email protected]>
AuthorDate: Tue Sep 22 14:14:24 2026 +0800
fix(global-index): stop reading index keys with a comparator that can panic
(#909)
---
.../tests/global_index_schema_evolution.rs | 107 ++++++++++
crates/paimon/src/btree/block.rs | 40 +++-
crates/paimon/src/btree/key_serde.rs | 237 +++++++++++++++++----
crates/paimon/src/btree/meta.rs | 104 +++++++--
crates/paimon/src/btree/query.rs | 32 ++-
crates/paimon/src/btree/reader.rs | 113 ++++++----
crates/paimon/src/btree/tests.rs | 27 ++-
crates/paimon/src/btree/var_len.rs | 56 +++++
crates/paimon/src/btree/writer.rs | 14 +-
.../paimon/src/table/bitmap_global_index_format.rs | 14 +-
.../paimon/src/table/bitmap_global_index_reader.rs | 109 ++++++----
.../paimon/src/table/bitmap_global_index_writer.rs | 52 +++--
crates/paimon/src/table/global_index_scanner.rs | 2 +-
.../src/table/global_index_scanner/all_match.rs | 58 +++--
.../paimon/src/table/global_index_scanner/entry.rs | 8 +-
.../src/table/global_index_scanner/reader.rs | 33 ++-
.../paimon/src/table/global_index_scanner/tests.rs | 6 +-
.../extraction.rs | 25 ++-
.../sorted_global_index_build_builder/tests.rs | 6 +-
.../sorted_global_index_build_builder/writer.rs | 2 +-
20 files changed, 798 insertions(+), 247 deletions(-)
diff --git
a/crates/integrations/datafusion/tests/global_index_schema_evolution.rs
b/crates/integrations/datafusion/tests/global_index_schema_evolution.rs
new file mode 100644
index 00000000..947b7e0e
--- /dev/null
+++ b/crates/integrations/datafusion/tests/global_index_schema_evolution.rs
@@ -0,0 +1,107 @@
+// 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.
+
+//! A sorted global index is read with a comparator built from the column's
*current*
+//! type, and nothing records the type the index was built with. Widening an
indexed
+//! column is allowed, so the index must be left alone once its keys can no
longer
+//! belong to that type.
+
+mod common;
+
+use common::{exec, row_count, setup_sql_context};
+
+async fn setup_indexed_table(
+ table_name: &str,
+ column_type: &str,
+) -> (tempfile::TempDir, paimon_datafusion::SQLContext) {
+ let (tmp, sql_context) = setup_sql_context().await;
+ exec(
+ &sql_context,
+ &format!(
+ "CREATE TABLE paimon.test_db.{table_name} (id {column_type}, name
VARCHAR(100)) WITH (\
+ 'row-tracking.enabled' = 'true',\
+ 'data-evolution.enabled' = 'true',\
+ 'global-index.enabled' = 'true',\
+ 'sorted-index.records-per-range' = '10'\
+ )"
+ ),
+ )
+ .await;
+ (tmp, sql_context)
+}
+
+#[tokio::test]
+async fn test_widening_an_indexed_int_column_keeps_answering_queries() {
+ let (_tmp, sql_context) = setup_indexed_table("gi_widen_int", "INT").await;
+ for id in 1..=40 {
+ exec(
+ &sql_context,
+ &format!("INSERT INTO paimon.test_db.gi_widen_int (id, name)
VALUES ({id}, 'n{id}')"),
+ )
+ .await;
+ }
+ exec(
+ &sql_context,
+ "CALL sys.create_global_index(table => 'test_db.gi_widen_int',
index_column => 'id')",
+ )
+ .await;
+ assert_eq!(
+ row_count(
+ &sql_context,
+ "SELECT * FROM paimon.test_db.gi_widen_int WHERE id = 7"
+ )
+ .await,
+ 1,
+ "the index must answer the query before the type change"
+ );
+
+ // Allowed: `UpdateColumnType` guards partition, primary-key, bucket-key
and
+ // primary-key-index columns, but not global-index columns.
+ exec(
+ &sql_context,
+ "ALTER TABLE paimon.test_db.gi_widen_int ALTER COLUMN id TYPE BIGINT",
+ )
+ .await;
+
+ // The index keys are still 4 bytes wide; the BIGINT comparator used to
read 8 and
+ // panic with "range end index 8 out of range for slice of length 4".
+ assert_eq!(
+ row_count(
+ &sql_context,
+ "SELECT * FROM paimon.test_db.gi_widen_int WHERE id = 7"
+ )
+ .await,
+ 1,
+ "widening an indexed column must fall back to a scan, not panic"
+ );
+ assert_eq!(
+ row_count(
+ &sql_context,
+ "SELECT * FROM paimon.test_db.gi_widen_int WHERE id > 35"
+ )
+ .await,
+ 5
+ );
+ assert_eq!(
+ row_count(
+ &sql_context,
+ "SELECT * FROM paimon.test_db.gi_widen_int WHERE id = 99"
+ )
+ .await,
+ 0
+ );
+}
diff --git a/crates/paimon/src/btree/block.rs b/crates/paimon/src/btree/block.rs
index 950995f7..6f5c513b 100644
--- a/crates/paimon/src/btree/block.rs
+++ b/crates/paimon/src/btree/block.rs
@@ -641,10 +641,11 @@ impl BlockReader {
/// Binary search for the given target key. Returns an iterator positioned
at the
/// first entry whose key >= target_key.
- /// The comparator compares two key byte slices.
- pub fn seek_and_iter<F>(&self, target_key: &[u8], cmp: &F) -> (bool,
BlockIter<'_>)
+ /// The comparator compares two key byte slices, and fails when the stored
keys are
+ /// not keys of the column's current type.
+ pub fn seek_and_iter<F>(&self, target_key: &[u8], cmp: &F) ->
io::Result<(bool, BlockIter<'_>)>
where
- F: Fn(&[u8], &[u8]) -> Ordering,
+ F: Fn(&[u8], &[u8]) -> io::Result<Ordering>,
{
let mut left: i32 = 0;
let mut right: i32 = self.record_count as i32 - 1;
@@ -657,7 +658,7 @@ impl BlockReader {
let byte_offset = self.seek_to_position(mid as usize);
let (key, _next_offset) = self.read_key_at(byte_offset);
- match cmp(key, target_key) {
+ match cmp(key, target_key)? {
Ordering::Equal => {
found = true;
best_index = Some(mid as usize);
@@ -675,7 +676,7 @@ impl BlockReader {
}
}
- match (best_index, best_offset) {
+ Ok(match (best_index, best_offset) {
(Some(idx), Some(off)) => (
found,
BlockIter {
@@ -692,7 +693,7 @@ impl BlockReader {
index: self.record_count,
},
),
- }
+ })
}
}
@@ -847,27 +848,46 @@ mod tests {
let block = writer.finish();
let reader = BlockReader::create(&block).unwrap();
- let cmp = |a: &[u8], b: &[u8]| a.cmp(b);
+ let cmp = |a: &[u8], b: &[u8]| Ok(a.cmp(b));
// Exact match
- let (found, mut iter) = reader.seek_and_iter(b"banana", &cmp);
+ let (found, mut iter) = reader.seek_and_iter(b"banana", &cmp).unwrap();
assert!(found);
let (k, v) = iter.next().unwrap();
assert_eq!(k, b"banana");
assert_eq!(v, b"2");
// Seek to position >= "bz" -> should land on "cherry"
- let (found, mut iter) = reader.seek_and_iter(b"bz", &cmp);
+ let (found, mut iter) = reader.seek_and_iter(b"bz", &cmp).unwrap();
assert!(!found);
let (k, _) = iter.next().unwrap();
assert_eq!(k, b"cherry");
// Seek past all entries
- let (found, iter) = reader.seek_and_iter(b"zzz", &cmp);
+ let (found, iter) = reader.seek_and_iter(b"zzz", &cmp).unwrap();
assert!(!found);
assert!(!iter.has_next());
}
+ /// A comparator that rejects the stored keys must abort the search, not
settle on
+ /// whatever entry the half-finished binary search last looked at.
+ #[test]
+ fn test_block_seek_propagates_a_comparator_failure() {
+ let mut writer = BlockWriter::new(1024);
+ writer.add(b"apple", b"1");
+ writer.add(b"banana", b"2");
+
+ let block = writer.finish();
+ let reader = BlockReader::create(&block).unwrap();
+
+ let cmp =
+ |_: &[u8], _: &[u8]|
Err(io::Error::new(io::ErrorKind::InvalidData, "not my key"));
+ let Err(error) = reader.seek_and_iter(b"banana", &cmp) else {
+ panic!("a failing comparator must abort the seek");
+ };
+ assert_eq!(error.kind(), io::ErrorKind::InvalidData);
+ }
+
#[test]
fn test_block_handle_roundtrip() {
let handle = BlockHandle::new(12345, 6789);
diff --git a/crates/paimon/src/btree/key_serde.rs
b/crates/paimon/src/btree/key_serde.rs
index d1b7834f..3ef651dd 100644
--- a/crates/paimon/src/btree/key_serde.rs
+++ b/crates/paimon/src/btree/key_serde.rs
@@ -19,7 +19,7 @@
//!
//! Reference:
[org.apache.paimon.globalindex.btree.KeySerializer](https://github.com/apache/paimon/blob/master/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/KeySerializer.java)
-use crate::btree::var_len::{decode_var_int_from_slice, encode_var_int};
+use crate::btree::var_len::{encode_var_int, try_decode_var_int_from_slice};
use crate::spec::{DataType, Datum, VariantType};
use std::cmp::Ordering;
@@ -29,85 +29,174 @@ const TIMESTAMP_COMPACT_PRECISION: u32 = 3;
const DECIMAL_COMPACT_PRECISION: u32 = 18;
/// Key comparator type alias.
-pub type KeyComparator = Box<dyn Fn(&[u8], &[u8]) -> Ordering + Send + Sync>;
+///
+/// Fallible because the bytes come from an index file that may have been
written when
+/// the column had a different type. [`make_key_comparator`] picks its arm
from the
+/// column's *current* type, and nothing records the type the index was built
with:
+/// neither `IndexManifestEntry` nor `IndexFileMeta` carries a schema id.
+/// `SchemaChange::UpdateColumnType` guards partition, primary-key, bucket-key
and
+/// primary-key-index columns but not global-index columns, so `INT` to
`BIGINT` is
+/// accepted and every stored key is then 4 bytes short of what the new arm
reads. That
+/// used to index out of bounds and panic inside a query; an `Err` lets the
caller give
+/// up on the index and fall back instead.
+pub type KeyComparator = Box<dyn Fn(&[u8], &[u8]) -> crate::Result<Ordering> +
Send + Sync>;
+
+/// A borrowed key comparator, for signatures that only call one.
[`KeyComparator`] is
+/// the owned form callers build from a [`DataType`].
+pub type DynKeyComparator<'a> = dyn Fn(&[u8], &[u8]) ->
crate::Result<Ordering> + 'a;
+
+/// Take the fixed-width body of a key, or explain why it cannot belong to
`type_name`.
+///
+/// The width must match exactly: [`serialize_datum`] emits exactly `N` bytes
for these
+/// types, so a longer key is as much a foreign encoding as a shorter one, and
silently
+/// comparing its prefix would answer the query from the wrong bytes.
+pub(crate) fn fixed_key_bytes<const N: usize>(
+ key: &[u8],
+ type_name: &str,
+) -> crate::Result<[u8; N]> {
+ key.try_into().map_err(|_| crate::Error::DataInvalid {
+ message: format!(
+ "Global index key of {} byte(s) cannot be a {type_name} key of
{N}; the index was \
+ built before the column's type changed and cannot be used",
+ key.len()
+ ),
+ source: None,
+ })
+}
+
+fn variable_key_body(key: &[u8], min_len: usize, type_name: &str) ->
crate::Result<()> {
+ if key.len() < min_len {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Global index key of {} byte(s) cannot be a {type_name} key of
at least \
+ {min_len}; the index was built before the column's type
changed and cannot be \
+ used",
+ key.len()
+ ),
+ source: None,
+ });
+ }
+ Ok(())
+}
+
+/// A [`KeyComparator`] failure travelling through a reader's
[`std::io::Result`].
+///
+/// The BTree and bitmap readers interleave key comparison with file I/O and
report both
+/// as [`std::io::Error`], but their caller has to treat the two differently:
a comparison
+/// failure means the stored bytes are not keys of this type, so the index
simply cannot
+/// answer and the query falls back to a scan, while a real I/O failure must
fail the
+/// query. Wrapping keeps both in one channel and lets
+/// [`is_key_comparison_failure`] tell them apart.
+#[derive(Debug)]
+pub struct KeyComparisonFailure {
+ message: String,
+}
+
+impl std::fmt::Display for KeyComparisonFailure {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.write_str(&self.message)
+ }
+}
+
+impl std::error::Error for KeyComparisonFailure {}
+
+/// Wrap a comparator failure for a reader that returns [`std::io::Result`].
+pub(crate) fn key_comparison_io_error(error: crate::Error) -> std::io::Error {
+ std::io::Error::new(
+ std::io::ErrorKind::InvalidData,
+ KeyComparisonFailure {
+ message: error.to_string(),
+ },
+ )
+}
+
+/// Whether `error` is a key comparison failure rather than a real I/O failure.
+pub(crate) fn is_key_comparison_failure(error: &std::io::Error) -> bool {
+ error
+ .get_ref()
+ .is_some_and(|source| source.is::<KeyComparisonFailure>())
+}
/// Create a key comparator based on the data type.
/// For fixed-size numeric types, compares by decoded value.
/// For variable-length types (string, bytes), uses lexicographic byte
comparison.
pub fn make_key_comparator(data_type: &DataType) -> KeyComparator {
match data_type {
- DataType::TinyInt(_) => Box::new(|a: &[u8], b: &[u8]| (a[0] as
i8).cmp(&(b[0] as i8))),
+ 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")?);
+ Ok(av.cmp(&bv))
+ }),
DataType::SmallInt(_) => Box::new(|a: &[u8], b: &[u8]| {
- let av = i16::from_le_bytes(a[..2].try_into().unwrap());
- let bv = i16::from_le_bytes(b[..2].try_into().unwrap());
- av.cmp(&bv)
+ let av = i16::from_le_bytes(fixed_key_bytes::<2>(a, "SMALLINT")?);
+ let bv = i16::from_le_bytes(fixed_key_bytes::<2>(b, "SMALLINT")?);
+ Ok(av.cmp(&bv))
}),
DataType::Int(_) | DataType::Date(_) | DataType::Time(_) => {
Box::new(|a: &[u8], b: &[u8]| {
- let av = i32::from_le_bytes(a[..4].try_into().unwrap());
- let bv = i32::from_le_bytes(b[..4].try_into().unwrap());
- av.cmp(&bv)
+ let av = i32::from_le_bytes(fixed_key_bytes::<4>(a, "INT")?);
+ let bv = i32::from_le_bytes(fixed_key_bytes::<4>(b, "INT")?);
+ Ok(av.cmp(&bv))
})
}
DataType::BigInt(_) => Box::new(|a: &[u8], b: &[u8]| {
- let av = i64::from_le_bytes(a[..8].try_into().unwrap());
- let bv = i64::from_le_bytes(b[..8].try_into().unwrap());
- av.cmp(&bv)
+ let av = i64::from_le_bytes(fixed_key_bytes::<8>(a, "BIGINT")?);
+ let bv = i64::from_le_bytes(fixed_key_bytes::<8>(b, "BIGINT")?);
+ Ok(av.cmp(&bv))
}),
DataType::Float(_) => Box::new(|a: &[u8], b: &[u8]| {
- let av = f32::from_le_bytes(a[..4].try_into().unwrap());
- let bv = f32::from_le_bytes(b[..4].try_into().unwrap());
- av.total_cmp(&bv)
+ let av = f32::from_le_bytes(fixed_key_bytes::<4>(a, "FLOAT")?);
+ let bv = f32::from_le_bytes(fixed_key_bytes::<4>(b, "FLOAT")?);
+ Ok(av.total_cmp(&bv))
}),
DataType::Double(_) => Box::new(|a: &[u8], b: &[u8]| {
- let av = f64::from_le_bytes(a[..8].try_into().unwrap());
- let bv = f64::from_le_bytes(b[..8].try_into().unwrap());
- av.total_cmp(&bv)
+ let av = f64::from_le_bytes(fixed_key_bytes::<8>(a, "DOUBLE")?);
+ let bv = f64::from_le_bytes(fixed_key_bytes::<8>(b, "DOUBLE")?);
+ Ok(av.total_cmp(&bv))
}),
DataType::Timestamp(t) if t.precision() > TIMESTAMP_COMPACT_PRECISION
=> {
// Non-compact: millis (8 bytes LE) + nanoOfMillisecond (varint)
- Box::new(|a: &[u8], b: &[u8]| {
- let a_millis = i64::from_le_bytes(a[..8].try_into().unwrap());
- let b_millis = i64::from_le_bytes(b[..8].try_into().unwrap());
- let (a_nanos, _) = decode_var_int_from_slice(a, 8);
- let (b_nanos, _) = decode_var_int_from_slice(b, 8);
- a_millis.cmp(&b_millis).then_with(|| a_nanos.cmp(&b_nanos))
- })
+ Box::new(|a: &[u8], b: &[u8]| compare_non_compact_timestamps(a, b))
}
DataType::LocalZonedTimestamp(t) if t.precision() >
TIMESTAMP_COMPACT_PRECISION => {
- Box::new(|a: &[u8], b: &[u8]| {
- let a_millis = i64::from_le_bytes(a[..8].try_into().unwrap());
- let b_millis = i64::from_le_bytes(b[..8].try_into().unwrap());
- let (a_nanos, _) = decode_var_int_from_slice(a, 8);
- let (b_nanos, _) = decode_var_int_from_slice(b, 8);
- a_millis.cmp(&b_millis).then_with(|| a_nanos.cmp(&b_nanos))
- })
+ Box::new(|a: &[u8], b: &[u8]| compare_non_compact_timestamps(a, b))
}
DataType::Decimal(d) if d.precision() > DECIMAL_COMPACT_PRECISION => {
// Non-compact Decimal keys use Java BigInteger.toByteArray()
bytes.
Box::new(|a: &[u8], b: &[u8]| {
-
decode_java_big_integer_i128(a).cmp(&decode_java_big_integer_i128(b))
+
Ok(decode_java_big_integer_i128(a)?.cmp(&decode_java_big_integer_i128(b)?))
})
}
// Compact Timestamp/LocalZonedTimestamp (precision <= 3): millis as
i64 LE
DataType::Timestamp(_) | DataType::LocalZonedTimestamp(_) => {
Box::new(|a: &[u8], b: &[u8]| {
- let av = i64::from_le_bytes(a[..8].try_into().unwrap());
- let bv = i64::from_le_bytes(b[..8].try_into().unwrap());
- av.cmp(&bv)
+ let av = i64::from_le_bytes(fixed_key_bytes::<8>(a,
"TIMESTAMP")?);
+ let bv = i64::from_le_bytes(fixed_key_bytes::<8>(b,
"TIMESTAMP")?);
+ Ok(av.cmp(&bv))
})
}
// Compact Decimal (precision <= 18): unscaled as i64 LE
DataType::Decimal(_) => Box::new(|a: &[u8], b: &[u8]| {
- let av = i64::from_le_bytes(a[..8].try_into().unwrap());
- let bv = i64::from_le_bytes(b[..8].try_into().unwrap());
- av.cmp(&bv)
+ let av = i64::from_le_bytes(fixed_key_bytes::<8>(a, "DECIMAL")?);
+ let bv = i64::from_le_bytes(fixed_key_bytes::<8>(b, "DECIMAL")?);
+ Ok(av.cmp(&bv))
}),
- // String, VarChar, Char, Bytes — lexicographic
- _ => Box::new(|a: &[u8], b: &[u8]| a.cmp(b)),
+ // String, VarChar, Char, Bytes — lexicographic, so any width is
readable
+ _ => Box::new(|a: &[u8], b: &[u8]| Ok(a.cmp(b))),
}
}
+/// Millis (8 bytes LE) then the varint nano-of-millisecond, both
bounds-checked.
+fn compare_non_compact_timestamps(a: &[u8], b: &[u8]) ->
crate::Result<Ordering> {
+ variable_key_body(a, 9, "TIMESTAMP")?;
+ variable_key_body(b, 9, "TIMESTAMP")?;
+ let a_millis = i64::from_le_bytes(fixed_key_bytes::<8>(&a[..8],
"TIMESTAMP")?);
+ let b_millis = i64::from_le_bytes(fixed_key_bytes::<8>(&b[..8],
"TIMESTAMP")?);
+ let (a_nanos, _) = try_decode_var_int_from_slice(a, 8)?;
+ let (b_nanos, _) = try_decode_var_int_from_slice(b, 8)?;
+ Ok(a_millis.cmp(&b_millis).then_with(|| a_nanos.cmp(&b_nanos)))
+}
+
/// Serialize a Datum to BTree key bytes (little-endian, matching Java
Paimon's KeySerializer).
pub fn serialize_datum(datum: &Datum, data_type: &DataType) -> Vec<u8> {
match datum {
@@ -174,16 +263,35 @@ fn encode_java_big_integer_i128(value: i128) -> Vec<u8> {
bytes[start..].to_vec()
}
-fn decode_java_big_integer_i128(bytes: &[u8]) -> i128 {
- if bytes.is_empty() {
- return 0;
+/// Read the Java `BigInteger.toByteArray()` bytes of a non-compact DECIMAL
key.
+///
+/// Anything outside 1..=16 bytes cannot be one:
[`encode_java_big_integer_i128`] never
+/// emits an empty slice, and more than 16 bytes does not fit the `i128`
unscaled value.
+/// Such a key belongs to the type the column had when the index was built.
+///
+/// The bound is all this arm can check, and 1..=16 covers every fixed-width
key this
+/// module writes, so a stale `INT` or `BIGINT` key is accepted here and read
as a
+/// big-endian magnitude -- `numeric` to `DECIMAL(p > 18)` is an implicit
cast, so that
+/// is reachable. It gives a wrong ordering rather than a panic, which puts it
in the
+/// same undetectable class as `INT` to `FLOAT`: telling a foreign key from a
real one
+/// needs the type the index was built with, and no index file records it.
+fn decode_java_big_integer_i128(bytes: &[u8]) -> crate::Result<i128> {
+ if bytes.is_empty() || bytes.len() > 16 {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Global index key of {} byte(s) cannot be a non-compact
DECIMAL key of 1 to 16; \
+ the index was built before the column's type changed and
cannot be used",
+ bytes.len()
+ ),
+ source: None,
+ });
}
let negative = bytes[0] & 0x80 != 0;
let mut value = if negative { -1 } else { 0 };
for &byte in bytes {
value = (value << 8) | i128::from(byte);
}
- value
+ Ok(value)
}
#[cfg(test)]
@@ -263,8 +371,8 @@ mod tests {
let key_128 = encode_java_big_integer_i128(128);
let key_minus_129 = encode_java_big_integer_i128(-129);
- assert_eq!(cmp(&key_127, &key_128), Ordering::Less);
- assert_eq!(cmp(&key_minus_129, &key_127), Ordering::Less);
+ assert_eq!(cmp(&key_127, &key_128).unwrap(), Ordering::Less);
+ assert_eq!(cmp(&key_minus_129, &key_127).unwrap(), Ordering::Less);
}
#[test]
@@ -272,4 +380,39 @@ mod tests {
let key = serialize_datum(&Datum::Int(42),
&DataType::Int(IntType::new()));
assert_eq!(key, 42i32.to_le_bytes());
}
+
+ /// Every arm that reads a fixed number of bytes must report a key of the
wrong
+ /// width instead of indexing out of bounds. Each of these used to panic
inside a
+ /// query after `ALTER COLUMN ... TYPE` widened an indexed column.
+ #[test]
+ fn test_comparator_rejects_keys_of_another_type_instead_of_panicking() {
+ use crate::spec::{BigIntType, TimestampType, VarCharType};
+
+ // An index built on INT, read after the column became BIGINT.
+ let big_int =
make_key_comparator(&DataType::BigInt(BigIntType::new()));
+ assert!(big_int(&[0; 4], &[0; 8]).is_err());
+ assert!(big_int(&[0; 8], &[0; 4]).is_err());
+ assert_eq!(big_int(&[0; 8], &[0; 8]).unwrap(), Ordering::Equal);
+
+ // An INT arm handed the 8-byte keys of a column that used to be
BIGINT.
+ let int = make_key_comparator(&DataType::Int(IntType::new()));
+ assert!(int(&[0; 8], &[0; 4]).is_err());
+
+ // An index built on TIMESTAMP(3), read after the column became
TIMESTAMP(6):
+ // the non-compact arm decodes a varint at offset 8.
+ let non_compact =
make_key_comparator(&DataType::Timestamp(TimestampType::new(6).unwrap()));
+ assert!(non_compact(&[0; 8], &[0; 9]).is_err());
+ assert_eq!(non_compact(&[0; 9], &[0; 9]).unwrap(), Ordering::Equal);
+ let compact =
make_key_comparator(&DataType::Timestamp(TimestampType::new(3).unwrap()));
+ assert!(compact(&[0; 9], &[0; 8]).is_err());
+
+ // Non-compact DECIMAL keys are BigInteger bytes: never empty, never
over 16.
+ let decimal =
make_key_comparator(&DataType::Decimal(DecimalType::new(20, 0).unwrap()));
+ assert!(decimal(&[], &[0x01]).is_err());
+ assert!(decimal(&[0; 17], &[0x01]).is_err());
+
+ // Character and byte keys are compared lexicographically, so no width
is wrong.
+ let varchar =
make_key_comparator(&DataType::VarChar(VarCharType::new(10).unwrap()));
+ assert_eq!(varchar(&[], &[0; 3]).unwrap(), Ordering::Less);
+ }
}
diff --git a/crates/paimon/src/btree/meta.rs b/crates/paimon/src/btree/meta.rs
index c8c0bbe8..e23892f5 100644
--- a/crates/paimon/src/btree/meta.rs
+++ b/crates/paimon/src/btree/meta.rs
@@ -24,6 +24,7 @@
//! ```
//! Null key flags distinguish empty serialized keys from absent keys.
+use crate::btree::key_serde::DynKeyComparator;
use crate::spec::PredicateOperator;
use std::cmp::Ordering;
use std::io;
@@ -76,66 +77,102 @@ impl BTreeIndexMeta {
}
/// File-level pruning: check if this BTree file may contain matching keys.
+ ///
+ /// Degradation: pruning predicate. A comparator failure means the stored
keys are
+ /// not keys of the column's current type, so this file's recorded bounds
order
+ /// nothing and prove nothing -- answer "may match" and leave the decision
to the
+ /// read. Turning the failure into "cannot match" would silently drop rows.
pub fn may_match(
&self,
op: PredicateOperator,
serialized_literals: &[Vec<u8>],
- cmp: &dyn Fn(&[u8], &[u8]) -> Ordering,
+ cmp: &DynKeyComparator<'_>,
) -> bool {
- match op {
+ self.try_may_match(op, serialized_literals, cmp)
+ .unwrap_or(true)
+ }
+
+ fn try_may_match(
+ &self,
+ op: PredicateOperator,
+ serialized_literals: &[Vec<u8>],
+ cmp: &DynKeyComparator<'_>,
+ ) -> crate::Result<bool> {
+ Ok(match op {
PredicateOperator::IsNull => self.has_nulls,
PredicateOperator::IsNotNull => !self.only_nulls(),
PredicateOperator::NotEq | PredicateOperator::NotIn => true,
_ => {
if self.only_nulls() {
- return false;
+ return Ok(false);
}
let (first_key, last_key) = match (&self.first_key,
&self.last_key) {
(Some(f), Some(l)) => (f.as_slice(), l.as_slice()),
- _ => return true,
+ _ => return Ok(true),
};
match op {
PredicateOperator::Eq => {
- cmp(&serialized_literals[0], first_key) !=
Ordering::Less
- && cmp(&serialized_literals[0], last_key) !=
Ordering::Greater
+ cmp(&serialized_literals[0], first_key)? !=
Ordering::Less
+ && cmp(&serialized_literals[0], last_key)? !=
Ordering::Greater
}
PredicateOperator::Lt => {
- cmp(first_key, &serialized_literals[0]) ==
Ordering::Less
+ cmp(first_key, &serialized_literals[0])? ==
Ordering::Less
}
PredicateOperator::LtEq => {
- cmp(first_key, &serialized_literals[0]) !=
Ordering::Greater
+ cmp(first_key, &serialized_literals[0])? !=
Ordering::Greater
}
PredicateOperator::Gt => {
- cmp(last_key, &serialized_literals[0]) ==
Ordering::Greater
+ cmp(last_key, &serialized_literals[0])? ==
Ordering::Greater
}
PredicateOperator::GtEq => {
- cmp(last_key, &serialized_literals[0]) !=
Ordering::Less
+ cmp(last_key, &serialized_literals[0])? !=
Ordering::Less
+ }
+ PredicateOperator::In => {
+ let mut any = false;
+ for key in serialized_literals {
+ if cmp(key, first_key)? != Ordering::Less
+ && cmp(key, last_key)? != Ordering::Greater
+ {
+ any = true;
+ break;
+ }
+ }
+ any
}
- PredicateOperator::In =>
serialized_literals.iter().any(|key| {
- cmp(key, first_key) != Ordering::Less
- && cmp(key, last_key) != Ordering::Greater
- }),
_ => true,
}
}
- }
+ })
}
/// File-level pruning for between: file may match if [first_key,
last_key] overlaps [from, to].
+ ///
+ /// Degradation: pruning predicate, for the same reason as
[`Self::may_match`].
pub fn may_match_between(
&self,
from_key: &[u8],
to_key: &[u8],
- cmp: &dyn Fn(&[u8], &[u8]) -> Ordering,
+ cmp: &DynKeyComparator<'_>,
) -> bool {
+ self.try_may_match_between(from_key, to_key, cmp)
+ .unwrap_or(true)
+ }
+
+ fn try_may_match_between(
+ &self,
+ from_key: &[u8],
+ to_key: &[u8],
+ cmp: &DynKeyComparator<'_>,
+ ) -> crate::Result<bool> {
if self.only_nulls() {
- return false;
+ return Ok(false);
}
let (first_key, last_key) = match (&self.first_key, &self.last_key) {
(Some(f), Some(l)) => (f.as_slice(), l.as_slice()),
- _ => return true,
+ _ => return Ok(true),
};
- cmp(first_key, to_key) != Ordering::Greater && cmp(last_key, from_key)
!= Ordering::Less
+ Ok(cmp(first_key, to_key)? != Ordering::Greater
+ && cmp(last_key, from_key)? != Ordering::Less)
}
/// Serialize to bytes (compatible with Java
SortedIndexFileMeta.serialize()).
@@ -306,4 +343,33 @@ mod tests {
let error = BTreeIndexMeta::deserialize(&encoded).unwrap_err();
assert_eq!(error.kind(), io::ErrorKind::InvalidData);
}
+
+ /// Pruning must stay conservative when the recorded keys cannot be
compared: the
+ /// index was built before the column's type changed, so its bounds prove
nothing.
+ /// Answering "cannot match" here would silently drop rows.
+ #[test]
+ fn
test_pruning_says_may_match_when_the_comparator_rejects_the_stored_keys() {
+ let meta = BTreeIndexMeta::new(Some(vec![0; 4]), Some(vec![9; 4]),
false);
+ let failing = |_: &[u8], _: &[u8]| {
+ Err(crate::Error::DataInvalid {
+ message: "stored keys are not keys of this type".to_string(),
+ source: None,
+ })
+ };
+
+ for op in [
+ PredicateOperator::Eq,
+ PredicateOperator::Lt,
+ PredicateOperator::LtEq,
+ PredicateOperator::Gt,
+ PredicateOperator::GtEq,
+ PredicateOperator::In,
+ ] {
+ assert!(
+ meta.may_match(op, &[vec![0; 8]], &failing),
+ "{op} must not prune the file"
+ );
+ }
+ assert!(meta.may_match_between(&[0; 8], &[9; 8], &failing));
+ }
}
diff --git a/crates/paimon/src/btree/query.rs b/crates/paimon/src/btree/query.rs
index cb7688c0..b5166b5a 100644
--- a/crates/paimon/src/btree/query.rs
+++ b/crates/paimon/src/btree/query.rs
@@ -42,7 +42,7 @@ pub trait IndexQuery: Send + Sync {
#[async_trait::async_trait]
impl<F> IndexQuery for BTreeIndexReader<F>
where
- F: Fn(&[u8], &[u8]) -> Ordering + Send + Sync,
+ F: Fn(&[u8], &[u8]) -> crate::Result<Ordering> + Send + Sync,
{
async fn query(
&self,
@@ -196,9 +196,14 @@ impl BetweenInfo<'_> {
&serialize_datum(self.from, self.data_type),
&serialize_datum(self.to, self.data_type),
) {
- Ordering::Greater => true,
- Ordering::Equal => !self.from_inclusive || !self.to_inclusive,
- Ordering::Less => false,
+ Ok(Ordering::Greater) => true,
+ Ok(Ordering::Equal) => !self.from_inclusive || !self.to_inclusive,
+ Ok(Ordering::Less) => false,
+ // Degradation: pure optimisation. This only shortcuts a provably
empty
+ // range to "no rows"; without an ordering we cannot prove that,
so say
+ // "not empty" and let the index read decide. Answering "empty" on
an
+ // error would drop rows.
+ Err(_) => false,
}
}
}
@@ -218,11 +223,20 @@ pub(crate) type ExtractBetweenResult<'a> = (
pub(crate) fn extract_between<'a>(
predicates: &[(PredicateOperator, &'a [Datum], &'a DataType)],
) -> ExtractBetweenResult<'a> {
+ // Degradation: pure optimisation. Merging bounds needs an ordering on the
+ // serialized literals; without one, hand every predicate back so each is
+ // evaluated on its own, exactly as when no complete range was found.
+ try_extract_between(predicates).unwrap_or_else(|_| (None,
predicates.to_vec()))
+}
+
+fn try_extract_between<'a>(
+ predicates: &[(PredicateOperator, &'a [Datum], &'a DataType)],
+) -> crate::Result<ExtractBetweenResult<'a>> {
let Some((_, _, data_type)) = predicates.first() else {
- return (None, Vec::new());
+ return Ok((None, Vec::new()));
};
if predicates.len() == 1 && predicates[0].0 != PredicateOperator::Between {
- return (None, predicates.to_vec());
+ return Ok((None, predicates.to_vec()));
}
let cmp = crate::btree::make_key_comparator(data_type);
let mut lower: Option<(&Datum, Vec<u8>, bool)> = None;
@@ -251,7 +265,7 @@ pub(crate) fn extract_between<'a>(
if let Some((value, inclusive)) = candidate {
let key = serialize_datum(value, data_type);
match bound {
- Some((_, existing, current_inclusive)) => match cmp(&key,
existing) {
+ Some((_, existing, current_inclusive)) => match cmp(&key,
existing)? {
Ordering::Equal => *current_inclusive &= inclusive,
order if order == tighter => *bound = Some((value,
key, inclusive)),
_ => {}
@@ -261,7 +275,7 @@ pub(crate) fn extract_between<'a>(
}
}
}
- match (lower, upper) {
+ Ok(match (lower, upper) {
(Some((from, _, from_inclusive)), Some((to, _, to_inclusive))) => (
Some(BetweenInfo {
from,
@@ -273,5 +287,5 @@ pub(crate) fn extract_between<'a>(
remaining,
),
_ => (None, predicates.to_vec()),
- }
+ })
}
diff --git a/crates/paimon/src/btree/reader.rs
b/crates/paimon/src/btree/reader.rs
index 1debf239..4e2f50e4 100644
--- a/crates/paimon/src/btree/reader.rs
+++ b/crates/paimon/src/btree/reader.rs
@@ -26,6 +26,7 @@
use crate::btree::block::{BlockHandle, BlockReader};
use crate::btree::bloom_filter::BloomFilter;
use crate::btree::footer::{BTreeFileFooter, BloomFilterHandle,
BTREE_FOOTER_ENCODED_LENGTH};
+use crate::btree::key_serde::key_comparison_io_error;
use crate::btree::meta::BTreeIndexMeta;
use crate::btree::posting_list;
use crate::btree::sst_file::{read_block_from_bytes, SstFileReader};
@@ -41,8 +42,28 @@ struct LazyBloomFilter {
filter: OnceCell<BloomFilter>,
}
+/// `slice::sort_by` with a fallible comparator. The first failure is reported
and the
+/// resulting order is unspecified, so the caller must not use the slice
afterwards.
+fn try_sort_by<T>(
+ values: &mut [T],
+ mut cmp: impl FnMut(&T, &T) -> io::Result<Ordering>,
+) -> io::Result<()> {
+ let mut failure = None;
+ values.sort_by(|left, right| match cmp(left, right) {
+ Ok(order) => order,
+ Err(error) => {
+ failure.get_or_insert(error);
+ Ordering::Equal
+ }
+ });
+ match failure {
+ Some(error) => Err(error),
+ None => Ok(()),
+ }
+}
+
/// BTree index reader with on-demand async data block loading.
-pub struct BTreeIndexReader<F: Fn(&[u8], &[u8]) -> Ordering> {
+pub struct BTreeIndexReader<F: Fn(&[u8], &[u8]) -> crate::Result<Ordering>> {
reader: Box<dyn FileRead>,
sst_reader: SstFileReader,
null_bitmap: RoaringTreemap,
@@ -53,7 +74,7 @@ pub struct BTreeIndexReader<F: Fn(&[u8], &[u8]) -> Ordering> {
file_version: u32,
}
-impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
+impl<F: Fn(&[u8], &[u8]) -> crate::Result<Ordering>> BTreeIndexReader<F> {
/// Open a BTree index reader from a FileRead and file metadata.
/// Only reads footer, index block, and null bitmap on open.
/// Data blocks are read on demand during queries.
@@ -115,6 +136,13 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
&self.null_bitmap
}
+ /// Compare two keys, reporting a stored key that cannot belong to the
column's
+ /// current type as a distinguishable I/O error so the caller can give up
on the
+ /// index instead of failing the query.
+ fn compare(&self, left: &[u8], right: &[u8]) -> io::Result<Ordering> {
+ (self.key_comparator)(left, right).map_err(key_comparison_io_error)
+ }
+
/// Collect all non-null row ids into a bitmap.
pub async fn all_non_null_rows(&self) -> io::Result<RoaringTreemap> {
if self.min_key.is_none() {
@@ -138,9 +166,9 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
return Ok(RoaringTreemap::new());
};
- let cmp = &self.key_comparator;
+ let cmp = |left: &[u8], right: &[u8]| self.compare(left, right);
let index_block = self.sst_reader.index_block();
- let (_, mut index_iter) = index_block.seek_and_iter(min_key, cmp);
+ let (_, mut index_iter) = index_block.seek_and_iter(min_key, &cmp)?;
let mut result = RoaringTreemap::new();
while let Some((_key, handle_bytes)) = index_iter.next() {
@@ -168,10 +196,10 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F>
{
from_inclusive: bool,
to_inclusive: bool,
) -> io::Result<RoaringTreemap> {
- let cmp = &self.key_comparator;
+ let cmp = |left: &[u8], right: &[u8]| self.compare(left, right);
let mut result = RoaringTreemap::new();
- match cmp(from, to) {
+ match self.compare(from, to)? {
Ordering::Greater => return Ok(result),
Ordering::Equal if !from_inclusive || !to_inclusive => return
Ok(result),
_ => {}
@@ -179,7 +207,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
// Seek in index block to find the first data block that may contain
`from`
let index_block = self.sst_reader.index_block();
- let (_, mut index_iter) = index_block.seek_and_iter(from, cmp);
+ let (_, mut index_iter) = index_block.seek_and_iter(from, &cmp)?;
// First data block: seek within it
let first_block = match index_iter.next() {
@@ -190,7 +218,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
None => return Ok(result),
};
- let (_, seeked) = first_block.seek_and_iter(from, cmp);
+ let (_, seeked) = first_block.seek_and_iter(from, &cmp)?;
let mut offset = seeked.offset;
// Iterate first block from seeked position
@@ -240,16 +268,15 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F>
{
to_inclusive: bool,
result: &mut RoaringTreemap,
) -> io::Result<bool> {
- let cmp = &self.key_comparator;
while *offset < block.data.len() {
let (key, value, next_offset) = block.read_entry_at(*offset);
*offset = next_offset;
- if !from_inclusive && cmp(key, from) == Ordering::Equal {
+ if !from_inclusive && self.compare(key, from)? == Ordering::Equal {
continue;
}
- let diff = cmp(key, to);
+ let diff = self.compare(key, to)?;
if diff == Ordering::Greater || (!to_inclusive && diff ==
Ordering::Equal) {
return Ok(true);
}
@@ -316,16 +343,16 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F>
{
/// Equal query: returns row ids for the given key.
pub async fn query_equal(&self, key: &[u8]) -> io::Result<RoaringTreemap> {
- let cmp = &self.key_comparator;
- if self
- .min_key
- .as_deref()
- .is_none_or(|min| cmp(key, min) == Ordering::Less)
- || self
- .max_key
- .as_deref()
- .is_none_or(|max| cmp(key, max) == Ordering::Greater)
- {
+ let cmp = |left: &[u8], right: &[u8]| self.compare(left, right);
+ let outside_bounds = match self.min_key.as_deref() {
+ None => true,
+ Some(min) if self.compare(key, min)? == Ordering::Less => true,
+ Some(_) => match self.max_key.as_deref() {
+ None => true,
+ Some(max) => self.compare(key, max)? == Ordering::Greater,
+ },
+ };
+ if outside_bounds {
return Ok(RoaringTreemap::new());
}
if !self.bloom_might_contain(key).await? {
@@ -333,13 +360,13 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F>
{
}
let index_block = self.sst_reader.index_block();
- let (_, mut index_iter) = index_block.seek_and_iter(key, cmp);
+ let (_, mut index_iter) = index_block.seek_and_iter(key, &cmp)?;
let Some((_last_key, handle_bytes)) = index_iter.next() else {
return Ok(RoaringTreemap::new());
};
let handle = BlockHandle::decode(handle_bytes)?;
let block = self.read_data_block(&handle).await?;
- let (found, mut entry_iter) = block.seek_and_iter(key, cmp);
+ let (found, mut entry_iter) = block.seek_and_iter(key, &cmp)?;
let mut result = RoaringTreemap::new();
if let (true, Some((_entry_key, value))) = (found, entry_iter.next()) {
posting_list::add_to(value, self.file_version, &mut result)?;
@@ -411,25 +438,39 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F>
{
return Ok(RoaringTreemap::new());
}
- let cmp = &self.key_comparator;
+ let cmp = |left: &[u8], right: &[u8]| self.compare(left, right);
let (Some(min_key), Some(max_key)) = (self.min_key.as_deref(),
self.max_key.as_deref())
else {
return Ok(RoaringTreemap::new());
};
- // Sort, deduplicate, and discard keys outside this file's bounds
before resolving blocks.
- let mut sorted_keys: Vec<&[u8]> = keys.to_vec();
- sorted_keys.sort_by(|a, b| cmp(a, b));
- sorted_keys.dedup_by(|a, b| cmp(a, b) == Ordering::Equal);
- sorted_keys.retain(|key| {
- cmp(key, min_key) != Ordering::Less && cmp(key, max_key) !=
Ordering::Greater
- });
- if sorted_keys.is_empty() {
+ // Discard keys outside this file's bounds, then sort and deduplicate
the rest
+ // before resolving blocks.
+ let mut sorted_keys: Vec<&[u8]> = Vec::with_capacity(keys.len());
+ for key in keys {
+ if self.compare(key, min_key)? != Ordering::Less
+ && self.compare(key, max_key)? != Ordering::Greater
+ {
+ sorted_keys.push(key);
+ }
+ }
+ try_sort_by(&mut sorted_keys, |left, right| self.compare(left,
right))?;
+ let mut unique_keys: Vec<&[u8]> =
Vec::with_capacity(sorted_keys.len());
+ for key in sorted_keys {
+ let duplicate = match unique_keys.last() {
+ Some(previous) => self.compare(previous, key)? ==
Ordering::Equal,
+ None => false,
+ };
+ if !duplicate {
+ unique_keys.push(key);
+ }
+ }
+ if unique_keys.is_empty() {
return Ok(RoaringTreemap::new());
}
- let mut bloom_matches = Vec::with_capacity(sorted_keys.len());
- for key in sorted_keys {
+ let mut bloom_matches = Vec::with_capacity(unique_keys.len());
+ for key in unique_keys {
if self.bloom_might_contain(key).await? {
bloom_matches.push(key);
}
@@ -443,7 +484,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
let index_block = self.sst_reader.index_block();
let mut target_blocks: Vec<(BlockHandle, Vec<&[u8]>)> = Vec::new();
for key in bloom_matches {
- let (_, mut index_iter) = index_block.seek_and_iter(key, cmp);
+ let (_, mut index_iter) = index_block.seek_and_iter(key, &cmp)?;
let Some((_last_key, handle_bytes)) = index_iter.next() else {
break;
};
@@ -463,7 +504,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
for (handle, block_keys) in target_blocks {
let block = self.read_data_block(&handle).await?;
for key in block_keys {
- let (found, mut entry_iter) = block.seek_and_iter(key, cmp);
+ let (found, mut entry_iter) = block.seek_and_iter(key, &cmp)?;
if let (true, Some((_entry_key, value))) = (found,
entry_iter.next()) {
posting_list::add_to(value, self.file_version, &mut
result)?;
}
diff --git a/crates/paimon/src/btree/tests.rs b/crates/paimon/src/btree/tests.rs
index 9e9d6c92..105e9b71 100644
--- a/crates/paimon/src/btree/tests.rs
+++ b/crates/paimon/src/btree/tests.rs
@@ -49,14 +49,14 @@ fn int_key(v: i32) -> Vec<u8> {
v.to_be_bytes().to_vec()
}
-fn int_cmp(a: &[u8], b: &[u8]) -> std::cmp::Ordering {
+fn int_cmp(a: &[u8], b: &[u8]) -> crate::Result<std::cmp::Ordering> {
let a_val = i32::from_be_bytes(a.try_into().unwrap());
let b_val = i32::from_be_bytes(b.try_into().unwrap());
- a_val.cmp(&b_val)
+ Ok(a_val.cmp(&b_val))
}
/// Helper: write entries, finish, then open a reader from the in-memory bytes.
-async fn write_and_open<F: Fn(&[u8], &[u8]) -> std::cmp::Ordering>(
+async fn write_and_open<F: Fn(&[u8], &[u8]) ->
crate::Result<std::cmp::Ordering>>(
buf: &VecFileWrite,
result: &crate::btree::writer::BTreeWriteResult,
cmp: F,
@@ -667,7 +667,7 @@ async fn test_prefix_query() {
}
let result = writer.finish().await.unwrap();
- let reader = write_and_open(&buf, &result, |a: &[u8], b: &[u8]|
a.cmp(b)).await;
+ let reader = write_and_open(&buf, &result, |a: &[u8], b: &[u8]|
Ok(a.cmp(b))).await;
let bm = reader.query_prefix(b"ap").await.unwrap();
assert_eq!(bm.len(), 3);
@@ -709,7 +709,7 @@ async fn test_string_fallback_scan_query() {
}
let result = writer.finish().await.unwrap();
- let reader = write_and_open(&buf, &result, |a: &[u8], b: &[u8]|
a.cmp(b)).await;
+ let reader = write_and_open(&buf, &result, |a: &[u8], b: &[u8]|
Ok(a.cmp(b))).await;
let data_type = DataType::VarChar(VarCharType::string_type());
let ends_with = reader
@@ -874,7 +874,7 @@ async fn test_string_keys() {
}
let result = writer.finish().await.unwrap();
- let reader = write_and_open(&buf, &result, |a, b| a.cmp(b)).await;
+ let reader = write_and_open(&buf, &result, |a: &[u8], b: &[u8]|
Ok(a.cmp(b))).await;
let bm = reader.query_equal(b"apple").await.unwrap();
assert!(bm.contains(0));
@@ -893,7 +893,7 @@ async fn test_string_keys_query() {
}
let result = writer.finish().await.unwrap();
- let reader = write_and_open(&buf, &result, |a, b| a.cmp(b)).await;
+ let reader = write_and_open(&buf, &result, |a: &[u8], b: &[u8]|
Ok(a.cmp(b))).await;
let bm = reader.query_equal(b"cherry").await.unwrap();
assert_eq!(bm.len(), 1);
@@ -1134,10 +1134,10 @@ fn le_int_key(v: i32) -> Vec<u8> {
v.to_le_bytes().to_vec()
}
-fn le_int_cmp(a: &[u8], b: &[u8]) -> std::cmp::Ordering {
+fn le_int_cmp(a: &[u8], b: &[u8]) -> crate::Result<std::cmp::Ordering> {
let a_val = i32::from_le_bytes(a.try_into().unwrap());
let b_val = i32::from_le_bytes(b.try_into().unwrap());
- a_val.cmp(&b_val)
+ Ok(a_val.cmp(&b_val))
}
fn load_testdata(name: &str) -> Vec<u8> {
@@ -1145,7 +1145,7 @@ fn load_testdata(name: &str) -> Vec<u8> {
std::fs::read(&path).unwrap_or_else(|e| panic!("Failed to read {path}:
{e}"))
}
-async fn open_testdata<F: Fn(&[u8], &[u8]) -> std::cmp::Ordering>(
+async fn open_testdata<F: Fn(&[u8], &[u8]) ->
crate::Result<std::cmp::Ordering>>(
name: &str,
meta: &BTreeIndexMeta,
cmp: F,
@@ -1218,7 +1218,12 @@ async fn test_java_compat_int_with_nulls() {
#[tokio::test]
async fn test_java_compat_varchar_no_compress() {
let meta = BTreeIndexMeta::new(Some(b"a".to_vec()),
Some(b"yyyy".to_vec()), false);
- let reader = open_testdata("btree_varchar_100_no_compress.bin", &meta, |a,
b| a.cmp(b)).await;
+ let reader = open_testdata(
+ "btree_varchar_100_no_compress.bin",
+ &meta,
+ |a: &[u8], b: &[u8]| Ok(a.cmp(b)),
+ )
+ .await;
let all = reader.all_non_null_rows().await.unwrap();
assert_eq!(all.len(), 100);
diff --git a/crates/paimon/src/btree/var_len.rs
b/crates/paimon/src/btree/var_len.rs
index 33e71cb1..c0b312cb 100644
--- a/crates/paimon/src/btree/var_len.rs
+++ b/crates/paimon/src/btree/var_len.rs
@@ -97,6 +97,10 @@ pub fn encode_var_int_to_slice(bytes: &mut [u8], offset:
usize, value: i32) -> u
}
/// Decode var-int from a byte slice, returning (value, bytes_consumed).
+///
+/// Panics on bytes that are not a var-int, so it is only for block-header
bytes the
+/// reader has already validated. Use [`try_decode_var_int_from_slice`] for
bytes whose
+/// shape is not known, such as a key read from an index file.
pub fn decode_var_int_from_slice(bytes: &[u8], offset: usize) -> (i32, usize) {
let mut result: u32 = 0;
let mut i = 0;
@@ -111,6 +115,42 @@ pub fn decode_var_int_from_slice(bytes: &[u8], offset:
usize) -> (i32, usize) {
panic!("Malformed integer");
}
+/// [`decode_var_int_from_slice`] for bytes that may not hold a var-int at all.
+///
+/// A BTree key is read with a comparator built from the column's *current*
type, so a
+/// key written before a type change can be too short for the var-int the new
arm
+/// expects. Running off the end or hitting a fifth continuation byte is then
a property
+/// of the stored data, not a bug, and must be reported instead of panicking.
+pub fn try_decode_var_int_from_slice(bytes: &[u8], offset: usize) ->
crate::Result<(i32, usize)> {
+ let mut result: u32 = 0;
+ let mut i = 0;
+ for shift in (0..32).step_by(7) {
+ let b = offset
+ .checked_add(i)
+ .and_then(|index| bytes.get(index))
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "Var-int at offset {offset} runs past the end of a {}-byte
slice",
+ bytes.len()
+ ),
+ source: None,
+ })?;
+ let b = *b as u32;
+ result |= (b & 0x7F) << shift;
+ i += 1;
+ if (b & 0x80) == 0 {
+ return Ok((result as i32, i));
+ }
+ }
+ Err(crate::Error::DataInvalid {
+ message: format!(
+ "Malformed var-int at offset {offset} of a {}-byte slice",
+ bytes.len()
+ ),
+ source: None,
+ })
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -143,5 +183,21 @@ mod tests {
let (decoded, consumed) = decode_var_int_from_slice(&buf, 0);
assert_eq!(300, decoded);
assert_eq!(written, consumed);
+ assert_eq!(
+ try_decode_var_int_from_slice(&buf, 0).unwrap(),
+ (300, written)
+ );
+ }
+
+ /// The infallible decoder panics on these; a key read from an index file
built
+ /// before the column's type changed reaches the fallible one instead.
+ #[test]
+ fn test_try_decode_var_int_reports_malformed_bytes_instead_of_panicking() {
+ // Offset past the end, and a truncated continuation.
+ assert!(try_decode_var_int_from_slice(&[], 0).is_err());
+ assert!(try_decode_var_int_from_slice(&[0x00], 1).is_err());
+ assert!(try_decode_var_int_from_slice(&[0x80], 0).is_err());
+ // Five continuation bytes never terminate an i32.
+ assert!(try_decode_var_int_from_slice(&[0x80, 0x80, 0x80, 0x80, 0x80],
0).is_err());
}
}
diff --git a/crates/paimon/src/btree/writer.rs
b/crates/paimon/src/btree/writer.rs
index 0ded1392..47708ef7 100644
--- a/crates/paimon/src/btree/writer.rs
+++ b/crates/paimon/src/btree/writer.rs
@@ -23,6 +23,7 @@
use crate::btree::block::{BlockCompressionType, BlockHandle};
use crate::btree::footer::BTreeFileFooter;
+use crate::btree::key_serde::key_comparison_io_error;
use crate::btree::meta::BTreeIndexMeta;
use crate::btree::posting_list;
use crate::btree::sst_file::SstFileWriter;
@@ -38,7 +39,7 @@ const BLOOM_FILTER_FPP: f64 = 0.05;
/// Usage:
/// 1. Call `write(key, row_id)` for each entry (keys must be sorted).
/// 2. Call `finish()` to close the file and get the index meta.
-pub struct BTreeIndexWriter<F: Fn(&[u8], &[u8]) -> Ordering> {
+pub struct BTreeIndexWriter<F: Fn(&[u8], &[u8]) -> crate::Result<Ordering>> {
sst_writer: SstFileWriter,
current_row_ids: Vec<i64>,
last_key: Option<Vec<u8>>,
@@ -57,7 +58,7 @@ pub struct BTreeWriteResult {
pub row_count: u64,
}
-impl BTreeIndexWriter<fn(&[u8], &[u8]) -> Ordering> {
+impl BTreeIndexWriter<fn(&[u8], &[u8]) -> crate::Result<Ordering>> {
pub fn new(
writer: Box<dyn FileWrite>,
block_size: usize,
@@ -84,13 +85,13 @@ impl BTreeIndexWriter<fn(&[u8], &[u8]) -> Ordering> {
first_key: None,
null_bitmap: None,
row_count: 0,
- key_comparator: |a, b| a.cmp(b),
+ key_comparator: |a, b| Ok(a.cmp(b)),
file_version: 1,
}
}
}
-impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexWriter<F> {
+impl<F: Fn(&[u8], &[u8]) -> crate::Result<Ordering>> BTreeIndexWriter<F> {
/// Create a writer with a custom key comparator.
pub fn with_comparator(
writer: Box<dyn FileWrite>,
@@ -179,7 +180,10 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexWriter<F> {
}
Some(k) => {
if let Some(ref last) = self.last_key {
- if (self.key_comparator)(k, last) != Ordering::Equal {
+ // The build side serializes both keys from the column's
current
+ // type, so a failure here is a real defect, not schema
evolution.
+ let order = (self.key_comparator)(k,
last).map_err(key_comparison_io_error)?;
+ if order != Ordering::Equal {
self.flush_row_ids().await?;
}
}
diff --git a/crates/paimon/src/table/bitmap_global_index_format.rs
b/crates/paimon/src/table/bitmap_global_index_format.rs
index 80fc5bec..a06ea995 100644
--- a/crates/paimon/src/table/bitmap_global_index_format.rs
+++ b/crates/paimon/src/table/bitmap_global_index_format.rs
@@ -16,7 +16,7 @@
//! Shared key and wire-format primitives for Java-compatible bitmap indexes.
-use crate::btree::key_serde::KeyComparator;
+use crate::btree::key_serde::{fixed_key_bytes, KeyComparator};
use crate::btree::{make_key_comparator, serialize_datum};
use crate::spec::{DataType, Datum, PredicateOperator};
use std::cmp::Ordering;
@@ -39,14 +39,14 @@ pub(super) struct BlockInfo {
pub(crate) fn make_bitmap_key_comparator(data_type: &DataType) ->
KeyComparator {
match data_type {
DataType::Float(_) => Box::new(|left, right| {
- let left = f32::from_le_bytes(left[..4].try_into().unwrap());
- let right = f32::from_le_bytes(right[..4].try_into().unwrap());
- compare_float_like_java(left, right)
+ let left = f32::from_le_bytes(fixed_key_bytes::<4>(left,
"FLOAT")?);
+ let right = f32::from_le_bytes(fixed_key_bytes::<4>(right,
"FLOAT")?);
+ Ok(compare_float_like_java(left, right))
}),
DataType::Double(_) => Box::new(|left, right| {
- let left = f64::from_le_bytes(left[..8].try_into().unwrap());
- let right = f64::from_le_bytes(right[..8].try_into().unwrap());
- compare_double_like_java(left, right)
+ let left = f64::from_le_bytes(fixed_key_bytes::<8>(left,
"DOUBLE")?);
+ let right = f64::from_le_bytes(fixed_key_bytes::<8>(right,
"DOUBLE")?);
+ Ok(compare_double_like_java(left, right))
}),
_ => make_key_comparator(data_type),
}
diff --git a/crates/paimon/src/table/bitmap_global_index_reader.rs
b/crates/paimon/src/table/bitmap_global_index_reader.rs
index 73e4c6ad..24d43f61 100644
--- a/crates/paimon/src/table/bitmap_global_index_reader.rs
+++ b/crates/paimon/src/table/bitmap_global_index_reader.rs
@@ -25,6 +25,7 @@ use super::bitmap_global_index_format::{
};
#[cfg(test)]
use super::bitmap_global_index_writer::BitmapGlobalIndexWriter;
+use crate::btree::key_serde::key_comparison_io_error;
use crate::btree::var_len::{decode_var_int, decode_var_long};
use crate::btree::{compute_crc32, decompress_block, BlockCompressionType};
use crate::io::FileRead;
@@ -62,6 +63,21 @@ pub(crate) struct BitmapGlobalIndexReader {
dictionary_block_cache: Mutex<HashMap<BlockInfo,
Arc<Vec<DictionaryEntry>>>>,
}
+/// A key comparator for this reader's `io::Result` signatures.
+///
+/// A dictionary key was written with the type the column had when the index
was built,
+/// so comparing it against a literal serialized from the column's current
type can fail.
+/// The failure is wrapped so the caller can tell it from a real I/O failure
and give up
+/// on the index rather than failing the query.
+type IoKeyComparator<'a> = dyn Fn(&[u8], &[u8]) -> io::Result<Ordering> + Send
+ Sync + 'a;
+
+fn io_key_comparator(
+ data_type: &DataType,
+) -> impl Fn(&[u8], &[u8]) -> io::Result<Ordering> + Send + Sync {
+ let cmp = make_bitmap_key_comparator(data_type);
+ move |left, right| cmp(left, right).map_err(key_comparison_io_error)
+}
+
impl BitmapGlobalIndexReader {
pub(crate) async fn open(reader: Box<dyn FileRead>, file_size: u64) ->
io::Result<Self> {
let footer = read_footer(reader.as_ref(), file_size).await?;
@@ -146,29 +162,39 @@ impl BitmapGlobalIndexReader {
PredicateOperator::IsNotNull => self.is_not_null().await,
PredicateOperator::Lt => {
let key = serialize_bitmap_datum(&literals[0], data_type);
- self.scan_dictionary(data_type, |candidate, cmp|
cmp(candidate, &key).is_lt())
- .await
+ self.scan_dictionary(
+ data_type,
+ |candidate, cmp| Ok(cmp(candidate, &key)?.is_lt()),
+ )
+ .await
}
PredicateOperator::LtEq => {
let key = serialize_bitmap_datum(&literals[0], data_type);
- self.scan_dictionary(data_type, |candidate, cmp|
!cmp(candidate, &key).is_gt())
- .await
+ self.scan_dictionary(data_type, |candidate, cmp| {
+ Ok(!cmp(candidate, &key)?.is_gt())
+ })
+ .await
}
PredicateOperator::Gt => {
let key = serialize_bitmap_datum(&literals[0], data_type);
- self.scan_dictionary(data_type, |candidate, cmp|
cmp(candidate, &key).is_gt())
- .await
+ self.scan_dictionary(
+ data_type,
+ |candidate, cmp| Ok(cmp(candidate, &key)?.is_gt()),
+ )
+ .await
}
PredicateOperator::GtEq => {
let key = serialize_bitmap_datum(&literals[0], data_type);
- self.scan_dictionary(data_type, |candidate, cmp|
!cmp(candidate, &key).is_lt())
- .await
+ self.scan_dictionary(data_type, |candidate, cmp| {
+ Ok(!cmp(candidate, &key)?.is_lt())
+ })
+ .await
}
PredicateOperator::Between => {
let from = serialize_bitmap_datum(&literals[0], data_type);
let to = serialize_bitmap_datum(&literals[1], data_type);
self.scan_dictionary(data_type, |candidate, cmp| {
- !cmp(candidate, &from).is_lt() && !cmp(candidate,
&to).is_gt()
+ Ok(!cmp(candidate, &from)?.is_lt() && !cmp(candidate,
&to)?.is_gt())
})
.await
}
@@ -178,7 +204,7 @@ impl BitmapGlobalIndexReader {
let to = serialize_bitmap_datum(&literals[1], data_type);
let inside = self
.scan_dictionary(data_type, |candidate, cmp| {
- !cmp(candidate, &from).is_lt() && !cmp(candidate,
&to).is_gt()
+ Ok(!cmp(candidate, &from)?.is_lt() && !cmp(candidate,
&to)?.is_gt())
})
.await?;
result -= inside;
@@ -195,7 +221,7 @@ impl BitmapGlobalIndexReader {
if prefix.is_empty() {
return self.is_not_null().await;
}
- self.scan_serialized_dictionary(|candidate|
candidate.starts_with(&prefix))
+ self.scan_serialized_dictionary(|candidate|
Ok(candidate.starts_with(&prefix)))
.await
}
PredicateOperator::EndsWith => {
@@ -209,7 +235,7 @@ impl BitmapGlobalIndexReader {
if suffix.is_empty() {
return self.is_not_null().await;
}
- self.scan_serialized_dictionary(|candidate|
candidate.ends_with(&suffix))
+ self.scan_serialized_dictionary(|candidate|
Ok(candidate.ends_with(&suffix)))
.await
}
PredicateOperator::Contains => {
@@ -223,7 +249,7 @@ impl BitmapGlobalIndexReader {
if needle.is_empty() {
return self.is_not_null().await;
}
- self.scan_serialized_dictionary(|candidate|
contains_bytes(candidate, &needle))
+ self.scan_serialized_dictionary(|candidate|
Ok(contains_bytes(candidate, &needle)))
.await
}
PredicateOperator::Like => {
@@ -235,7 +261,8 @@ impl BitmapGlobalIndexReader {
}
let pattern = string_literal(literals, op)?.to_string();
self.scan_serialized_dictionary(|candidate| {
- std::str::from_utf8(candidate).is_ok_and(|value|
like_match(value, &pattern))
+ Ok(std::str::from_utf8(candidate)
+ .is_ok_and(|value| like_match(value, &pattern)))
})
.await
}
@@ -254,10 +281,10 @@ impl BitmapGlobalIndexReader {
return self.is_not_null().await;
}
self.scan_dictionary(data_type, |candidate, cmp| {
- let from_cmp = cmp(candidate, from);
- let to_cmp = cmp(candidate, to);
- (from_cmp.is_gt() || (from_inclusive && from_cmp.is_eq()))
- && (to_cmp.is_lt() || (to_inclusive && to_cmp.is_eq()))
+ let from_cmp = cmp(candidate, from)?;
+ let to_cmp = cmp(candidate, to)?;
+ Ok((from_cmp.is_gt() || (from_inclusive && from_cmp.is_eq()))
+ && (to_cmp.is_lt() || (to_inclusive && to_cmp.is_eq())))
})
.await
}
@@ -271,14 +298,14 @@ impl BitmapGlobalIndexReader {
}
async fn equal(&self, key: &[u8], data_type: &DataType) ->
io::Result<RoaringTreemap> {
- let logical_cmp = make_bitmap_key_comparator(data_type);
- self.equal_with_comparator(key, logical_cmp.as_ref()).await
+ let logical_cmp = io_key_comparator(data_type);
+ self.equal_with_comparator(key, &logical_cmp).await
}
async fn equal_with_comparator(
&self,
key: &[u8],
- logical_cmp: &(dyn Fn(&[u8], &[u8]) -> Ordering + Send + Sync),
+ logical_cmp: &IoKeyComparator<'_>,
) -> io::Result<RoaringTreemap> {
match self.find_bitmap_block(key, logical_cmp).await? {
Some(block) => self.read_bitmap(block).await,
@@ -291,12 +318,10 @@ impl BitmapGlobalIndexReader {
sorted_keys.sort();
sorted_keys.dedup();
- let logical_cmp = make_bitmap_key_comparator(data_type);
+ let logical_cmp = io_key_comparator(data_type);
let mut result = RoaringTreemap::new();
for key in sorted_keys {
- result |= self
- .equal_with_comparator(&key, logical_cmp.as_ref())
- .await?;
+ result |= self.equal_with_comparator(&key, &logical_cmp).await?;
}
Ok(result)
}
@@ -304,21 +329,21 @@ impl BitmapGlobalIndexReader {
async fn scan_dictionary(
&self,
data_type: &DataType,
- predicate: impl Fn(&[u8], &dyn Fn(&[u8], &[u8]) -> Ordering) -> bool,
+ predicate: impl Fn(&[u8], &IoKeyComparator<'_>) -> io::Result<bool>,
) -> io::Result<RoaringTreemap> {
- let cmp = make_bitmap_key_comparator(data_type);
- self.scan_serialized_dictionary(|candidate| predicate(candidate,
cmp.as_ref()))
+ let cmp = io_key_comparator(data_type);
+ self.scan_serialized_dictionary(|candidate| predicate(candidate, &cmp))
.await
}
async fn scan_serialized_dictionary(
&self,
- predicate: impl Fn(&[u8]) -> bool,
+ predicate: impl Fn(&[u8]) -> io::Result<bool>,
) -> io::Result<RoaringTreemap> {
let mut result = RoaringTreemap::new();
for block_meta in &self.dictionary_blocks {
for entry in
self.read_dictionary_block(block_meta.block).await?.iter() {
- if predicate(&entry.key) {
+ if predicate(&entry.key)? {
result |= self.read_bitmap(entry.bitmap_block).await?;
}
}
@@ -329,13 +354,13 @@ impl BitmapGlobalIndexReader {
async fn find_bitmap_block(
&self,
key: &[u8],
- logical_cmp: &(dyn Fn(&[u8], &[u8]) -> Ordering + Send + Sync),
+ logical_cmp: &IoKeyComparator<'_>,
) -> io::Result<Option<BlockInfo>> {
- let Some(block_meta) = self.find_dictionary_block_meta(key,
logical_cmp) else {
+ let Some(block_meta) = self.find_dictionary_block_meta(key,
logical_cmp)? else {
return Ok(None);
};
for entry in
self.read_dictionary_block(block_meta.block).await?.iter() {
- match logical_cmp(&entry.key, key) {
+ match logical_cmp(&entry.key, key)? {
Ordering::Equal => return Ok(Some(entry.bitmap_block)),
Ordering::Greater => return Ok(None),
Ordering::Less => {}
@@ -347,23 +372,24 @@ impl BitmapGlobalIndexReader {
fn find_dictionary_block_meta(
&self,
key: &[u8],
- compare: impl Fn(&[u8], &[u8]) -> Ordering,
- ) -> Option<&DictionaryBlockMeta> {
+ compare: impl Fn(&[u8], &[u8]) -> io::Result<Ordering>,
+ ) -> io::Result<Option<&DictionaryBlockMeta>> {
if self.dictionary_blocks.is_empty() {
- return None;
+ return Ok(None);
}
let mut low = 0usize;
let mut high = self.dictionary_blocks.len();
while low < high {
let mid = (low + high) / 2;
- if compare(&self.dictionary_blocks[mid].first_key, key) !=
Ordering::Greater {
+ if compare(&self.dictionary_blocks[mid].first_key, key)? !=
Ordering::Greater {
low = mid + 1;
} else {
high = mid;
}
}
- low.checked_sub(1)
- .and_then(|index| self.dictionary_blocks.get(index))
+ Ok(low
+ .checked_sub(1)
+ .and_then(|index| self.dictionary_blocks.get(index)))
}
async fn read_dictionary_block(
@@ -1006,9 +1032,10 @@ mod tests {
];
for (op, literals, expected) in cases {
let key = serialize_bitmap_datum(&literals[0], &data_type);
- let cmp = make_bitmap_key_comparator(&data_type);
+ let cmp = io_key_comparator(&data_type);
let expected_block = reader
- .find_dictionary_block_meta(&key, cmp.as_ref())
+ .find_dictionary_block_meta(&key, cmp)
+ .unwrap()
.unwrap()
.block;
let was_cached = reader
diff --git a/crates/paimon/src/table/bitmap_global_index_writer.rs
b/crates/paimon/src/table/bitmap_global_index_writer.rs
index 9283c200..a235c4dc 100644
--- a/crates/paimon/src/table/bitmap_global_index_writer.rs
+++ b/crates/paimon/src/table/bitmap_global_index_writer.rs
@@ -17,6 +17,7 @@
//! Writer for Java Paimon's `BitmapGlobalIndexFormat`.
use super::bitmap_global_index_format::{BlockInfo, MAGIC, VERSION};
+use crate::btree::key_serde::key_comparison_io_error;
use crate::btree::var_len::{encode_var_int, encode_var_long};
use crate::btree::{compress_block, compute_crc32, BTreeIndexMeta,
BlockCompressionType};
use crate::io::FileWrite;
@@ -32,7 +33,7 @@ pub(crate) struct BitmapWriteResult {
pub(crate) row_count: u64,
}
-pub(crate) struct BitmapGlobalIndexWriter<F: Fn(&[u8], &[u8]) -> Ordering> {
+pub(crate) struct BitmapGlobalIndexWriter<F: Fn(&[u8], &[u8]) ->
crate::Result<Ordering>> {
writer: Box<dyn FileWrite>,
dictionary_block_size: usize,
compression_type: BlockCompressionType,
@@ -46,7 +47,7 @@ pub(crate) struct BitmapGlobalIndexWriter<F: Fn(&[u8], &[u8])
-> Ordering> {
row_count: u64,
}
-impl<F: Fn(&[u8], &[u8]) -> Ordering> BitmapGlobalIndexWriter<F> {
+impl<F: Fn(&[u8], &[u8]) -> crate::Result<Ordering>>
BitmapGlobalIndexWriter<F> {
#[cfg(test)]
pub(crate) fn new(
writer: Box<dyn FileWrite>,
@@ -99,7 +100,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering>
BitmapGlobalIndexWriter<F> {
let row_id = relative_row_id as u64;
self.non_null_rows.insert(row_id);
self.bitmaps.entry(key.to_vec()).or_default().insert(row_id);
- self.update_min_max(key);
+ self.update_min_max(key)?;
}
None => {
self.null_rows.insert(relative_row_id as u64);
@@ -119,7 +120,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering>
BitmapGlobalIndexWriter<F> {
}
let row_id = relative_row_id as u64;
self.bitmaps.entry(key.to_vec()).or_default().insert(row_id);
- self.update_min_max(key);
+ self.update_min_max(key)?;
Ok(())
}
@@ -141,7 +142,21 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering>
BitmapGlobalIndexWriter<F> {
let mut bitmaps = std::mem::take(&mut self.bitmaps)
.into_iter()
.collect::<Vec<_>>();
- bitmaps.sort_by(|(left, _), (right, _)| (self.key_comparator)(left,
right));
+ // The build side serializes every key from the column's current type,
so a
+ // comparison failure here is a real defect rather than schema
evolution.
+ let mut failure = None;
+ bitmaps.sort_by(
+ |(left, _), (right, _)| match (self.key_comparator)(left, right) {
+ Ok(order) => order,
+ Err(error) => {
+ failure.get_or_insert(error);
+ Ordering::Equal
+ }
+ },
+ );
+ if let Some(error) = failure {
+ return Err(key_comparison_io_error(error));
+ }
let mut bytes = Vec::new();
write_bitmap_index_bytes(
@@ -168,21 +183,26 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering>
BitmapGlobalIndexWriter<F> {
})
}
- fn update_min_max(&mut self, key: &[u8]) {
- if self
- .first_key
- .as_ref()
- .is_none_or(|existing| (self.key_comparator)(key,
existing).is_lt())
- {
+ fn update_min_max(&mut self, key: &[u8]) -> io::Result<()> {
+ let replaces_first = match &self.first_key {
+ None => true,
+ Some(existing) => (self.key_comparator)(key, existing)
+ .map_err(key_comparison_io_error)?
+ .is_lt(),
+ };
+ if replaces_first {
self.first_key = Some(key.to_vec());
}
- if self
- .last_key
- .as_ref()
- .is_none_or(|existing| (self.key_comparator)(key,
existing).is_gt())
- {
+ let replaces_last = match &self.last_key {
+ None => true,
+ Some(existing) => (self.key_comparator)(key, existing)
+ .map_err(key_comparison_io_error)?
+ .is_gt(),
+ };
+ if replaces_last {
self.last_key = Some(key.to_vec());
}
+ Ok(())
}
}
diff --git a/crates/paimon/src/table/global_index_scanner.rs
b/crates/paimon/src/table/global_index_scanner.rs
index 5fbeffa6..8c721b8c 100644
--- a/crates/paimon/src/table/global_index_scanner.rs
+++ b/crates/paimon/src/table/global_index_scanner.rs
@@ -54,7 +54,7 @@ use tokio::sync::Semaphore;
#[cfg(test)]
use std::sync::atomic::{AtomicUsize as TestAtomicUsize, Ordering as
TestOrdering};
-type BoxedCmp = Box<dyn Fn(&[u8], &[u8]) -> Ordering + Send + Sync>;
+type BoxedCmp = Box<dyn Fn(&[u8], &[u8]) -> Result<Ordering> + Send + Sync>;
const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS";
diff --git a/crates/paimon/src/table/global_index_scanner/all_match.rs
b/crates/paimon/src/table/global_index_scanner/all_match.rs
index 57da267b..1b60fa73 100644
--- a/crates/paimon/src/table/global_index_scanner/all_match.rs
+++ b/crates/paimon/src/table/global_index_scanner/all_match.rs
@@ -18,7 +18,8 @@
//! Prove complete scalar BTree domains match without opening index files.
use super::entry::{sorted_entry_meta, GlobalIndexEntry, GlobalIndexFileKind};
-use crate::btree::{make_key_comparator, serialize_datum};
+use crate::btree::key_serde::KeyComparator;
+use crate::btree::{make_key_comparator, serialize_datum, BTreeIndexMeta};
use crate::spec::{DataType, Datum, PredicateOperator};
use std::cmp::Ordering::{Equal, Greater, Less};
use std::collections::HashMap;
@@ -69,28 +70,12 @@ pub(super) fn all_matching_entries(
}
for (predicate_index, (op, values, cmp)) in
comparisons.iter().enumerate() {
let all_match = files.iter().all(|&index| {
- let meta = sorted_entry_meta(entries[index]);
- let (Some(first), Some(last)) = (&meta.first_key,
&meta.last_key) else {
- return false;
- };
- !meta.has_nulls && {
- if cmp(first, last) == Greater {
- return false;
- }
- match (op, values.as_slice()) {
- (PredicateOperator::Eq, [value]) => {
- cmp(first, value) == Equal && cmp(last, value) ==
Equal
- }
- (PredicateOperator::Lt, [value]) => cmp(last, value)
== Less,
- (PredicateOperator::LtEq, [value]) => cmp(last, value)
!= Greater,
- (PredicateOperator::Gt, [value]) => cmp(first, value)
== Greater,
- (PredicateOperator::GtEq, [value]) => cmp(first,
value) != Less,
- (PredicateOperator::Between, [from, to]) => {
- cmp(first, from) != Less && cmp(last, to) !=
Greater
- }
- _ => false,
- }
- }
+ // Degradation: pure optimisation. This only proves that a
file's whole
+ // row range matches so the posting list need not be decoded.
Without an
+ // ordering nothing is proven, so the entry is simply not an
all-match
+ // and the ordinary query path handles it.
+ entry_all_matches(sorted_entry_meta(entries[index]), *op,
values, cmp)
+ .unwrap_or(false)
});
if all_match {
for &index in &files {
@@ -101,3 +86,30 @@ pub(super) fn all_matching_entries(
}
result
}
+
+fn entry_all_matches(
+ meta: &BTreeIndexMeta,
+ op: PredicateOperator,
+ values: &[Vec<u8>],
+ cmp: &KeyComparator,
+) -> crate::Result<bool> {
+ let (Some(first), Some(last)) = (&meta.first_key, &meta.last_key) else {
+ return Ok(false);
+ };
+ if meta.has_nulls || cmp(first, last)? == Greater {
+ return Ok(false);
+ }
+ Ok(match (op, values) {
+ (PredicateOperator::Eq, [value]) => {
+ cmp(first, value)? == Equal && cmp(last, value)? == Equal
+ }
+ (PredicateOperator::Lt, [value]) => cmp(last, value)? == Less,
+ (PredicateOperator::LtEq, [value]) => cmp(last, value)? != Greater,
+ (PredicateOperator::Gt, [value]) => cmp(first, value)? == Greater,
+ (PredicateOperator::GtEq, [value]) => cmp(first, value)? != Less,
+ (PredicateOperator::Between, [from, to]) => {
+ cmp(first, from)? != Less && cmp(last, to)? != Greater
+ }
+ _ => false,
+ })
+}
diff --git a/crates/paimon/src/table/global_index_scanner/entry.rs
b/crates/paimon/src/table/global_index_scanner/entry.rs
index d402f69b..4d21e502 100644
--- a/crates/paimon/src/table/global_index_scanner/entry.rs
+++ b/crates/paimon/src/table/global_index_scanner/entry.rs
@@ -17,12 +17,12 @@
//! Manifest entry parsing and file-level pruning metadata.
+use crate::btree::key_serde::DynKeyComparator;
use crate::btree::BTreeIndexMeta;
use crate::spec::{DataType, PredicateOperator};
use
crate::table::bitmap_global_index_format::is_bitmap_floating_residual_sensitive_op;
use crate::table::index_file_path::IndexFileLocation;
use crate::{Error, Result};
-use std::cmp::Ordering;
use std::collections::HashMap;
/// A resolved global index entry with parsed metadata.
@@ -167,7 +167,7 @@ pub(super) fn bitmap_meta_may_match(
op: PredicateOperator,
data_type: &DataType,
serialized_literals: &[Vec<u8>],
- cmp: &dyn Fn(&[u8], &[u8]) -> Ordering,
+ cmp: &DynKeyComparator<'_>,
) -> bool {
if is_floating_point(data_type) &&
is_bitmap_floating_residual_sensitive_op(op) {
!meta.only_nulls()
@@ -181,7 +181,7 @@ pub(super) fn bitmap_meta_may_match_between(
data_type: &DataType,
from_key: &[u8],
to_key: &[u8],
- cmp: &dyn Fn(&[u8], &[u8]) -> Ordering,
+ cmp: &DynKeyComparator<'_>,
) -> bool {
if is_floating_point(data_type)
&& is_bitmap_floating_residual_sensitive_op(PredicateOperator::Between)
@@ -196,7 +196,7 @@ pub(super) fn multivalue_meta_may_match(
meta: &BTreeIndexMeta,
op: PredicateOperator,
serialized_literals: &[Vec<u8>],
- cmp: &dyn Fn(&[u8], &[u8]) -> Ordering,
+ cmp: &DynKeyComparator<'_>,
) -> bool {
match op {
PredicateOperator::ArrayContains => {
diff --git a/crates/paimon/src/table/global_index_scanner/reader.rs
b/crates/paimon/src/table/global_index_scanner/reader.rs
index 246eeccd..fc011e10 100644
--- a/crates/paimon/src/table/global_index_scanner/reader.rs
+++ b/crates/paimon/src/table/global_index_scanner/reader.rs
@@ -22,6 +22,7 @@ use super::entry::{
};
use super::query_plan::{add_file_size, EntryQueryPlan, EntryQueryResult,
FallbackScanPlan};
use super::{BoxedCmp, GlobalIndexScanner};
+use crate::btree::key_serde::is_key_comparison_failure;
use crate::btree::query::{BetweenInfo, IndexQuery};
use crate::btree::{make_key_comparator, serialize_datum, BTreeIndexMeta,
BTreeIndexReader};
use crate::fm_index::FMGlobalIndexReader;
@@ -129,7 +130,7 @@ impl GlobalIndexScanner {
};
let from_key = serialize_key(between.from, between.data_type);
let to_key = serialize_key(between.to, between.data_type);
- let bitmap = reader
+ let bitmap = match reader
.as_ref()
.expect("reader is opened when between matches")
.range_query(
@@ -140,7 +141,22 @@ impl GlobalIndexScanner {
between.to_inclusive,
)
.await
- .map_err(|error| Self::query_error(entry, error))?;
+ {
+ Ok(bitmap) => bitmap,
+ // Degradation: this predicate is not evaluated by the global
index at
+ // all. The stored keys are not keys of the column's current
type, so
+ // this file cannot answer; declining makes `evaluate_leaf`
return
+ // `Ok(None)` and the predicate falls through to the read
pipeline.
+ // Returning no rows would silently drop rows, and failing the
query
+ // would turn a readable table into an error.
+ Err(error) if is_key_comparison_failure(&error) => {
+ return Ok(EntryQueryResult {
+ bitmap: None,
+ declined: true,
+ })
+ }
+ Err(error) => return Err(Self::query_error(entry, error)),
+ };
file_result = Some(bitmap);
}
@@ -152,13 +168,18 @@ impl GlobalIndexScanner {
.fetch_add(1, super::TestOrdering::SeqCst);
}
let (op, literals, data_type) = &effective_predicates[idx];
- let Some(bitmap) = reader
+ let queried = reader
.as_ref()
.expect("reader is opened when predicates match")
.query(*op, literals, data_type)
- .await
- .map_err(|error| Self::query_error(entry, error))?
- else {
+ .await;
+ let Some(bitmap) = (match queried {
+ Ok(bitmap) => bitmap,
+ // Degradation: this predicate is not evaluated by the global
index at
+ // all, for the same reason as the between query above.
+ Err(error) if is_key_comparison_failure(&error) => None,
+ Err(error) => return Err(Self::query_error(entry, error)),
+ }) else {
return Ok(EntryQueryResult {
bitmap: None,
declined: true,
diff --git a/crates/paimon/src/table/global_index_scanner/tests.rs
b/crates/paimon/src/table/global_index_scanner/tests.rs
index 4bd731ab..fc055c95 100644
--- a/crates/paimon/src/table/global_index_scanner/tests.rs
+++ b/crates/paimon/src/table/global_index_scanner/tests.rs
@@ -1929,12 +1929,12 @@ fn legacy_floating_comparator(data_type: &DataType) ->
BoxedCmp {
DataType::Float(_) => Box::new(|left, right| {
let left = f32::from_le_bytes(left.try_into().unwrap());
let right = f32::from_le_bytes(right.try_into().unwrap());
- left.total_cmp(&right)
+ Ok(left.total_cmp(&right))
}),
DataType::Double(_) => Box::new(|left, right| {
let left = f64::from_le_bytes(left.try_into().unwrap());
let right = f64::from_le_bytes(right.try_into().unwrap());
- left.total_cmp(&right)
+ Ok(left.total_cmp(&right))
}),
_ => unreachable!("legacy floating comparator requires Float or
Double"),
}
@@ -1955,7 +1955,7 @@ async fn assert_legacy_floating_btree(
.collect::<Vec<_>>();
rows.push((zero_key, 3));
let cmp = legacy_floating_comparator(&data_type);
- rows.sort_by(|left, right| cmp(&left.0, &right.0));
+ rows.sort_by(|left, right| cmp(&left.0, &right.0).unwrap());
let expected_first_key = rows.first().unwrap().0.clone();
let expected_last_key = rows.last().unwrap().0.clone();
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 f50799c9..c136f796 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
@@ -20,6 +20,7 @@
use super::planning::SortedGlobalIndexShard;
use super::validation::checked_row_count;
use super::{SerializeKeyFn, SortedIndexKeyRow};
+use crate::btree::key_serde::DynKeyComparator;
use crate::spec::{
extract_datum_from_array, extract_datum_from_arrow, DataField, DataType,
ROW_ID_FIELD_NAME,
};
@@ -328,16 +329,30 @@ pub(super) fn extract_multivalue_index_rows_from_batches(
Ok(rows)
}
+/// Sort the extracted rows with the index's key comparator.
+///
+/// Both keys come from [`super::make_index_key_codec`] for the column's
current type, so
+/// a comparison failure is a real defect rather than an index built before a
type change;
+/// it aborts the build instead of degrading.
pub(super) fn sort_index_rows(
rows: &mut [SortedIndexKeyRow],
- cmp: &dyn Fn(&[u8], &[u8]) -> Ordering,
-) {
+ cmp: &DynKeyComparator<'_>,
+) -> Result<()> {
+ let mut failure = None;
rows.sort_by(|left, right| match (&left.0, &right.0) {
(None, None) => left.1.cmp(&right.1),
(None, Some(_)) => Ordering::Less,
(Some(_), None) => Ordering::Greater,
- (Some(left_key), Some(right_key)) => {
- cmp(left_key, right_key).then_with(|| left.1.cmp(&right.1))
- }
+ (Some(left_key), Some(right_key)) => match cmp(left_key, right_key) {
+ Ok(order) => order.then_with(|| left.1.cmp(&right.1)),
+ Err(error) => {
+ failure.get_or_insert(error);
+ left.1.cmp(&right.1)
+ }
+ },
});
+ match failure {
+ Some(error) => Err(error),
+ None => Ok(()),
+ }
}
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 cace6bec..aa0d0551 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
@@ -475,13 +475,13 @@ fn
test_index_key_codec_scopes_java_nan_semantics_to_bitmap() {
let btree_nan_key = btree_serialize(&negative_nan, &data_type);
let zero_key = btree_serialize(&zero, &data_type);
assert_eq!(btree_nan_key, raw_nan_key);
- assert!(btree_cmp(&btree_nan_key, &zero_key).is_lt());
+ assert!(btree_cmp(&btree_nan_key, &zero_key).unwrap().is_lt());
let (bitmap_cmp, bitmap_serialize) =
make_index_key_codec(BITMAP_GLOBAL_INDEX_TYPE, &data_type);
let bitmap_nan_key = bitmap_serialize(&negative_nan, &data_type);
assert_eq!(bitmap_nan_key, canonical_nan_key);
- assert!(bitmap_cmp(&bitmap_nan_key, &zero_key).is_gt());
+ assert!(bitmap_cmp(&bitmap_nan_key, &zero_key).unwrap().is_gt());
}
assert_codec(
@@ -528,7 +528,7 @@ fn test_sort_index_rows_orders_nulls_then_keys() {
];
let cmp = make_key_comparator(&DataType::Int(IntType::new()));
- sort_index_rows(&mut rows, &cmp);
+ sort_index_rows(&mut rows, &cmp).unwrap();
assert_eq!(
rows,
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 aa71e501..8ef05b2d 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
@@ -81,7 +81,7 @@ impl SortedGlobalIndexBuildBuilder<'_> {
.await?
};
if !rows.is_empty() {
- sort_index_rows(&mut rows, &cmp);
+ sort_index_rows(&mut rows, &cmp)?;
}
self.table
