You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by be...@apache.org on 2026/10/21 13:00:37 UTC
(fluss) branch main updated: [rust] Reuse predicate binding and add local Arrow evaluation (#4717)
This is an automated email from the ASF dual-hosted git repository.
beryllw pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 75e919b1a [rust] Reuse predicate binding and add local Arrow evaluation (#4717)
75e919b1a is described below
commit 75e919b1ac57b5b3af48a0a92386dcbd5b95f93a
Author: naivedogger <ya...@gmail.com>
AuthorDate: Wed Oct 21 21:00:30 2026 +0800
[rust] Reuse predicate binding and add local Arrow evaluation (#4717)
* [rust] Reuse predicate binding and add local Arrow evaluation
---
.../crates/fluss/src/predicate/arrow_eval.rs | 1445 ++++++++++++++++++++
fluss-rust/crates/fluss/src/predicate/binding.rs | 403 ++++++
fluss-rust/crates/fluss/src/predicate/coercion.rs | 204 +++
fluss-rust/crates/fluss/src/predicate/mod.rs | 106 +-
fluss-rust/crates/fluss/src/predicate/numeric.rs | 454 ++++++
fluss-rust/crates/fluss/src/predicate/pb.rs | 205 +--
fluss-rust/crates/fluss/src/record/statistics.rs | 44 +-
fluss-rust/crates/fluss/src/row/column_vector.rs | 11 +
fluss-rust/crates/fluss/src/util/mod.rs | 15 +
fluss-rust/crates/fluss/src/util/partition.rs | 120 ++
10 files changed, 2812 insertions(+), 195 deletions(-)
diff --git a/fluss-rust/crates/fluss/src/predicate/arrow_eval.rs b/fluss-rust/crates/fluss/src/predicate/arrow_eval.rs
new file mode 100644
index 000000000..8f8505f09
--- /dev/null
+++ b/fluss-rust/crates/fluss/src/predicate/arrow_eval.rs
@@ -0,0 +1,1445 @@
+// 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.
+
+//! Exact Arrow evaluation of schema-bound core predicates.
+
+use super::coercion::{validate_leaf, validate_leaf_shape};
+use crate::error::Error::IllegalArgument;
+use crate::error::Result;
+use crate::metadata::{DataField, DataType as FlussDataType};
+use crate::predicate::{BoundPredicate, CompoundFunction, LeafFunction, Literal};
+use crate::record::to_arrow_type;
+use crate::row::column_vector::{TimestampColumn, TypedColumn};
+use crate::row::{Datum, TimestampLtz, TimestampNtz};
+use crate::util::partition::parse_value_of_type;
+use crate::util::{canonicalize_nan, compare_floats};
+use arrow::array::{Array, ArrayRef, AsArray, BooleanArray, BooleanBuilder, make_builder};
+use arrow::compute::kernels::boolean::{and, is_not_null, is_null, or};
+use arrow::datatypes::{DataType as ArrowDataType, Float32Type, Float64Type};
+use arrow::record_batch::RecordBatch;
+use arrow::row::{RowConverter, SortField};
+use std::cmp::Ordering;
+use std::collections::HashSet;
+
+impl BoundPredicate {
+ /// Prepares an owned evaluator, converting constants and building membership sets once.
+ ///
+ /// Rejects malformed leaf shapes and errors while preparing constants. The evaluator is
+ /// immutable and can be shared across batch readers; its buffers
+ /// are released when it is dropped. Batch column names and types are checked
+ /// on each evaluation. Prefer this over `evaluate_batch` for a stream of batches.
+ #[doc(hidden)]
+ pub fn into_evaluator(self) -> Result<PredicateEvaluator> {
+ Ok(PredicateEvaluator {
+ root: PreparedPredicate::new(&self)?,
+ })
+ }
+
+ /// Evaluates a leaf against a non-null, encoded Fluss partition value.
+ ///
+ /// Uses the same exact filter semantics as [`Self::evaluate_batch`]. Returns
+ /// an error for non-leaves, invalid encodings, or unsupported types.
+ /// Temporal values retain their core precision, independently of Arrow units.
+ /// A pruning caller must retain the partition on error.
+ #[doc(hidden)]
+ pub fn evaluate_partition_value(&self, value: &str) -> Result<bool> {
+ let Self::Leaf {
+ field_name,
+ data_type,
+ function,
+ literals,
+ ..
+ } = self
+ else {
+ return Err(IllegalArgument {
+ message: "partition value evaluation requires a leaf predicate".into(),
+ });
+ };
+ let datum = parse_value_of_type(value, data_type)?;
+ validate_leaf_shape(*function, field_name, literals)?;
+ // Partition strings are core values, not Arrow rows. In particular a
+ // timestamp must not be narrowed to the column's physical unit here.
+ match &datum {
+ Datum::TimestampNtz(value) => {
+ return evaluate_value(*value, *function, literals, |literal| match literal {
+ Literal::TimestampNtz(value) => Ok(*value),
+ _ => Err(literal_mismatch()),
+ });
+ }
+ Datum::TimestampLtz(value) => {
+ return evaluate_value(*value, *function, literals, |literal| match literal {
+ Literal::TimestampLtz(value) => Ok(*value),
+ _ => Err(literal_mismatch()),
+ });
+ }
+ Datum::Time(value) => {
+ return evaluate_value(value.get_inner(), *function, literals, |literal| {
+ match literal {
+ Literal::Time(value) => Ok(*value),
+ _ => Err(literal_mismatch()),
+ }
+ });
+ }
+ _ => {}
+ }
+ let array = datum_array(&datum, data_type)?;
+ let batch = RecordBatch::try_from_iter([(field_name.as_str(), array)])?;
+ Ok(self.evaluate_batch(&batch)?.value(0))
+ }
+
+ /// Evaluates this predicate exactly against an Arrow record batch.
+ ///
+ /// Returns a non-null filter mask. Except for `IS NULL`, null input values
+ /// do not match; `NOT IN` with a null literal matches no rows.
+ /// Prepares a temporary evaluator per call; use [`Self::into_evaluator`]
+ /// to reuse converted constants and membership sets across batches.
+ #[doc(hidden)]
+ pub fn evaluate_batch(&self, batch: &RecordBatch) -> Result<BooleanArray> {
+ PreparedPredicate::new(self)?.evaluate_batch(batch)
+ }
+}
+
+fn predicate_column<'a>(
+ batch: &'a RecordBatch,
+ field_name: &str,
+ data_type: &FlussDataType,
+) -> Result<&'a ArrayRef> {
+ let array = batch
+ .column_by_name(field_name)
+ .ok_or_else(|| IllegalArgument {
+ message: format!("Arrow batch is missing predicate column '{field_name}'"),
+ })?;
+ let expected_arrow_type = to_arrow_type(data_type)?;
+ if array.data_type() != &expected_arrow_type {
+ return Err(IllegalArgument {
+ message: format!(
+ "predicate column '{field_name}' has Arrow type {}, expected {expected_arrow_type}",
+ array.data_type()
+ ),
+ });
+ }
+
+ Ok(array)
+}
+
+fn compare(
+ array: &ArrayRef,
+ function: LeafFunction,
+ operand: &Operand,
+ data_type: &FlussDataType,
+) -> Result<BooleanArray> {
+ let column = TypedColumn::build(array.as_ref(), data_type)?;
+ match (&column, operand) {
+ (TypedColumn::Timestamp(_), Operand::Literal(Literal::TimestampNtz(bound))) => {
+ timestamp_mask(array, &column, TypedColumn::get_timestamp_ntz, |value| {
+ matches_order(
+ value.map_or_else(|side| side, |value| value.cmp(bound)),
+ function,
+ )
+ })
+ }
+ (TypedColumn::TimestampLtz(_), Operand::Literal(Literal::TimestampLtz(bound))) => {
+ timestamp_mask(array, &column, TypedColumn::get_timestamp_ltz, |value| {
+ matches_order(
+ value.map_or_else(|side| side, |value| value.cmp(bound)),
+ function,
+ )
+ })
+ }
+ (TypedColumn::Time(time), Operand::Literal(Literal::Time(bound))) => Ok(compare_values(
+ (0..array.len()).map(|i| array.is_valid(i).then(|| time.get_nanos(i))),
+ i128::from(*bound) * 1_000_000,
+ function,
+ |a, b| a.cmp(&b),
+ )),
+ (TypedColumn::Binary(values), Operand::Literal(Literal::Bytes(bound))) => Ok(
+ compare_values(values.iter(), bound.as_slice(), function, compare_binary),
+ ),
+ (TypedColumn::Bytes(values), Operand::Literal(Literal::Bytes(bound))) => Ok(
+ compare_values(values.iter(), bound.as_slice(), function, compare_binary),
+ ),
+ (_, Operand::Scalar(constant)) => compare_scalar(&column, function, constant),
+ _ => Err(literal_mismatch()),
+ }
+}
+
+fn literal_mismatch() -> crate::error::Error {
+ IllegalArgument {
+ message: "predicate literal does not match its column type".into(),
+ }
+}
+
+fn compare_scalar(
+ column: &TypedColumn,
+ function: LeafFunction,
+ constant: &TypedColumn,
+) -> Result<BooleanArray> {
+ macro_rules! compare {
+ ($array:expr, $bound:expr, $cmp:expr) => {
+ compare_values($array.iter(), $bound.value(0), function, $cmp)
+ };
+ ($array:expr, $bound:expr) => {
+ compare!($array, $bound, |a, b| a.cmp(&b))
+ };
+ }
+ Ok(match (column, constant) {
+ (TypedColumn::Boolean(a), TypedColumn::Boolean(b)) => compare!(a, b),
+ (TypedColumn::TinyInt(a), TypedColumn::TinyInt(b)) => compare!(a, b),
+ (TypedColumn::SmallInt(a), TypedColumn::SmallInt(b)) => compare!(a, b),
+ (TypedColumn::Int(a), TypedColumn::Int(b)) => compare!(a, b),
+ (TypedColumn::BigInt(a), TypedColumn::BigInt(b)) => compare!(a, b),
+ (TypedColumn::Date(a), TypedColumn::Date(b)) => compare!(a, b),
+ (TypedColumn::Float(a), TypedColumn::Float(b)) => {
+ compare!(a, b, |a, b| compare_floats(f64::from(a), f64::from(b)))
+ }
+ (TypedColumn::Double(a), TypedColumn::Double(b)) => compare!(a, b, compare_floats),
+ (TypedColumn::String(a), TypedColumn::String(b)) => match function {
+ LeafFunction::StartsWith | LeafFunction::EndsWith | LeafFunction::Contains => a
+ .iter()
+ .map(|value| value.is_some_and(|value| matches_string(value, b.value(0), function)))
+ .collect(),
+ _ => compare!(a, b),
+ },
+ // Binding has already aligned the scale. Compare unscaled values rather
+ // than allocating a core Decimal/BigDecimal for every row.
+ (TypedColumn::Decimal(a, _), TypedColumn::Decimal(b, _)) => compare!(a, b),
+ _ => return Err(literal_mismatch()),
+ })
+}
+
+// Arrow seconds span a wider range than core timestamps. Classify only those
+// out-of-range rows before using the existing core timestamp accessors.
+fn timestamp_mask<T>(
+ array: &ArrayRef,
+ column: &TypedColumn,
+ read: impl Fn(&TypedColumn, usize) -> Result<T>,
+ matches: impl Fn(std::result::Result<T, Ordering>) -> bool,
+) -> Result<BooleanArray> {
+ let mut mask = BooleanBuilder::with_capacity(array.len());
+ for i in 0..array.len() {
+ if array.is_null(i) {
+ mask.append_value(false);
+ continue;
+ }
+ if let TypedColumn::Timestamp(TimestampColumn::Second(seconds))
+ | TypedColumn::TimestampLtz(TimestampColumn::Second(seconds)) = column
+ {
+ let value = seconds.value(i);
+ if value.checked_mul(1_000).is_none() {
+ mask.append_value(matches(Err(value.cmp(&0))));
+ continue;
+ }
+ }
+ mask.append_value(matches(Ok(read(column, i)?)));
+ }
+ Ok(mask.finish())
+}
+
+fn evaluate_value<T: Ord + Copy>(
+ value: T,
+ function: LeafFunction,
+ literals: &[Literal],
+ extract: impl Fn(&Literal) -> Result<T>,
+) -> Result<bool> {
+ match function {
+ LeafFunction::IsNull => Ok(false),
+ LeafFunction::IsNotNull => Ok(true),
+ LeafFunction::In | LeafFunction::NotIn => {
+ if function == LeafFunction::NotIn && literals.iter().any(Literal::is_null) {
+ return Ok(false);
+ }
+ let mut found = false;
+ for literal in literals.iter().filter(|literal| !literal.is_null()) {
+ found |= value == extract(literal)?;
+ }
+ Ok(found != (function == LeafFunction::NotIn))
+ }
+ LeafFunction::StartsWith | LeafFunction::EndsWith | LeafFunction::Contains => {
+ Err(literal_mismatch())
+ }
+ _ => Ok(matches_order(value.cmp(&extract(&literals[0])?), function)),
+ }
+}
+
+fn matches_order(ordering: Ordering, function: LeafFunction) -> bool {
+ match function {
+ LeafFunction::Equal => ordering.is_eq(),
+ LeafFunction::NotEqual => !ordering.is_eq(),
+ LeafFunction::LessThan => ordering.is_lt(),
+ LeafFunction::LessOrEqual => !ordering.is_gt(),
+ LeafFunction::GreaterThan => ordering.is_gt(),
+ LeafFunction::GreaterOrEqual => !ordering.is_lt(),
+ _ => unreachable!("non-comparison leaf reached compare"),
+ }
+}
+
+fn compare_values<T: Copy>(
+ values: impl Iterator<Item = Option<T>>,
+ constant: T,
+ function: LeafFunction,
+ compare: impl Fn(T, T) -> Ordering,
+) -> BooleanArray {
+ values
+ .map(|value| value.is_some_and(|value| matches_order(compare(value, constant), function)))
+ .collect()
+}
+
+fn compare_binary(left: &[u8], right: &[u8]) -> Ordering {
+ // Java byte[] comparison is signed, unlike Arrow's binary ordering.
+ left.iter()
+ .map(|b| *b as i8)
+ .cmp(right.iter().map(|b| *b as i8))
+}
+
+fn matches_string(value: &str, expected: &str, function: LeafFunction) -> bool {
+ match function {
+ LeafFunction::StartsWith => value.starts_with(expected),
+ LeafFunction::EndsWith => value.ends_with(expected),
+ LeafFunction::Contains => value.contains(expected),
+ _ => unreachable!("non-string function reached string evaluator"),
+ }
+}
+
+/// Builds one scalar value, rather than repeating the literal for every input row.
+fn literal_array(literal: &Literal, data_type: &FlussDataType) -> Result<ArrayRef> {
+ datum_array(&literal.to_datum(), data_type)
+}
+
+fn datum_array(datum: &Datum<'_>, data_type: &FlussDataType) -> Result<ArrayRef> {
+ let arrow_type = to_arrow_type(data_type)?;
+ // Temporal predicates use core values directly, not the row writer's
+ // narrowing conversions.
+ let mut builder = make_builder(&arrow_type, 1);
+ datum.append_to(builder.as_mut(), data_type, &arrow_type)?;
+ Ok(builder.finish())
+}
+
+// Constant folding here is for a filter mask, not a nullable SQL expression.
+fn constant_filter(array: &ArrayRef, matches: bool) -> Result<BooleanArray> {
+ if matches {
+ is_not_null(array.as_ref()).map_err(Into::into)
+ } else {
+ Ok(BooleanArray::from(vec![false; array.len()]))
+ }
+}
+
+/// An immutable evaluator with constants and membership sets prepared for repeated batches.
+#[doc(hidden)]
+#[derive(Debug)]
+pub struct PredicateEvaluator {
+ root: PreparedPredicate,
+}
+
+impl PredicateEvaluator {
+ /// Returns a non-null filter mask, using the same semantics and batch schema
+ /// checks as [`BoundPredicate::evaluate_batch`].
+ pub fn evaluate_batch(&self, batch: &RecordBatch) -> Result<BooleanArray> {
+ self.root.evaluate_batch(batch)
+ }
+}
+
+#[derive(Debug)]
+enum PreparedPredicate {
+ AlwaysTrue,
+ Leaf {
+ field_name: String,
+ data_type: FlussDataType,
+ function: LeafFunction,
+ operand: Operand,
+ },
+ Compound {
+ function: CompoundFunction,
+ children: Vec<PreparedPredicate>,
+ },
+}
+
+impl PreparedPredicate {
+ fn new(predicate: &BoundPredicate) -> Result<Self> {
+ Ok(match predicate {
+ BoundPredicate::AlwaysTrue => Self::AlwaysTrue,
+ BoundPredicate::Compound { function, children } => Self::Compound {
+ function: *function,
+ children: children.iter().map(Self::new).collect::<Result<_>>()?,
+ },
+ BoundPredicate::Leaf {
+ field_name,
+ data_type,
+ function,
+ literals,
+ ..
+ } => {
+ validate_leaf(
+ *function,
+ &DataField::new(field_name, data_type.clone(), None),
+ literals,
+ )?;
+ Self::Leaf {
+ field_name: field_name.clone(),
+ data_type: data_type.clone(),
+ function: *function,
+ operand: Operand::new(*function, literals, data_type)?,
+ }
+ }
+ })
+ }
+
+ fn evaluate_batch(&self, batch: &RecordBatch) -> Result<BooleanArray> {
+ match self {
+ Self::AlwaysTrue => Ok(BooleanArray::from(vec![true; batch.num_rows()])),
+ Self::Leaf {
+ field_name,
+ data_type,
+ function,
+ operand,
+ } => {
+ let array = predicate_column(batch, field_name, data_type)?;
+ match function {
+ LeafFunction::IsNull => is_null(array.as_ref()).map_err(Into::into),
+ LeafFunction::IsNotNull => is_not_null(array.as_ref()).map_err(Into::into),
+ _ => match operand {
+ Operand::Constant(matches) => constant_filter(array, *matches),
+ Operand::Membership(set) => set.evaluate(array, data_type, *function),
+ _ => compare(array, *function, operand, data_type),
+ },
+ }
+ }
+ Self::Compound { function, children } => combine_masks(
+ *function,
+ batch.num_rows(),
+ children.iter().map(|child| child.evaluate_batch(batch)),
+ ),
+ }
+ }
+}
+
+#[derive(Debug)]
+enum Operand {
+ None,
+ Constant(bool),
+ Literal(Literal),
+ Scalar(Box<TypedColumn>),
+ Membership(MembershipSet),
+}
+
+impl Operand {
+ fn new(
+ function: LeafFunction,
+ literals: &[Literal],
+ data_type: &FlussDataType,
+ ) -> Result<Self> {
+ match function {
+ LeafFunction::IsNull | LeafFunction::IsNotNull => return Ok(Self::None),
+ LeafFunction::In | LeafFunction::NotIn => {
+ return Ok(if literals.is_empty() {
+ Self::Constant(function == LeafFunction::NotIn)
+ } else if function == LeafFunction::NotIn && literals.iter().any(Literal::is_null) {
+ Self::Constant(false)
+ } else {
+ Self::Membership(MembershipSet::new(literals, data_type)?)
+ });
+ }
+ _ => {}
+ }
+ let literal = &literals[0];
+ match (data_type, literal) {
+ (FlussDataType::Binary(_) | FlussDataType::Bytes(_), Literal::Bytes(_))
+ | (FlussDataType::Time(_), Literal::Time(_))
+ | (FlussDataType::Timestamp(_), Literal::TimestampNtz(_))
+ | (FlussDataType::TimestampLTz(_), Literal::TimestampLtz(_)) => {
+ // Comparison bounds keep their precision and length; row-writer
+ // narrowing and fixed-width checks apply only to stored values.
+ Ok(Self::Literal(literal.clone()))
+ }
+ (
+ FlussDataType::Binary(_)
+ | FlussDataType::Bytes(_)
+ | FlussDataType::Time(_)
+ | FlussDataType::Timestamp(_)
+ | FlussDataType::TimestampLTz(_),
+ _,
+ ) => Err(literal_mismatch()),
+ _ => Ok(Self::Scalar(Box::new(TypedColumn::build(
+ literal_array(literal, data_type)?.as_ref(),
+ data_type,
+ )?))),
+ }
+ }
+}
+
+fn combine_masks(
+ function: CompoundFunction,
+ len: usize,
+ masks: impl Iterator<Item = Result<BooleanArray>>,
+) -> Result<BooleanArray> {
+ let mut result = BooleanArray::from(vec![function == CompoundFunction::And; len]);
+ for mask in masks {
+ let mask = mask?;
+ result = match function {
+ CompoundFunction::And => and(&result, &mask),
+ CompoundFunction::Or => or(&result, &mask),
+ }?;
+ }
+ Ok(result)
+}
+
+#[derive(Debug)]
+enum MembershipSet {
+ Bytes(HashSet<Vec<u8>>),
+ Timestamps(HashSet<TimestampNtz>),
+ TimestampsLtz(HashSet<TimestampLtz>),
+ Times(HashSet<i128>),
+ Floats(HashSet<u64>),
+ Rows {
+ converter: RowConverter,
+ values: HashSet<Box<[u8]>>,
+ },
+}
+
+impl MembershipSet {
+ fn new(literals: &[Literal], data_type: &FlussDataType) -> Result<Self> {
+ macro_rules! core_set {
+ ($literal:ident, $variant:ident, $value:ident => $convert:expr) => {{
+ let values = literals
+ .iter()
+ .filter(|literal| !literal.is_null())
+ .map(|literal| match literal {
+ Literal::$literal($value) => Ok($convert),
+ _ => Err(literal_mismatch()),
+ })
+ .collect::<Result<HashSet<_>>>()?;
+ return Ok(Self::$variant(values));
+ }};
+ }
+ match data_type {
+ FlussDataType::Binary(_) | FlussDataType::Bytes(_) => {
+ core_set!(Bytes, Bytes, value => value.clone())
+ }
+ FlussDataType::Timestamp(_) => core_set!(TimestampNtz, Timestamps, value => *value),
+ FlussDataType::TimestampLTz(_) => {
+ core_set!(TimestampLtz, TimestampsLtz, value => *value)
+ }
+ FlussDataType::Time(_) => {
+ core_set!(Time, Times, value => i128::from(*value) * 1_000_000)
+ }
+ _ => {}
+ }
+ let arrow_type = to_arrow_type(data_type)?;
+ if matches!(arrow_type, ArrowDataType::Float32 | ArrowDataType::Float64) {
+ let mut values = HashSet::with_capacity(literals.len());
+ for literal in literals.iter().filter(|literal| !literal.is_null()) {
+ let scalar = literal_array(literal, data_type)?;
+ let value = match arrow_type {
+ ArrowDataType::Float32 => {
+ f64::from(scalar.as_primitive::<Float32Type>().value(0))
+ }
+ _ => scalar.as_primitive::<Float64Type>().value(0),
+ };
+ values.insert(canonicalize_nan(value).to_bits());
+ }
+ return Ok(Self::Floats(values));
+ }
+ let converter = RowConverter::new(vec![SortField::new(arrow_type)])?;
+ let mut values = HashSet::with_capacity(literals.len());
+ for literal in literals.iter().filter(|literal| !literal.is_null()) {
+ let scalar = literal_array(literal, data_type)?;
+ let rows = converter.convert_columns(&[scalar])?;
+ values.insert(rows.row(0).as_ref().into());
+ }
+ Ok(Self::Rows { converter, values })
+ }
+
+ fn evaluate(
+ &self,
+ array: &ArrayRef,
+ data_type: &FlussDataType,
+ function: LeafFunction,
+ ) -> Result<BooleanArray> {
+ let negate = function == LeafFunction::NotIn;
+ let (converter, values) = match self {
+ Self::Bytes(values) => {
+ let matches = |value: Option<&[u8]>| {
+ value.is_some_and(|value| values.contains(value) != negate)
+ };
+ return Ok(match TypedColumn::build(array.as_ref(), data_type)? {
+ TypedColumn::Binary(array) => array.iter().map(matches).collect(),
+ TypedColumn::Bytes(array) => array.iter().map(matches).collect(),
+ _ => return Err(literal_mismatch()),
+ });
+ }
+ Self::Timestamps(values) => {
+ let column = TypedColumn::build(array.as_ref(), data_type)?;
+ return timestamp_mask(array, &column, TypedColumn::get_timestamp_ntz, |value| {
+ value.is_ok_and(|value| values.contains(&value)) != negate
+ });
+ }
+ Self::TimestampsLtz(values) => {
+ let column = TypedColumn::build(array.as_ref(), data_type)?;
+ return timestamp_mask(array, &column, TypedColumn::get_timestamp_ltz, |value| {
+ value.is_ok_and(|value| values.contains(&value)) != negate
+ });
+ }
+ Self::Times(values) => {
+ let TypedColumn::Time(time) = TypedColumn::build(array.as_ref(), data_type)? else {
+ return Err(literal_mismatch());
+ };
+ return Ok((0..array.len())
+ .map(|i| array.is_valid(i) && (values.contains(&time.get_nanos(i)) != negate))
+ .collect());
+ }
+ Self::Floats(values) => {
+ // Canonicalize only NaNs; signed-zero keys remain distinct.
+ // Probe fixed-size keys without allocating per row or copying a batch.
+ let matches = |value: Option<f64>| {
+ value.is_some_and(|value| {
+ values.contains(&canonicalize_nan(value).to_bits()) != negate
+ })
+ };
+ return Ok(match array.data_type() {
+ ArrowDataType::Float32 => array
+ .as_primitive::<Float32Type>()
+ .iter()
+ .map(|value| matches(value.map(f64::from)))
+ .collect(),
+ ArrowDataType::Float64 => array
+ .as_primitive::<Float64Type>()
+ .iter()
+ .map(matches)
+ .collect(),
+ _ => unreachable!("membership column type was validated"),
+ });
+ }
+ Self::Rows { converter, values } => (converter, values),
+ };
+ if values.is_empty() {
+ return constant_filter(array, negate);
+ }
+ // For the remaining scalar types, Arrow row encoding agrees with Fluss equality.
+ let rows = converter.convert_columns(std::slice::from_ref(array))?;
+ Ok(rows
+ .iter()
+ .enumerate()
+ .map(|(index, row)| array.is_valid(index) && (values.contains(row.as_ref()) != negate))
+ .collect())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::metadata::{DataField, DataTypes, RowType};
+ use crate::predicate::tests::assert_comparisons;
+ use crate::predicate::{Literal, Predicate, col};
+ use crate::row::{TimestampLtz, TimestampNtz};
+ use arrow::array::{
+ ArrayRef, Decimal128Array, FixedSizeBinaryArray, Int32Array, Int64Array, StringArray,
+ new_empty_array,
+ };
+ use arrow::compute::cast;
+ use arrow::compute::kernels::boolean::not;
+ use std::sync::Arc;
+
+ #[test]
+ fn partition_values_use_typed_parsing_and_exact_filters() {
+ use crate::util::partition::convert_value_of_type;
+ let cases = [
+ (DataTypes::tinyint(), Literal::Int8(-1), "-1"),
+ (DataTypes::smallint(), Literal::Int16(256), "256"),
+ (DataTypes::int(), Literal::Int32(-1000), "-1000"),
+ (
+ DataTypes::bigint(),
+ Literal::Int64(i64::MAX),
+ "9223372036854775807",
+ ),
+ (DataTypes::boolean(), Literal::Bool(true), "true"),
+ (DataTypes::float(), Literal::Float32(1.5), "1_5"),
+ (DataTypes::double(), Literal::Float64(-1.5), "-1_5"),
+ (DataTypes::double(), Literal::Float64(-0.0), "-0_0"),
+ (DataTypes::float(), Literal::Float32(f32::INFINITY), "Inf"),
+ (DataTypes::float(), Literal::Float32(-f32::NAN), "NaN"),
+ (DataTypes::double(), Literal::Float64(-f64::NAN), "NaN"),
+ (
+ DataTypes::double(),
+ Literal::Float64(f64::NEG_INFINITY),
+ "-Inf",
+ ),
+ (DataTypes::string(), Literal::String("US".into()), "US"),
+ (DataTypes::char(2), Literal::String("US".into()), "US"),
+ (DataTypes::bytes(), Literal::Bytes(vec![0x00, 0xff]), "00ff"),
+ (
+ DataTypes::binary(2),
+ Literal::Bytes(vec![0x00, 0xff]),
+ "00ff",
+ ),
+ (DataTypes::date(), Literal::Date(-1), "1969-12-31"),
+ (DataTypes::date(), Literal::Date(19_782), "2024-02-29"),
+ (
+ DataTypes::time_with_precision(3),
+ Literal::Time(3_723_004),
+ "01-02-03_004",
+ ),
+ (
+ DataTypes::timestamp_with_precision(9),
+ Literal::TimestampNtz(TimestampNtz::from_millis_nanos(-1, 900_000).unwrap()),
+ "1969-12-31-23-59-59_9999",
+ ),
+ (
+ DataTypes::timestamp_ltz_with_precision(9),
+ Literal::TimestampLtz(TimestampLtz::from_millis_nanos(1, 900_000).unwrap()),
+ "1970-01-01-00-00-00_0019",
+ ),
+ ];
+ for (data_type, literal, encoded) in cases {
+ let datum = literal.to_datum();
+ assert_eq!(
+ parse_value_of_type(encoded, &data_type).unwrap(),
+ datum,
+ "{encoded}"
+ );
+ let round_trip = convert_value_of_type(&datum, &data_type).unwrap();
+ let row_type = RowType::new(vec![DataField::new("p", data_type, None)]);
+ for (predicate, expected) in [
+ (col("p").eq(literal.clone()), true),
+ (col("p").ne(literal.clone()), false),
+ (col("p").is_in([literal.clone()]), true),
+ (col("p").not_in([literal.clone(), Literal::Null]), false),
+ (col("p").is_null(), false),
+ (col("p").is_not_null(), true),
+ ] {
+ let bound = BoundPredicate::bind(Some(&predicate), &row_type).unwrap();
+ for value in [encoded, round_trip.as_str()] {
+ assert_eq!(
+ bound.evaluate_partition_value(value).unwrap(),
+ expected,
+ "{bound:?}: {value}"
+ );
+ }
+ }
+ }
+ }
+
+ #[test]
+ fn partition_evaluation_rejects_invalid_values_and_preserves_temporal_precision() {
+ for (data_type, value) in [
+ (DataTypes::tinyint(), "128"),
+ (DataTypes::boolean(), "invalid"),
+ (DataTypes::bytes(), "0ff"),
+ (DataTypes::binary(2), "ff"),
+ (DataTypes::bytes(), "é"),
+ (DataTypes::date(), "2026-02-30"),
+ (DataTypes::time(), "24-00-00_000"),
+ (DataTypes::timestamp(), "1970-01-01-00-00-00_1234567890"),
+ ] {
+ let row_type = RowType::new(vec![DataField::new("p", data_type, None)]);
+ let bound = BoundPredicate::bind(Some(&col("p").is_not_null()), &row_type).unwrap();
+ assert!(bound.evaluate_partition_value(value).is_err(), "{value}");
+ }
+ for precision in [0, 3, 6, 9] {
+ for (data_type, literal) in [
+ (
+ DataTypes::timestamp_with_precision(precision),
+ Literal::TimestampNtz(TimestampNtz::from_millis_nanos(1, 900_000).unwrap()),
+ ),
+ (
+ DataTypes::timestamp_ltz_with_precision(precision),
+ Literal::TimestampLtz(TimestampLtz::from_millis_nanos(1, 900_000).unwrap()),
+ ),
+ ] {
+ let row_type = RowType::new(vec![DataField::new("p", data_type, None)]);
+ for (predicate, expected) in [
+ (col("p").lt(literal.clone()), [true, false, false, false]),
+ (col("p").eq(literal.clone()), [false, true, false, false]),
+ (
+ col("p").is_in([literal.clone()]),
+ [false, true, false, false],
+ ),
+ (col("p").not_in([literal]), [true, false, true, true]),
+ ] {
+ let bound = BoundPredicate::bind(Some(&predicate), &row_type).unwrap();
+ for (value, expected) in [
+ "1970-01-01-00-00-00_001",
+ "1970-01-01-00-00-00_0019",
+ "1970-01-01-00-00-00_002",
+ "3000-01-01",
+ ]
+ .into_iter()
+ .zip(expected)
+ {
+ assert_eq!(
+ bound.evaluate_partition_value(value).unwrap(),
+ expected,
+ "{bound:?}: {value}"
+ );
+ }
+ }
+ }
+ }
+ }
+
+ #[test]
+ fn binary_comparisons_use_signed_bytes() {
+ use Ordering::{Equal, Greater, Less};
+ use arrow::array::BinaryArray;
+ let bytes: &[Option<&[u8]>] = &[
+ Some(&[0x80]),
+ Some(&[0xff]),
+ Some(&[0]),
+ Some(&[0x7f]),
+ None,
+ ];
+ for (data_type, array) in [
+ (
+ DataTypes::bytes(),
+ std::sync::Arc::new(BinaryArray::from(bytes.to_vec())) as ArrayRef,
+ ),
+ (
+ DataTypes::binary(1),
+ std::sync::Arc::new(
+ FixedSizeBinaryArray::try_from_sparse_iter_with_size(bytes.iter().copied(), 1)
+ .unwrap(),
+ ) as ArrayRef,
+ ),
+ ] {
+ assert_comparisons(
+ data_type,
+ array,
+ Literal::Bytes(vec![0]),
+ vec![Some(Less), Some(Less), Some(Equal), Some(Greater), None],
+ );
+ }
+ assert_comparisons(
+ DataTypes::bytes(),
+ std::sync::Arc::new(BinaryArray::from(vec![
+ Some(&[][..]),
+ Some(&[0][..]),
+ Some(&[0, 0x80][..]),
+ Some(&[0, 0][..]),
+ None,
+ ])),
+ Literal::Bytes(vec![0, 0]),
+ vec![Some(Less), Some(Less), Some(Less), Some(Equal), None],
+ );
+ }
+
+ #[test]
+ fn nan_signs_and_payloads_share_one_equality_class() {
+ use arrow::array::{Float32Array, Float64Array};
+ // Explicit Java expectations, independent of Arrow/total_cmp ordering.
+ let bits = [
+ 0x7ff8_0000_0000_0000,
+ 0xfff8_0000_0000_0000,
+ 0x7ff8_0000_2000_0000,
+ 0xfff8_0000_2000_0000,
+ 0x7ff0_0000_2000_0000,
+ ];
+ for single in [false, true] {
+ let (data_type, array): (_, ArrayRef) = if single {
+ (
+ DataTypes::float(),
+ std::sync::Arc::new(Float32Array::from(
+ bits.iter()
+ .map(|bits| Some(f64::from_bits(*bits) as f32))
+ .chain([None])
+ .collect::<Vec<_>>(),
+ )),
+ )
+ } else {
+ (
+ DataTypes::double(),
+ std::sync::Arc::new(Float64Array::from(
+ bits.iter()
+ .map(|bits| Some(f64::from_bits(*bits)))
+ .chain([None])
+ .collect::<Vec<_>>(),
+ )),
+ )
+ };
+ for literal in [f64::NAN, f64::from_bits(bits[3])] {
+ assert_comparisons(
+ data_type.clone(),
+ array.clone(),
+ Literal::Float64(literal),
+ vec![
+ Some(Ordering::Equal),
+ Some(Ordering::Equal),
+ Some(Ordering::Equal),
+ Some(Ordering::Equal),
+ Some(Ordering::Equal),
+ None,
+ ],
+ );
+ }
+ }
+ }
+
+ // Independent comparison path for membership differential tests.
+ fn evaluate_membership_linear(
+ array: &ArrayRef,
+ data_type: &FlussDataType,
+ function: LeafFunction,
+ literals: &[Literal],
+ len: usize,
+ ) -> Result<BooleanArray> {
+ let mut result = BooleanArray::from(vec![false; len]);
+ for literal in literals.iter().filter(|literal| !literal.is_null()) {
+ let operand = Operand::new(
+ LeafFunction::Equal,
+ std::slice::from_ref(literal),
+ data_type,
+ )?;
+ let mask = compare(array, LeafFunction::Equal, &operand, data_type)?;
+ result = or(&result, &mask)?;
+ }
+
+ if function == LeafFunction::In {
+ return Ok(result);
+ }
+
+ let non_null = is_not_null(array.as_ref())?;
+ and(&non_null, ¬(&result)?).map_err(Into::into)
+ }
+
+ fn row_type() -> RowType {
+ RowType::new(vec![
+ DataField::new("id", DataTypes::int(), None),
+ DataField::new("name", DataTypes::string(), None),
+ ])
+ }
+
+ fn batch() -> RecordBatch {
+ RecordBatch::try_from_iter(vec![
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![Some(1), None, Some(3)])) as ArrayRef,
+ ),
+ (
+ "name",
+ Arc::new(StringArray::from(vec![
+ Some("alpha"),
+ None,
+ Some("alphabet"),
+ ])) as ArrayRef,
+ ),
+ ])
+ .unwrap()
+ }
+
+ fn evaluate(predicate: Predicate) -> Vec<bool> {
+ BoundPredicate::bind(Some(&predicate), &row_type())
+ .unwrap()
+ .evaluate_batch(&batch())
+ .unwrap()
+ .values()
+ .iter()
+ .collect()
+ }
+
+ #[test]
+ fn evaluates_string_functions_and_null_inputs_exactly() {
+ assert_eq!(
+ evaluate(col("name").starts_with("alpha")),
+ vec![true, false, true]
+ );
+ assert_eq!(
+ evaluate(col("name").ends_with("bet")),
+ vec![false, false, true]
+ );
+ assert_eq!(
+ evaluate(col("name").contains("ph")),
+ vec![true, false, true]
+ );
+ }
+
+ #[test]
+ fn compound_predicates_treat_null_leaf_results_as_false() {
+ assert_eq!(
+ evaluate(col("id").eq(1_i32).or(col("name").is_null())),
+ vec![true, true, false]
+ );
+ }
+
+ #[test]
+ fn in_and_not_in_match_core_null_semantics() {
+ for (predicate, expected) in [
+ (
+ col("id").is_in(vec![Some(1_i32), None]),
+ [true, false, false],
+ ),
+ (
+ col("id").not_in(vec![Some(1_i32), None]),
+ [false, false, false],
+ ),
+ (col("id").is_in(Vec::<i32>::new()), [false, false, false]),
+ (col("id").not_in(Vec::<i32>::new()), [true, false, true]),
+ ] {
+ assert_eq!(evaluate(predicate), expected);
+ }
+ }
+
+ #[test]
+ fn manually_constructed_leaves_reject_invalid_shapes_and_types() {
+ let batch = RecordBatch::try_from_iter([(
+ "c",
+ Arc::new(StringArray::from(vec![Some("a"), None])) as ArrayRef,
+ )])
+ .unwrap();
+ let leaf = |function, literals| BoundPredicate::Leaf {
+ field_index: 0,
+ field_id: 0,
+ field_name: "c".into(),
+ data_type: DataTypes::string(),
+ function,
+ literals,
+ };
+ for function in [
+ LeafFunction::Equal,
+ LeafFunction::NotEqual,
+ LeafFunction::LessThan,
+ LeafFunction::LessOrEqual,
+ LeafFunction::GreaterThan,
+ LeafFunction::GreaterOrEqual,
+ LeafFunction::StartsWith,
+ LeafFunction::EndsWith,
+ LeafFunction::Contains,
+ ] {
+ for literals in [
+ vec![],
+ vec![Literal::String("a".into()); 2],
+ vec![Literal::Null],
+ ] {
+ let predicate = leaf(function, literals);
+ assert!(matches!(
+ predicate.evaluate_batch(&batch),
+ Err(IllegalArgument { .. })
+ ));
+ assert!(matches!(
+ predicate.into_evaluator(),
+ Err(IllegalArgument { .. })
+ ));
+ }
+ }
+ for function in [LeafFunction::IsNull, LeafFunction::IsNotNull] {
+ assert!(
+ leaf(function, vec![Literal::Null])
+ .evaluate_batch(&batch)
+ .is_err()
+ );
+ }
+ let mut predicate = leaf(LeafFunction::StartsWith, vec![Literal::Int32(1)]);
+ if let BoundPredicate::Leaf { data_type, .. } = &mut predicate {
+ *data_type = DataTypes::int();
+ }
+ assert!(predicate.into_evaluator().is_err());
+ }
+
+ fn assert_membership_strategies(
+ data_type: FlussDataType,
+ array: ArrayRef,
+ literals: Vec<Literal>,
+ ) {
+ let row_type = RowType::new(vec![DataField::new("c", data_type.clone(), None)]);
+ let batch = RecordBatch::try_from_iter([("c", array.clone())]).unwrap();
+ for function in [LeafFunction::In, LeafFunction::NotIn] {
+ for with_null in [false, true] {
+ let mut literals = literals.clone();
+ literals.extend(literals.clone());
+ if with_null {
+ literals.push(Literal::Null);
+ }
+ let predicate = Predicate::Leaf {
+ field: "c".into(),
+ function,
+ literals,
+ };
+ let bound = BoundPredicate::bind(Some(&predicate), &row_type).unwrap();
+ let BoundPredicate::Leaf { literals, .. } = &bound else {
+ unreachable!()
+ };
+ let expected = if function == LeafFunction::NotIn && with_null {
+ BooleanArray::from(vec![false; array.len()])
+ } else {
+ evaluate_membership_linear(&array, &data_type, function, literals, array.len())
+ .unwrap()
+ };
+ assert_eq!(bound.evaluate_batch(&batch).unwrap(), expected);
+ let evaluator = bound.into_evaluator().unwrap();
+ assert_eq!(evaluator.evaluate_batch(&batch).unwrap(), expected);
+ assert_eq!(
+ evaluator.evaluate_batch(&batch.slice(0, 0)).unwrap().len(),
+ 0
+ );
+ }
+ }
+ }
+
+ #[test]
+ fn hashed_membership_matches_linear_comparison_for_scalar_types() {
+ use arrow::array::{BinaryArray, Float64Array};
+ let nan = f64::from_bits(f64::NAN.to_bits() + (1 << 30));
+ for single in [false, true] {
+ let values = vec![
+ Some(-0.0),
+ Some(0.0),
+ Some(f64::NAN),
+ Some(-f64::NAN),
+ Some(nan),
+ Some(1.0),
+ Some(2.0),
+ None,
+ ];
+ let data_type = if single {
+ DataTypes::float()
+ } else {
+ DataTypes::double()
+ };
+ let array = cast(
+ &Float64Array::from(values),
+ &to_arrow_type(&data_type).unwrap(),
+ )
+ .unwrap();
+ let literals = [-0.0, nan, 1.0]
+ .into_iter()
+ .map(|v| {
+ if single {
+ Literal::Float32(v as f32)
+ } else {
+ Literal::Float64(v)
+ }
+ })
+ .collect();
+ assert_membership_strategies(data_type, array, literals);
+ }
+ assert_membership_strategies(
+ DataTypes::int(),
+ Arc::new(Int32Array::from(vec![Some(-1), Some(0), Some(1), None])),
+ vec![Literal::Int32(-1), Literal::Int32(0)],
+ );
+ assert_membership_strategies(
+ DataTypes::string(),
+ Arc::new(StringArray::from(vec![
+ Some(""),
+ Some("a"),
+ Some("ab"),
+ Some("你好"),
+ None,
+ ])),
+ vec![Literal::String("a".into()), Literal::String("你好".into())],
+ );
+ assert_membership_strategies(
+ DataTypes::bytes(),
+ Arc::new(BinaryArray::from(vec![
+ Some(b"a".as_slice()),
+ Some(b"ab".as_slice()),
+ None,
+ ])),
+ vec![Literal::Bytes(b"a".to_vec())],
+ );
+ assert_membership_strategies(
+ DataTypes::binary(2),
+ Arc::new(
+ FixedSizeBinaryArray::try_from_sparse_iter_with_size(
+ [Some(b"ab".as_slice()), Some(b"cd".as_slice()), None].into_iter(),
+ 2,
+ )
+ .unwrap(),
+ ),
+ vec![Literal::Bytes(b"ab".to_vec())],
+ );
+ let values = vec![Some(-(10_i128.pow(38) - 1)), Some(123), Some(124), None];
+ assert_membership_strategies(
+ DataTypes::decimal(38, 2),
+ Arc::new(
+ Decimal128Array::from(values)
+ .with_precision_and_scale(38, 2)
+ .unwrap(),
+ ),
+ vec![Literal::Decimal(
+ crate::row::Decimal::from_unscaled_bytes(&123_i128.to_be_bytes(), 38, 2).unwrap(),
+ )],
+ );
+ }
+
+ #[test]
+ fn prepared_evaluator_reuses_constants_and_validates_each_batch() {
+ fn assert_send_sync<T: Send + Sync>() {}
+ assert_send_sync::<PredicateEvaluator>();
+ let predicate = col("id")
+ .is_in(0..128)
+ .and(col("id").ge(1))
+ .and(col("name").starts_with("alpha"))
+ .and(col("name").is_not_null());
+ let bound = BoundPredicate::bind(Some(&predicate), &row_type()).unwrap();
+ let evaluator = bound.clone().into_evaluator().unwrap();
+ for (input, expected) in [
+ (batch(), vec![true, false, true]),
+ (batch().slice(1, 2), vec![false, true]),
+ (batch().slice(0, 0), vec![]),
+ (
+ RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![0, 2, 3])) as ArrayRef),
+ (
+ "name",
+ Arc::new(StringArray::from(vec!["alpha", "beta", "alpha"])) as ArrayRef,
+ ),
+ ])
+ .unwrap(),
+ vec![false, false, true],
+ ),
+ ] {
+ let expected = BooleanArray::from(expected);
+ assert_eq!(evaluator.evaluate_batch(&input).unwrap(), expected);
+ assert_eq!(bound.evaluate_batch(&input).unwrap(), expected);
+ }
+ let missing = RecordBatch::try_from_iter([("name", batch().column(1).clone())]).unwrap();
+ let wrong = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int64Array::from(vec![1])) as ArrayRef),
+ ("name", Arc::new(StringArray::from(vec!["a"])) as ArrayRef),
+ ])
+ .unwrap();
+ // Constant folding must not bypass validation of the referenced column.
+ for predicate in [
+ predicate,
+ col("id").is_in(Vec::<i32>::new()),
+ col("id").not_in([Literal::Null]),
+ ] {
+ let bound = BoundPredicate::bind(Some(&predicate), &row_type()).unwrap();
+ let evaluator = bound.clone().into_evaluator().unwrap();
+ for invalid in [&missing, &wrong] {
+ assert!(bound.evaluate_batch(invalid).is_err());
+ assert!(evaluator.evaluate_batch(invalid).is_err());
+ }
+ }
+ }
+
+ #[test]
+ fn binary_literals_allow_different_lengths_but_columns_keep_their_width() {
+ use Ordering::{Equal, Greater, Less};
+ use arrow::array::BinaryArray;
+ let values = [Some(&[1, 2, 3, 4][..]), None];
+ for (data_type, array) in [
+ (
+ DataTypes::bytes(),
+ Arc::new(BinaryArray::from(values.to_vec())) as ArrayRef,
+ ),
+ (
+ DataTypes::binary(4),
+ Arc::new(
+ FixedSizeBinaryArray::try_from_sparse_iter_with_size(values.into_iter(), 4)
+ .unwrap(),
+ ) as ArrayRef,
+ ),
+ ] {
+ for (literal, ordering) in [
+ (vec![], Greater),
+ (vec![1, 2, 3], Greater),
+ (vec![1, 2, 3, 4], Equal),
+ (vec![1, 2, 3, 4, 0], Less),
+ (vec![1, 2, 4], Less),
+ (vec![1, 2, 3, 0x80], Greater),
+ ] {
+ assert_comparisons(
+ data_type.clone(),
+ array.clone(),
+ Literal::Bytes(literal),
+ vec![Some(ordering), None],
+ );
+ }
+ assert_membership_strategies(
+ data_type,
+ array,
+ vec![vec![], vec![1, 2, 3], vec![1, 2, 3, 4], vec![1, 2, 3, 4, 0]]
+ .into_iter()
+ .map(Literal::Bytes)
+ .collect(),
+ );
+ }
+ let row_type = RowType::new(vec![DataField::new("c", DataTypes::binary(4), None)]);
+ let bound =
+ BoundPredicate::bind(Some(&col("c").ne(Literal::Bytes(vec![1, 2, 3]))), &row_type)
+ .unwrap();
+ assert!(bound.evaluate_partition_value("01020304").unwrap());
+ assert!(bound.evaluate_partition_value("010203").is_err());
+ let evaluator = bound.clone().into_evaluator().unwrap();
+ for wrong_type in [DataTypes::bytes(), DataTypes::binary(3)] {
+ let batch = RecordBatch::try_from_iter([(
+ "c",
+ new_empty_array(&to_arrow_type(&wrong_type).unwrap()),
+ )])
+ .unwrap();
+ assert!(bound.evaluate_batch(&batch).is_err());
+ assert!(evaluator.evaluate_batch(&batch).is_err());
+ }
+ }
+
+ #[test]
+ fn scalar_comparisons_preserve_nulls_and_operators() {
+ assert_comparisons(
+ DataTypes::int(),
+ Arc::new(Int32Array::from(vec![Some(1), None, Some(2), Some(3)])),
+ Literal::Int32(2),
+ vec![
+ Some(Ordering::Less),
+ None,
+ Some(Ordering::Equal),
+ Some(Ordering::Greater),
+ ],
+ );
+ }
+
+ #[test]
+ fn core_converted_scalars_preserve_comparison_semantics() {
+ for (data_type, literal) in [
+ (DataTypes::boolean(), Literal::Bool(true)),
+ (DataTypes::int(), Literal::Int32(-3)),
+ (DataTypes::double(), Literal::Float64(-2.5)),
+ (DataTypes::string(), Literal::String("hello".into())),
+ (DataTypes::bytes(), Literal::Bytes(vec![0, 255])),
+ (DataTypes::binary(4), Literal::Bytes(vec![1, 2, 3, 4])),
+ (DataTypes::date(), Literal::Date(-1)),
+ (DataTypes::time_with_precision(0), Literal::Time(1_000)),
+ (DataTypes::time_with_precision(9), Literal::Time(1)),
+ ] {
+ let scalar = literal_array(&literal, &data_type).unwrap();
+ assert_eq!(scalar.len(), 1);
+ assert_eq!(scalar.data_type(), &to_arrow_type(&data_type).unwrap());
+ let indices = arrow::array::UInt32Array::from(vec![Some(0), None, Some(0)]);
+ let array = arrow::compute::take(scalar.as_ref(), &indices, None).unwrap();
+ assert_comparisons(
+ data_type,
+ array,
+ literal,
+ vec![Some(Ordering::Equal), None, Some(Ordering::Equal)],
+ );
+ }
+ }
+
+ #[test]
+ fn temporal_comparisons_match_exact_nanosecond_ordering() {
+ for (precision, millis, nanos) in [
+ (0_u32, 1_500, 0),
+ (0, -1_500, 0),
+ (1, 1, 1),
+ (3, -1, 999_999),
+ (3, i64::from(i32::MIN), 0),
+ (6, 1, 1),
+ (6, -9_223_372_036_854_776, 192_000),
+ (6, i64::MAX, 0),
+ (9, 1, 1),
+ (9, i64::from(i32::MAX), 0),
+ (9, -9_223_372_036_855, 224_191),
+ (9, -9_223_372_036_855, 224_192),
+ (9, 9_223_372_036_854, 775_807),
+ (9, 9_223_372_036_854, 775_808),
+ ] {
+ let tick_nanos = 10_i128.pow(9 - precision.div_ceil(3) * 3);
+ let mut cases = vec![
+ (
+ DataTypes::timestamp_with_precision(precision),
+ Literal::TimestampNtz(TimestampNtz::from_millis_nanos(millis, nanos).unwrap()),
+ ),
+ (
+ DataTypes::timestamp_ltz_with_precision(precision),
+ Literal::TimestampLtz(TimestampLtz::from_millis_nanos(millis, nanos).unwrap()),
+ ),
+ ];
+ if nanos == 0 {
+ if let Ok(millis) = i32::try_from(millis) {
+ cases.push((
+ DataTypes::time_with_precision(precision),
+ Literal::Time(millis),
+ ));
+ }
+ }
+ let literal_nanos = i128::from(millis) * 1_000_000 + i128::from(nanos);
+ for (data_type, literal) in cases {
+ let arrow_type = to_arrow_type(&data_type).unwrap();
+ let time32 = matches!(arrow_type, ArrowDataType::Time32(_));
+ let (min, max) = if time32 {
+ (i128::from(i32::MIN), i128::from(i32::MAX))
+ } else {
+ (i128::from(i64::MIN), i128::from(i64::MAX))
+ };
+ let near = literal_nanos / tick_nanos;
+ let values: Vec<_> = [min, max, -2, -1, 0, 1, 2, near - 1, near, near + 1]
+ .into_iter()
+ .filter(|v| (min..=max).contains(v))
+ .map(|v| Some(v as i64))
+ .chain([None])
+ .collect();
+ let integers: ArrayRef = if time32 {
+ Arc::new(Int32Array::from(
+ values
+ .iter()
+ .map(|v| v.map(|v| v as i32))
+ .collect::<Vec<_>>(),
+ ))
+ } else {
+ Arc::new(Int64Array::from(values.clone()))
+ };
+ let array = cast(integers.as_ref(), &arrow_type).unwrap();
+ // Compare in the exact nanosecond domain, independently of the core timestamp accessors.
+ let orderings = values
+ .iter()
+ .map(|v| v.map(|v| (i128::from(v) * tick_nanos).cmp(&literal_nanos)))
+ .collect();
+ assert_comparisons(data_type, array, literal, orderings);
+ }
+ }
+ }
+
+ #[test]
+ fn temporal_membership_keeps_exact_literals() {
+ let data_type = DataTypes::timestamp_with_precision(0);
+ let row_type = RowType::new(vec![DataField::new("c", data_type.clone(), None)]);
+ let array = cast(
+ &Int64Array::from(vec![Some(1), None, Some(2)]),
+ &to_arrow_type(&data_type).unwrap(),
+ )
+ .unwrap();
+ let batch = RecordBatch::try_from_iter([("c", array)]).unwrap();
+ let literals = vec![
+ Literal::TimestampNtz(TimestampNtz::new(1_500)),
+ Literal::TimestampNtz(TimestampNtz::new(2_000)),
+ ];
+ let mut with_null = literals.clone();
+ with_null.push(Literal::Null);
+ for (predicate, expected) in [
+ (col("c").is_in(literals.clone()), [false, false, true]),
+ (col("c").not_in(literals), [true, false, false]),
+ (col("c").is_in(with_null.clone()), [false, false, true]),
+ (col("c").not_in(with_null), [false, false, false]),
+ ] {
+ let mask = BoundPredicate::bind(Some(&predicate), &row_type)
+ .unwrap()
+ .evaluate_batch(&batch)
+ .unwrap();
+ assert_eq!(mask.iter().collect::<Vec<_>>(), expected.map(Some));
+ }
+ }
+
+ #[test]
+ fn decimal_literals_preserve_signed_unscaled_values() {
+ for unscaled in [0, i128::from(i64::MAX) + 1, -(10_i128.pow(38) - 1)] {
+ let value =
+ crate::row::Decimal::from_unscaled_bytes(&unscaled.to_be_bytes(), 38, 3).unwrap();
+ let array =
+ literal_array(&Literal::Decimal(value), &DataTypes::decimal(38, 3)).unwrap();
+ let array = array.as_any().downcast_ref::<Decimal128Array>().unwrap();
+ assert_eq!(array.values().as_ref(), &[unscaled]);
+ assert_eq!(array.data_type(), &ArrowDataType::Decimal128(38, 3));
+ }
+ }
+}
diff --git a/fluss-rust/crates/fluss/src/predicate/binding.rs b/fluss-rust/crates/fluss/src/predicate/binding.rs
new file mode 100644
index 000000000..1a3451cfa
--- /dev/null
+++ b/fluss-rust/crates/fluss/src/predicate/binding.rs
@@ -0,0 +1,403 @@
+// 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.
+
+//! Schema-bound predicates for local evaluation and conservative pruning.
+
+use super::coercion::{compound_name, mismatch, resolve_field, unsupported_column, validate_leaf};
+use crate::error::Error::IllegalArgument;
+use crate::error::Result;
+use crate::metadata::{DataField, DataType, RowType};
+use crate::predicate::{CompoundFunction, LeafFunction, Literal, Predicate};
+use std::collections::HashSet;
+
+/// A core predicate resolved against a concrete Fluss row type.
+///
+/// This type is public only so sibling workspace crates can share binding,
+/// coercion, pruning, and exact evaluation semantics.
+#[doc(hidden)]
+#[derive(Debug, Clone, PartialEq)]
+pub enum BoundPredicate {
+ AlwaysTrue,
+ Leaf {
+ field_index: usize,
+ field_name: String,
+ field_id: i32,
+ data_type: DataType,
+ function: LeafFunction,
+ literals: Vec<Literal>,
+ },
+ Compound {
+ function: CompoundFunction,
+ children: Vec<BoundPredicate>,
+ },
+}
+
+impl BoundPredicate {
+ /// Binds an optional predicate to `row_type`, rewriting numeric bounds exactly.
+ /// Numeric comparisons use the literal's value, not a lossy cast to the column
+ /// type. Floating comparisons follow Fluss: all NaNs compare equal and above
+ /// infinity, while signed zeros remain distinct. Other physical representations
+ /// are checked by the evaluator.
+ #[doc(hidden)]
+ pub fn bind(predicate: Option<&Predicate>, row_type: &RowType) -> Result<Self> {
+ match predicate {
+ Some(predicate) => Self::bind_predicate(predicate, row_type),
+ None => Ok(Self::AlwaysTrue),
+ }
+ }
+
+ fn bind_predicate(predicate: &Predicate, row_type: &RowType) -> Result<Self> {
+ match predicate {
+ Predicate::Leaf {
+ field,
+ function,
+ literals,
+ } => {
+ let (field_index, data_field) = resolve_field(row_type, field)?;
+ validate_leaf(*function, data_field, literals)?;
+ let (function, literals) =
+ match super::numeric::bind(data_field, *function, literals)? {
+ Some(rewritten) => rewritten,
+ None => (
+ *function,
+ literals
+ .iter()
+ .map(|literal| bind_literal(data_field, literal))
+ .collect::<Result<Vec<_>>>()?,
+ ),
+ };
+ Ok(Self::Leaf {
+ field_index,
+ field_name: field.clone(),
+ field_id: data_field.field_id(),
+ data_type: data_field.data_type().clone(),
+ function,
+ literals,
+ })
+ }
+ Predicate::Compound { function, children } => {
+ if children.is_empty() {
+ return Err(IllegalArgument {
+ message: format!("{} predicate has no children", compound_name(*function)),
+ });
+ }
+ Ok(Self::Compound {
+ function: *function,
+ children: children
+ .iter()
+ .map(|child| Self::bind_predicate(child, row_type))
+ .collect::<Result<Vec<_>>>()?,
+ })
+ }
+ }
+ }
+
+ /// Returns a conservative predicate for Fluss log-batch pruning.
+ ///
+ /// Uses the rewritten bounds, not the original literals, and excludes filters
+ /// whose server-side statistics semantics may discard matching rows. This
+ /// method does not negotiate backend capabilities. `None` means no pruning;
+ /// callers must still apply this bound predicate to the returned rows.
+ /// A partially supported `OR` is never pushed down.
+ #[doc(hidden)]
+ pub fn to_pushdown_predicate(&self) -> Option<Predicate> {
+ match self {
+ Self::AlwaysTrue => None,
+ Self::Leaf {
+ field_name,
+ data_type,
+ function,
+ literals,
+ ..
+ } => {
+ // TIME statistics store milliseconds even for finer-precision columns.
+ let null_checks_only =
+ matches!(data_type, DataType::Float(_) | DataType::Double(_))
+ || matches!(data_type, DataType::Time(time) if time.precision() > 3);
+ if null_checks_only
+ && !matches!(function, LeafFunction::IsNull | LeafFunction::IsNotNull)
+ {
+ return None;
+ }
+ // Server prefix pruning uses UTF-16 ordering, unlike UTF-8 min/max.
+ if matches!(function, LeafFunction::StartsWith) {
+ return None;
+ }
+ Some(Predicate::Leaf {
+ field: field_name.clone(),
+ function: *function,
+ literals: literals.clone(),
+ })
+ }
+ Self::Compound {
+ function: CompoundFunction::And,
+ children,
+ } => Predicate::and_all(children.iter().filter_map(Self::to_pushdown_predicate)),
+ Self::Compound {
+ function: CompoundFunction::Or,
+ children,
+ } => Predicate::or_all(
+ children
+ .iter()
+ .map(Self::to_pushdown_predicate)
+ .collect::<Option<Vec<_>>>()?,
+ ),
+ }
+ }
+
+ /// Returns referenced table field indexes in first-seen order.
+ #[doc(hidden)]
+ pub fn referenced_field_indexes(&self) -> Vec<usize> {
+ let mut indexes = Vec::new();
+ let mut seen = HashSet::new();
+ self.collect_referenced_field_indexes(&mut indexes, &mut seen);
+ indexes
+ }
+
+ fn collect_referenced_field_indexes(
+ &self,
+ indexes: &mut Vec<usize>,
+ seen: &mut HashSet<usize>,
+ ) {
+ match self {
+ Self::AlwaysTrue => {}
+ Self::Leaf { field_index, .. } => {
+ if seen.insert(*field_index) {
+ indexes.push(*field_index);
+ }
+ }
+ Self::Compound { children, .. } => {
+ for child in children {
+ child.collect_referenced_field_indexes(indexes, seen);
+ }
+ }
+ }
+ }
+}
+
+fn bind_literal(field: &DataField, literal: &Literal) -> Result<Literal> {
+ if matches!(literal, Literal::Null) {
+ return Ok(Literal::Null);
+ }
+
+ // Numeric bounds are normalized before this point. Temporal bounds keep
+ // their precision for comparison using the core timestamp types.
+ Ok(match (field.data_type(), literal) {
+ (DataType::Boolean(_), Literal::Bool(_))
+ | (DataType::Char(_) | DataType::String(_), Literal::String(_))
+ | (DataType::Binary(_) | DataType::Bytes(_), Literal::Bytes(_))
+ | (DataType::Date(_), Literal::Date(_))
+ | (DataType::Time(_), Literal::Time(_))
+ | (DataType::Timestamp(_), Literal::TimestampNtz(_))
+ | (DataType::TimestampLTz(_), Literal::TimestampLtz(_)) => literal.clone(),
+ (DataType::Array(_) | DataType::Map(_) | DataType::Row(_), _) => {
+ return Err(unsupported_column(field));
+ }
+ _ => return Err(mismatch(field, literal)),
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::metadata::{DataField, DataTypes};
+ use crate::predicate::col;
+
+ fn row_type() -> RowType {
+ RowType::new(vec![
+ DataField::with_field_id("id", DataTypes::int(), None, 3),
+ DataField::with_field_id("name", DataTypes::string(), None, 5),
+ ])
+ }
+
+ #[test]
+ fn binds_field_identity_and_coerced_literals_once() {
+ let bound = BoundPredicate::bind(Some(&col("id").eq(1_i8)), &row_type()).unwrap();
+
+ assert_eq!(
+ bound,
+ BoundPredicate::Leaf {
+ field_index: 0,
+ field_name: "id".to_string(),
+ field_id: 3,
+ data_type: DataTypes::int(),
+ function: LeafFunction::Equal,
+ literals: vec![Literal::Int32(1)],
+ }
+ );
+ }
+
+ #[test]
+ fn discovers_referenced_fields_in_first_seen_order() {
+ let predicate = col("name").starts_with("A").and(col("id").gt(1_i32));
+ let bound = BoundPredicate::bind(Some(&predicate), &row_type()).unwrap();
+
+ assert_eq!(bound.referenced_field_indexes(), vec![1, 0]);
+ }
+
+ #[test]
+ fn pushdown_keeps_only_statistics_safe_conditions() {
+ for (data_type, functions, literal, safe) in [
+ (
+ DataTypes::time_with_precision(3),
+ vec![LeafFunction::GreaterThan],
+ Literal::Time(0),
+ true,
+ ),
+ (
+ DataTypes::time_with_precision(6),
+ vec![LeafFunction::GreaterThan],
+ Literal::Time(0),
+ false,
+ ),
+ (
+ DataTypes::time_with_precision(9),
+ vec![
+ LeafFunction::Equal,
+ LeafFunction::NotEqual,
+ LeafFunction::LessThan,
+ LeafFunction::LessOrEqual,
+ LeafFunction::GreaterThan,
+ LeafFunction::GreaterOrEqual,
+ LeafFunction::In,
+ LeafFunction::NotIn,
+ ],
+ Literal::Time(0),
+ false,
+ ),
+ (
+ DataTypes::float(),
+ vec![LeafFunction::Equal],
+ Literal::Float32(0.0),
+ false,
+ ),
+ (
+ DataTypes::double(),
+ vec![LeafFunction::Equal],
+ Literal::Float64(0.0),
+ false,
+ ),
+ (
+ DataTypes::string(),
+ vec![LeafFunction::StartsWith],
+ Literal::String("\u{e000}".into()),
+ false,
+ ),
+ (
+ DataTypes::char(4),
+ vec![LeafFunction::StartsWith],
+ Literal::String("A".into()),
+ false,
+ ),
+ (
+ DataTypes::string(),
+ vec![LeafFunction::Equal, LeafFunction::LessThan],
+ Literal::String("A".into()),
+ true,
+ ),
+ ] {
+ let row_type = RowType::new(vec![DataField::new("c", data_type, None)]);
+ let pushdown = |predicate: &Predicate| {
+ BoundPredicate::bind(Some(predicate), &row_type)
+ .unwrap()
+ .to_pushdown_predicate()
+ };
+ let not_null = col("c").is_not_null();
+ assert_eq!(pushdown(¬_null), Some(not_null.clone()));
+ assert_eq!(pushdown(&col("c").is_null()), Some(col("c").is_null()));
+ for function in functions {
+ let predicate = Predicate::Leaf {
+ field: "c".into(),
+ function,
+ literals: vec![literal.clone()],
+ };
+ assert_eq!(
+ pushdown(&predicate),
+ safe.then(|| predicate.clone()),
+ "{row_type:?}, {function:?}"
+ );
+ if !safe {
+ assert_eq!(
+ pushdown(&predicate.clone().and(not_null.clone())),
+ Some(not_null.clone())
+ );
+ assert_eq!(pushdown(&predicate.or(not_null.clone())), None);
+ }
+ }
+ }
+
+ let row_type = RowType::new(vec![DataField::new("c", DataTypes::string(), None)]);
+ let batch = arrow::record_batch::RecordBatch::try_from_iter([(
+ "c",
+ std::sync::Arc::new(arrow::array::StringArray::from(vec![
+ Some("\u{e000}"),
+ Some("\u{10000}"),
+ None,
+ ])) as arrow::array::ArrayRef,
+ )])
+ .unwrap();
+ let bound =
+ BoundPredicate::bind(Some(&col("c").starts_with("\u{e000}")), &row_type).unwrap();
+ assert_eq!(
+ bound
+ .into_evaluator()
+ .unwrap()
+ .evaluate_batch(&batch)
+ .unwrap()
+ .values()
+ .iter()
+ .collect::<Vec<_>>(),
+ vec![true, false, false]
+ );
+ }
+
+ #[test]
+ fn rejects_invalid_shape_before_any_consumer_uses_the_predicate() {
+ let predicate = Predicate::Leaf {
+ field: "id".to_string(),
+ function: LeafFunction::IsNull,
+ literals: vec![Literal::Int32(1)],
+ };
+
+ let error = BoundPredicate::bind(Some(&predicate), &row_type()).unwrap_err();
+
+ assert!(error.to_string().contains("expects 0 literal(s), got 1"));
+ }
+
+ #[test]
+ fn local_binding_does_not_require_protocol_field_ids() {
+ let row_type = RowType::new(vec![DataField::new("id", DataTypes::int(), None)]);
+ let bound = BoundPredicate::bind(Some(&col("id").eq(1_i8)), &row_type).unwrap();
+
+ assert!(matches!(
+ bound,
+ BoundPredicate::Leaf { field_index: 0, ref literals, .. }
+ if literals == &[Literal::Int32(1)]
+ ));
+ }
+
+ #[test]
+ fn folds_unrepresentable_integer_equality_without_rounding() {
+ let row_type = RowType::new(vec![DataField::new("value", DataTypes::double(), None)]);
+
+ let bound = BoundPredicate::bind(Some(&col("value").eq(i64::MAX)), &row_type).unwrap();
+ assert!(
+ matches!(bound, BoundPredicate::Leaf { function: LeafFunction::In, literals, .. } if literals.is_empty())
+ );
+ BoundPredicate::bind(Some(&col("value").eq(9_007_199_254_740_994_i64)), &row_type).unwrap();
+ }
+}
diff --git a/fluss-rust/crates/fluss/src/predicate/coercion.rs b/fluss-rust/crates/fluss/src/predicate/coercion.rs
new file mode 100644
index 000000000..680912c44
--- /dev/null
+++ b/fluss-rust/crates/fluss/src/predicate/coercion.rs
@@ -0,0 +1,204 @@
+// 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.
+
+//! Field validation and literal coercion shared by wire encoding and local binding.
+
+use crate::error::Error::IllegalArgument;
+use crate::error::Result;
+use crate::metadata::{DataField, DataType, RowType};
+use crate::predicate::{CompoundFunction, LeafFunction, Literal};
+use crate::row::Decimal;
+
+pub(super) fn resolve_field<'a>(
+ row_type: &'a RowType,
+ name: &str,
+) -> Result<(usize, &'a DataField)> {
+ row_type
+ .get_field_index(name)
+ .map(|index| (index, &row_type.fields()[index]))
+ .ok_or_else(|| IllegalArgument {
+ message: format!(
+ "filter column '{}' does not exist in the table schema, available columns: {:?}",
+ name,
+ row_type.get_field_names()
+ ),
+ })
+}
+
+pub(super) fn validate_leaf(
+ function: LeafFunction,
+ field: &DataField,
+ literals: &[Literal],
+) -> Result<()> {
+ validate_leaf_shape(function, field.name(), literals)?;
+
+ if matches!(
+ function,
+ LeafFunction::StartsWith | LeafFunction::EndsWith | LeafFunction::Contains
+ ) && !matches!(field.data_type(), DataType::Char(_) | DataType::String(_))
+ {
+ return Err(IllegalArgument {
+ message: format!(
+ "{function:?} requires a character string column, but column '{}' is {}",
+ field.name(),
+ field.data_type()
+ ),
+ });
+ }
+ Ok(())
+}
+
+pub(super) fn validate_leaf_shape(
+ function: LeafFunction,
+ field_name: &str,
+ literals: &[Literal],
+) -> Result<()> {
+ let expected = match function {
+ LeafFunction::In | LeafFunction::NotIn => return Ok(()),
+ LeafFunction::IsNull | LeafFunction::IsNotNull => 0,
+ _ => 1,
+ };
+ if literals.len() != expected {
+ return Err(IllegalArgument {
+ message: format!(
+ "{function:?} on column '{}' expects {expected} literal(s), got {}",
+ field_name,
+ literals.len()
+ ),
+ });
+ }
+
+ if literals.first() == Some(&Literal::Null) {
+ return Err(IllegalArgument {
+ message: format!(
+ "{function:?} on column '{}' cannot take a null literal, use is_null()/is_not_null()",
+ field_name
+ ),
+ });
+ }
+
+ Ok(())
+}
+
+pub(super) fn integer_value(field: &DataField, literal: &Literal) -> Result<i64> {
+ match literal {
+ Literal::Int8(value) => Ok(i64::from(*value)),
+ Literal::Int16(value) => Ok(i64::from(*value)),
+ Literal::Int32(value) => Ok(i64::from(*value)),
+ Literal::Int64(value) => Ok(*value),
+ _ => Err(mismatch(field, literal)),
+ }
+}
+
+pub(super) fn integer_in_range(field: &DataField, literal: &Literal, max: i64) -> Result<i64> {
+ let value = integer_value(field, literal)?;
+ if value > max || value < -max - 1 {
+ return Err(IllegalArgument {
+ message: format!(
+ "filter literal {value} is out of range for column '{}' of type {}",
+ field.name(),
+ field.data_type()
+ ),
+ });
+ }
+ Ok(value)
+}
+
+pub(super) fn float32(field: &DataField, literal: &Literal) -> Result<f32> {
+ if let Literal::Float32(value) = literal {
+ return Ok(*value);
+ }
+ let value = float64(field, literal)?;
+ let narrowed = value as f32;
+ if f64::from(narrowed) != value && !value.is_nan() {
+ return Err(inexact(field, value));
+ }
+ Ok(narrowed)
+}
+
+pub(super) fn float64(field: &DataField, literal: &Literal) -> Result<f64> {
+ match literal {
+ Literal::Float64(value) => Ok(*value),
+ Literal::Float32(value) => Ok(f64::from(*value)),
+ Literal::Int8(_) | Literal::Int16(_) | Literal::Int32(_) | Literal::Int64(_) => {
+ let value = integer_value(field, literal)?;
+ let widened = value as f64;
+ if widened as i128 != i128::from(value) {
+ return Err(inexact(field, widened));
+ }
+ Ok(widened)
+ }
+ _ => Err(mismatch(field, literal)),
+ }
+}
+
+fn inexact(field: &DataField, value: f64) -> crate::error::Error {
+ IllegalArgument {
+ message: format!(
+ "filter literal {value} cannot be represented exactly by column '{}' of type {}",
+ field.name(),
+ field.data_type()
+ ),
+ }
+}
+
+pub(super) fn rescale(
+ field: &DataField,
+ value: &Decimal,
+ precision: u32,
+ scale: u32,
+) -> Result<Decimal> {
+ let big_decimal = value.to_big_decimal();
+ let rescaled = Decimal::from_big_decimal(big_decimal.clone(), precision, scale)?;
+ if rescaled.to_big_decimal() != big_decimal {
+ return Err(IllegalArgument {
+ message: format!(
+ "filter literal {big_decimal} does not fit column '{}' of type {}",
+ field.name(),
+ field.data_type()
+ ),
+ });
+ }
+ Ok(rescaled)
+}
+
+pub(super) fn mismatch(field: &DataField, literal: &Literal) -> crate::error::Error {
+ IllegalArgument {
+ message: format!(
+ "filter literal {literal:?} does not match column '{}' of type {}",
+ field.name(),
+ field.data_type()
+ ),
+ }
+}
+
+pub(super) fn unsupported_column(field: &DataField) -> crate::error::Error {
+ IllegalArgument {
+ message: format!(
+ "filter on column '{}' of type {} is not supported",
+ field.name(),
+ field.data_type()
+ ),
+ }
+}
+
+pub(super) fn compound_name(function: CompoundFunction) -> &'static str {
+ match function {
+ CompoundFunction::And => "AND",
+ CompoundFunction::Or => "OR",
+ }
+}
diff --git a/fluss-rust/crates/fluss/src/predicate/mod.rs b/fluss-rust/crates/fluss/src/predicate/mod.rs
index feb6aefb9..b87c2bb92 100644
--- a/fluss-rust/crates/fluss/src/predicate/mod.rs
+++ b/fluss-rust/crates/fluss/src/predicate/mod.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-//! Predicates for server-side filter pushdown on log scans.
+//! Predicates for server-side filter pushdown and local Arrow evaluation.
//!
//! ```rust
//! use fluss::predicate::col;
@@ -31,11 +31,19 @@
//! Setting a predicate on a scan is only supported on log scans over tables
//! with the ARROW log format.
+mod arrow_eval;
+mod binding;
+mod coercion;
+mod numeric;
mod pb;
+#[doc(hidden)]
+pub use arrow_eval::PredicateEvaluator;
+#[doc(hidden)]
+pub use binding::BoundPredicate;
pub(crate) use pb::to_pb_predicate;
-use crate::row::{Decimal, TimestampLtz, TimestampNtz};
+use crate::row::{Date, Datum, Decimal, Time, TimestampLtz, TimestampNtz};
/// Comparison applied by a [`Predicate::Leaf`] to one column.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
@@ -84,6 +92,36 @@ pub enum Literal {
TimestampLtz(TimestampLtz),
}
+impl Literal {
+ /// Returns whether this is a null literal.
+ #[doc(hidden)]
+ pub fn is_null(&self) -> bool {
+ matches!(self, Self::Null)
+ }
+
+ /// Converts this literal into a Fluss row datum.
+ #[doc(hidden)]
+ pub fn to_datum(&self) -> Datum<'static> {
+ match self {
+ Self::Null => Datum::Null,
+ Self::Bool(value) => (*value).into(),
+ Self::Int8(value) => (*value).into(),
+ Self::Int16(value) => (*value).into(),
+ Self::Int32(value) => (*value).into(),
+ Self::Int64(value) => (*value).into(),
+ Self::Float32(value) => (*value).into(),
+ Self::Float64(value) => (*value).into(),
+ Self::String(value) => value.clone().into(),
+ Self::Bytes(value) => value.clone().into(),
+ Self::Decimal(value) => value.clone().into(),
+ Self::Date(value) => Date::new(*value).into(),
+ Self::Time(value) => Time::new(*value).into(),
+ Self::TimestampNtz(value) => (*value).into(),
+ Self::TimestampLtz(value) => (*value).into(),
+ }
+ }
+}
+
macro_rules! impl_from_literal {
($($ty:ty => $variant:ident),* $(,)?) => {
$(
@@ -316,6 +354,70 @@ impl ColumnRef {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::metadata::{DataField, DataType, RowType};
+ use arrow::array::{Array, ArrayRef};
+ use arrow::record_batch::RecordBatch;
+ use std::cmp::Ordering;
+
+ pub(super) fn assert_comparisons(
+ data_type: DataType,
+ array: ArrayRef,
+ literal: Literal,
+ orderings: Vec<Option<Ordering>>,
+ ) {
+ let row_type = RowType::new(vec![DataField::with_field_id("c", data_type, None, 7)]);
+ let batch = RecordBatch::try_from_iter([("c", array)]).unwrap();
+ for function in [
+ LeafFunction::Equal,
+ LeafFunction::NotEqual,
+ LeafFunction::LessThan,
+ LeafFunction::LessOrEqual,
+ LeafFunction::GreaterThan,
+ LeafFunction::GreaterOrEqual,
+ LeafFunction::In,
+ LeafFunction::NotIn,
+ ] {
+ let predicate = Predicate::Leaf {
+ field: "c".into(),
+ function,
+ literals: vec![literal.clone()],
+ };
+ let bound = BoundPredicate::bind(Some(&predicate), &row_type).unwrap();
+ let actual = bound.evaluate_batch(&batch).unwrap();
+ let expected: Vec<_> = orderings
+ .iter()
+ .map(|ordering| {
+ ordering.is_some_and(|ordering| match function {
+ LeafFunction::Equal | LeafFunction::In => ordering == Ordering::Equal,
+ LeafFunction::NotEqual | LeafFunction::NotIn => ordering != Ordering::Equal,
+ LeafFunction::LessThan => ordering == Ordering::Less,
+ LeafFunction::LessOrEqual => ordering != Ordering::Greater,
+ LeafFunction::GreaterThan => ordering == Ordering::Greater,
+ LeafFunction::GreaterOrEqual => ordering != Ordering::Less,
+ _ => unreachable!(),
+ })
+ })
+ .collect();
+ assert_eq!(
+ actual.values().iter().collect::<Vec<_>>(),
+ expected,
+ "{predicate:?}, {row_type:?}"
+ );
+ let evaluator = bound.clone().into_evaluator().unwrap();
+ assert_eq!(evaluator.evaluate_batch(&batch).unwrap(), actual);
+ assert_eq!(
+ evaluator.evaluate_batch(&batch.slice(0, 0)).unwrap().len(),
+ 0
+ );
+ assert_eq!(actual.null_count(), 0);
+ assert_eq!(bound.evaluate_batch(&batch.slice(0, 0)).unwrap().len(), 0);
+ if let Some(pushdown) = bound.to_pushdown_predicate() {
+ to_pb_predicate(&pushdown, &row_type).unwrap();
+ let rebound = BoundPredicate::bind(Some(&pushdown), &row_type).unwrap();
+ assert_eq!(rebound.evaluate_batch(&batch).unwrap(), actual);
+ }
+ }
+ }
fn leaf(field: &str, function: LeafFunction, literals: Vec<Literal>) -> Predicate {
Predicate::Leaf {
diff --git a/fluss-rust/crates/fluss/src/predicate/numeric.rs b/fluss-rust/crates/fluss/src/predicate/numeric.rs
new file mode 100644
index 000000000..8530a7506
--- /dev/null
+++ b/fluss-rust/crates/fluss/src/predicate/numeric.rs
@@ -0,0 +1,454 @@
+// 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.
+
+//! Exact numeric boundary rewriting at bind time; wire coercion remains unchanged.
+
+use super::coercion::{float32, float64, integer_in_range, integer_value, mismatch, rescale};
+use crate::error::Result;
+use crate::metadata::{DataField, DataType};
+use crate::predicate::{LeafFunction, Literal};
+use crate::row::Decimal;
+use bigdecimal::num_traits::ToPrimitive;
+use bigdecimal::{BigDecimal, RoundingMode};
+use std::cmp::Ordering;
+
+// Value compares the converted boundary to the original literal; it must be
+// an adjacent representable value. Outside records the literal's side of the range.
+enum Bound {
+ Value(Literal, Ordering),
+ Outside(Ordering),
+}
+
+pub(super) fn bind(
+ field: &DataField,
+ function: LeafFunction,
+ literals: &[Literal],
+) -> Result<Option<(LeafFunction, Vec<Literal>)>> {
+ if !matches!(
+ field.data_type(),
+ DataType::TinyInt(_)
+ | DataType::SmallInt(_)
+ | DataType::Int(_)
+ | DataType::BigInt(_)
+ | DataType::Float(_)
+ | DataType::Double(_)
+ | DataType::Decimal(_)
+ ) {
+ return Ok(None);
+ }
+ let rewritten = match function {
+ LeafFunction::IsNull | LeafFunction::IsNotNull => (function, vec![]),
+ LeafFunction::In | LeafFunction::NotIn => {
+ let mut values = Vec::with_capacity(literals.len());
+ for literal in literals {
+ if literal.is_null() {
+ values.push(Literal::Null);
+ } else if let Bound::Value(value, Ordering::Equal) = convert(field, literal)? {
+ values.push(value);
+ }
+ }
+ (function, values)
+ }
+ _ => rewrite(function, convert(field, &literals[0])?),
+ };
+ Ok(Some(rewritten))
+}
+
+fn rewrite(function: LeafFunction, bound: Bound) -> (LeafFunction, Vec<Literal>) {
+ use LeafFunction::*;
+ let (value, direction) = match bound {
+ Bound::Outside(side) => {
+ return constant(match function {
+ Equal => false,
+ NotEqual => true,
+ LessThan | LessOrEqual => side == Ordering::Greater,
+ GreaterThan | GreaterOrEqual => side == Ordering::Less,
+ _ => unreachable!("non-comparison in numeric rewrite"),
+ });
+ }
+ Bound::Value(value, direction) => (value, direction),
+ };
+ let function = match (function, direction) {
+ (_, Ordering::Equal) => function,
+ (Equal, _) => return constant(false),
+ (NotEqual, _) => return constant(true),
+ (LessThan | LessOrEqual, Ordering::Less) => LessOrEqual,
+ (LessThan | LessOrEqual, Ordering::Greater) => LessThan,
+ (GreaterThan | GreaterOrEqual, Ordering::Less) => GreaterThan,
+ (GreaterThan | GreaterOrEqual, Ordering::Greater) => GreaterOrEqual,
+ _ => unreachable!("non-comparison in numeric rewrite"),
+ };
+ (function, vec![value])
+}
+
+// Keep field identity and null filtering without adding new predicate node kinds.
+fn constant(matches: bool) -> (LeafFunction, Vec<Literal>) {
+ (
+ if matches {
+ LeafFunction::IsNotNull
+ } else {
+ LeafFunction::In
+ },
+ vec![],
+ )
+}
+
+fn convert(field: &DataField, literal: &Literal) -> Result<Bound> {
+ // Preserve existing coercions for exact values and special floats.
+ if let Ok(value) = exact_coercion(field, literal) {
+ return Ok(Bound::Value(value, Ordering::Equal));
+ }
+ let value = match literal {
+ Literal::Int8(v) => BigDecimal::from(*v),
+ Literal::Int16(v) => BigDecimal::from(*v),
+ Literal::Int32(v) => BigDecimal::from(*v),
+ Literal::Int64(v) => BigDecimal::from(*v),
+ Literal::Decimal(v) => v.to_big_decimal(),
+ Literal::Float32(v) => return convert_float_literal(field, literal, f64::from(*v)),
+ Literal::Float64(v) => return convert_float_literal(field, literal, *v),
+ _ => return Err(mismatch(field, literal)),
+ };
+ convert_finite(field, literal, &value)
+}
+
+fn convert_float_literal(field: &DataField, literal: &Literal, value: f64) -> Result<Bound> {
+ if value.is_infinite() {
+ return Ok(Bound::Outside(if value.is_sign_negative() {
+ Ordering::Less
+ } else {
+ Ordering::Greater
+ }));
+ }
+ // Decode the binary value exactly: formatting the float itself as decimal
+ // text would hide its rounding error. NaN has no integer/decimal value.
+ let value = BigDecimal::try_from(value).map_err(|_| mismatch(field, literal))?;
+ convert_finite(field, literal, &value)
+}
+
+fn convert_finite(field: &DataField, original: &Literal, value: &BigDecimal) -> Result<Bound> {
+ let data_type = field.data_type();
+ if matches!(data_type, DataType::Float(_) | DataType::Double(_)) {
+ // Parse the exact decimal expansion directly into the target width;
+ // an intermediate f64 could double-round a boundary intended for f32.
+ let text = value.to_string();
+ let (literal, rounded) = if matches!(data_type, DataType::Float(_)) {
+ let narrowed = text.parse::<f32>().map_err(|_| mismatch(field, original))?;
+ (Literal::Float32(narrowed), f64::from(narrowed))
+ } else {
+ let rounded = text.parse::<f64>().map_err(|_| mismatch(field, original))?;
+ (Literal::Float64(rounded), rounded)
+ };
+ // Infinity is a representable float boundary, not an out-of-domain bound:
+ // the column may itself contain infinities or NaNs ordered beyond them.
+ let direction = if rounded.is_infinite() {
+ if rounded.is_sign_negative() {
+ Ordering::Less
+ } else {
+ Ordering::Greater
+ }
+ } else {
+ BigDecimal::try_from(rounded)
+ .map_err(|_| mismatch(field, original))?
+ .cmp(value)
+ };
+ return Ok(Bound::Value(literal, direction));
+ }
+ let (min, max, scale) = match data_type {
+ DataType::TinyInt(_) => (i128::from(i8::MIN), i128::from(i8::MAX), 0),
+ DataType::SmallInt(_) => (i128::from(i16::MIN), i128::from(i16::MAX), 0),
+ DataType::Int(_) => (i128::from(i32::MIN), i128::from(i32::MAX), 0),
+ DataType::BigInt(_) => (i128::from(i64::MIN), i128::from(i64::MAX), 0),
+ DataType::Decimal(t) => {
+ let max = 10_i128.pow(t.precision()) - 1;
+ (-max, max, i64::from(t.scale()))
+ }
+ _ => unreachable!("non-numeric field in numeric conversion"),
+ };
+ for (limit, side) in [(min, Ordering::Less), (max, Ordering::Greater)] {
+ if value.cmp(&BigDecimal::new(limit.into(), scale)) == side {
+ return Ok(Bound::Outside(side));
+ }
+ }
+ let rounded = value.with_scale_round(scale, RoundingMode::Floor);
+ let direction = rounded.cmp(value);
+ let literal = match data_type {
+ DataType::Decimal(t) => Literal::Decimal(Decimal::from_big_decimal(
+ rounded,
+ t.precision(),
+ t.scale(),
+ )?),
+ _ => {
+ let integer = rounded.to_i64().ok_or_else(|| mismatch(field, original))?;
+ exact_coercion(field, &Literal::Int64(integer))?
+ }
+ };
+ Ok(Bound::Value(literal, direction))
+}
+
+fn exact_coercion(field: &DataField, literal: &Literal) -> Result<Literal> {
+ Ok(match field.data_type() {
+ DataType::TinyInt(_) => {
+ Literal::Int8(integer_in_range(field, literal, i8::MAX as i64)? as i8)
+ }
+ DataType::SmallInt(_) => {
+ Literal::Int16(integer_in_range(field, literal, i16::MAX as i64)? as i16)
+ }
+ DataType::Int(_) => {
+ Literal::Int32(integer_in_range(field, literal, i32::MAX as i64)? as i32)
+ }
+ DataType::BigInt(_) => Literal::Int64(integer_value(field, literal)?),
+ DataType::Float(_) => Literal::Float32(float32(field, literal)?),
+ DataType::Double(_) => Literal::Float64(float64(field, literal)?),
+ DataType::Decimal(t) => match literal {
+ Literal::Decimal(v) => Literal::Decimal(rescale(field, v, t.precision(), t.scale())?),
+ _ => return Err(mismatch(field, literal)),
+ },
+ _ => unreachable!("non-numeric field in numeric conversion"),
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::metadata::{DataTypes, RowType};
+ use crate::predicate::tests::assert_comparisons;
+ use crate::predicate::{BoundPredicate, Predicate, col, to_pb_predicate};
+ use arrow::array::{ArrayRef, Decimal128Array, Float64Array, Int64Array};
+ use arrow::compute::cast;
+ use arrow::record_batch::RecordBatch;
+ use std::sync::Arc;
+
+ fn decimal(text: &str) -> Literal {
+ let value: BigDecimal = text.parse().unwrap();
+ let (_, scale) = value.as_bigint_and_exponent();
+ Literal::Decimal(Decimal::from_big_decimal(value, 38, scale as u32).unwrap())
+ }
+
+ #[test]
+ fn numeric_boundaries_preserve_comparison_results() {
+ use Ordering::{Equal as E, Greater as G, Less as L};
+ let decimal_max = 10_i128.pow(38) - 1;
+ // Decimal samples are unscaled; other samples are exact integer values.
+ // Expected orderings compare each sample to the original literal.
+ for (data_type, values, literal, expected) in [
+ (
+ DataTypes::tinyint(),
+ [-128, -1, 127],
+ Literal::Float64(-1.3),
+ [L, G, G],
+ ),
+ (
+ DataTypes::smallint(),
+ [-32768, 1, 32767],
+ Literal::Float64(1.3),
+ [L, L, G],
+ ),
+ (
+ DataTypes::int(),
+ [i32::MIN as i128, 0, i32::MAX as i128],
+ Literal::Int64(i64::MAX),
+ [L, L, L],
+ ),
+ (
+ DataTypes::tinyint(),
+ [-128, 0, 127],
+ Literal::Int64(-129),
+ [G, G, G],
+ ),
+ (
+ DataTypes::bigint(),
+ [i64::MIN as i128, i64::MAX as i128 - 1, i64::MAX as i128],
+ decimal("9223372036854775806.9"),
+ [L, L, G],
+ ),
+ (
+ DataTypes::decimal(3, 2),
+ [-124, 123, 124],
+ decimal("1.234"),
+ [L, L, G],
+ ),
+ (
+ DataTypes::decimal(3, 2),
+ [-124, -123, 0],
+ decimal("-1.234"),
+ [L, G, G],
+ ),
+ (
+ DataTypes::decimal(38, 0),
+ [-decimal_max, 0, decimal_max],
+ decimal("99999999999999999999999999999999999999"),
+ [L, L, E],
+ ),
+ (
+ DataTypes::decimal(38, 38),
+ [-decimal_max, 0, decimal_max],
+ decimal("0.000000000000000000000000000000000000001"),
+ [L, L, G],
+ ),
+ (
+ DataTypes::float(),
+ [16_777_216, 16_777_218, 16_777_220],
+ Literal::Int64(16_777_219),
+ [L, L, G],
+ ),
+ (
+ DataTypes::float(),
+ [16_777_216, 16_777_218, 16_777_220],
+ Literal::Int64(16_777_217),
+ [L, G, G],
+ ),
+ (
+ DataTypes::double(),
+ [
+ 9_007_199_254_740_992,
+ 9_007_199_254_740_994,
+ 9_007_199_254_740_996,
+ ],
+ Literal::Int64(9_007_199_254_740_995),
+ [L, L, G],
+ ),
+ (
+ DataTypes::double(),
+ [
+ 9_007_199_254_740_992,
+ 9_007_199_254_740_994,
+ 9_007_199_254_740_996,
+ ],
+ Literal::Int64(9_007_199_254_740_993),
+ [L, G, G],
+ ),
+ ] {
+ let array = if let DataType::Decimal(t) = &data_type {
+ Arc::new(
+ Decimal128Array::from(
+ values
+ .into_iter()
+ .map(Some)
+ .chain([None])
+ .collect::<Vec<_>>(),
+ )
+ .with_precision_and_scale(t.precision() as u8, t.scale() as i8)
+ .unwrap(),
+ ) as ArrayRef
+ } else {
+ cast(
+ &Int64Array::from(
+ values
+ .into_iter()
+ .map(|v| Some(i64::try_from(v).unwrap()))
+ .chain([None])
+ .collect::<Vec<_>>(),
+ ),
+ &crate::record::to_arrow_type(&data_type).unwrap(),
+ )
+ .unwrap()
+ };
+ assert_comparisons(
+ data_type,
+ array,
+ literal,
+ expected.into_iter().map(Some).chain([None]).collect(),
+ );
+ }
+ }
+
+ #[test]
+ fn floating_special_values_follow_fluss_ordering() {
+ use Ordering::{Equal as E, Greater as G, Less as L};
+ let values = Float64Array::from(vec![
+ Some(-f64::NAN),
+ Some(f64::NEG_INFINITY),
+ Some(-0.0),
+ Some(0.0),
+ Some(f64::INFINITY),
+ Some(f64::NAN),
+ None,
+ ]);
+ for data_type in [DataTypes::float(), DataTypes::double()] {
+ let array = cast(&values, &crate::record::to_arrow_type(&data_type).unwrap()).unwrap();
+ for (literal, expected) in [
+ (-0.0, [G, L, E, G, G, G]),
+ (0.0, [G, L, L, E, G, G]),
+ (f64::NAN, [E, L, L, L, L, E]),
+ (f64::MAX, [G, L, L, L, G, G]),
+ (-f64::MAX, [G, L, G, G, G, G]),
+ (f64::from_bits(1), [G, L, L, L, G, G]),
+ (-f64::from_bits(1), [G, L, G, G, G, G]),
+ ] {
+ assert_comparisons(
+ data_type.clone(),
+ array.clone(),
+ Literal::Float64(literal),
+ expected.into_iter().map(Some).chain([None]).collect(),
+ );
+ }
+ }
+ }
+
+ #[test]
+ fn membership_preserves_null_and_drops_unrepresentable_values() {
+ let row_type = RowType::new(vec![DataField::new("c", DataTypes::int(), None)]);
+ let batch = RecordBatch::try_from_iter([(
+ "c",
+ cast(
+ &Int64Array::from(vec![Some(1), Some(2), None]),
+ &arrow::datatypes::DataType::Int32,
+ )
+ .unwrap(),
+ )])
+ .unwrap();
+ let values = vec![
+ Literal::Float64(1.3),
+ Literal::Int64(2),
+ Literal::Int64(i64::MAX),
+ Literal::Null,
+ ];
+ for (predicate, expected) in [
+ (col("c").is_in(values.clone()), [false, true, false]),
+ (col("c").not_in(values), [false, false, false]),
+ ] {
+ let mask = BoundPredicate::bind(Some(&predicate), &row_type)
+ .unwrap()
+ .evaluate_batch(&batch)
+ .unwrap();
+ assert_eq!(mask.iter().collect::<Vec<_>>(), expected.map(Some));
+ }
+ assert!(BoundPredicate::bind(Some(&col("c").eq(f64::NAN)), &row_type).is_err());
+ assert!(BoundPredicate::bind(Some(&col("c").eq("1.3")), &row_type).is_err());
+ }
+
+ #[test]
+ fn pushdown_uses_rewritten_bounds_without_changing_wire_coercion() {
+ let row_type = RowType::new(vec![
+ DataField::with_field_id("i", DataTypes::int(), None, 0),
+ DataField::with_field_id("f", DataTypes::float(), None, 1),
+ ]);
+ let predicate = col("i").ge(1.3);
+ assert!(to_pb_predicate(&predicate, &row_type).is_err());
+ let bind = |p: Predicate| {
+ BoundPredicate::bind(Some(&p), &row_type)
+ .unwrap()
+ .to_pushdown_predicate()
+ };
+ assert_eq!(bind(predicate.clone()), Some(col("i").gt(1_i32)));
+ assert_eq!(
+ bind(predicate.clone().and(col("f").ge(1.3))),
+ Some(col("i").gt(1_i32))
+ );
+ assert_eq!(bind(predicate.or(col("f").ge(1.3))), None);
+ }
+}
diff --git a/fluss-rust/crates/fluss/src/predicate/pb.rs b/fluss-rust/crates/fluss/src/predicate/pb.rs
index d4373b231..0b38d22cf 100644
--- a/fluss-rust/crates/fluss/src/predicate/pb.rs
+++ b/fluss-rust/crates/fluss/src/predicate/pb.rs
@@ -21,12 +21,15 @@
//! coerced to the declared type of its column, since the server reads the log
//! unfiltered when a literal type root does not match the field's.
+use super::coercion::{
+ compound_name, float32, float64, integer_in_range, integer_value, mismatch, rescale,
+ unsupported_column, validate_leaf,
+};
use crate::error::Error::IllegalArgument;
use crate::error::Result;
use crate::metadata::{DataField, DataType, RowType, UNASSIGNED_FIELD_ID};
use crate::predicate::{CompoundFunction, LeafFunction, Literal, Predicate};
use crate::proto::{PbCompoundPredicate, PbLeafPredicate, PbLiteralValue, PbPredicate};
-use crate::row::Decimal;
/// Wire codes for `PbPredicate.type`.
const PREDICATE_TYPE_LEAF: i32 = 0;
@@ -132,17 +135,7 @@ pub(crate) fn to_pb_predicate(predicate: &Predicate, row_type: &RowType) -> Resu
}
fn resolve_field<'a>(row_type: &'a RowType, name: &str) -> Result<&'a DataField> {
- let field = row_type
- .fields()
- .iter()
- .find(|field| field.name() == name)
- .ok_or_else(|| IllegalArgument {
- message: format!(
- "filter column '{}' does not exist in the table schema, available columns: {:?}",
- name,
- row_type.get_field_names()
- ),
- })?;
+ let (_, field) = super::coercion::resolve_field(row_type, name)?;
// The server resolves the column by field id, so an unassigned one would be
// unresolvable there.
@@ -154,52 +147,6 @@ fn resolve_field<'a>(row_type: &'a RowType, name: &str) -> Result<&'a DataField>
Ok(field)
}
-/// Checks the literal count and kind a function requires, so that a predicate
-/// the server could only evaluate to garbage is rejected before it is sent.
-fn validate_leaf(function: LeafFunction, field: &DataField, literals: &[Literal]) -> Result<()> {
- let expected = match function {
- // `IN`/`NOT IN` accept any count: empty ones match nothing/everything.
- LeafFunction::In | LeafFunction::NotIn => return Ok(()),
- LeafFunction::IsNull | LeafFunction::IsNotNull => 0,
- _ => 1,
- };
- if literals.len() != expected {
- return Err(IllegalArgument {
- message: format!(
- "{function:?} on column '{}' expects {expected} literal(s), got {}",
- field.name(),
- literals.len()
- ),
- });
- }
-
- // A null literal is only meaningful to `IN`/`NOT IN`, which skip it; every
- // other function compares against it and the server errors out.
- if literals.first() == Some(&Literal::Null) {
- return Err(IllegalArgument {
- message: format!(
- "{function:?} on column '{}' cannot take a null literal, use is_null()/is_not_null()",
- field.name()
- ),
- });
- }
-
- if matches!(
- function,
- LeafFunction::StartsWith | LeafFunction::EndsWith | LeafFunction::Contains
- ) && !matches!(field.data_type(), DataType::Char(_) | DataType::String(_))
- {
- return Err(IllegalArgument {
- message: format!(
- "{function:?} requires a character string column, but column '{}' is {}",
- field.name(),
- field.data_type()
- ),
- });
- }
- Ok(())
-}
-
fn to_pb_literal(field: &DataField, literal: &Literal) -> Result<PbLiteralValue> {
let data_type = field.data_type();
let mut pb = PbLiteralValue {
@@ -280,118 +227,6 @@ fn to_pb_literal(field: &DataField, literal: &Literal) -> Result<PbLiteralValue>
Ok(pb)
}
-/// Reads an integer literal of any width, so that a column can accept the one
-/// Rust inferred for it.
-fn integer_value(field: &DataField, literal: &Literal) -> Result<i64> {
- match literal {
- Literal::Int8(value) => Ok(*value as i64),
- Literal::Int16(value) => Ok(*value as i64),
- Literal::Int32(value) => Ok(*value as i64),
- Literal::Int64(value) => Ok(*value),
- _ => Err(mismatch(field, literal)),
- }
-}
-
-/// Reads an integer literal the column's type must hold, where `max` is its
-/// upper bound and `-max - 1` its lower one.
-fn integer_in_range(field: &DataField, literal: &Literal, max: i64) -> Result<i64> {
- let value = integer_value(field, literal)?;
- if value > max || value < -max - 1 {
- return Err(IllegalArgument {
- message: format!(
- "filter literal {value} is out of range for column '{}' of type {}",
- field.name(),
- field.data_type()
- ),
- });
- }
- Ok(value)
-}
-
-fn float32(field: &DataField, literal: &Literal) -> Result<f32> {
- if let Literal::Float32(value) = literal {
- return Ok(*value);
- }
- let value = float64(field, literal)?;
- let narrowed = value as f32;
- if narrowed as f64 != value && !value.is_nan() {
- return Err(inexact(field, value));
- }
- Ok(narrowed)
-}
-
-fn float64(field: &DataField, literal: &Literal) -> Result<f64> {
- match literal {
- Literal::Float64(value) => Ok(*value),
- Literal::Float32(value) => Ok(*value as f64),
- Literal::Int8(_) | Literal::Int16(_) | Literal::Int32(_) | Literal::Int64(_) => {
- let value = integer_value(field, literal)?;
- let widened = value as f64;
- if widened as i64 != value {
- return Err(inexact(field, widened));
- }
- Ok(widened)
- }
- _ => Err(mismatch(field, literal)),
- }
-}
-
-/// Rejects a conversion that would move the comparison bound, since the server
-/// prunes batches by that bound and a shifted one drops matching rows.
-fn inexact(field: &DataField, value: f64) -> crate::error::Error {
- IllegalArgument {
- message: format!(
- "filter literal {value} cannot be represented exactly by column '{}' of type {}",
- field.name(),
- field.data_type()
- ),
- }
-}
-
-fn rescale(field: &DataField, value: &Decimal, precision: u32, scale: u32) -> Result<Decimal> {
- let big_decimal = value.to_big_decimal();
- let rescaled = Decimal::from_big_decimal(big_decimal.clone(), precision, scale)?;
- // `from_big_decimal` rounds, which would move a comparison bound; only an
- // exact change of scale is safe.
- if rescaled.to_big_decimal() != big_decimal {
- return Err(IllegalArgument {
- message: format!(
- "filter literal {big_decimal} does not fit column '{}' of type {}",
- field.name(),
- field.data_type()
- ),
- });
- }
- Ok(rescaled)
-}
-
-fn mismatch(field: &DataField, literal: &Literal) -> crate::error::Error {
- IllegalArgument {
- message: format!(
- "filter literal {literal:?} does not match column '{}' of type {}",
- field.name(),
- field.data_type()
- ),
- }
-}
-
-fn unsupported_column(field: &DataField) -> crate::error::Error {
- IllegalArgument {
- message: format!(
- "filter on column '{}' of type {} is not supported",
- field.name(),
- field.data_type()
- ),
- }
-}
-
-fn compound_name(function: CompoundFunction) -> &'static str {
- match function {
- CompoundFunction::And => "AND",
- CompoundFunction::Or => "OR",
- }
-}
-
#[cfg(test)]
mod tests {
use super::*;
@@ -400,8 +235,8 @@ mod tests {
DoubleType, FloatType, IntType, RowType, SmallIntType, StringType, TimeType,
TimestampLTzType, TimestampType, TinyIntType,
};
- use crate::predicate::col;
- use crate::row::{TimestampLtz, TimestampNtz};
+ use crate::predicate::{Literal, col};
+ use crate::row::{Decimal, TimestampLtz, TimestampNtz};
use bigdecimal::BigDecimal;
use std::str::FromStr;
@@ -566,21 +401,21 @@ mod tests {
fn encodes_both_timestamp_kinds() {
let ntz = encode_literal(
DataType::Timestamp(TimestampType::new(6).unwrap()),
- col("c").eq(TimestampNtz::from_millis_nanos(1_700_000_000_000, 123_456).unwrap()),
+ col("c").eq(TimestampNtz::from_millis_nanos(1_700_000_000_000, 123_000).unwrap()),
)
.unwrap();
assert_eq!(ntz.literal_type, 12);
assert_eq!(ntz.timestamp_millis_value, Some(1_700_000_000_000));
- assert_eq!(ntz.timestamp_nano_of_millis_value, Some(123_456));
+ assert_eq!(ntz.timestamp_nano_of_millis_value, Some(123_000));
let ltz = encode_literal(
DataType::TimestampLTz(TimestampLTzType::new(6).unwrap()),
- col("c").eq(TimestampLtz::from_millis_nanos(1_700_000_000_000, 123_456).unwrap()),
+ col("c").eq(TimestampLtz::from_millis_nanos(1_700_000_000_000, 123_000).unwrap()),
)
.unwrap();
assert_eq!(ltz.literal_type, 13);
assert_eq!(ltz.timestamp_millis_value, Some(1_700_000_000_000));
- assert_eq!(ltz.timestamp_nano_of_millis_value, Some(123_456));
+ assert_eq!(ltz.timestamp_nano_of_millis_value, Some(123_000));
}
#[test]
@@ -680,9 +515,15 @@ mod tests {
None,
)]);
- let error = to_pb_predicate(&col("c").eq(1i32), &row_type).unwrap_err();
-
- assert!(error.to_string().contains("no schema field id"));
+ for predicate in [
+ col("c").eq(1i32),
+ col("c").eq(1i32).and(col("missing").eq(2i32)),
+ ] {
+ // Preserve the encoder's leaf-first validation order instead of
+ // binding the entire expression before checking protocol field IDs.
+ let error = to_pb_predicate(&predicate, &row_type).unwrap_err();
+ assert!(error.to_string().contains("no schema field id"));
+ }
}
#[test]
@@ -728,8 +569,10 @@ mod tests {
DataType::Double(DoubleType::new()),
DataType::Float(FloatType::new()),
] {
- let error = encode_literal(data_type, col("c").eq(too_precise)).unwrap_err();
- assert!(error.to_string().contains("cannot be represented exactly"));
+ for value in [too_precise, i64::MAX] {
+ let error = encode_literal(data_type.clone(), col("c").eq(value)).unwrap_err();
+ assert!(error.to_string().contains("cannot be represented exactly"));
+ }
}
}
diff --git a/fluss-rust/crates/fluss/src/record/statistics.rs b/fluss-rust/crates/fluss/src/record/statistics.rs
index 80f4ceeb5..f5896baac 100644
--- a/fluss-rust/crates/fluss/src/record/statistics.rs
+++ b/fluss-rust/crates/fluss/src/record/statistics.rs
@@ -312,27 +312,17 @@ fn column_bounds(column: &dyn Array, data_type: &DataType) -> Result<Option<Colu
.as_any()
.downcast_ref::<PrimitiveArray<$arrow_ty>>()
.ok_or_else(|| unexpected_array(column, data_type))?;
- let java_cmp =
- |a: <$arrow_ty as arrow::datatypes::ArrowPrimitiveType>::Native,
- b: <$arrow_ty as arrow::datatypes::ArrowPrimitiveType>::Native| {
- match (a.is_nan(), b.is_nan()) {
- (true, true) => std::cmp::Ordering::Equal,
- (true, false) => std::cmp::Ordering::Greater,
- (false, true) => std::cmp::Ordering::Less,
- (false, false) => a.total_cmp(&b),
- }
- };
let mut bounds = None;
for value in array.iter().flatten() {
bounds = Some(match bounds {
None => (value, value),
Some((min, max)) => (
- if java_cmp(value, min).is_lt() {
+ if crate::util::compare_floats(f64::from(value), f64::from(min)).is_lt() {
value
} else {
min
},
- if java_cmp(value, max).is_gt() {
+ if crate::util::compare_floats(f64::from(value), f64::from(max)).is_gt() {
value
} else {
max
@@ -955,6 +945,36 @@ mod tests {
);
}
+ #[test]
+ fn submillisecond_time_statistics_do_not_allow_exact_filter_pushdown() {
+ use crate::predicate::{BoundPredicate, Literal, col};
+
+ let data_type = DataTypes::time_with_precision(9);
+ let array = Arc::new(Time64NanosecondArray::from(vec![Some(500_000), None]));
+ let (min, max) =
+ single_column_rows(data_type.clone(), array.data_type().clone(), array.clone());
+ // Actual serialized bounds lose the fraction: a server evaluating t > 0
+ // against this maximum would reject a batch containing a matching row.
+ assert_eq!(i32::from_le_bytes(min[8..12].try_into().unwrap()), 0);
+ assert_eq!(i32::from_le_bytes(max[8..12].try_into().unwrap()), 0);
+
+ let row_type = RowType::new(vec![DataField::new("v", data_type, None)]);
+ let batch = RecordBatch::try_from_iter([("v", array as arrow::array::ArrayRef)]).unwrap();
+ let bound = BoundPredicate::bind(Some(&col("v").gt(Literal::Time(0))), &row_type).unwrap();
+ assert!(bound.to_pushdown_predicate().is_none());
+ assert_eq!(
+ bound
+ .into_evaluator()
+ .unwrap()
+ .evaluate_batch(&batch)
+ .unwrap()
+ .values()
+ .iter()
+ .collect::<Vec<_>>(),
+ vec![true, false]
+ );
+ }
+
#[test]
fn scales_a_second_precision_timestamp_to_millis() {
let (min, max) = single_column_rows(
diff --git a/fluss-rust/crates/fluss/src/row/column_vector.rs b/fluss-rust/crates/fluss/src/row/column_vector.rs
index bcd13f5f1..313a28bc8 100644
--- a/fluss-rust/crates/fluss/src/row/column_vector.rs
+++ b/fluss-rust/crates/fluss/src/row/column_vector.rs
@@ -513,6 +513,17 @@ impl TimeColumn {
}
}
+ /// Exact time-of-day value for filters. Unlike `get_millis`, this preserves
+ /// sub-millisecond input and does not narrow Arrow's physical integer range.
+ pub(crate) fn get_nanos(&self, row_id: usize) -> i128 {
+ match self {
+ Self::Second(a) => i128::from(a.value(row_id)) * 1_000_000_000,
+ Self::Millisecond(a) => i128::from(a.value(row_id)) * 1_000_000,
+ Self::Microsecond(a) => i128::from(a.value(row_id)) * 1_000,
+ Self::Nanosecond(a) => i128::from(a.value(row_id)),
+ }
+ }
+
fn get_millis(&self, row_id: usize) -> i32 {
match self {
Self::Second(a) => a.value(row_id) * 1000,
diff --git a/fluss-rust/crates/fluss/src/util/mod.rs b/fluss-rust/crates/fluss/src/util/mod.rs
index ee8dde4a4..0801178a0 100644
--- a/fluss-rust/crates/fluss/src/util/mod.rs
+++ b/fluss-rust/crates/fluss/src/util/mod.rs
@@ -27,6 +27,21 @@ use std::hash::Hash;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
+/// Java Float/Double comparison treats every NaN as one largest value while
+/// distinguishing signed zeros. Widening an f32 to f64 preserves this ordering.
+pub(crate) fn compare_floats(a: f64, b: f64) -> std::cmp::Ordering {
+ canonicalize_nan(a).total_cmp(&canonicalize_nan(b))
+}
+
+/// Canonicalizes comparison/hash keys only; stored values retain their bits.
+pub(crate) fn canonicalize_nan(value: f64) -> f64 {
+ if value.is_nan() {
+ f64::from_bits(0x7ff8_0000_0000_0000)
+ } else {
+ value
+ }
+}
+
pub fn current_time_ms() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
diff --git a/fluss-rust/crates/fluss/src/util/partition.rs b/fluss-rust/crates/fluss/src/util/partition.rs
index ccc71a6b0..dd0b707d4 100644
--- a/fluss-rust/crates/fluss/src/util/partition.rs
+++ b/fluss-rust/crates/fluss/src/util/partition.rs
@@ -189,6 +189,126 @@ pub fn convert_value_of_type(value: &Datum, data_type: &DataType) -> Result<Stri
}
}
+/// Parses a non-null partition value using the format of [`convert_value_of_type`].
+/// Temporal values retain their full encoded precision; this does not apply
+/// column-precision rounding. Invalid encodings and unsupported types are errors.
+pub(crate) fn parse_value_of_type<'a>(value: &'a str, data_type: &DataType) -> Result<Datum<'a>> {
+ let parsed = (|| -> Option<Datum<'a>> {
+ Some(match data_type {
+ DataType::Char(_) | DataType::String(_) => Datum::String(value.into()),
+ DataType::Boolean(_) if value.eq_ignore_ascii_case("true") => Datum::Bool(true),
+ DataType::Boolean(_) if value.eq_ignore_ascii_case("false") => Datum::Bool(false),
+ DataType::TinyInt(_) => Datum::Int8(value.parse().ok()?),
+ DataType::SmallInt(_) => Datum::Int16(value.parse().ok()?),
+ DataType::Int(_) => Datum::Int32(value.parse().ok()?),
+ DataType::BigInt(_) => Datum::Int64(value.parse().ok()?),
+ DataType::Float(_) => {
+ Datum::Float32(value.replace('_', ".").parse::<f32>().ok()?.into())
+ }
+ DataType::Double(_) => {
+ Datum::Float64(value.replace('_', ".").parse::<f64>().ok()?.into())
+ }
+ DataType::Binary(_) | DataType::Bytes(_) => {
+ if value.len() % 2 != 0 {
+ return None;
+ }
+ let bytes = value
+ .as_bytes()
+ .chunks_exact(2)
+ .map(|pair| {
+ let high = char::from(pair[0]).to_digit(16)?;
+ let low = char::from(pair[1]).to_digit(16)?;
+ Some((high * 16 + low) as u8)
+ })
+ .collect::<Option<Vec<_>>>()?;
+ Datum::Blob(bytes.into())
+ }
+ DataType::Date(_) => {
+ let date = value.parse::<jiff::civil::Date>().ok()?;
+ let days = UNIX_EPOCH_DATE.until(date).ok()?.get_days();
+ Datum::Date(Date::new(days))
+ }
+ DataType::Time(_) => {
+ let (time, millis) = value.split_once('_')?;
+ let parts = time
+ .split('-')
+ .map(str::parse::<i8>)
+ .collect::<std::result::Result<Vec<_>, _>>()
+ .ok()?;
+ let [hour, minute, second] = parts.as_slice() else {
+ return None;
+ };
+ let millis: i32 = millis.parse().ok()?;
+ if !(0..1000).contains(&millis) {
+ return None;
+ }
+ let time =
+ jiff::civil::Time::new(*hour, *minute, *second, millis * 1_000_000).ok()?;
+ Datum::Time(Time::new(
+ i32::from(time.hour()) * 3_600_000
+ + i32::from(time.minute()) * 60_000
+ + i32::from(time.second()) * 1000
+ + millis,
+ ))
+ }
+ DataType::Timestamp(_) | DataType::TimestampLTz(_) => {
+ let (datetime, fraction) = value.split_once('_').unwrap_or((value, ""));
+ if fraction.len() > 9 || !fraction.bytes().all(|c| c.is_ascii_digit()) {
+ return None;
+ }
+ let nanos = if fraction.is_empty() {
+ 0
+ } else {
+ fraction.parse::<i32>().ok()? * 10_i32.pow(9 - fraction.len() as u32)
+ };
+ let negative_year = datetime.starts_with('-');
+ let parts = datetime
+ .strip_prefix('-')
+ .unwrap_or(datetime)
+ .split('-')
+ .map(str::parse::<i16>)
+ .collect::<std::result::Result<Vec<_>, _>>()
+ .ok()?;
+ let [year, month, day, time @ ..] = parts.as_slice() else {
+ return None;
+ };
+ if time.len() > 3 {
+ return None;
+ }
+ let year = if negative_year {
+ year.checked_neg()?
+ } else {
+ *year
+ };
+ let datetime = jiff::civil::DateTime::new(
+ year,
+ (*month).try_into().ok()?,
+ (*day).try_into().ok()?,
+ (*time.first().unwrap_or(&0)).try_into().ok()?,
+ (*time.get(1).unwrap_or(&0)).try_into().ok()?,
+ (*time.get(2).unwrap_or(&0)).try_into().ok()?,
+ 0,
+ )
+ .ok()?;
+ let instant = datetime.to_zoned(jiff::tz::TimeZone::UTC).ok()?.timestamp();
+ // Split the positive fraction from the whole second, matching
+ // Fluss's floor-based representation before the Unix epoch.
+ let millis = instant.as_millisecond() + i64::from(nanos / 1_000_000);
+ let nanos = nanos % 1_000_000;
+ if matches!(data_type, DataType::Timestamp(_)) {
+ Datum::TimestampNtz(TimestampNtz::from_millis_nanos(millis, nanos).ok()?)
+ } else {
+ Datum::TimestampLtz(TimestampLtz::from_millis_nanos(millis, nanos).ok()?)
+ }
+ }
+ _ => return None,
+ })
+ })();
+ parsed.ok_or_else(|| IllegalArgument {
+ message: format!("Invalid or unsupported partition value {value:?} for {data_type:?}"),
+ })
+}
+
#[cfg(test)]
mod tests {
use super::*;