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 717e3d39 fix(file_index): skip pruning for narrowing integer schema
changes (#806)
717e3d39 is described below
commit 717e3d39ca60cbcdb0c6f3090417813f04eacbc6
Author: QuakeWang <[email protected]>
AuthorDate: Fri Sep 11 14:25:09 2026 +0800
fix(file_index): skip pruning for narrowing integer schema changes (#806)
---
crates/paimon/src/file_index/evaluator.rs | 136 +++++++++++++++++++--
crates/paimon/src/table/data_file_reader.rs | 183 ++++++++++++++++++++++++++++
2 files changed, 312 insertions(+), 7 deletions(-)
diff --git a/crates/paimon/src/file_index/evaluator.rs
b/crates/paimon/src/file_index/evaluator.rs
index ce84aa75..51c5e82a 100644
--- a/crates/paimon/src/file_index/evaluator.rs
+++ b/crates/paimon/src/file_index/evaluator.rs
@@ -186,7 +186,9 @@ fn devolve_literals(
if same_type_ignoring_nullability(table_type, data_type) {
return Some(literals.to_vec());
}
- if !is_integer_type(table_type) || !is_integer_type(data_type) {
+ // Narrowing can turn non-null file values into NULL, so reject it even for
+ // predicates without literals, such as IS NULL and IS NOT NULL.
+ if integer_width(table_type)? <= integer_width(data_type)? {
return None;
}
literals
@@ -208,11 +210,14 @@ fn same_type_ignoring_nullability(left: &DataType, right:
&DataType) -> bool {
}
}
-fn is_integer_type(data_type: &DataType) -> bool {
- matches!(
- data_type,
- DataType::TinyInt(_) | DataType::SmallInt(_) | DataType::Int(_) |
DataType::BigInt(_)
- )
+fn integer_width(data_type: &DataType) -> Option<u8> {
+ match data_type {
+ DataType::TinyInt(_) => Some(8),
+ DataType::SmallInt(_) => Some(16),
+ DataType::Int(_) => Some(32),
+ DataType::BigInt(_) => Some(64),
+ _ => None,
+ }
}
fn integer_value(data_type: &DataType, datum: &Datum) -> Option<i64> {
@@ -262,7 +267,8 @@ mod tests {
use crate::io::FileIOBuilder;
use crate::spec::stats::BinaryTableStats;
use crate::spec::{
- BigIntType, FloatType, IntType, PredicateBuilder, PredicateOperator,
VarCharType,
+ BigIntType, FloatType, IntType, PredicateBuilder, PredicateOperator,
SmallIntType,
+ TinyIntType, VarCharType,
};
fn field(id: i32, name: &str, data_type: DataType) -> DataField {
@@ -337,6 +343,122 @@ mod tests {
));
}
+ #[test]
+ fn test_devolve_literals_preserves_safe_integer_schema_changes() {
+ let types = [
+ (
+ DataType::TinyInt(TinyIntType::new()),
+ i64::from(i8::MIN),
+ i64::from(i8::MAX),
+ ),
+ (
+ DataType::SmallInt(SmallIntType::new()),
+ i64::from(i16::MIN),
+ i64::from(i16::MAX),
+ ),
+ (
+ DataType::Int(IntType::new()),
+ i64::from(i32::MIN),
+ i64::from(i32::MAX),
+ ),
+ (DataType::BigInt(BigIntType::new()), i64::MIN, i64::MAX),
+ ];
+ for (data_index, (data_type, min, max)) in types.iter().enumerate() {
+ let literals = vec![
+ integer_datum(data_type, *min).unwrap(),
+ integer_datum(data_type, *max).unwrap(),
+ ];
+ for table_nullable in [false, true] {
+ for data_nullable in [false, true] {
+ let table_type =
data_type.copy_with_nullable(table_nullable).unwrap();
+ let data_type =
data_type.copy_with_nullable(data_nullable).unwrap();
+ assert_eq!(
+ devolve_literals(&table_type, &data_type, &literals),
+ Some(literals.clone())
+ );
+ assert_eq!(devolve_literals(&table_type, &data_type, &[]),
Some(vec![]));
+ }
+ }
+ for (table_type, _, _) in &types[data_index + 1..] {
+ let table_literals = vec![
+ integer_datum(table_type, *min).unwrap(),
+ integer_datum(table_type, *max).unwrap(),
+ ];
+ assert_eq!(
+ devolve_literals(table_type, data_type, &table_literals),
+ Some(literals.clone())
+ );
+ assert_eq!(devolve_literals(table_type, data_type, &[]),
Some(vec![]));
+ for overflow in [min - 1, max + 1] {
+ let mut mixed_literals = table_literals.clone();
+ mixed_literals.push(integer_datum(table_type,
overflow).unwrap());
+ assert_eq!(
+ devolve_literals(table_type, data_type,
&mixed_literals),
+ None
+ );
+ }
+ }
+ }
+ }
+
+ #[test]
+ fn test_remap_predicate_falls_back_for_narrowing_integer_schema_changes() {
+ let types = [
+ (DataType::TinyInt(TinyIntType::new()), Datum::TinyInt(127)),
+ (
+ DataType::SmallInt(SmallIntType::new()),
+ Datum::SmallInt(127),
+ ),
+ (DataType::Int(IntType::new()), Datum::Int(127)),
+ (DataType::BigInt(BigIntType::new()), Datum::Long(127)),
+ ];
+ for (table_index, (table_type, literal)) in types.iter().enumerate() {
+ let table_fields = vec![field(0, "value", table_type.clone())];
+ let builder = PredicateBuilder::new(&table_fields);
+ for (data_type, _) in &types[table_index + 1..] {
+ let data_fields = vec![field(0, "value", data_type.clone())];
+ for predicate in [
+ builder.equal("value", literal.clone()).unwrap(),
+ builder.is_null("value").unwrap(),
+ builder.is_not_null("value").unwrap(),
+ ] {
+ assert!(
+ remap_predicate(&table_fields, &data_fields,
&predicate).is_none(),
+ "{data_type:?} -> {table_type:?}: {predicate:?} must
fall back"
+ );
+ }
+ }
+ }
+ }
+
+ #[test]
+ fn
test_remap_narrowing_predicate_keeps_safe_and_child_but_rejects_or_and_not() {
+ let table_fields = vec![
+ field(0, "id", DataType::Int(IntType::new())),
+ field(1, "value", DataType::TinyInt(TinyIntType::new())),
+ ];
+ let data_fields = vec![
+ table_fields[0].clone(),
+ field(1, "value", DataType::Int(IntType::new())),
+ ];
+ let builder = PredicateBuilder::new(&table_fields);
+ let safe = builder.equal("id", Datum::Int(1)).unwrap();
+ let narrowing = builder.is_null("value").unwrap();
+ let combined = Predicate::and(vec![safe.clone(), narrowing.clone()]);
+
+ assert_eq!(
+ remap_predicate(&table_fields, &data_fields, &combined),
+ Some(safe.clone())
+ );
+ for predicate in [
+ Predicate::or(vec![safe, narrowing.clone()]),
+ Predicate::negate(narrowing),
+ Predicate::negate(combined),
+ ] {
+ assert!(remap_predicate(&table_fields, &data_fields,
&predicate).is_none());
+ }
+ }
+
#[test]
fn
test_remap_predicate_uses_field_id_for_rename_reorder_and_integer_devolution() {
let table_fields = vec![
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index e6221cc0..667b681e 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -2437,6 +2437,189 @@ mod tests {
assert!(batches.is_empty());
}
+ #[tokio::test]
+ async fn
test_file_index_narrowing_integer_schema_changes_preserve_query_results() {
+ use std::collections::HashMap;
+
+ use apache_avro::types::Value;
+ use arrow_array::Int8Array;
+
+ use crate::catalog::Identifier;
+ use crate::spec::TinyIntType;
+ use crate::table::{Table, TableRead};
+
+ let old_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("value", DataType::Int(IntType::new()))
+ .build()
+ .unwrap(),
+ );
+ let current_schema = old_schema
+ .apply_changes(vec![SchemaChange::update_column_type(
+ "value".to_string(),
+ DataType::TinyInt(TinyIntType::new()),
+ )])
+ .unwrap();
+ let avro_schema = apache_avro::Schema::parse_str(
+
r#"{"type":"record","name":"row","fields":[{"name":"value","type":["null","int"]}]}"#,
+ )
+ .unwrap();
+ for (case, values, expected_values) in [
+ ("overflow_only", vec![Some(383)], vec![None]),
+ (
+ "mixed",
+ vec![Some(127), Some(383), None],
+ vec![Some(127), None, None],
+ ),
+ ] {
+ let mut avro_writer = apache_avro::Writer::new(&avro_schema,
Vec::new());
+ let mut index_writer = FileIndexerFactory::create_writer(
+ BITMAP_INDEX,
+ old_schema.fields()[0].data_type().clone(),
+ &Options::new(),
+ )
+ .unwrap();
+ for value in &values {
+ let (tag, avro_value) = match value {
+ Some(value) => (1, Value::Int(*value)),
+ None => (0, Value::Null),
+ };
+ avro_writer
+ .append(Value::Record(vec![(
+ "value".to_string(),
+ Value::Union(tag, Box::new(avro_value)),
+ )]))
+ .unwrap();
+ index_writer.write(value.map(Datum::Int).as_ref()).unwrap();
+ }
+ let data = Bytes::from(avro_writer.into_inner().unwrap());
+ let indexes = HashMap::from([(
+ "value".to_string(),
+ HashMap::from([(
+ BITMAP_INDEX.to_string(),
+ Some(index_writer.serialized_bytes().unwrap()),
+ )]),
+ )]);
+ let index =
write_column_indexes(&format!("memory:/narrowing_{case}_index"), indexes)
+ .await
+ .unwrap()
+ .to_input_file()
+ .read()
+ .await
+ .unwrap();
+
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table_path = format!("memory:/file_index_narrowing_{case}");
+ let bucket_path = format!("{table_path}/bucket-0");
+ let file_name = "part-0.avro";
+ file_io
+ .new_output(&format!("{bucket_path}/{file_name}"))
+ .unwrap()
+ .write(data.clone())
+ .await
+ .unwrap();
+ let schema_manager = SchemaManager::new(file_io.clone(),
table_path.clone());
+ let schema_path = schema_manager.schema_path(old_schema.id());
+ file_io
+ .mkdirs(schema_path.rsplit_once('/').unwrap().0)
+ .await
+ .unwrap();
+ file_io
+ .new_output(&schema_path)
+ .unwrap()
+ .write(Bytes::from(serde_json::to_vec(&old_schema).unwrap()))
+ .await
+ .unwrap();
+ let mut file = data_file(
+ file_name,
+ data.len() as i64,
+ values.len() as i64,
+ old_schema.id(),
+ );
+ file.embedded_index = Some(index.to_vec());
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(crate::spec::BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(bucket_path)
+ .with_total_buckets(1)
+ .with_data_files(vec![file])
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "narrowing"),
+ table_path,
+ current_schema.clone(),
+ None,
+ );
+ let builder = PredicateBuilder::new(current_schema.fields());
+ for (query, predicate, expected) in [
+ ("all", Predicate::AlwaysTrue, expected_values.clone()),
+ (
+ "IS NULL",
+ builder.is_null("value").unwrap(),
+ expected_values
+ .iter()
+ .copied()
+ .filter(Option::is_none)
+ .collect(),
+ ),
+ (
+ "IS NOT NULL",
+ builder.is_not_null("value").unwrap(),
+ expected_values
+ .iter()
+ .copied()
+ .filter(Option::is_some)
+ .collect(),
+ ),
+ (
+ "= 127",
+ builder.equal("value", Datum::TinyInt(127)).unwrap(),
+ expected_values
+ .iter()
+ .copied()
+ .filter(|value| *value == Some(127))
+ .collect(),
+ ),
+ ] {
+ for enabled in [false, true] {
+ let table = table.copy_with_options(HashMap::from([(
+ "file-index.read.enabled".to_string(),
+ enabled.to_string(),
+ )]));
+ let batches = TableRead::new(
+ &table,
+ current_schema.fields().to_vec(),
+ vec![predicate.clone()],
+ )
+ .to_arrow(std::slice::from_ref(&split))
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let actual = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int8Array>()
+ .unwrap()
+ .iter()
+ })
+ .collect::<Vec<_>>();
+ assert_eq!(
+ actual, expected,
+ "case={case}, query={query}, enabled={enabled}"
+ );
+ }
+ }
+ }
+ }
+
#[tokio::test]
async fn test_file_index_nested_not_with_added_column_falls_back() {
let old_schema = TableSchema::new(
