You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "naivedogger (via GitHub)" <gi...@apache.org> on 2026/10/19 04:05:36 UTC
[PR] [rust] Reuse predicate binding and add local Arrow evaluation [fluss]
naivedogger opened a new pull request, #4717:
URL: https://github.com/apache/fluss/pull/4717
Closes #4715
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Reuse predicate binding and add local Arrow evaluation [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw merged PR #4717:
URL: https://github.com/apache/fluss/pull/4717
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Reuse predicate binding and add local Arrow evaluation [fluss]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on code in PR #4717:
URL: https://github.com/apache/fluss/pull/4717#discussion_r4332218500
##########
fluss-rust/crates/fluss/src/predicate/arrow_eval.rs:
##########
@@ -0,0 +1,1382 @@
+// 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::Temporal(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::Temporal(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::Temporal(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),
+ )),
+ (_, 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),
+ },
+ (TypedColumn::Binary(a), TypedColumn::Binary(b)) => compare!(a, b, compare_binary),
+ (TypedColumn::Bytes(a), TypedColumn::Bytes(b)) => compare!(a, b, compare_binary),
+ // 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)?;
Review Comment:
Good catch, the width check was incorrectly applied to comparison literals. Fixed in this PR: binary comparisons now use byte slices with signed-byte ordering, and `IN`/`NOT IN` use raw byte keys, while column-width validation remains unchanged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Reuse predicate binding and add local Arrow evaluation [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on code in PR #4717:
URL: https://github.com/apache/fluss/pull/4717#discussion_r4333526241
##########
fluss-rust/crates/fluss/src/predicate/binding.rs:
##########
@@ -0,0 +1,279 @@
+// 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. Floating comparisons
+ /// remain excluded from this pushdown path; 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,
+ ..
+ } => {
+ if matches!(data_type, DataType::Float(_) | DataType::Double(_))
Review Comment:
Could we skip pushdown here for TIME precision > 3, except null checks? For a TIME(9) value of 500,000 ns, `t > Time(0)` matches locally, but the serialized min/max are both 0 ms, so the server drops a matching batch. A regression test using the actual statistics would help.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Reuse predicate binding and add local Arrow evaluation [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4717:
URL: https://github.com/apache/fluss/pull/4717#discussion_r4320228385
##########
fluss-rust/crates/fluss/src/predicate/arrow_eval.rs:
##########
@@ -0,0 +1,484 @@
+// 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 crate::error::Error::IllegalArgument;
+use crate::error::Result;
+use crate::metadata::DataType as FlussDataType;
+use crate::predicate::{BoundPredicate, CompoundFunction, LeafFunction, Literal};
+use crate::record::to_arrow_type;
+use arrow::array::{Array, ArrayRef, BooleanArray, Int64Array, Scalar, StringArray, make_builder};
+use arrow::compute::cast;
+use arrow::compute::kernels::boolean::{and, is_not_null, is_null, not, or};
+use arrow::compute::kernels::cmp::{eq, gt, gt_eq, lt, lt_eq, neq};
+use arrow::datatypes::{DataType as ArrowDataType, TimeUnit};
+use arrow::record_batch::RecordBatch;
+
+impl BoundPredicate {
+ /// Evaluates this predicate exactly against an Arrow record batch.
+ ///
+ /// Leaf predicates follow Fluss predicate semantics: null input values
+ /// evaluate to false, and `NOT IN` with a null literal matches no rows.
+ #[doc(hidden)]
+ pub fn evaluate_batch(&self, batch: &RecordBatch) -> Result<BooleanArray> {
+ match self {
+ Self::AlwaysTrue => Ok(BooleanArray::from(vec![true; batch.num_rows()])),
+ Self::Compound { function, children } => {
+ let initial = match function {
+ CompoundFunction::And => true,
+ CompoundFunction::Or => false,
+ };
+ let mut result = BooleanArray::from(vec![initial; batch.num_rows()]);
+ for child in children {
+ let child = child.evaluate_batch(batch)?;
+ result = match function {
+ CompoundFunction::And => and(&result, &child),
+ CompoundFunction::Or => or(&result, &child),
+ }?;
+ }
+ Ok(result)
+ }
+ Self::Leaf {
+ field_name,
+ data_type,
+ function,
+ literals,
+ ..
+ } => evaluate_leaf(batch, field_name, data_type, *function, literals),
+ }
+ }
+}
+
+fn evaluate_leaf(
+ batch: &RecordBatch,
+ field_name: &str,
+ data_type: &FlussDataType,
+ function: LeafFunction,
+ literals: &[Literal],
+) -> Result<BooleanArray> {
+ 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()
+ ),
+ });
+ }
+
+ 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),
+ LeafFunction::In | LeafFunction::NotIn => {
+ evaluate_membership(array, data_type, function, literals, batch.num_rows())
+ }
+ LeafFunction::StartsWith | LeafFunction::EndsWith | LeafFunction::Contains => {
+ evaluate_string_function(array, function, &literals[0])
+ }
+ _ => compare(array, function, &literals[0], data_type),
Review Comment:
nit: a BoundPredicate built by hand with no value panics here (literals[0]).
Shall we return an error instead?
##########
fluss-rust/crates/fluss/src/predicate/binding.rs:
##########
@@ -0,0 +1,241 @@
+// 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, float32, float64, integer_in_range, integer_value, mismatch, rescale,
+ 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` using Fluss literal coercion rules.
+ /// Consumers validate backend-specific physical representations separately.
+ #[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)?;
+ Ok(Self::Leaf {
+ field_index,
+ field_name: field.clone(),
+ field_id: data_field.field_id(),
+ data_type: data_field.data_type().clone(),
+ function: *function,
+ literals: literals
+ .iter()
+ .map(|literal| bind_literal(data_field, literal))
+ .collect::<Result<Vec<_>>>()?,
+ })
+ }
+ 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 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> {
Review Comment:
Some filters fail while reading, e.g. ts >= 1.5 s on a TIMESTAMP(0) column. Java never fails on a filter: LakeSource.withFilters returns what it can't apply as remaining, and the engine keeps it.
Shall we do the same here, so with_filter can hand back filters it can't apply instead of failing? WDYT?
##########
fluss-rust/crates/fluss/src/predicate/arrow_eval.rs:
##########
@@ -0,0 +1,484 @@
+// 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 crate::error::Error::IllegalArgument;
+use crate::error::Result;
+use crate::metadata::DataType as FlussDataType;
+use crate::predicate::{BoundPredicate, CompoundFunction, LeafFunction, Literal};
+use crate::record::to_arrow_type;
+use arrow::array::{Array, ArrayRef, BooleanArray, Int64Array, Scalar, StringArray, make_builder};
+use arrow::compute::cast;
+use arrow::compute::kernels::boolean::{and, is_not_null, is_null, not, or};
+use arrow::compute::kernels::cmp::{eq, gt, gt_eq, lt, lt_eq, neq};
+use arrow::datatypes::{DataType as ArrowDataType, TimeUnit};
+use arrow::record_batch::RecordBatch;
+
+impl BoundPredicate {
+ /// Evaluates this predicate exactly against an Arrow record batch.
+ ///
+ /// Leaf predicates follow Fluss predicate semantics: null input values
+ /// evaluate to false, and `NOT IN` with a null literal matches no rows.
+ #[doc(hidden)]
+ pub fn evaluate_batch(&self, batch: &RecordBatch) -> Result<BooleanArray> {
+ match self {
+ Self::AlwaysTrue => Ok(BooleanArray::from(vec![true; batch.num_rows()])),
+ Self::Compound { function, children } => {
+ let initial = match function {
+ CompoundFunction::And => true,
+ CompoundFunction::Or => false,
+ };
+ let mut result = BooleanArray::from(vec![initial; batch.num_rows()]);
+ for child in children {
+ let child = child.evaluate_batch(batch)?;
+ result = match function {
+ CompoundFunction::And => and(&result, &child),
+ CompoundFunction::Or => or(&result, &child),
+ }?;
+ }
+ Ok(result)
+ }
+ Self::Leaf {
+ field_name,
+ data_type,
+ function,
+ literals,
+ ..
+ } => evaluate_leaf(batch, field_name, data_type, *function, literals),
+ }
+ }
+}
+
+fn evaluate_leaf(
+ batch: &RecordBatch,
+ field_name: &str,
+ data_type: &FlussDataType,
+ function: LeafFunction,
+ literals: &[Literal],
+) -> Result<BooleanArray> {
+ 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()
+ ),
+ });
+ }
+
+ 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),
+ LeafFunction::In | LeafFunction::NotIn => {
+ evaluate_membership(array, data_type, function, literals, batch.num_rows())
+ }
+ LeafFunction::StartsWith | LeafFunction::EndsWith | LeafFunction::Contains => {
+ evaluate_string_function(array, function, &literals[0])
+ }
+ _ => compare(array, function, &literals[0], data_type),
+ }
+}
+
+fn evaluate_membership(
+ array: &ArrayRef,
+ data_type: &FlussDataType,
+ function: LeafFunction,
+ literals: &[Literal],
+ len: usize,
+) -> Result<BooleanArray> {
+ if function == LeafFunction::NotIn && literals.iter().any(Literal::is_null) {
+ return Ok(BooleanArray::from(vec![false; len]));
+ }
+
+ let mut result = BooleanArray::from(vec![false; len]);
+ for literal in literals.iter().filter(|literal| !literal.is_null()) {
Review Comment:
IN scans the column once per value, mb hashset is better here for long lists?
##########
fluss-rust/crates/fluss/src/predicate/mod.rs:
##########
@@ -84,6 +89,59 @@ 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)
+ }
+
+ /// Returns an integer representation useful for conservative pruning.
+ #[doc(hidden)]
+ pub fn as_integer(&self) -> Option<i128> {
Review Comment:
as_integer cuts timestamps to whole milliseconds (1.9 ms -> 1 ms), so we could skip matching rows. Java compares timestamps exactly. What is the idea here I wonder?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Reuse predicate binding and add local Arrow evaluation [fluss]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on PR #4717:
URL: https://github.com/apache/fluss/pull/4717#issuecomment-6290743478
@fresh-borzoni, thank you for the helpful review! I've addressed your comments. For accepted/remaining, I'd keep the core evaluator strict and add negotiation at the engine-adapter layer in a follow-up. PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Reuse predicate binding and add local Arrow evaluation [fluss]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on code in PR #4717:
URL: https://github.com/apache/fluss/pull/4717#discussion_r4333836133
##########
fluss-rust/crates/fluss/src/predicate/binding.rs:
##########
@@ -0,0 +1,279 @@
+// 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. Floating comparisons
+ /// remain excluded from this pushdown path; 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,
+ ..
+ } => {
+ if matches!(data_type, DataType::Float(_) | DataType::Double(_))
Review Comment:
Agreed, TIME predicates with precision > 3 now stay out of log-batch pushdown, except for null checks; exact filtering still runs on the returned rows. I also excluded `StartsWith` from pushdown because the server's prefix check uses UTF-16 ordering against UTF-8 min/max bounds, keeping that safeguard client-side as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Reuse predicate binding and add local Arrow evaluation [fluss]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on PR #4717:
URL: https://github.com/apache/fluss/pull/4717#issuecomment-6309937073
@beryllw Thanks for the detailed review! I have push a commit to address your comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Reuse predicate binding and add local Arrow evaluation [fluss]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on code in PR #4717:
URL: https://github.com/apache/fluss/pull/4717#discussion_r4324862391
##########
fluss-rust/crates/fluss/src/predicate/binding.rs:
##########
@@ -0,0 +1,241 @@
+// 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, float32, float64, integer_in_range, integer_value, mismatch, rescale,
+ 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` using Fluss literal coercion rules.
+ /// Consumers validate backend-specific physical representations separately.
+ #[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)?;
+ Ok(Self::Leaf {
+ field_index,
+ field_name: field.clone(),
+ field_id: data_field.field_id(),
+ data_type: data_field.data_type().clone(),
+ function: *function,
+ literals: literals
+ .iter()
+ .map(|literal| bind_literal(data_field, literal))
+ .collect::<Result<Vec<_>>>()?,
+ })
+ }
+ 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 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> {
Review Comment:
I agree that engine adapters should return accepted/remaining filters so the engine can evaluate the remainder; I'd add that in the adapter follow-up and keep the core exact evaluator strict about unsupported predicates. I've fixed the precision case in this PR: `ts >= 1.5s` on `TIMESTAMP(0)` now compares exactly, keeping `2s` and rejecting `1s` without an error.
##########
fluss-rust/crates/fluss/src/predicate/arrow_eval.rs:
##########
@@ -0,0 +1,484 @@
+// 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 crate::error::Error::IllegalArgument;
+use crate::error::Result;
+use crate::metadata::DataType as FlussDataType;
+use crate::predicate::{BoundPredicate, CompoundFunction, LeafFunction, Literal};
+use crate::record::to_arrow_type;
+use arrow::array::{Array, ArrayRef, BooleanArray, Int64Array, Scalar, StringArray, make_builder};
+use arrow::compute::cast;
+use arrow::compute::kernels::boolean::{and, is_not_null, is_null, not, or};
+use arrow::compute::kernels::cmp::{eq, gt, gt_eq, lt, lt_eq, neq};
+use arrow::datatypes::{DataType as ArrowDataType, TimeUnit};
+use arrow::record_batch::RecordBatch;
+
+impl BoundPredicate {
+ /// Evaluates this predicate exactly against an Arrow record batch.
+ ///
+ /// Leaf predicates follow Fluss predicate semantics: null input values
+ /// evaluate to false, and `NOT IN` with a null literal matches no rows.
+ #[doc(hidden)]
+ pub fn evaluate_batch(&self, batch: &RecordBatch) -> Result<BooleanArray> {
+ match self {
+ Self::AlwaysTrue => Ok(BooleanArray::from(vec![true; batch.num_rows()])),
+ Self::Compound { function, children } => {
+ let initial = match function {
+ CompoundFunction::And => true,
+ CompoundFunction::Or => false,
+ };
+ let mut result = BooleanArray::from(vec![initial; batch.num_rows()]);
+ for child in children {
+ let child = child.evaluate_batch(batch)?;
+ result = match function {
+ CompoundFunction::And => and(&result, &child),
+ CompoundFunction::Or => or(&result, &child),
+ }?;
+ }
+ Ok(result)
+ }
+ Self::Leaf {
+ field_name,
+ data_type,
+ function,
+ literals,
+ ..
+ } => evaluate_leaf(batch, field_name, data_type, *function, literals),
+ }
+ }
+}
+
+fn evaluate_leaf(
+ batch: &RecordBatch,
+ field_name: &str,
+ data_type: &FlussDataType,
+ function: LeafFunction,
+ literals: &[Literal],
+) -> Result<BooleanArray> {
+ 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()
+ ),
+ });
+ }
+
+ 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),
+ LeafFunction::In | LeafFunction::NotIn => {
+ evaluate_membership(array, data_type, function, literals, batch.num_rows())
+ }
+ LeafFunction::StartsWith | LeafFunction::EndsWith | LeafFunction::Contains => {
+ evaluate_string_function(array, function, &literals[0])
+ }
+ _ => compare(array, function, &literals[0], data_type),
Review Comment:
I've added shared leaf validation before accessing the literals, so a manually constructed predicate with a missing value now returns an error instead of panicking. Both evaluation entry points use the same validation.
##########
fluss-rust/crates/fluss/src/predicate/mod.rs:
##########
@@ -84,6 +89,59 @@ 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)
+ }
+
+ /// Returns an integer representation useful for conservative pruning.
+ #[doc(hidden)]
+ pub fn as_integer(&self) -> Option<i128> {
Review Comment:
I intended this helper for partition pruning, but you're right that truncating timestamps makes that unsafe. I've removed `as_integer`; both batch filtering and timestamp partition evaluation now use the core timestamp types' exact comparison.
##########
fluss-rust/crates/fluss/src/predicate/arrow_eval.rs:
##########
@@ -0,0 +1,484 @@
+// 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 crate::error::Error::IllegalArgument;
+use crate::error::Result;
+use crate::metadata::DataType as FlussDataType;
+use crate::predicate::{BoundPredicate, CompoundFunction, LeafFunction, Literal};
+use crate::record::to_arrow_type;
+use arrow::array::{Array, ArrayRef, BooleanArray, Int64Array, Scalar, StringArray, make_builder};
+use arrow::compute::cast;
+use arrow::compute::kernels::boolean::{and, is_not_null, is_null, not, or};
+use arrow::compute::kernels::cmp::{eq, gt, gt_eq, lt, lt_eq, neq};
+use arrow::datatypes::{DataType as ArrowDataType, TimeUnit};
+use arrow::record_batch::RecordBatch;
+
+impl BoundPredicate {
+ /// Evaluates this predicate exactly against an Arrow record batch.
+ ///
+ /// Leaf predicates follow Fluss predicate semantics: null input values
+ /// evaluate to false, and `NOT IN` with a null literal matches no rows.
+ #[doc(hidden)]
+ pub fn evaluate_batch(&self, batch: &RecordBatch) -> Result<BooleanArray> {
+ match self {
+ Self::AlwaysTrue => Ok(BooleanArray::from(vec![true; batch.num_rows()])),
+ Self::Compound { function, children } => {
+ let initial = match function {
+ CompoundFunction::And => true,
+ CompoundFunction::Or => false,
+ };
+ let mut result = BooleanArray::from(vec![initial; batch.num_rows()]);
+ for child in children {
+ let child = child.evaluate_batch(batch)?;
+ result = match function {
+ CompoundFunction::And => and(&result, &child),
+ CompoundFunction::Or => or(&result, &child),
+ }?;
+ }
+ Ok(result)
+ }
+ Self::Leaf {
+ field_name,
+ data_type,
+ function,
+ literals,
+ ..
+ } => evaluate_leaf(batch, field_name, data_type, *function, literals),
+ }
+ }
+}
+
+fn evaluate_leaf(
+ batch: &RecordBatch,
+ field_name: &str,
+ data_type: &FlussDataType,
+ function: LeafFunction,
+ literals: &[Literal],
+) -> Result<BooleanArray> {
+ 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()
+ ),
+ });
+ }
+
+ 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),
+ LeafFunction::In | LeafFunction::NotIn => {
+ evaluate_membership(array, data_type, function, literals, batch.num_rows())
+ }
+ LeafFunction::StartsWith | LeafFunction::EndsWith | LeafFunction::Contains => {
+ evaluate_string_function(array, function, &literals[0])
+ }
+ _ => compare(array, function, &literals[0], data_type),
+ }
+}
+
+fn evaluate_membership(
+ array: &ArrayRef,
+ data_type: &FlussDataType,
+ function: LeafFunction,
+ literals: &[Literal],
+ len: usize,
+) -> Result<BooleanArray> {
+ if function == LeafFunction::NotIn && literals.iter().any(Literal::is_null) {
+ return Ok(BooleanArray::from(vec![false; len]));
+ }
+
+ let mut result = BooleanArray::from(vec![false; len]);
+ for literal in literals.iter().filter(|literal| !literal.is_null()) {
Review Comment:
I've switched `IN`/`NOT IN` to hash-based lookup. The reusable evaluator builds the set once and reuses it across batches.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Reuse predicate binding and add local Arrow evaluation [fluss]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on PR #4717:
URL: https://github.com/apache/fluss/pull/4717#issuecomment-6261295439
cc @fresh-borzoni @beryllw, this is the first PR for FIP-48.
Would appreciate your review!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Reuse predicate binding and add local Arrow evaluation [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on code in PR #4717:
URL: https://github.com/apache/fluss/pull/4717#discussion_r4324946049
##########
fluss-rust/crates/fluss/src/predicate/arrow_eval.rs:
##########
@@ -0,0 +1,1382 @@
+// 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::Temporal(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::Temporal(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::Temporal(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),
+ )),
+ (_, 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),
+ },
+ (TypedColumn::Binary(a), TypedColumn::Binary(b)) => compare!(a, b, compare_binary),
+ (TypedColumn::Bytes(a), TypedColumn::Bytes(b)) => compare!(a, b, compare_binary),
+ // 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)?;
Review Comment:
Could we allow binary literals with a different length from the column? For a `BINARY(4)` column containing `[1, 2, 3, 4]`, this should match:
```rust
let predicate = col("b").ne(Literal::Bytes(vec![1, 2, 3]));
let bound = BoundPredicate::bind(Some(&predicate), &row_type)?;
bound.into_evaluator()?; // Fails: literal length != 4.
```
Java supports this comparison. Could we compare byte slices directly, preserving the signed-byte ordering, and use raw byte keys for `IN`/`NOT IN`? Width validation should apply to stored values only. A regression test with different-length literals would help.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org