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 8b18dda8 feat: add Java-compatible array predicate pushdown (#732)
8b18dda8 is described below
commit 8b18dda8d4fc44ba9d54360bbb93b571b8970ddf
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Aug 19 19:05:26 2026 +0800
feat: add Java-compatible array predicate pushdown (#732)
---
Cargo.lock | 1 +
.../integrations/datafusion/src/filter_pushdown.rs | 548 ++++++++++++++++++++-
crates/paimon/Cargo.toml | 1 +
crates/paimon/src/arrow/format/orc.rs | 5 +-
crates/paimon/src/arrow/format/parquet.rs | 24 +
crates/paimon/src/arrow/residual.rs | 405 ++++++++++++++-
crates/paimon/src/btree/query.rs | 6 +
crates/paimon/src/predicate_stats.rs | 38 +-
crates/paimon/src/spec/binary_row.rs | 34 ++
crates/paimon/src/spec/predicate.rs | 266 +++++++++-
.../paimon/src/table/bitmap_global_index_reader.rs | 146 +++++-
11 files changed, 1434 insertions(+), 40 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 02d2665a..627ac1d4 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4587,6 +4587,7 @@ dependencies = [
"log",
"lru 0.18.2",
"lz4_flex 0.13.1",
+ "lzokay-native",
"md-5 0.10.6",
"opendal-core",
"opendal-http-transport-reqwest",
diff --git a/crates/integrations/datafusion/src/filter_pushdown.rs
b/crates/integrations/datafusion/src/filter_pushdown.rs
index 33fd2aca..cdbe6460 100644
--- a/crates/integrations/datafusion/src/filter_pushdown.rs
+++ b/crates/integrations/datafusion/src/filter_pushdown.rs
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+use datafusion::arrow::array::Array;
+use datafusion::arrow::datatypes::DataType as ArrowDataType;
use datafusion::common::{Column, ScalarValue};
use datafusion::logical_expr::expr::{InList, ScalarFunction};
use datafusion::logical_expr::{
@@ -168,6 +170,13 @@ impl<'a> FilterTranslator<'a> {
// must keep its residual filter for NULL / three-valued semantics.
Expr::Not(inner) => {
let inner = self.translate(inner.as_ref())?;
+ // A positive inexact predicate is only guaranteed to be a
+ // conservative superset. Negating it would turn that into a
+ // subset and could remove rows before DataFusion's residual
+ // runs (notably for floating-array NaN payload semantics).
+ if inner.requires_residual {
+ return None;
+ }
Some(TranslatedPredicate {
predicate: Predicate::negate(inner.predicate),
requires_residual: true,
@@ -311,6 +320,18 @@ impl<'a> FilterTranslator<'a> {
}
fn translate_scalar_function(&self, func: &ScalarFunction) ->
Option<TranslatedPredicate> {
+ if matches!(
+ func.name(),
+ "array_has"
+ | "list_has"
+ | "array_has_any"
+ | "list_has_any"
+ | "arrays_overlap"
+ | "array_has_all"
+ | "list_has_all"
+ ) {
+ return self.translate_array_function(func);
+ }
// DataFusion built-in UDFs surfaced from `LIKE 'x%' / '%x' / '%x%'`
// rewrites and direct `starts_with(col, 'x') / ends_with / contains`
// calls. Only `(col, literal)` shapes are handled; anything else
@@ -334,6 +355,104 @@ impl<'a> FilterTranslator<'a> {
self.exact(predicate)
}
+ fn translate_array_function(&self, func: &ScalarFunction) ->
Option<TranslatedPredicate> {
+ if func.args.len() != 2 {
+ return None;
+ }
+ let (field, comparison_element_type) =
self.resolve_array_field(&func.args[0])?;
+ let DataType::Array(array_type) = field.data_type() else {
+ return None;
+ };
+ let predicate = match func.name() {
+ "array_has" | "list_has" => {
+ let literal = extract_array_scalar_literal(
+ &func.args[1],
+ array_type.element_type(),
+ &comparison_element_type,
+ )?;
+ self.predicate_builder
+ .array_contains(field.name(), literal)
+ .ok()?
+ }
+ "array_has_any" | "list_has_any" | "arrays_overlap" => {
+ let literals = extract_array_literals(
+ &func.args[1],
+ array_type.element_type(),
+ &comparison_element_type,
+ )?;
+ self.predicate_builder
+ .arrays_overlap(field.name(), literals)
+ .ok()?
+ }
+ "array_has_all" | "list_has_all" => {
+ let literals = extract_array_literals(
+ &func.args[1],
+ array_type.element_type(),
+ &comparison_element_type,
+ )?;
+ // DataFusion 54's empty-needle fast path currently returns
true
+ // even for a NULL haystack, while Paimon/Java
ARRAY_CONTAINS_ALL
+ // returns false for NULL arrays. Pushing it would remove rows
+ // before DataFusion can apply its own semantics.
+ if literals.is_empty() {
+ return None;
+ }
+ self.predicate_builder
+ .array_contains_all(field.name(), literals)
+ .ok()?
+ }
+ _ => return None,
+ };
+ Some(TranslatedPredicate {
+ predicate,
+ // Paimon's core residual follows Java Float.compare /
Double.compare
+ // and canonicalizes all NaNs. DataFusion's Arrow equality keeps
NaN
+ // payloads distinct, so retain its residual for floating arrays.
+ requires_residual: matches!(
+ array_type.element_type(),
+ DataType::Float(_) | DataType::Double(_)
+ ),
+ })
+ }
+
+ /// Resolve an ARRAY column, accepting only the lossless element-wise casts
+ /// inserted by DataFusion's array function type coercion.
+ fn resolve_array_field(&self, expr: &Expr) -> Option<(&'a DataField,
ArrowDataType)> {
+ match expr {
+ Expr::Column(_) => {
+ let field = self.resolve_field(expr)?;
+ let DataType::Array(array_type) = field.data_type() else {
+ return None;
+ };
+ let comparison_type =
+
paimon::arrow::paimon_type_to_arrow(array_type.element_type()).ok()?;
+ Some((field, comparison_type))
+ }
+ Expr::Cast(cast) => {
+ let field = self.resolve_field(cast.expr.as_ref())?;
+ let DataType::Array(array_type) = field.data_type() else {
+ return None;
+ };
+ // Paimon ARRAY columns are Arrow List values. DataFusion's
+ // numeric coercion keeps that container and only widens its
+ // element. In particular, List -> FixedSizeList is
+ // value-dependent and must not be erased here.
+ let ArrowDataType::List(target_field) = cast.field.data_type()
else {
+ return None;
+ };
+ if array_type.element_type().is_nullable() &&
!target_field.is_nullable() {
+ return None;
+ }
+ let target_element = target_field.data_type();
+ if !is_lossless_array_element_cast(array_type.element_type(),
target_element) {
+ return None;
+ }
+ Some((field, target_element.clone()))
+ }
+ _ => None,
+ }
+ }
+
fn translate_like(&self, like: &Like) -> Option<TranslatedPredicate> {
// ILIKE has no equivalent in Paimon's predicate model.
if like.case_insensitive {
@@ -398,6 +517,114 @@ fn extract_scalar_literal(expr: &Expr) ->
Option<&ScalarValue> {
}
}
+fn extract_array_scalar_literal(
+ expr: &Expr,
+ element_type: &DataType,
+ comparison_element_type: &ArrowDataType,
+) -> Option<Datum> {
+ match expr {
+ Expr::Literal(scalar, _) if !scalar.is_null() => {
+ scalar_to_array_datum(scalar, element_type)
+ }
+ Expr::Cast(cast) if cast.field.data_type() == comparison_element_type
=> {
+ let scalar = extract_scalar_literal(cast.expr.as_ref())?;
+ if !is_lossless_arrow_scalar_cast(&scalar.data_type(),
comparison_element_type) {
+ return None;
+ }
+ scalar_to_array_datum(scalar, element_type)
+ }
+ _ => None,
+ }
+}
+
+fn extract_array_literals(
+ expr: &Expr,
+ element_type: &DataType,
+ comparison_element_type: &ArrowDataType,
+) -> Option<Vec<Datum>> {
+ match expr {
+ Expr::ScalarFunction(function) if matches!(function.name(),
"make_array" | "make_list") => {
+ function
+ .args
+ .iter()
+ .map(|expr| {
+ extract_array_scalar_literal(expr, element_type,
comparison_element_type)
+ })
+ .collect()
+ }
+ Expr::Cast(cast)
+ if arrow_list_element_type(cast.field.data_type()) ==
Some(comparison_element_type) =>
+ {
+ extract_array_literals(cast.expr.as_ref(), element_type,
comparison_element_type)
+ }
+ Expr::Literal(scalar, _) => {
+ let values = match scalar {
+ ScalarValue::List(list) if !list.is_null(0) => list.value(0),
+ ScalarValue::LargeList(list) if !list.is_null(0) =>
list.value(0),
+ ScalarValue::FixedSizeList(list) if !list.is_null(0) =>
list.value(0),
+ _ => return None,
+ };
+ (0..values.len())
+ .map(|index| {
+ let scalar = ScalarValue::try_from_array(values.as_ref(),
index).ok()?;
+ if scalar.is_null() {
+ return None;
+ }
+ scalar_to_array_datum(&scalar, element_type)
+ })
+ .collect()
+ }
+ _ => None,
+ }
+}
+
+fn scalar_to_array_datum(scalar: &ScalarValue, element_type: &DataType) ->
Option<Datum> {
+ if let (DataType::Float(_), ScalarValue::Float64(Some(value))) =
(element_type, scalar) {
+ let narrowed = *value as f32;
+ return ((narrowed as f64).to_bits() ==
value.to_bits()).then_some(Datum::Float(narrowed));
+ }
+ scalar_to_datum(scalar, element_type)
+}
+
+fn arrow_list_element_type(data_type: &ArrowDataType) ->
Option<&ArrowDataType> {
+ match data_type {
+ ArrowDataType::List(field) => Some(field.data_type()),
+ _ => None,
+ }
+}
+
+fn is_lossless_array_element_cast(source: &DataType, target: &ArrowDataType)
-> bool {
+ if paimon::arrow::paimon_type_to_arrow(source).ok().as_ref() ==
Some(target) {
+ return true;
+ }
+ matches!(
+ (source, target),
+ (
+ DataType::TinyInt(_),
+ ArrowDataType::Int16 | ArrowDataType::Int32 | ArrowDataType::Int64
+ ) | (
+ DataType::SmallInt(_),
+ ArrowDataType::Int32 | ArrowDataType::Int64
+ ) | (DataType::Int(_), ArrowDataType::Int64)
+ | (DataType::Float(_), ArrowDataType::Float64)
+ )
+}
+
+fn is_lossless_arrow_scalar_cast(source: &ArrowDataType, target:
&ArrowDataType) -> bool {
+ source == target
+ || matches!(
+ (source, target),
+ (
+ ArrowDataType::Int8,
+ ArrowDataType::Int16 | ArrowDataType::Int32 |
ArrowDataType::Int64
+ ) | (
+ ArrowDataType::Int16,
+ ArrowDataType::Int32 | ArrowDataType::Int64
+ ) | (ArrowDataType::Int32, ArrowDataType::Int64)
+ | (ArrowDataType::Float32, ArrowDataType::Float64)
+ )
+}
+
fn reverse_comparison_operator(op: Operator) -> Option<Operator> {
match op {
Operator::Eq => Some(Operator::Eq),
@@ -546,7 +773,8 @@ mod tests {
use paimon::catalog::Identifier;
use paimon::io::FileIOBuilder;
use paimon::spec::{
- IntType, LocalZonedTimestampType, Schema, TableSchema, TimeType,
TimestampType, VarCharType,
+ ArrayType, BigIntType, FloatType, IntType, LocalZonedTimestampType,
PredicateOperator,
+ Schema, SmallIntType, TableSchema, TimeType, TimestampType,
VarCharType,
};
use paimon::table::Table;
@@ -567,6 +795,10 @@ mod tests {
"lzts_col",
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(9).unwrap()),
)
+ .column(
+ "items",
+
DataType::Array(ArrayType::new(DataType::Int(IntType::new()))),
+ )
.partition_keys(["dt", "hr"])
.build()
.unwrap(),
@@ -603,6 +835,320 @@ mod tests {
}
}
+ #[test]
+ fn test_translate_datafusion_array_membership_functions() {
+ use datafusion::functions_nested::expr_fn::{
+ array_has, array_has_all, array_has_any, make_array,
+ };
+
+ let column = Expr::Column(Column::from_name("items"));
+ let cases = [
+ (
+ array_has(column.clone(), lit(2)),
+ PredicateOperator::ArrayContains,
+ vec![Datum::Int(2)],
+ ),
+ (
+ array_has_any(column.clone(), make_array(vec![lit(1),
lit(3)])),
+ PredicateOperator::ArraysOverlap,
+ vec![Datum::Int(1), Datum::Int(3)],
+ ),
+ (
+ array_has_all(column, make_array(vec![lit(2), lit(2),
lit(4)])),
+ PredicateOperator::ArrayContainsAll,
+ vec![Datum::Int(2), Datum::Int(2), Datum::Int(4)],
+ ),
+ ];
+
+ let fields = test_fields();
+ for (filter, expected_op, expected_literals) in cases {
+ let predicate = build_pushed_predicate(&[filter], &fields)
+ .expect("array membership function should translate");
+ assert!(matches!(
+ predicate,
+ Predicate::Leaf { op, literals, .. }
+ if op == expected_op && literals == expected_literals
+ ));
+ }
+ }
+
+ #[test]
+ fn test_empty_array_has_all_falls_open_for_datafusion_null_semantics() {
+ use datafusion::functions_nested::expr_fn::{array_has_all, make_array};
+
+ let filter = array_has_all(
+ Expr::Column(Column::from_name("items")),
+ make_array(Vec::<Expr>::new()),
+ );
+ let fields = test_fields();
+
+ assert!(build_pushed_predicate(std::slice::from_ref(&filter),
&fields).is_none());
+ assert_eq!(
+ classify_filter_pushdown(&filter, &fields, true,
is_exact_filter_pushdown),
+ TableProviderFilterPushDown::Unsupported
+ );
+ }
+
+ #[test]
+ fn test_float_array_membership_keeps_datafusion_residual() {
+ use datafusion::functions_nested::expr_fn::array_has;
+
+ let fields = vec![DataField::new(
+ 1,
+ "items".to_string(),
+ DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ )];
+ let filter = array_has(
+ Expr::Column(Column::from_name("items")),
+ lit(f32::from_bits(0x7fc0_1234)),
+ );
+ let analysis = analyze_filters(std::slice::from_ref(&filter), &fields,
true);
+
+ assert!(analysis.pushed_predicate.is_some());
+ assert!(analysis.requires_residual);
+ assert_eq!(
+ classify_filter_pushdown(&filter, &fields, true, |_| true),
+ TableProviderFilterPushDown::Inexact
+ );
+ }
+
+ #[test]
+ fn test_negated_inexact_float_array_membership_falls_open() {
+ use datafusion::functions_nested::expr_fn::array_has;
+
+ let fields = vec![DataField::new(
+ 1,
+ "items".to_string(),
+ DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ )];
+ let filter = Expr::Not(Box::new(array_has(
+ Expr::Column(Column::from_name("items")),
+ lit(f32::from_bits(0xffc0_1234)),
+ )));
+
+ // Paimon/Java considers every NaN payload equal, while DataFusion
+ // distinguishes payloads. Negating that inexact positive predicate
+ // would turn its safe superset into a subset and silently drop rows.
+ assert!(build_pushed_predicate(std::slice::from_ref(&filter),
&fields).is_none());
+ assert_eq!(
+ classify_filter_pushdown(&filter, &fields, true, |_| true),
+ TableProviderFilterPushDown::Unsupported
+ );
+ }
+
+ #[test]
+ fn test_translate_analyzer_inserted_array_widening_casts() {
+ use datafusion::arrow::datatypes::DataType as ArrowDataType;
+ use datafusion::functions_nested::expr_fn::{array_has, array_has_any,
make_array};
+ use datafusion::logical_expr::Cast;
+
+ let fields = test_fields();
+ let widened_items = Expr::Cast(Cast::new(
+ Box::new(Expr::Column(Column::from_name("items"))),
+ ArrowDataType::new_list(ArrowDataType::Int64, true),
+ ));
+ let cases = [
+ (
+ array_has(widened_items.clone(), lit(2_i64)),
+ PredicateOperator::ArrayContains,
+ vec![Datum::Int(2)],
+ ),
+ (
+ array_has_any(widened_items, make_array(vec![lit(1_i64),
lit(3_i64)])),
+ PredicateOperator::ArraysOverlap,
+ vec![Datum::Int(1), Datum::Int(3)],
+ ),
+ ];
+
+ for (filter, expected_op, expected_literals) in cases {
+ let predicate = build_pushed_predicate(&[filter], &fields)
+ .expect("lossless analyzer-inserted widening cast should
translate");
+ assert!(matches!(
+ predicate,
+ Predicate::Leaf { op, literals, .. }
+ if op == expected_op && literals == expected_literals
+ ));
+ }
+
+ let out_of_range = array_has(
+ Expr::Cast(Cast::new(
+ Box::new(Expr::Column(Column::from_name("items"))),
+ ArrowDataType::new_list(ArrowDataType::Int64, true),
+ )),
+ lit(i64::MAX),
+ );
+ assert!(build_pushed_predicate(&[out_of_range], &fields).is_none());
+
+ let fixed_size_cast = array_has(
+ Expr::Cast(Cast::new(
+ Box::new(Expr::Column(Column::from_name("items"))),
+ ArrowDataType::new_fixed_size_list(ArrowDataType::Int64, 2,
true),
+ )),
+ lit(2_i64),
+ );
+ assert!(build_pushed_predicate(&[fixed_size_cast], &fields).is_none());
+
+ let non_nullable_elements = array_has(
+ Expr::Cast(Cast::new(
+ Box::new(Expr::Column(Column::from_name("items"))),
+ ArrowDataType::new_list(ArrowDataType::Int64, false),
+ )),
+ lit(2_i64),
+ );
+ assert!(build_pushed_predicate(&[non_nullable_elements],
&fields).is_none());
+
+ let fixed_size_literals = array_has_any(
+ Expr::Column(Column::from_name("items")),
+ Expr::Cast(Cast::new(
+ Box::new(make_array(vec![lit(1_i32), lit(3_i32), lit(5_i32)])),
+ ArrowDataType::new_fixed_size_list(ArrowDataType::Int32, 2,
true),
+ )),
+ );
+ assert!(build_pushed_predicate(&[fixed_size_literals],
&fields).is_none());
+
+ let long_fields = vec![DataField::new(
+ 1,
+ "items".to_string(),
+
DataType::Array(ArrayType::new(DataType::BigInt(BigIntType::new()))),
+ )];
+ let widened_literal = Expr::Cast(Cast::new(Box::new(lit(2_i32)),
ArrowDataType::Int64));
+ let predicate = build_pushed_predicate(
+ &[array_has(
+ Expr::Column(Column::from_name("items")),
+ widened_literal,
+ )],
+ &long_fields,
+ )
+ .expect("losslessly widened scalar literal should translate");
+ assert!(matches!(
+ predicate,
+ Predicate::Leaf { literals, .. } if literals ==
vec![Datum::Long(2)]
+ ));
+
+ let float_fields = vec![DataField::new(
+ 1,
+ "items".to_string(),
+ DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ )];
+ let widened_floats = Expr::Cast(Cast::new(
+ Box::new(Expr::Column(Column::from_name("items"))),
+ ArrowDataType::new_list(ArrowDataType::Float64, true),
+ ));
+ assert!(build_pushed_predicate(
+ &[array_has(widened_floats.clone(), lit(1.0_f64))],
+ &float_fields,
+ )
+ .is_some());
+ assert!(
+ build_pushed_predicate(&[array_has(widened_floats, lit(1.1_f64))],
&float_fields)
+ .is_none()
+ );
+ }
+
+ #[tokio::test]
+ async fn test_translate_array_membership_after_datafusion_sql_analysis() {
+ use datafusion::arrow::datatypes::{
+ DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema,
+ };
+ use datafusion::datasource::empty::EmptyTable;
+ use datafusion::logical_expr::LogicalPlan;
+ use datafusion::prelude::SessionContext;
+ use std::sync::Arc;
+
+ fn filter_expr(plan: &LogicalPlan) -> Option<&Expr> {
+ match plan {
+ LogicalPlan::Filter(filter) => Some(&filter.predicate),
+ LogicalPlan::TableScan(scan) => scan.filters.first(),
+ other => other.inputs().into_iter().find_map(filter_expr),
+ }
+ }
+
+ let ctx = SessionContext::new();
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![ArrowField::new(
+ "items",
+ ArrowDataType::new_list(ArrowDataType::Int32, true),
+ true,
+ )]));
+ ctx.register_table("t", Arc::new(EmptyTable::new(arrow_schema)))
+ .unwrap();
+ let fields = vec![DataField::new(
+ 1,
+ "items".to_string(),
+ DataType::Array(ArrayType::new(DataType::Int(IntType::new()))),
+ )];
+ let cases = [
+ (
+ "SELECT * FROM t WHERE array_has(items, 2)",
+ PredicateOperator::ArrayContains,
+ vec![Datum::Int(2)],
+ ),
+ (
+ "SELECT * FROM t WHERE array_has_any(items, [1, 3])",
+ PredicateOperator::ArraysOverlap,
+ vec![Datum::Int(1), Datum::Int(3)],
+ ),
+ (
+ "SELECT * FROM t WHERE array_has_all(items, [1, 3])",
+ PredicateOperator::ArrayContainsAll,
+ vec![Datum::Int(1), Datum::Int(3)],
+ ),
+ ];
+ for (sql, expected_op, expected_literals) in cases {
+ let plan = ctx.state().create_logical_plan(sql).await.unwrap();
+ let plan = ctx.state().optimize(&plan).unwrap();
+ let filter = filter_expr(&plan).expect("optimized plan should
retain the filter");
+ assert!(filter.to_string().contains("CAST"));
+
+ let predicate =
build_pushed_predicate(std::slice::from_ref(filter), &fields)
+ .expect("analyzed SQL array predicate should translate");
+ assert!(matches!(
+ predicate,
+ Predicate::Leaf { op, literals, .. }
+ if op == expected_op && literals == expected_literals
+ ));
+ }
+
+ let numeric_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new(
+ "small_items",
+ ArrowDataType::new_list(ArrowDataType::Int16, true),
+ true,
+ ),
+ ArrowField::new(
+ "float_items",
+ ArrowDataType::new_list(ArrowDataType::Float32, true),
+ true,
+ ),
+ ]));
+ ctx.register_table("numeric_arrays",
Arc::new(EmptyTable::new(numeric_schema)))
+ .unwrap();
+ let numeric_fields = vec![
+ DataField::new(
+ 1,
+ "small_items".to_string(),
+
DataType::Array(ArrayType::new(DataType::SmallInt(SmallIntType::new()))),
+ ),
+ DataField::new(
+ 2,
+ "float_items".to_string(),
+
DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ ),
+ ];
+ for sql in [
+ "SELECT * FROM numeric_arrays WHERE array_has(small_items, 2)",
+ "SELECT * FROM numeric_arrays WHERE array_has(float_items, 1.0)",
+ ] {
+ let plan = ctx.state().create_logical_plan(sql).await.unwrap();
+ let plan = ctx.state().optimize(&plan).unwrap();
+ let filter = filter_expr(&plan).expect("optimized plan should
retain the filter");
+ assert!(filter.to_string().contains("CAST"));
+ assert!(
+ build_pushed_predicate(std::slice::from_ref(filter),
&numeric_fields).is_some(),
+ "analyzed numeric ARRAY predicate should translate: {filter}"
+ );
+ }
+ }
+
#[test]
fn test_translate_time32_millisecond_literal() {
let filter =
Expr::Column(Column::from_name("time_col")).eq(Expr::Literal(
diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml
index 4c0bd8dd..5283ea27 100644
--- a/crates/paimon/Cargo.toml
+++ b/crates/paimon/Cargo.toml
@@ -102,6 +102,7 @@ crc32fast = "1"
zstd = "0.13"
snap = "1"
lz4_flex = "0.13"
+lzokay-native = { version = "0.1", default-features = false, features =
["decompress"] }
arrow-array = { workspace = true }
arrow-arith = { workspace = true }
arrow-buffer = { workspace = true }
diff --git a/crates/paimon/src/arrow/format/orc.rs
b/crates/paimon/src/arrow/format/orc.rs
index 6ebbeb5f..fd2e3811 100644
--- a/crates/paimon/src/arrow/format/orc.rs
+++ b/crates/paimon/src/arrow/format/orc.rs
@@ -296,7 +296,10 @@ fn build_orc_leaf_predicate(
| PredicateOperator::Contains
| PredicateOperator::Like
| PredicateOperator::Between
- | PredicateOperator::NotBetween => None,
+ | PredicateOperator::NotBetween
+ | PredicateOperator::ArrayContains
+ | PredicateOperator::ArraysOverlap
+ | PredicateOperator::ArrayContainsAll => None,
}
}
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index 76c76892..444ff8f8 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -903,6 +903,9 @@ fn predicate_supported_for_parquet_row_filter(op:
PredicateOperator) -> bool {
| PredicateOperator::Like
| PredicateOperator::Between
| PredicateOperator::NotBetween
+ | PredicateOperator::ArrayContains
+ | PredicateOperator::ArraysOverlap
+ | PredicateOperator::ArrayContainsAll
)
}
@@ -968,6 +971,27 @@ fn parquet_row_filter_literals_supported(
}
Ok(true)
}
+ PredicateOperator::ArrayContains
+ | PredicateOperator::ArraysOverlap
+ | PredicateOperator::ArrayContainsAll => {
+ let DataType::Array(array_type) = file_data_type else {
+ return Ok(false);
+ };
+ if matches!(op, PredicateOperator::ArrayContains) &&
literals.len() != 1 {
+ return Ok(false);
+ }
+ for literal in literals {
+ if crate::arrow::residual::literal_scalar_for_arrow_filter(
+ literal,
+ array_type.element_type(),
+ )?
+ .is_none()
+ {
+ return Ok(false);
+ }
+ }
+ Ok(true)
+ }
}
}
diff --git a/crates/paimon/src/arrow/residual.rs
b/crates/paimon/src/arrow/residual.rs
index 93159af6..5ddeb95d 100644
--- a/crates/paimon/src/arrow/residual.rs
+++ b/crates/paimon/src/arrow/residual.rs
@@ -49,9 +49,9 @@ use crate::spec::{is_row_id_column, DataField, DataType,
Datum, Predicate, Predi
use crate::Error;
use arrow_array::{
Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Datum as
ArrowDatum, Decimal128Array,
- Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array,
RecordBatch, Scalar,
- StringArray, Time32MillisecondArray, TimestampMicrosecondArray,
TimestampMillisecondArray,
- TimestampNanosecondArray,
+ FixedSizeListArray, Float32Array, Float64Array, Int16Array, Int32Array,
Int64Array, Int8Array,
+ LargeListArray, ListArray, RecordBatch, Scalar, StringArray,
Time32MillisecondArray,
+ TimestampMicrosecondArray, TimestampMillisecondArray,
TimestampNanosecondArray,
};
use arrow_ord::cmp::{
eq as arrow_eq, gt as arrow_gt, gt_eq as arrow_gt_eq, lt as arrow_lt,
lt_eq as arrow_lt_eq,
@@ -327,6 +327,14 @@ pub(crate) fn evaluate_exact_leaf_predicate(
op: PredicateOperator,
literals: &[Datum],
) -> Result<BooleanArray, ArrowError> {
+ if matches!(
+ op,
+ PredicateOperator::ArrayContains
+ | PredicateOperator::ArraysOverlap
+ | PredicateOperator::ArrayContainsAll
+ ) {
+ return evaluate_array_membership_predicate(array, data_type, op,
literals);
+ }
// Decimals are compared by mathematical value across scales (Paimon
// `datum_cmp`/`decimal_cmp`). Arrow scalar comparison requires the
literal to
// be representable at the column scale, which fails for a finer-scale
literal
@@ -378,6 +386,176 @@ pub(crate) fn evaluate_exact_leaf_predicate(
PredicateOperator::Between | PredicateOperator::NotBetween => {
evaluate_between_predicate(array, data_type, op, literals)
}
+ PredicateOperator::ArrayContains
+ | PredicateOperator::ArraysOverlap
+ | PredicateOperator::ArrayContainsAll => unreachable!("handled before
scalar dispatch"),
+ }
+}
+
+fn evaluate_array_membership_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ let DataType::Array(array_type) = data_type else {
+ return Err(ArrowError::ComputeError(format!(
+ "array predicate {op} requires an ARRAY column, got {data_type:?}"
+ )));
+ };
+ if matches!(op, PredicateOperator::ArrayContains) && literals.len() != 1 {
+ return Err(unconvertible_literal_error(op, data_type));
+ }
+
+ #[derive(Clone, Copy)]
+ enum ListLayout<'a> {
+ List(&'a ListArray),
+ Large(&'a LargeListArray),
+ Fixed(&'a FixedSizeListArray),
+ }
+ let layout = if let Some(list) =
array.as_any().downcast_ref::<ListArray>() {
+ ListLayout::List(list)
+ } else if let Some(list) = array.as_any().downcast_ref::<LargeListArray>()
{
+ ListLayout::Large(list)
+ } else if let Some(list) =
array.as_any().downcast_ref::<FixedSizeListArray>() {
+ ListLayout::Fixed(list)
+ } else {
+ return Err(ArrowError::ComputeError(format!(
+ "array predicate {op} requires an Arrow list array, got {:?}",
+ array.data_type()
+ )));
+ };
+ let values = match layout {
+ ListLayout::List(list) => list.values(),
+ ListLayout::Large(list) => list.values(),
+ ListLayout::Fixed(list) => list.values(),
+ };
+
+ let mut row_ranges = Vec::with_capacity(array.len());
+ for row in 0..array.len() {
+ if array.is_null(row) {
+ row_ranges.push(None);
+ continue;
+ }
+ let (start, end) = match layout {
+ ListLayout::List(list) => {
+ let offsets = list.value_offsets();
+ (
+ usize::try_from(offsets[row]),
+ usize::try_from(offsets[row + 1]),
+ )
+ }
+ ListLayout::Large(list) => {
+ let offsets = list.value_offsets();
+ (
+ usize::try_from(offsets[row]),
+ usize::try_from(offsets[row + 1]),
+ )
+ }
+ ListLayout::Fixed(list) => {
+ let start = usize::try_from(list.value_offset(row));
+ let end = usize::try_from(list.value_offset(row) +
list.value_length());
+ (start, end)
+ }
+ };
+ let (start, end) = match (start, end) {
+ (Ok(start), Ok(end)) => (start, end),
+ _ => {
+ return Err(ArrowError::ComputeError(
+ "array predicate encountered a negative list
offset".to_string(),
+ ))
+ }
+ };
+ row_ranges.push(Some((start, end)));
+ }
+
+ // Fold one child mask at a time. Besides bounding temporary memory to one
+ // flattened-child mask, the specialized equality below preserves Java's
+ // Float.compare / Double.compare NaN semantics and decimal by-value
+ // comparison for nested elements.
+ let contains_all = matches!(op, PredicateOperator::ArrayContainsAll);
+ let mut result = row_ranges
+ .iter()
+ .map(|range| range.is_some() && contains_all)
+ .collect::<Vec<_>>();
+ for literal in literals {
+ let mask = evaluate_array_element_equality(values,
array_type.element_type(), literal, op)?;
+ for (row, range) in row_ranges.iter().enumerate() {
+ let Some((start, end)) = range else {
+ continue;
+ };
+ if (contains_all && !result[row]) || (!contains_all &&
result[row]) {
+ continue;
+ }
+ let matches_literal = (*start..*end).any(|element|
mask.value(element));
+ if contains_all {
+ result[row] &= matches_literal;
+ } else {
+ result[row] |= matches_literal;
+ }
+ }
+ }
+ Ok(BooleanArray::from(result))
+}
+
+fn evaluate_array_element_equality(
+ values: &ArrayRef,
+ element_type: &DataType,
+ literal: &Datum,
+ op: PredicateOperator,
+) -> Result<BooleanArray, ArrowError> {
+ match element_type {
+ DataType::Float(_) => {
+ let expected = float32_literal(literal)
+ .ok_or_else(|| unconvertible_literal_error(op, element_type))?;
+ let values = values
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .ok_or_else(|| {
+ ArrowError::ComputeError(
+ "FLOAT array elements require an Arrow
Float32Array".to_string(),
+ )
+ })?;
+ Ok(BooleanArray::from_iter(values.iter().map(|value| {
+ Some(value.is_some_and(|value| {
+ (value.is_nan() && expected.is_nan()) || value.to_bits()
== expected.to_bits()
+ }))
+ })))
+ }
+ DataType::Double(_) => {
+ let expected = float64_literal(literal)
+ .ok_or_else(|| unconvertible_literal_error(op, element_type))?;
+ let values = values
+ .as_any()
+ .downcast_ref::<Float64Array>()
+ .ok_or_else(|| {
+ ArrowError::ComputeError(
+ "DOUBLE array elements require an Arrow
Float64Array".to_string(),
+ )
+ })?;
+ Ok(BooleanArray::from_iter(values.iter().map(|value| {
+ Some(value.is_some_and(|value| {
+ (value.is_nan() && expected.is_nan()) || value.to_bits()
== expected.to_bits()
+ }))
+ })))
+ }
+ DataType::Decimal(_) => Ok(sanitize_filter_mask(evaluate_decimal_leaf(
+ values,
+ PredicateOperator::Eq,
+ std::slice::from_ref(literal),
+ )?)),
+ _ => {
+ let Some(scalar) = literal_scalar_for_arrow_filter(literal,
element_type)
+ .map_err(|error| ArrowError::ComputeError(error.to_string()))?
+ else {
+ return Err(unconvertible_literal_error(op, element_type));
+ };
+ Ok(sanitize_filter_mask(evaluate_column_predicate(
+ values,
+ &scalar,
+ PredicateOperator::Eq,
+ )?))
+ }
}
}
@@ -409,8 +587,7 @@ fn evaluate_decimal_leaf(
// Compare one column value (as a Datum::Decimal at the column scale)
against a
// literal Datum. `precision` is irrelevant to `datum_cmp` (it compares by
- // value), so any value is fine. `None` means the cross-scale normalization
- // overflowed i128 — surface it rather than silently drop rows.
+ // value), so any value is fine.
let cmp = |v: i128, lit: &Datum| -> Result<Ordering, ArrowError> {
let cell = Datum::Decimal {
unscaled: v,
@@ -419,7 +596,7 @@ fn evaluate_decimal_leaf(
};
crate::spec::datum_cmp(&cell, lit).ok_or_else(|| {
ArrowError::ComputeError(
- "decimal comparison overflowed while normalizing
scales".to_string(),
+ "decimal column compared against a non-decimal
literal".to_string(),
)
})
};
@@ -743,7 +920,10 @@ fn evaluate_column_predicate(
| PredicateOperator::In
| PredicateOperator::NotIn
| PredicateOperator::Between
- | PredicateOperator::NotBetween =>
Ok(BooleanArray::new_null(column.len())),
+ | PredicateOperator::NotBetween
+ | PredicateOperator::ArrayContains
+ | PredicateOperator::ArraysOverlap
+ | PredicateOperator::ArrayContainsAll =>
Ok(BooleanArray::new_null(column.len())),
}
}
@@ -991,7 +1171,13 @@ fn float64_literal(literal: &Datum) -> Option<f64> {
#[cfg(test)]
mod tests {
use super::*;
- use crate::spec::{row_id_leaf, IntType, VarCharType, ROW_ID_FIELD_NAME};
+ use crate::spec::{
+ row_id_leaf, ArrayType, DecimalType, DoubleType, FloatType, IntType,
VarCharType,
+ ROW_ID_FIELD_NAME,
+ };
+ use arrow_array::builder::{
+ Decimal128Builder, Float32Builder, Float64Builder, Int32Builder,
ListBuilder,
+ };
use arrow_array::{Int32Array, StringArray};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
use std::sync::Arc;
@@ -1049,6 +1235,209 @@ mod tests {
RecordBatch::try_new(schema,
vec![Arc::new(StringArray::from(values))]).unwrap()
}
+ fn int_list_array() -> ArrayRef {
+ let element = Arc::new(ArrowField::new("element",
ArrowDataType::Int32, true));
+ let mut builder =
ListBuilder::new(Int32Builder::new()).with_field(element);
+
+ builder.values().append_value(1);
+ builder.values().append_null();
+ builder.values().append_value(2);
+ builder.values().append_value(2);
+ builder.append(true);
+
+ builder.values().append_value(2);
+ builder.values().append_value(3);
+ builder.append(true);
+
+ builder.append(true); // empty array
+ builder.append(false); // null array
+ Arc::new(builder.finish())
+ }
+
+ fn bool_values(mask: &BooleanArray) -> Vec<bool> {
+ mask.iter().map(|value| value.unwrap_or(false)).collect()
+ }
+
+ #[test]
+ fn test_array_membership_predicates_match_java_multivalue_semantics() {
+ let array = int_list_array();
+ let data_type =
DataType::Array(ArrayType::new(DataType::Int(IntType::new())));
+
+ let contains = evaluate_exact_leaf_predicate(
+ &array,
+ &data_type,
+ PredicateOperator::ArrayContains,
+ &[Datum::Int(2)],
+ )
+ .unwrap();
+ assert_eq!(bool_values(&contains), vec![true, true, false, false]);
+
+ let overlap = evaluate_exact_leaf_predicate(
+ &array,
+ &data_type,
+ PredicateOperator::ArraysOverlap,
+ &[Datum::Int(9), Datum::Int(2), Datum::Int(2)],
+ )
+ .unwrap();
+ assert_eq!(bool_values(&overlap), vec![true, true, false, false]);
+
+ let contains_all = evaluate_exact_leaf_predicate(
+ &array,
+ &data_type,
+ PredicateOperator::ArrayContainsAll,
+ &[Datum::Int(1), Datum::Int(2), Datum::Int(1)],
+ )
+ .unwrap();
+ assert_eq!(bool_values(&contains_all), vec![true, false, false,
false]);
+
+ let contains_all_empty = evaluate_exact_leaf_predicate(
+ &array,
+ &data_type,
+ PredicateOperator::ArrayContainsAll,
+ &[],
+ )
+ .unwrap();
+ assert_eq!(
+ bool_values(&contains_all_empty),
+ vec![true, true, true, false]
+ );
+ }
+
+ #[test]
+ fn test_array_membership_uses_java_nan_and_signed_zero_semantics() {
+ let float_element = Arc::new(ArrowField::new("element",
ArrowDataType::Float32, true));
+ let mut floats =
ListBuilder::new(Float32Builder::new()).with_field(float_element);
+ floats.values().append_value(f32::from_bits(0x7fc0_0001));
+ floats.append(true);
+ floats.values().append_value(0.0);
+ floats.append(true);
+ floats.values().append_value(-0.0);
+ floats.append(true);
+ let floats: ArrayRef = Arc::new(floats.finish());
+ let float_type =
DataType::Array(ArrayType::new(DataType::Float(FloatType::new())));
+
+ let nan = evaluate_exact_leaf_predicate(
+ &floats,
+ &float_type,
+ PredicateOperator::ArrayContains,
+ &[Datum::Float(f32::from_bits(0xffc0_1234))],
+ )
+ .unwrap();
+ assert_eq!(bool_values(&nan), vec![true, false, false]);
+ let positive_zero = evaluate_exact_leaf_predicate(
+ &floats,
+ &float_type,
+ PredicateOperator::ArrayContains,
+ &[Datum::Float(0.0)],
+ )
+ .unwrap();
+ assert_eq!(bool_values(&positive_zero), vec![false, true, false]);
+
+ let double_element = Arc::new(ArrowField::new("element",
ArrowDataType::Float64, true));
+ let mut doubles =
ListBuilder::new(Float64Builder::new()).with_field(double_element);
+ doubles
+ .values()
+ .append_value(f64::from_bits(0x7ff8_0000_0000_0001));
+ doubles.append(true);
+ doubles.values().append_value(-0.0);
+ doubles.append(true);
+ let doubles: ArrayRef = Arc::new(doubles.finish());
+ let double_type =
DataType::Array(ArrayType::new(DataType::Double(DoubleType::new())));
+ let double_nan = evaluate_exact_leaf_predicate(
+ &doubles,
+ &double_type,
+ PredicateOperator::ArraysOverlap,
+ &[Datum::Double(f64::from_bits(0xfff8_0000_0000_4321))],
+ )
+ .unwrap();
+ assert_eq!(bool_values(&double_nan), vec![true, false]);
+ }
+
+ #[test]
+ fn test_decimal_array_membership_compares_by_value_across_scales() {
+ let decimal_arrow_type = ArrowDataType::Decimal128(10, 2);
+ let element = Arc::new(ArrowField::new("element",
decimal_arrow_type.clone(), true));
+ let values =
Decimal128Builder::new().with_data_type(decimal_arrow_type);
+ let mut decimals = ListBuilder::new(values).with_field(element);
+ decimals.values().append_value(100);
+ decimals.values().append_value(200);
+ decimals.append(true);
+ decimals.values().append_value(110);
+ decimals.append(true);
+ decimals.append(false);
+ let decimals: ArrayRef = Arc::new(decimals.finish());
+ let decimal_type = DataType::Array(ArrayType::new(DataType::Decimal(
+ DecimalType::new(10, 2).unwrap(),
+ )));
+
+ let contains = evaluate_exact_leaf_predicate(
+ &decimals,
+ &decimal_type,
+ PredicateOperator::ArrayContains,
+ &[Datum::Decimal {
+ unscaled: 10,
+ precision: 2,
+ scale: 1,
+ }],
+ )
+ .unwrap();
+ assert_eq!(bool_values(&contains), vec![true, false, false]);
+
+ let contains_all = evaluate_exact_leaf_predicate(
+ &decimals,
+ &decimal_type,
+ PredicateOperator::ArrayContainsAll,
+ &[
+ Datum::Decimal {
+ unscaled: 1,
+ precision: 1,
+ scale: 0,
+ },
+ Datum::Decimal {
+ unscaled: 2,
+ precision: 1,
+ scale: 0,
+ },
+ ],
+ )
+ .unwrap();
+ assert_eq!(bool_values(&contains_all), vec![true, false, false]);
+ }
+
+ #[test]
+ fn test_decimal_array_membership_handles_extreme_scale_differences() {
+ let decimal_arrow_type = ArrowDataType::Decimal128(38, 38);
+ let element = Arc::new(ArrowField::new("element",
decimal_arrow_type.clone(), true));
+ let values =
Decimal128Builder::new().with_data_type(decimal_arrow_type);
+ let mut decimals = ListBuilder::new(values).with_field(element);
+ decimals.values().append_value(10_i128.pow(38) - 1);
+ decimals.append(true);
+ let decimals: ArrayRef = Arc::new(decimals.finish());
+ let decimal_type = DataType::Array(ArrayType::new(DataType::Decimal(
+ DecimalType::new(38, 38).unwrap(),
+ )));
+ let two = Datum::Decimal {
+ unscaled: 2,
+ precision: 1,
+ scale: 0,
+ };
+
+ for op in [
+ PredicateOperator::ArrayContains,
+ PredicateOperator::ArraysOverlap,
+ PredicateOperator::ArrayContainsAll,
+ ] {
+ let mask = evaluate_exact_leaf_predicate(
+ &decimals,
+ &decimal_type,
+ op,
+ std::slice::from_ref(&two),
+ )
+ .expect("valid cross-scale decimal comparison must not overflow");
+ assert_eq!(bool_values(&mask), vec![false]);
+ }
+ }
+
fn int_values(batch: &RecordBatch) -> Vec<i32> {
batch
.column(0)
diff --git a/crates/paimon/src/btree/query.rs b/crates/paimon/src/btree/query.rs
index df4c2756..2573dc16 100644
--- a/crates/paimon/src/btree/query.rs
+++ b/crates/paimon/src/btree/query.rs
@@ -138,6 +138,12 @@ where
})
.await
}
+ PredicateOperator::ArrayContains
+ | PredicateOperator::ArraysOverlap
+ | PredicateOperator::ArrayContainsAll => Err(io::Error::new(
+ io::ErrorKind::Unsupported,
+ format!("BTree index does not support {op}"),
+ )),
}
}
}
diff --git a/crates/paimon/src/predicate_stats.rs
b/crates/paimon/src/predicate_stats.rs
index 57d01e76..d2b8308f 100644
--- a/crates/paimon/src/predicate_stats.rs
+++ b/crates/paimon/src/predicate_stats.rs
@@ -69,6 +69,15 @@ pub(crate) fn data_leaf_may_match<T: StatsAccessor>(
PredicateOperator::NotIn => {
return true;
}
+ PredicateOperator::ArrayContains => {
+ return all_null != Some(true);
+ }
+ PredicateOperator::ArraysOverlap => {
+ return !literals.is_empty() && all_null != Some(true);
+ }
+ PredicateOperator::ArrayContainsAll => {
+ return all_null != Some(true);
+ }
PredicateOperator::EndsWith | PredicateOperator::Contains => {
// String min/max ordering carries no information about suffix /
// substring matches, so fail open.
@@ -203,7 +212,10 @@ pub(crate) fn data_leaf_may_match<T: StatsAccessor>(
| PredicateOperator::EndsWith
| PredicateOperator::Contains
| PredicateOperator::Between
- | PredicateOperator::NotBetween => true,
+ | PredicateOperator::NotBetween
+ | PredicateOperator::ArrayContains
+ | PredicateOperator::ArraysOverlap
+ | PredicateOperator::ArrayContainsAll => true,
}
}
@@ -301,7 +313,10 @@ pub(crate) fn data_leaf_must_match<T: StatsAccessor>(
| PredicateOperator::StartsWith
| PredicateOperator::EndsWith
| PredicateOperator::Contains
- | PredicateOperator::Like => false,
+ | PredicateOperator::Like
+ | PredicateOperator::ArrayContains
+ | PredicateOperator::ArraysOverlap
+ | PredicateOperator::ArrayContainsAll => false,
}
}
@@ -908,6 +923,25 @@ mod tests {
));
}
+ #[test]
+ fn array_contains_prunes_all_null_file_like_java() {
+ let dt =
DataType::Array(crate::spec::ArrayType::new(DataType::Int(IntType::new())));
+ let stats = MockStats {
+ row_count: 10,
+ null_count: Some(10),
+ min: None,
+ max: None,
+ };
+ assert!(!data_leaf_may_match(
+ 0,
+ &dt,
+ &dt,
+ PredicateOperator::ArrayContains,
+ &[Datum::Int(1)],
+ &stats,
+ ));
+ }
+
#[test]
fn between_falls_open_when_stats_missing() {
let dt = DataType::Int(IntType::new());
diff --git a/crates/paimon/src/spec/binary_row.rs
b/crates/paimon/src/spec/binary_row.rs
index 236cf91e..4c1b816b 100644
--- a/crates/paimon/src/spec/binary_row.rs
+++ b/crates/paimon/src/spec/binary_row.rs
@@ -945,6 +945,13 @@ pub fn extract_datum_from_arrow(
.ok_or_else(|| type_mismatch_err("Date", col_idx))?;
Datum::Date(arr.value(row_idx))
}
+ DataType::Time(_) => {
+ let arr = col
+ .as_any()
+ .downcast_ref::<arrow_array::Time32MillisecondArray>()
+ .ok_or_else(|| type_mismatch_err("Time", col_idx))?;
+ Datum::Time(arr.value(row_idx))
+ }
DataType::Decimal(d) => {
let arr = col
.as_any()
@@ -1556,6 +1563,33 @@ mod tests {
use super::*;
use crate::variant::GenericVariant;
+ #[test]
+ fn test_extract_time_datum_from_arrow() {
+ let schema =
+
std::sync::Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
+ "time_col",
+
arrow_schema::DataType::Time32(arrow_schema::TimeUnit::Millisecond),
+ true,
+ )]));
+ let batch = arrow_array::RecordBatch::try_new(
+ schema,
+ vec![std::sync::Arc::new(
+ arrow_array::Time32MillisecondArray::from(vec![Some(12_345),
None]),
+ )],
+ )
+ .unwrap();
+ let data_type = DataType::Time(crate::spec::TimeType::new(3).unwrap());
+
+ assert_eq!(
+ extract_datum_from_arrow(&batch, 0, 0, &data_type).unwrap(),
+ Some(Datum::Time(12_345))
+ );
+ assert_eq!(
+ extract_datum_from_arrow(&batch, 1, 0, &data_type).unwrap(),
+ None
+ );
+ }
+
#[test]
fn test_empty_binary_row() {
let row = BinaryRow::new(0);
diff --git a/crates/paimon/src/spec/predicate.rs
b/crates/paimon/src/spec/predicate.rs
index 66946abd..c168e769 100644
--- a/crates/paimon/src/spec/predicate.rs
+++ b/crates/paimon/src/spec/predicate.rs
@@ -196,18 +196,59 @@ pub(crate) fn datum_cmp(lhs: &Datum, rhs: &Datum) ->
Option<Ordering> {
/// Compare two decimals by mathematical value.
///
-/// Normalizes both to the larger scale, then compares unscaled values.
-/// E.g. `(10, scale=1)` vs `(100, scale=2)` → both represent 1.0 → equal.
+/// Compares the sign, decimal exponent, then significant digits. Padding the
+/// shorter significand with trailing zeroes is equivalent to normalizing both
+/// values to the larger scale, without overflowing `i128`.
fn decimal_cmp(ua: i128, sa: u32, ub: i128, sb: u32) -> Option<Ordering> {
if sa == sb {
- return ua.partial_cmp(&ub);
+ return Some(ua.cmp(&ub));
}
- let (na, nb) = if sa < sb {
- (ua.checked_mul(pow10_i128(sb - sa))?, ub)
+ let normalized = if sa < sb {
+ 10_i128
+ .checked_pow(sb - sa)
+ .and_then(|factor| ua.checked_mul(factor))
+ .map(|scaled| scaled.cmp(&ub))
} else {
- (ua, ub.checked_mul(pow10_i128(sa - sb))?)
+ 10_i128
+ .checked_pow(sa - sb)
+ .and_then(|factor| ub.checked_mul(factor))
+ .map(|scaled| ua.cmp(&scaled))
};
- na.partial_cmp(&nb)
+ if normalized.is_some() {
+ return normalized;
+ }
+
+ let sign_a = ua.signum();
+ let sign_b = ub.signum();
+ if sign_a != sign_b {
+ return Some(sign_a.cmp(&sign_b));
+ }
+ if sign_a == 0 {
+ return Some(Ordering::Equal);
+ }
+
+ let digits_a = ua.unsigned_abs().to_string();
+ let digits_b = ub.unsigned_abs().to_string();
+ let exponent_a = digits_a.len() as i64 - i64::from(sa);
+ let exponent_b = digits_b.len() as i64 - i64::from(sb);
+ let mut magnitude = exponent_a.cmp(&exponent_b);
+ if magnitude == Ordering::Equal {
+ let a = digits_a.as_bytes();
+ let b = digits_b.as_bytes();
+ for index in 0..a.len().max(b.len()) {
+ let digit_a = a.get(index).copied().unwrap_or(b'0');
+ let digit_b = b.get(index).copied().unwrap_or(b'0');
+ magnitude = digit_a.cmp(&digit_b);
+ if magnitude != Ordering::Equal {
+ break;
+ }
+ }
+ }
+ Some(if sign_a < 0 {
+ magnitude.reverse()
+ } else {
+ magnitude
+ })
}
/// Match Java `CompareUtils.compare(byte[], byte[])`, which compares bytes as
@@ -216,19 +257,6 @@ fn java_bytes_cmp(a: &[u8], b: &[u8]) -> Ordering {
a.cmp(b)
}
-/// 10^exp as i128. Returns i128::MAX for exponents that would overflow.
-fn pow10_i128(exp: u32) -> i128 {
- const MAX_EXP: u32 = 38; // 10^38 fits in i128
- if exp > MAX_EXP {
- return i128::MAX;
- }
- let mut result: i128 = 1;
- for _ in 0..exp {
- result = result.saturating_mul(10);
- }
- result
-}
-
// PredicateOperator
// ---------------------------------------------------------------------------
@@ -248,6 +276,9 @@ pub enum PredicateOperator {
StartsWith,
EndsWith,
Contains,
+ ArrayContains,
+ ArraysOverlap,
+ ArrayContainsAll,
Like,
Between,
NotBetween,
@@ -269,6 +300,9 @@ impl fmt::Display for PredicateOperator {
Self::StartsWith => write!(f, "STARTS_WITH"),
Self::EndsWith => write!(f, "ENDS_WITH"),
Self::Contains => write!(f, "CONTAINS"),
+ Self::ArrayContains => write!(f, "ARRAY_CONTAINS"),
+ Self::ArraysOverlap => write!(f, "ARRAYS_OVERLAP"),
+ Self::ArrayContainsAll => write!(f, "ARRAY_CONTAINS_ALL"),
Self::Like => write!(f, "LIKE"),
Self::Between => write!(f, "BETWEEN"),
Self::NotBetween => write!(f, "NOT BETWEEN"),
@@ -560,7 +594,10 @@ impl fmt::Display for Predicate {
write!(f, "{column} {op}")?;
match op {
PredicateOperator::IsNull | PredicateOperator::IsNotNull
=> {}
- PredicateOperator::In | PredicateOperator::NotIn => {
+ PredicateOperator::In
+ | PredicateOperator::NotIn
+ | PredicateOperator::ArraysOverlap
+ | PredicateOperator::ArrayContainsAll => {
write!(f, " (")?;
for (i, lit) in literals.iter().enumerate() {
if i > 0 {
@@ -722,6 +759,17 @@ fn parse_rest_leaf(
.ok_or_else(|| rest_json_err(format!("unknown field `{field}`")))?;
let function = str_prop(value, "function")?;
+ let literal_data_type = match function {
+ "ARRAY_CONTAINS" | "ARRAYS_OVERLAP" | "ARRAY_CONTAINS_ALL" => match
data_type {
+ DataType::Array(array) => array.element_type(),
+ other => {
+ return Err(rest_json_err(format!(
+ "{function} requires an ARRAY field, got {other:?}"
+ )))
+ }
+ },
+ _ => data_type,
+ };
// Convert every literal up front (like Java `deserializeLiterals`) so a
// malformed extra literal fails the parse. JSON null -> `None` (SQL NULL).
let literals: Vec<Option<Datum>> = value
@@ -731,7 +779,7 @@ fn parse_rest_leaf(
.iter()
.map(|v| match v {
serde_json::Value::Null => Ok(None),
- v => Ok(Some(json_to_datum(v, data_type)?)),
+ v => Ok(Some(json_to_datum(v, literal_data_type)?)),
})
.collect::<Result<Vec<_>>>()?;
// Binary/ternary leaves test false for every row on a null literal
@@ -788,6 +836,22 @@ fn parse_rest_leaf(
"STARTS_WITH" => binary(one_literal()?, &|l|
builder.starts_with(field, l)),
"ENDS_WITH" => binary(one_literal()?, &|l| builder.ends_with(field,
l)),
"CONTAINS" => binary(one_literal()?, &|l| builder.contains(field, l)),
+ "ARRAY_CONTAINS" => binary(one_literal()?, &|l|
builder.array_contains(field, l)),
+ "ARRAYS_OVERLAP" => {
+ let datums =
literals.iter().flatten().cloned().collect::<Vec<_>>();
+ if datums.is_empty() {
+ Ok(Predicate::AlwaysFalse)
+ } else {
+ builder.arrays_overlap(field, datums)
+ }
+ }
+ "ARRAY_CONTAINS_ALL" => {
+ if literals.iter().any(Option::is_none) {
+ Ok(Predicate::AlwaysFalse)
+ } else {
+ builder.array_contains_all(field,
literals.iter().flatten().cloned().collect())
+ }
+ }
"LIKE" => match one_literal()? {
Some(l) => {
if let Datum::String(pattern) = &l {
@@ -1131,6 +1195,25 @@ impl PredicateBuilder {
self.string_leaf(field, PredicateOperator::Contains, pattern)
}
+ /// Build an element-membership predicate over an ARRAY column.
+ pub fn array_contains(&self, field: &str, element: Datum) ->
Result<Predicate> {
+ self.array_leaf(field, PredicateOperator::ArrayContains, vec![element])
+ }
+
+ /// Build an any-element membership predicate over an ARRAY column.
+ pub fn arrays_overlap(&self, field: &str, elements: Vec<Datum>) ->
Result<Predicate> {
+ self.array_leaf(field, PredicateOperator::ArraysOverlap, elements)
+ }
+
+ /// Build an all-elements membership predicate over an ARRAY column.
+ ///
+ /// An empty literal list is retained as a leaf: it matches non-null
arrays,
+ /// while the multivalue index must fall back because it cannot distinguish
+ /// null arrays from empty arrays.
+ pub fn array_contains_all(&self, field: &str, elements: Vec<Datum>) ->
Result<Predicate> {
+ self.array_leaf(field, PredicateOperator::ArrayContainsAll, elements)
+ }
+
/// `field LIKE '<pattern>'` with optional `escape` character (default
`\`).
/// Mirrors Java `LikeOptimization`: rewrites `prefix%` / `%suffix` /
/// `%mid%` / no-wildcard patterns into [`PredicateOperator::StartsWith`] /
@@ -1219,6 +1302,31 @@ impl PredicateBuilder {
self.leaf(field, op, vec![pattern])
}
+ fn array_leaf(
+ &self,
+ field: &str,
+ op: PredicateOperator,
+ literals: Vec<Datum>,
+ ) -> Result<Predicate> {
+ let (index, canonical, data_type) = self.resolve_field(field)?;
+ let DataType::Array(array_type) = &data_type else {
+ return Err(Error::ConfigInvalid {
+ message: format!("{op} requires an ARRAY field, got
{data_type:?}"),
+ });
+ };
+ Self::validate_literal_count(op, &literals)?;
+ for literal in &literals {
+ validate_datum_matches_type(literal, array_type.element_type())?;
+ }
+ Ok(Predicate::Leaf {
+ column: canonical,
+ index,
+ data_type,
+ op,
+ literals,
+ })
+ }
+
// -- internal --
/// Resolve field name to index + type, validate literals, and build a
leaf predicate.
@@ -1302,6 +1410,7 @@ impl PredicateBuilder {
message: format!("{op} expects at least 1 literal, got 0"),
});
}
+ PredicateOperator::ArraysOverlap |
PredicateOperator::ArrayContainsAll => return Ok(()),
PredicateOperator::Between | PredicateOperator::NotBetween => (2,
literals.len()),
_ => (1, literals.len()),
};
@@ -1562,6 +1671,11 @@ fn eval_leaf(op: PredicateOperator, datum:
Option<&Datum>, literals: &[Datum]) -
},
PredicateOperator::Between => eval_between(val, literals),
PredicateOperator::NotBetween => !eval_between(val, literals),
+ PredicateOperator::ArrayContains
+ | PredicateOperator::ArraysOverlap
+ | PredicateOperator::ArrayContainsAll => {
+ unreachable!("array predicates are evaluated against Arrow
list arrays")
+ }
// IsNull/IsNotNull are handled in the outer match above.
PredicateOperator::IsNull | PredicateOperator::IsNotNull =>
unreachable!(),
}
@@ -2328,6 +2442,33 @@ mod tests {
assert_eq!(a, b);
}
+ #[test]
+ fn test_decimal_extreme_scale_comparison_does_not_overflow() {
+ let two = Datum::Decimal {
+ unscaled: 2,
+ precision: 1,
+ scale: 0,
+ };
+ let below_one = Datum::Decimal {
+ unscaled: 10_i128.pow(38) - 1,
+ precision: 38,
+ scale: 38,
+ };
+ assert!(two > below_one);
+
+ let negative_two = Datum::Decimal {
+ unscaled: -2,
+ precision: 1,
+ scale: 0,
+ };
+ let above_negative_one = Datum::Decimal {
+ unscaled: -(10_i128.pow(38) - 1),
+ precision: 38,
+ scale: 38,
+ };
+ assert!(negative_two < above_negative_one);
+ }
+
// ======================== PartialOrd ========================
#[test]
@@ -3168,6 +3309,87 @@ mod tests {
assert!(Predicate::from_rest_json(json, &fields).is_ok());
}
+ #[test]
+ fn test_array_predicate_builder_uses_element_literals() {
+ let fields = vec![
+ DataField::new(
+ 0,
+ "items".to_string(),
+ DataType::Array(ArrayType::new(DataType::Int(IntType::new()))),
+ ),
+ DataField::new(1, "id".to_string(), DataType::Int(IntType::new())),
+ ];
+ let builder = PredicateBuilder::new(&fields);
+
+ let contains = builder.array_contains("items", Datum::Int(2)).unwrap();
+ assert!(matches!(
+ contains,
+ Predicate::Leaf {
+ op: PredicateOperator::ArrayContains,
+ literals,
+ ..
+ } if literals == vec![Datum::Int(2)]
+ ));
+ let overlap = builder
+ .arrays_overlap("items", vec![Datum::Int(1), Datum::Int(3)])
+ .unwrap();
+ assert!(matches!(
+ overlap,
+ Predicate::Leaf {
+ op: PredicateOperator::ArraysOverlap,
+ literals,
+ ..
+ } if literals == vec![Datum::Int(1), Datum::Int(3)]
+ ));
+ assert!(builder.array_contains_all("items", vec![]).is_ok());
+ assert!(builder.array_contains("id", Datum::Int(2)).is_err());
+ assert!(builder
+ .array_contains("items", Datum::String("2".to_string()))
+ .is_err());
+ }
+
+ #[test]
+ fn test_from_rest_json_parses_java_array_predicates_and_nulls() {
+ let fields = vec![DataField::new(
+ 0,
+ "items".to_string(),
+ DataType::Array(ArrayType::new(DataType::Int(IntType::new()))),
+ )];
+
+ for (function, literals, expected_op) in [
+ ("ARRAY_CONTAINS", "[2]", PredicateOperator::ArrayContains),
+ ("ARRAYS_OVERLAP", "[1,2]", PredicateOperator::ArraysOverlap),
+ (
+ "ARRAY_CONTAINS_ALL",
+ "[1,2]",
+ PredicateOperator::ArrayContainsAll,
+ ),
+ ] {
+ let parsed = Predicate::from_rest_json(
+ &rest_leaf_json_typed(function, "items", "ARRAY<INT>",
literals),
+ &fields,
+ )
+ .unwrap();
+ assert!(matches!(
+ parsed,
+ Predicate::Leaf { op, .. } if op == expected_op
+ ));
+ }
+
+ for (function, literals) in [
+ ("ARRAY_CONTAINS", "[null]"),
+ ("ARRAYS_OVERLAP", "[null]"),
+ ("ARRAY_CONTAINS_ALL", "[1,null]"),
+ ] {
+ let parsed = Predicate::from_rest_json(
+ &rest_leaf_json_typed(function, "items", "ARRAY<INT>",
literals),
+ &fields,
+ )
+ .unwrap();
+ assert!(matches!(parsed, Predicate::AlwaysFalse));
+ }
+ }
+
#[test]
fn test_from_rest_json_rejects_empty_compounds() {
let fields = test_fields();
diff --git a/crates/paimon/src/table/bitmap_global_index_reader.rs
b/crates/paimon/src/table/bitmap_global_index_reader.rs
index 151aac8e..ab867b78 100644
--- a/crates/paimon/src/table/bitmap_global_index_reader.rs
+++ b/crates/paimon/src/table/bitmap_global_index_reader.rs
@@ -274,6 +274,12 @@ impl BitmapGlobalIndexReader {
return self.is_not_null().await;
}
match op {
+ PredicateOperator::ArrayContains
+ | PredicateOperator::ArraysOverlap
+ | PredicateOperator::ArrayContainsAll => Err(io::Error::new(
+ io::ErrorKind::Unsupported,
+ format!("Scalar bitmap index does not support {op}"),
+ )),
PredicateOperator::Eq => {
let key = serialize_bitmap_datum(&literals[0], data_type);
self.equal(&key, data_type).await
@@ -654,7 +660,7 @@ async fn read_compressible_block(reader: &dyn FileRead,
block: BlockInfo) -> io:
BlockCompressionType::None => Ok(block_bytes.to_vec()),
BlockCompressionType::Zstd => {
let mut cursor = Cursor::new(block_bytes);
- let uncompressed_size = decode_var_int(&mut cursor)? as usize;
+ let uncompressed_size = decode_uncompressed_size(&mut cursor)?;
let compressed_start = cursor.position() as usize;
let compressed_data = &block_bytes[compressed_start..];
let mut decompressed = vec![0u8; uncompressed_size];
@@ -670,14 +676,80 @@ async fn read_compressible_block(reader: &dyn FileRead,
block: BlockInfo) -> io:
}
Ok(decompressed)
}
- _ => Err(io::Error::new(
- io::ErrorKind::Unsupported,
+ BlockCompressionType::Lz4 | BlockCompressionType::Lzo => {
+ let mut cursor = Cursor::new(block_bytes);
+ let uncompressed_size = decode_uncompressed_size(&mut cursor)?;
+ let compressed_start = cursor.position() as usize;
+ decompress_java_header_block(
+ &block_bytes[compressed_start..],
+ uncompressed_size,
+ compression_type,
+ )
+ }
+ }
+}
+
+fn decode_uncompressed_size(input: &mut impl Read) -> io::Result<usize> {
+ let size = decode_var_int(input)?;
+ usize::try_from(size).map_err(|_| {
+ io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!("Invalid bitmap block uncompressed size: {size}"),
+ )
+ })
+}
+
+/// Decode Java's LZ4/LZO block envelope:
+/// `[compressed_len: i32 LE][original_len: i32 LE][raw codec payload]`.
+fn decompress_java_header_block(
+ block: &[u8],
+ expected_size: usize,
+ compression_type: BlockCompressionType,
+) -> io::Result<Vec<u8>> {
+ if block.len() < 8 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ "Compressed bitmap block is shorter than the Java codec header",
+ ));
+ }
+ let compressed_len = i32::from_le_bytes(block[..4].try_into().unwrap());
+ let original_len = i32::from_le_bytes(block[4..8].try_into().unwrap());
+ let (compressed_len, original_len) = match (
+ usize::try_from(compressed_len),
+ usize::try_from(original_len),
+ ) {
+ (Ok(compressed_len), Ok(original_len))
+ if original_len == expected_size
+ && compressed_len <= block.len().saturating_sub(8)
+ && ((original_len == 0) == (compressed_len == 0)) =>
+ {
+ (compressed_len, original_len)
+ }
+ _ => {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ "Invalid Java LZ4/LZO bitmap block lengths",
+ ))
+ }
+ };
+ let payload = &block[8..8 + compressed_len];
+ let decompressed = match compression_type {
+ BlockCompressionType::Lz4 => lz4_flex::block::decompress(payload,
original_len)
+ .map_err(|error| io::Error::new(io::ErrorKind::InvalidData,
error))?,
+ BlockCompressionType::Lzo => lzokay_native::decompress_all(payload,
Some(original_len))
+ .map_err(|error| io::Error::new(io::ErrorKind::InvalidData,
error))?,
+ _ => unreachable!("only Java header codecs use this decoder"),
+ };
+ if decompressed.len() != original_len {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
format!(
- "Bitmap global index compression type {:?} is not supported",
- compression_type
+ "Bitmap block decompressed size mismatch: expected
{original_len}, got {}",
+ decompressed.len()
),
- )),
+ ));
}
+ Ok(decompressed)
}
fn compute_crc32(data: &[u8], compression_type: BlockCompressionType) -> u32 {
@@ -1047,6 +1119,68 @@ mod tests {
(reader, entries)
}
+ fn java_codec_envelope(compressed: &[u8], original_len: usize) -> Vec<u8> {
+ let mut encoded = Vec::with_capacity(8 + compressed.len());
+
encoded.extend_from_slice(&i32::try_from(compressed.len()).unwrap().to_le_bytes());
+
encoded.extend_from_slice(&i32::try_from(original_len).unwrap().to_le_bytes());
+ encoded.extend_from_slice(compressed);
+ encoded
+ }
+
+ #[tokio::test]
+ async fn test_decode_java_lz4_and_lzo_bitmap_blocks() {
+ use base64::Engine;
+
+ let original = b"java-compatible multivalue bitmap block".repeat(16);
+
+ let lz4 = lz4_flex::block::compress(&original);
+ assert_eq!(
+ decompress_java_header_block(
+ &java_codec_envelope(&lz4, original.len()),
+ original.len(),
+ BlockCompressionType::Lz4,
+ )
+ .unwrap(),
+ original
+ );
+
+ // Produced by Airlift 2.0.3 `LzoCompressor`, the implementation used
+ // by Java Paimon. Keep this as a cross-language golden rather than a
+ // Rust self-roundtrip.
+ let lzo = base64::engine::general_purpose::STANDARD
+
.decode("OGphdmEtY29tcGF0aWJsZSBtdWx0aXZhbHVlIGJpdG1hcCBibG9jayAAACWYAAJibG9jaxEAAA==")
+ .unwrap();
+ assert_eq!(
+ decompress_java_header_block(
+ &java_codec_envelope(&lzo, original.len()),
+ original.len(),
+ BlockCompressionType::Lzo,
+ )
+ .unwrap(),
+ original
+ );
+
+ let mut block = Vec::new();
+ encode_var_int(&mut block,
i32::try_from(original.len()).unwrap()).unwrap();
+ block.extend_from_slice(&java_codec_envelope(&lzo, original.len()));
+ let block_len = block.len();
+ let crc = compute_crc32(&block, BlockCompressionType::Lzo);
+ block.push(BlockCompressionType::Lzo as u8);
+ block.extend_from_slice(&crc.to_le_bytes());
+ assert_eq!(
+ read_compressible_block(
+ &BytesFileRead(Bytes::from(block)),
+ BlockInfo {
+ offset: 0,
+ length: block_len,
+ },
+ )
+ .await
+ .unwrap(),
+ original
+ );
+ }
+
#[test]
fn test_bitmap_floating_residual_sensitive_operator_set() {
for op in [
