You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "a-agmon (via GitHub)" <gi...@apache.org> on 2024/08/28 06:57:08 UTC
[PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
a-agmon opened a new pull request, #588:
URL: https://github.com/apache/iceberg-rust/pull/588
This PR partially closes #585
- Adds datafusion filters to the `IcebergTableScan` struct
- Converts datafusion filters (or `Exp`) to Iceberg `Predicate` and apply them to the TableScan On data fusion `execute()`.
- As detailed in the code it currently supports binary expressions (e.g., `(X > 1 AND Y = 'something') OR Z > 100)`, and any combination of them. Unsupported filters will be ignored.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "FANNG1 (via GitHub)" <gi...@apache.org>.
FANNG1 commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1738462529
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -82,6 +82,26 @@ impl TableProvider for IcebergTableProvider {
Ok(Arc::new(IcebergTableScan::new(
self.table.clone(),
self.schema.clone(),
+ _filters,
)))
}
+
+ fn supports_filters_pushdown(
+ &self,
+ filters: &[&Expr],
+ ) -> std::result::Result<Vec<TableProviderFilterPushDown>, datafusion::error::DataFusionError>
+ {
+ let filter_support = filters
+ .iter()
+ .map(|e| {
+ if let Expr::BinaryExpr(BinaryExpr { .. }) = e {
+ TableProviderFilterPushDown::Inexact
Review Comment:
why not return `exact`?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "FANNG1 (via GitHub)" <gi...@apache.org>.
FANNG1 commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1740096447
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,3 +147,223 @@ async fn get_batch_stream(
Ok(Box::pin(stream))
}
+
+/// Converts DataFusion filters ([`Expr`]) to an iceberg [`Predicate`].
+/// If none of the filters could be converted, return `None` which adds no predicates to the scan operation.
+/// If the conversion was successful, return the converted predicates combined with an AND operator.
+fn convert_filters_to_predicate(filters: &[Expr]) -> Option<Predicate> {
+ PredicateConverter.visit_many(filters)
+}
+
+#[cfg(test)]
+mod tests {
Review Comment:
It seems there are too many tests which make it hard to review and maintain. may be we could classify it as:
1. whether a simple expression (without `and` `or` expression) could be converted to a predication. 1 test for the cases for yes, 1 test for the cases for no.
2. a combination of `AND`, `OR`, convertible Expr, un-convertible Expr test cases.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "FANNG1 (via GitHub)" <gi...@apache.org>.
FANNG1 commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1738456745
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -82,6 +82,26 @@ impl TableProvider for IcebergTableProvider {
Ok(Arc::new(IcebergTableScan::new(
self.table.clone(),
self.schema.clone(),
+ _filters,
)))
}
+
+ fn supports_filters_pushdown(
+ &self,
+ filters: &[&Expr],
+ ) -> std::result::Result<Vec<TableProviderFilterPushDown>, datafusion::error::DataFusionError>
+ {
+ let filter_support = filters
+ .iter()
+ .map(|e| {
+ if let Expr::BinaryExpr(BinaryExpr { .. }) = e {
Review Comment:
using match to better support other Expr?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1749163419
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
Review Comment:
I basically designed the conversion of filters to be greedy, which means we need to try and convert as many filters as we can in order to scan less files.
suppose we have a case like this:
```
SELECT * FROM T WHERE A > 10 AND REGEXP_LIKE(name, '[a-z]A[a-zA-Z]{2}');
```
This will translate into an expr that looks something like this
```
BinaryExpr(BinaryExpr AND RegExpExpr)
```
After we processed both leafs: `( BinaryExpr and RegExpExpr)`, we have on the stack just one valid expression.
Then we get to the parent: `BinaryExpr(A AND B)`, we try to get the 2 child expressions from the stack, but in this case we get only one (`A > 10`) because the second one is not supported (`REGEXP_LIKE`).
In this case, because its a logical **AND** expr, we proceed with just the valid predicate. If it was an **OR** expr that was wrapping both then we would have failed the whole predicate.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1749166477
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ if let Expr::BinaryExpr(binary) = node {
+ match (&*binary.left, &binary.op, &*binary.right) {
+ // process simple expressions (involving a column, operator and literal)
+ (Expr::Column(col), op, Expr::Literal(lit)) => {
+ let col_pred = self.convert_column_expr(col, op, lit);
+ self.stack.push_back(col_pred);
+ }
+ // process compound expressions (involving AND or OR and children)
+ (_left, op, _right) if matches!(op, Operator::And | Operator::Or) => {
+ let right_pred = self.stack.pop_back().flatten();
+ let left_pred = self.stack.pop_back().flatten();
+ let children: Vec<_> = [left_pred, right_pred].into_iter().flatten().collect();
+ let compound_pred = self.convert_compound_expr(&children, op);
+ self.stack.push_back(compound_pred);
+ }
+ _ => {}
Review Comment:
Done
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1749163419
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
Review Comment:
I basically designed the conversion of filters to be greedy, which means we need to try and convert as many filters as we can in order to scan less files.
suppose we have a case like this:
```
SELECT * FROM T WHERE A > 10 AND REGEXP_LIKE(name, '[a-z]A[a-zA-Z]{2}');
```
This will translate into an expr that looks something like this
```
BinaryExpr(BinaryExpr AND RegExpExpr)
```
After we processed both leafs: `( BinaryExpr and RegExpExpr)`, we have on the stack just one valid expression.
Then we get to the parent: BinaryExpr(A AND B), we try to get the 2 child expressions from the stack, but in this case we get only one (`A > 10`) because the second one is not supported (`REGEXP_LIKE`).
In this case, because its a logical **AND** expr, we proceed with just the valid predicate. If it was an **OR** expr that was wrapping both then we would have failed the whole predicate.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#issuecomment-2323283411
you can see what I mean in this test case:
```rust
#[test]
fn test_predicate_conversion_with_unsupported_condition_or() {
let sql = "(foo > 1 and bar in ('test', 'test2')) or foo < 0 ";
let df_schema = create_test_schema();
let expr = SessionContext::new()
.parse_sql_expr(sql, &df_schema)
.unwrap();
let predicate = convert_filters_to_predicate(&[expr]).unwrap();
let expected_predicate = Predicate::or(
Reference::new("foo").greater_than(Datum::long(1)),
Reference::new("foo").less_than(Datum::long(0)),
);
assert_eq!(predicate, expected_predicate);
}
```
although we don't support `bar in ('test', 'test2')` there is no reason to fail the entire conversion and not let you enjoy filtering files according to what we can support.
Do you see my point here? or perhaps I misunderstood 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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1740552449
##########
crates/integrations/datafusion/src/physical_plan/predicate_converter.rs:
##########
@@ -0,0 +1,156 @@
+// 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.
+
+use datafusion::arrow::datatypes::DataType;
+use datafusion::logical_expr::{BinaryExpr, Cast, Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+#[derive(Default)]
+pub struct PredicateConverter;
Review Comment:
Why we need this struct?
##########
crates/integrations/datafusion/src/physical_plan/predicate_converter.rs:
##########
Review Comment:
Is there any specific reason we don't use [TreeNodeVisitor](https://docs.rs/datafusion/latest/datafusion/common/tree_node/trait.TreeNodeVisitor.html) to implement this?
##########
crates/integrations/datafusion/src/physical_plan/predicate_converter.rs:
##########
Review Comment:
Also please add some unit tests.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1739722426
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,3 +150,231 @@ async fn get_batch_stream(
Ok(Box::pin(stream))
}
+
+/// convert DataFusion filters ([`Expr`]) to an iceberg [`Predicate`]
+/// if none of the filters could be converted, return `None`
+/// if the conversion was successful, return the converted predicates combined with an AND operator
+fn convert_filters_to_predicate(filters: &[Expr]) -> Option<Predicate> {
+ filters
+ .iter()
+ .filter_map(expr_to_predicate)
+ .reduce(Predicate::and)
+}
+
+/// Recuresivly converting DataFusion filters ( in a [`Expr`]) to an Iceberg [`Predicate`].
+///
+/// This function currently handles the conversion of DataFusion expression of the following types:
+///
+/// 1. Simple binary expressions (e.g., "column < value")
+/// 2. Compound AND expressions (e.g., "x < 1 AND y > 10")
+/// 3. Compound OR expressions (e.g., "x < 1 OR y > 10")
+///
+/// For AND expressions, if one part of the expression can't be converted,
+/// the function will still return a predicate for the part that can be converted.
+/// For OR expressions, if any part can't be converted, the entire expression
+/// will fail to convert.
+///
+/// # Arguments
+///
+/// * `expr` - A reference to a DataFusion [`Expr`] to be converted.
+///
+/// # Returns
+///
+/// * `Some(Predicate)` if the expression could be successfully converted.
+/// * `None` if the expression couldn't be converted to an Iceberg predicate.
+fn expr_to_predicate(expr: &Expr) -> Option<Predicate> {
Review Comment:
I agree. I'm not a DataFusion expert but I think this makes sense being implemented as a visitor of `Expr`, probably by implementing a https://docs.rs/datafusion-common/41.0.0/datafusion_common/tree_node/trait.TreeNodeVisitor.html
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#issuecomment-2323298188
> Aah I see! That makes complete sense. So, what you are saying is, we do the best we can to filter via metadata within iceberg, but whatever operations we can't handle in iceberg will get applied at row-level anyway once the data is passed back to DataFusion? If so, then I withdraw my objection as this seems totally sensible 👍
Precisely!
DataFusion is pretty fast in scanning and processing parquet files and record batches, but I think that the major performance boost that Iceberg can bring in is by using its metadata to filter out and prune data files. So we make best effort to prune using predicate and let DF handle the rest (some of my tests on huge tables show this to gain a great performance boost)
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "FANNG1 (via GitHub)" <gi...@apache.org>.
FANNG1 commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1738462529
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -82,6 +82,26 @@ impl TableProvider for IcebergTableProvider {
Ok(Arc::new(IcebergTableScan::new(
self.table.clone(),
self.schema.clone(),
+ _filters,
)))
}
+
+ fn supports_filters_pushdown(
+ &self,
+ filters: &[&Expr],
+ ) -> std::result::Result<Vec<TableProviderFilterPushDown>, datafusion::error::DataFusionError>
+ {
+ let filter_support = filters
+ .iter()
+ .map(|e| {
+ if let Expr::BinaryExpr(BinaryExpr { .. }) = e {
+ TableProviderFilterPushDown::Inexact
Review Comment:
why not return `exact`?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1749163419
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
Review Comment:
I basically designed the conversion of filters to be greedy, which means we need to try and convert as many filters as we can in order to scan less files.
suppose we have a case like this:
```
SELECT * FROM T WHERE A > 10 AND REGEXP_LIKE(name, '[a-z]A[a-zA-Z]{2}');
```
This will translate into an expr that looks something like this
```
BinaryExpr(BinaryExpr AND RegExpExpr)
```
After we processed both leafs: `( BinaryExpr and RegExpExpr)`, we have on the stack just one valis expression.
Then we get to the parent: BinaryExpr(A AND B), we try to get the 2 child expressions from the stack, but in this case we get only one (`A > 10`) because the second one is not supported (`REGEXP_LIKE`).
In this case, because its a logical **AND** expr, we proceed with just the valid predicate. If it was an **OR** expr that was wrapping both then we would have failed the whole predicate.
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ if let Expr::BinaryExpr(binary) = node {
+ match (&*binary.left, &binary.op, &*binary.right) {
+ // process simple expressions (involving a column, operator and literal)
+ (Expr::Column(col), op, Expr::Literal(lit)) => {
Review Comment:
Thanks for the comment @liurenjie1024
I have added support, and test, for both "directions" of binary statement: `X > 1` and `1 < X`.
But I think there is something more to be clarified:
Both `x>1` and `1<x` are binary statements.
When the binary statement `foo > 1` is processed then there are 3 main leafs to handle:
```
1 - Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" })
```
This ^ will be ignored as its not a binary expr (its a column expr)
```
2 - Literal(Int64(1))
```
This ^ will also be ignored as its not a binary expr (its a literal expr)
```
3 - BinaryExpr(BinaryExpr { left: Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" }), op: Gt, right: Literal(Int64(1)) })
```
This binary expr will be captured, deconstructed, and processed by this match arm
```
match (&*binary.left, &binary.op, &*binary.right) {
(Expr::Column(col), op, Expr::Literal(lit)) => {
...
}
```
Therefore, in order to support `a>6` we dont need to capture `Expr:Column`, only the binary statement that includes it.
On the other hand, I dont think we want to be too flexible because there are expr we want to avoid. For example, `Expr::SimilarTo` (regexp comparison) is something I think we should leave to DataFusion to handle. TBH, I think there are very few Exp we want to handle, rather than leaving them to DataFusion, which are mostly where Iceberg can improve the datafusion scan by using predicates at the metadata level and pruning data files to be scanned.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1749162764
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ if let Expr::BinaryExpr(binary) = node {
+ match (&*binary.left, &binary.op, &*binary.right) {
+ // process simple expressions (involving a column, operator and literal)
+ (Expr::Column(col), op, Expr::Literal(lit)) => {
Review Comment:
Thanks for the comment @liurenjie1024
I have added support for both "directions" of binary statement: `X > 1` and `1 < X`.
But I think there is something more to be clarified:
Both `x>1` and `1<x` are binary statements.
When the binary statement `foo > 1` is processed then there are 3 main leafs to handle:
```
1 - Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" })
```
This ^ will be ignored as its not a binary expr (its a column expr)
```
2 - Literal(Int64(1))
```
This ^ will also be ignored as its not a binary expr (its a literal expr)
```
3 - BinaryExpr(BinaryExpr { left: Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" }), op: Gt, right: Literal(Int64(1)) })
```
This binary expr will be captured, deconstructed, and processed by this match arm
```
match (&*binary.left, &binary.op, &*binary.right) {
(Expr::Column(col), op, Expr::Literal(lit)) => {
...
}
```
Therefore, in order to support `a>6` we dont need to capture `Expr:Column`, only the binary statement that includes it.
On the other hand, I dont think we want to be too flexible because there are expr we want to avoid. For example, `Expr::SimilarTo` (regexp comparison) is something I think we should leave to DataFusion to handle. TBH, I think there are very few Exp we want to handle, rather than leaving them to DataFusion, which are mostly where Iceberg can improve the datafusion scan by using predicates at the metadata level and pruning data files to be scanned.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1739863150
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,3 +150,231 @@ async fn get_batch_stream(
Ok(Box::pin(stream))
}
+
+/// convert DataFusion filters ([`Expr`]) to an iceberg [`Predicate`]
+/// if none of the filters could be converted, return `None`
+/// if the conversion was successful, return the converted predicates combined with an AND operator
+fn convert_filters_to_predicate(filters: &[Expr]) -> Option<Predicate> {
+ filters
+ .iter()
+ .filter_map(expr_to_predicate)
+ .reduce(Predicate::and)
+}
+
+/// Recuresivly converting DataFusion filters ( in a [`Expr`]) to an Iceberg [`Predicate`].
+///
+/// This function currently handles the conversion of DataFusion expression of the following types:
+///
+/// 1. Simple binary expressions (e.g., "column < value")
+/// 2. Compound AND expressions (e.g., "x < 1 AND y > 10")
+/// 3. Compound OR expressions (e.g., "x < 1 OR y > 10")
+///
+/// For AND expressions, if one part of the expression can't be converted,
+/// the function will still return a predicate for the part that can be converted.
+/// For OR expressions, if any part can't be converted, the entire expression
+/// will fail to convert.
+///
+/// # Arguments
+///
+/// * `expr` - A reference to a DataFusion [`Expr`] to be converted.
+///
+/// # Returns
+///
+/// * `Some(Predicate)` if the expression could be successfully converted.
+/// * `None` if the expression couldn't be converted to an Iceberg predicate.
+fn expr_to_predicate(expr: &Expr) -> Option<Predicate> {
Review Comment:
Thanks @sdd, @FANNG1 , I appriciate it
I will try to refactor this, though I am wondering whether Visitor will indeed be the most suitable. Let me know what you think, but I think visitor shines when we have to run different logic on different kinds of elements (chained in some way) while we want to keep the logic in one place - i.e., the Visitor. whereas here we have one kind of element - Expr - which is an enum that can be deconstructed in different ways, for example -
```rust
Expr::Column(col), op, Expr::Literal(lit)
OR
Expr::BinaryExpr(left_expr), Operator::Or, Expr::BinaryExpr(right_expr)
```
so what Im trying to say is that using visitor will simply move the matching complexity to another place - to the visitor.
Does this make sense?
I will continue to try and refactor this but please let me know what you think
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1749162764
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ if let Expr::BinaryExpr(binary) = node {
+ match (&*binary.left, &binary.op, &*binary.right) {
+ // process simple expressions (involving a column, operator and literal)
+ (Expr::Column(col), op, Expr::Literal(lit)) => {
Review Comment:
Thanks for the comment @liurenjie1024
I have added support, and test, for both "directions" of binary statement: `X > 1` and `1 < X`.
But I think there is something more to be clarified:
Both `x>1` and `1<x` are binary statements.
When the binary statement `foo > 1` is processed then there are 3 main leafs to handle:
```
1 - Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" })
```
This ^ will be ignored as its not a binary expr (its a column expr)
```
2 - Literal(Int64(1))
```
This ^ will also be ignored as its not a binary expr (its a literal expr)
```
3 - BinaryExpr(BinaryExpr { left: Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" }), op: Gt, right: Literal(Int64(1)) })
```
This ^ binary expr will be captured, deconstructed, and processed by this match arm
```
match (&*binary.left, &binary.op, &*binary.right) {
(Expr::Column(col), op, Expr::Literal(lit)) => {
...
}
```
Therefore, in order to support `a>6` or `6<a` we dont need to capture `Expr:Column`, only the binary statement that includes it.
On the other hand, I dont think we want to be too flexible because there are expr we want to avoid. For example, `Expr::SimilarTo` (regexp comparison) is something I think we should leave to DataFusion to handle. TBH, I think there are very few Exp we want to handle, rather than leaving them to DataFusion, which are mostly where Iceberg can improve the datafusion scan by using predicates at the metadata level and pruning data files to be scanned.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1738793217
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -82,6 +82,26 @@ impl TableProvider for IcebergTableProvider {
Ok(Arc::new(IcebergTableScan::new(
self.table.clone(),
self.schema.clone(),
+ _filters,
Review Comment:
Thanks. will fix.
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -82,6 +82,26 @@ impl TableProvider for IcebergTableProvider {
Ok(Arc::new(IcebergTableScan::new(
self.table.clone(),
self.schema.clone(),
+ _filters,
)))
}
+
+ fn supports_filters_pushdown(
+ &self,
+ filters: &[&Expr],
+ ) -> std::result::Result<Vec<TableProviderFilterPushDown>, datafusion::error::DataFusionError>
+ {
+ let filter_support = filters
+ .iter()
+ .map(|e| {
+ if let Expr::BinaryExpr(BinaryExpr { .. }) = e {
Review Comment:
sure.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1749163419
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
Review Comment:
I basically designed the conversion of filters to be greedy, which means we need to try and convert as many filters as we can in order to scan less files.
suppose we have a case like this:
```
SELECT * FROM T WHERE A > 10 AND REGEXP_LIKE(name, '[a-z]A[a-zA-Z]{2}');
```
This will translate into an expr that looks something like this
```
BinaryExpr(BinaryExpr AND RegExpExpr)
```
After we processed both leafs: `( BinaryExpr and RegExpExpr)`, we have on the stack just one valid expression because the `RegExpExpr` is not supported.
Then we get to the parent: `BinaryExpr(A AND B)`, we try to get the 2 child expressions from the stack, but in this case we get only one (`A > 10`) because the second one is not supported (`REGEXP_LIKE`).
In this case, because its a logical **AND** expr, we proceed with just one valid predicate. Thats the case you are highlighinting - cases where we process logical AND expr where just one side is valid. in this case - we are processing with just one predicate.
If it was an **OR** expr that was wrapping both then we would have failed the whole predicate.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#issuecomment-2325251287
Thanks much for the review and comments @liurenjie1024
I refactored the code to be based on `TreeNodeVisitor` now, and all unit tests there.
I initially thought that `TreeNodeVisitor` will add an unnecessary layer of complexity but it actually seems to simplify it in some sense.
cc @sdd @FANNG1
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "FANNG1 (via GitHub)" <gi...@apache.org>.
FANNG1 commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1738455466
##########
crates/integrations/datafusion/src/table.rs:
##########
@@ -82,6 +82,26 @@ impl TableProvider for IcebergTableProvider {
Ok(Arc::new(IcebergTableScan::new(
self.table.clone(),
self.schema.clone(),
+ _filters,
Review Comment:
please use `filters`
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#issuecomment-2348277213
Hi @liurenjie1024 ,
just fixed a few conflicts due to https://github.com/apache/iceberg-rust/pull/594 (@FANNG1)
Should be good now
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#issuecomment-2336631735
Thank you very much for the review and comments, @liurenjie1024
I think I have addressed all your comments and questions and would be happy if you can take a another look.
I just want to make a comment which might explain some of my design choices.
I think that the main value of converting data fusion expr to iceberg predicates, is that by pushing down some predicate to an iceberg `scan` we can prune data files and have data fusion scan less data. So, if there is some Expr that we can convert to Iceberg Predicate and then scan less files we should do it. Therefore I tried to implement this as greedy as possible - trying to convert as much relevant expr as possible, even when its just part of all the filters, and basically leave to data fusion to process the rest (which can sometimes be done better there)
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1749162764
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ if let Expr::BinaryExpr(binary) = node {
+ match (&*binary.left, &binary.op, &*binary.right) {
+ // process simple expressions (involving a column, operator and literal)
+ (Expr::Column(col), op, Expr::Literal(lit)) => {
Review Comment:
Thanks for the comment @liurenjie1024
I have added support, and test, for both "directions" of binary statement: `X > 1` and `1 < X`.
But I think there is something more to be clarified:
Both `x>1` and `1<x` are binary statements.
When the binary statement `foo > 1` is processed then there are 3 main leafs to handle:
```
1 - Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" })
```
This ^ will be ignored as its not a binary expr (its a column expr)
```
2 - Literal(Int64(1))
```
This ^ will also be ignored as its not a binary expr (its a literal expr)
```
3 - BinaryExpr(BinaryExpr { left: Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" }), op: Gt, right: Literal(Int64(1)) })
```
This binary expr will be captured, deconstructed, and processed by this match arm
```
match (&*binary.left, &binary.op, &*binary.right) {
(Expr::Column(col), op, Expr::Literal(lit)) => {
...
}
```
Therefore, in order to support `a>6` or `6<a` we dont need to capture `Expr:Column`, only the binary statement that includes it.
On the other hand, I dont think we want to be too flexible because there are expr we want to avoid. For example, `Expr::SimilarTo` (regexp comparison) is something I think we should leave to DataFusion to handle. TBH, I think there are very few Exp we want to handle, rather than leaving them to DataFusion, which are mostly where Iceberg can improve the datafusion scan by using predicates at the metadata level and pruning data files to be scanned.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#issuecomment-2323281873
> My main concern is that it seems like you have taken a "soft-fail" approach, where, for unsupported operators or nodes, you return None and proceed by effectively ignoring as-yet-unimplemented operations? I think the whole thing should return a Result rather than an Option, and when an unimplemented aspect of Expr is encountered, we should fail, returning a Err with ErrorKind::Unimplemented. This prevents confusion and unexpected behaviour by users whose datafusion queries succeed but give them unexpected results due to partial filters.
Thanks for the review and comment, @sdd !
I Will be happy to update this accordingly (will also learn more carefully the visitor approach in the code base to perhaps refactor this further in the future).
I do have one question though. The idea I have followed here is that the conversion process operates in an eager-like mode. It works through the expressions, some of which are not implemented yet, but some of which will not be implemented by design (for example: `Expr::ScalarFunction(ScalarFunction)`), and basically returns a predicate that might not reflect exactly your filters but will filter as much as possible to speed up the query.
for example, suppose that we have an expression like:
`(A > 1 AND B = func(X)) OR (A < 0)`
and we know that `B = func(X)` is not an expression support or can evaluate. So the idea here of soft-fail, is to not fail the entire conversion but to return (A > 1) OR (A < 0), which will still contain unnecessary data but will filter out many files that are not relevant and let you enjoy the metadata-based file pruning.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#issuecomment-2324171975
Hey @liurenjie1024 ,
I think this one is ready to go :)
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1749162764
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ if let Expr::BinaryExpr(binary) = node {
+ match (&*binary.left, &binary.op, &*binary.right) {
+ // process simple expressions (involving a column, operator and literal)
+ (Expr::Column(col), op, Expr::Literal(lit)) => {
Review Comment:
Thanks for the comment @liurenjie1024
I have added support, and test, for both "directions" of binary statement: `X > 1` and `1 < X`.
But I think there is something more to be clarified:
Both `x>1` and `1<x` are binary statements.
When the binary statement `foo > 1` is processed then there are 3 main leafs to handle:
```
1 - Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" })
```
This ^ will be ignored as its not a binary expr (its a column expr)
```
2 - Literal(Int64(1))
```
This ^ will also be ignored as its not a binary expr (its a literal expr)
```
3 - BinaryExpr(BinaryExpr { left: Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" }), op: Gt, right: Literal(Int64(1)) })
```
This ^ binary expr will be captured, deconstructed, and processed by this match arm
```
match (&*binary.left, &binary.op, &*binary.right) {
(Expr::Column(col), op, Expr::Literal(lit)) => {
...
}
```
Therefore, in order to support `a>6` or `6<a` we dont need to capture `Expr:Column`, only the binary statement that includes it.
On the other hand, I dont think we want to be too flexible because there are expr we want to avoid. For example, `Expr::SimilarTo` (regexp comparison) is something I think we should leave to DataFusion to handle. TBH, I think there are actually very few Exp we want to handle, rather than leaving them to DataFusion, which are mostly where Iceberg can improve the datafusion scan by using predicates at the metadata level and pruning data files to be scanned.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1749162684
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
Review Comment:
Done
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
Review Comment:
Done
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ if let Expr::BinaryExpr(binary) = node {
+ match (&*binary.left, &binary.op, &*binary.right) {
+ // process simple expressions (involving a column, operator and literal)
+ (Expr::Column(col), op, Expr::Literal(lit)) => {
Review Comment:
I have added support for both "directions" of binary statement: `X > 1` and `1 < X`.
But I think there is something more to be clarified:
Both `x>1` and `1<x` are binary statements.
When the binary statement `foo > 1` is processed then there are 3 main leafs to handle:
```
Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" })
```
This ^ will be ignored as its not a binary expr (its a column expr)
```
Literal(Int64(1))
```
This ^ will also be ignored as its not a binary expr (its a literal expr)
```
BinaryExpr(BinaryExpr { left: Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" }), op: Gt, right: Literal(Int64(1)) })
```
This binary expr will be captured, deconstructed, and processed by this match arm
```
match (&*binary.left, &binary.op, &*binary.right) {
(Expr::Column(col), op, Expr::Literal(lit)) => {
...
}
```
Therefore, in order to support `a>6` we dont need to capture `Expr:Column`, only the binary statement that includes it.
On the other hand, I dont think we want to be too flexible because there are expr we want to avoid. For example, `Expr::SimilarTo` (regexp comparison) is something I think we should leave to DataFusion to handle. TBH, I think there are very few Exp we want to handle, rather than leaving them to DataFusion, which are mostly where Iceberg can improve the datafusion scan by using predicates at the metadata level and pruning data files to be scanned.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1749162764
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ if let Expr::BinaryExpr(binary) = node {
+ match (&*binary.left, &binary.op, &*binary.right) {
+ // process simple expressions (involving a column, operator and literal)
+ (Expr::Column(col), op, Expr::Literal(lit)) => {
Review Comment:
Thanks for the comment @liurenjie1024
I have added support for both "directions" of binary statement: `X > 1` and `1 < X`.
But I think there is something more to be clarified:
Both `x>1` and `1<x` are binary statements.
When the binary statement `foo > 1` is processed then there are 3 main leafs to handle:
```
Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" })
```
This ^ will be ignored as its not a binary expr (its a column expr)
```
Literal(Int64(1))
```
This ^ will also be ignored as its not a binary expr (its a literal expr)
```
BinaryExpr(BinaryExpr { left: Column(Column { relation: Some(Bare { table: "my_table" }), name: "foo" }), op: Gt, right: Literal(Int64(1)) })
```
This binary expr will be captured, deconstructed, and processed by this match arm
```
match (&*binary.left, &binary.op, &*binary.right) {
(Expr::Column(col), op, Expr::Literal(lit)) => {
...
}
```
Therefore, in order to support `a>6` we dont need to capture `Expr:Column`, only the binary statement that includes it.
On the other hand, I dont think we want to be too flexible because there are expr we want to avoid. For example, `Expr::SimilarTo` (regexp comparison) is something I think we should leave to DataFusion to handle. TBH, I think there are very few Exp we want to handle, rather than leaving them to DataFusion, which are mostly where Iceberg can improve the datafusion scan by using predicates at the metadata level and pruning data files to be scanned.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1749163419
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
Review Comment:
I basically designed the conversion of filters to be greedy, which means we need to try and convert as many filters as we can in order to scan less files.
suppose we have a case like this:
```
SELECT * FROM T WHERE A > 10 AND REGEXP_LIKE(name, '[a-z]A[a-zA-Z]{2}');
```
This will translate into an expr that looks something like this
```
BinaryExpr(BinaryExpr AND RegExpExpr)
```
After we processed both leafs: `( BinaryExpr and RegExpExpr)`, we have on the stack just one valid expression.
Then we get to the parent: `BinaryExpr(A AND B)`, we try to get the 2 child expressions from the stack, but in this case we get only one (`A > 10`) because the second one is not supported (`REGEXP_LIKE`).
In this case, because its a logical **AND** expr, we proceed with just one valid predicate. Thats the case you are highlighinting - cases where we process logical AND expr where just one side is valid. in this case - we are processing with just one predicate.
If it was an **OR** expr that was wrapping both then we would have failed the whole predicate.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "FANNG1 (via GitHub)" <gi...@apache.org>.
FANNG1 commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1738539115
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,3 +150,231 @@ async fn get_batch_stream(
Ok(Box::pin(stream))
}
+
+/// convert DataFusion filters ([`Expr`]) to an iceberg [`Predicate`]
+/// if none of the filters could be converted, return `None`
+/// if the conversion was successful, return the converted predicates combined with an AND operator
+fn convert_filters_to_predicate(filters: &[Expr]) -> Option<Predicate> {
+ filters
+ .iter()
+ .filter_map(expr_to_predicate)
+ .reduce(Predicate::and)
+}
+
+/// Recuresivly converting DataFusion filters ( in a [`Expr`]) to an Iceberg [`Predicate`].
+///
+/// This function currently handles the conversion of DataFusion expression of the following types:
+///
+/// 1. Simple binary expressions (e.g., "column < value")
+/// 2. Compound AND expressions (e.g., "x < 1 AND y > 10")
+/// 3. Compound OR expressions (e.g., "x < 1 OR y > 10")
+///
+/// For AND expressions, if one part of the expression can't be converted,
+/// the function will still return a predicate for the part that can be converted.
+/// For OR expressions, if any part can't be converted, the entire expression
+/// will fail to convert.
+///
+/// # Arguments
+///
+/// * `expr` - A reference to a DataFusion [`Expr`] to be converted.
+///
+/// # Returns
+///
+/// * `Some(Predicate)` if the expression could be successfully converted.
+/// * `None` if the expression couldn't be converted to an Iceberg predicate.
+fn expr_to_predicate(expr: &Expr) -> Option<Predicate> {
Review Comment:
Maybe we could use `visitor pattern` to make code more cleaner to avoid code bomb if supporting more expressions.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "sdd (via GitHub)" <gi...@apache.org>.
sdd commented on PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#issuecomment-2323295148
Aah I see! That makes complete sense. So, what you are saying is, we do the best we can to filter via metadata within iceberg, but whatever operations we can't handle in iceberg will get applied at row-level anyway once the data is passed back to DataFusion? If so, then I withdraw my objection as this seems totally sensible 👍
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#issuecomment-2323262045
Hey @sdd, @FANNG1 and @liurenjie1024
Please see updated PR. I have refactored `predicate_converter` into a separate module that follows a more visitor-ish approach.
The `scan` mod is now clean and convert predicates using:
```rust
fn convert_filters_to_predicate(filters: &[Expr]) -> Option<Predicate> {
PredicateConverter.visit_many(filters)
}
```
`PredicateConverter` is now more modular and allows adding more expressions and patterns. Though I think that it currently covers quite a lot of the cases that are commonly supported in pushdown strategies. Most of the complexity is just in identifying and matching the particular patterns of the `Expr` we need to deal with in different ways.
Please let me know what do you think about this
also note that I have added a fix to the schema mod that correctly converts Date type to its Arrow type.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1749056511
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ if let Expr::BinaryExpr(binary) = node {
+ match (&*binary.left, &binary.op, &*binary.right) {
+ // process simple expressions (involving a column, operator and literal)
+ (Expr::Column(col), op, Expr::Literal(lit)) => {
Review Comment:
Will this visitor visit leaf nodes such as `Expr::Column`? This pattern doesn't seem flexible to me, for example, I believe we could also support things like `a > 6`, `6 < a`.
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
Review Comment:
Ditto.
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
Review Comment:
```suggestion
fn f_down(&mut self, _node: &Expr) -> Result<TreeNodeRecursion, DataFusionError> {
```
I would suggest to use concrete type to make it more readable.
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
Review Comment:
I'm a little confusing about this case, when will this be valid?
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ if let Expr::BinaryExpr(binary) = node {
+ match (&*binary.left, &binary.op, &*binary.right) {
+ // process simple expressions (involving a column, operator and literal)
+ (Expr::Column(col), op, Expr::Literal(lit)) => {
+ let col_pred = self.convert_column_expr(col, op, lit);
+ self.stack.push_back(col_pred);
+ }
+ // process compound expressions (involving AND or OR and children)
+ (_left, op, _right) if matches!(op, Operator::And | Operator::Or) => {
+ let right_pred = self.stack.pop_back().flatten();
+ let left_pred = self.stack.pop_back().flatten();
+ let children: Vec<_> = [left_pred, right_pred].into_iter().flatten().collect();
+ let compound_pred = self.convert_compound_expr(&children, op);
+ self.stack.push_back(compound_pred);
+ }
+ _ => {}
+ }
+ };
+ Ok(TreeNodeRecursion::Continue)
+ }
+}
+
+const MILLIS_PER_DAY: i64 = 24 * 60 * 60 * 1000;
+/// Convert a scalar value to an iceberg datum.
+fn scalar_value_to_datum(value: &ScalarValue) -> Option<Datum> {
+ match value {
+ ScalarValue::Int8(Some(v)) => Some(Datum::int(*v as i32)),
+ ScalarValue::Int16(Some(v)) => Some(Datum::int(*v as i32)),
+ ScalarValue::Int32(Some(v)) => Some(Datum::int(*v)),
+ ScalarValue::Int64(Some(v)) => Some(Datum::long(*v)),
+ ScalarValue::Float32(Some(v)) => Some(Datum::double(*v as f64)),
+ ScalarValue::Float64(Some(v)) => Some(Datum::double(*v)),
+ ScalarValue::Utf8(Some(v)) => Some(Datum::string(v.clone())),
+ ScalarValue::LargeUtf8(Some(v)) => Some(Datum::string(v.clone())),
+ ScalarValue::Date32(Some(v)) => Some(Datum::date(*v)),
+ ScalarValue::Date64(Some(v)) => Some(Datum::date((*v / MILLIS_PER_DAY) as i32)),
+ _ => None,
+ }
+}
+
+/// convert the data fusion Exp to an iceberg [`Predicate`]
+fn binary_op_to_predicate(reference: Reference, op: &Operator, datum: Datum) -> Predicate {
+ match op {
+ Operator::Eq => reference.equal_to(datum),
+ Operator::NotEq => reference.not_equal_to(datum),
+ Operator::Lt => reference.less_than(datum),
+ Operator::LtEq => reference.less_than_or_equal_to(datum),
+ Operator::Gt => reference.greater_than(datum),
+ Operator::GtEq => reference.greater_than_or_equal_to(datum),
+ _ => Predicate::AlwaysTrue,
+ }
+}
+
+#[cfg(test)]
+mod tests {
Review Comment:
Thanks for all these tests, really appreciate it!
##########
crates/integrations/datafusion/src/physical_plan/expr_to_predicate.rs:
##########
@@ -0,0 +1,312 @@
+// 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.
+
+use std::collections::VecDeque;
+
+use datafusion::common::tree_node::{TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::Column;
+use datafusion::error::DataFusionError;
+use datafusion::logical_expr::{Expr, Operator};
+use datafusion::scalar::ScalarValue;
+use iceberg::expr::{Predicate, Reference};
+use iceberg::spec::Datum;
+
+pub struct ExprToPredicateVisitor {
+ stack: VecDeque<Option<Predicate>>,
+}
+impl ExprToPredicateVisitor {
+ /// Create a new predicate conversion visitor.
+ pub fn new() -> Self {
+ Self {
+ stack: VecDeque::new(),
+ }
+ }
+ /// Get the predicate from the stack.
+ pub fn get_predicate(&self) -> Option<Predicate> {
+ self.stack
+ .iter()
+ .filter_map(|opt| opt.clone())
+ .reduce(Predicate::and)
+ }
+
+ /// Convert a column expression to an iceberg predicate.
+ fn convert_column_expr(
+ &self,
+ col: &Column,
+ op: &Operator,
+ lit: &ScalarValue,
+ ) -> Option<Predicate> {
+ let reference = Reference::new(col.name.clone());
+ let datum = scalar_value_to_datum(lit)?;
+ Some(binary_op_to_predicate(reference, op, datum))
+ }
+
+ /// Convert a compound expression to an iceberg predicate.
+ ///
+ /// The strategy is to support the following cases:
+ /// - if its an AND expression then the result will be the valid predicates, whether there are 2 or just 1
+ /// - if its an OR expression then a predicate will be returned only if there are 2 valid predicates on both sides
+ fn convert_compound_expr(&self, valid_preds: &[Predicate], op: &Operator) -> Option<Predicate> {
+ let valid_preds_count = valid_preds.len();
+ match (op, valid_preds_count) {
+ (Operator::And, 1) => valid_preds.first().cloned(),
+ (Operator::And, 2) => Some(Predicate::and(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ (Operator::Or, 2) => Some(Predicate::or(
+ valid_preds[0].clone(),
+ valid_preds[1].clone(),
+ )),
+ _ => None,
+ }
+ }
+}
+
+// Implement TreeNodeVisitor for ExprToPredicateVisitor
+impl<'n> TreeNodeVisitor<'n> for ExprToPredicateVisitor {
+ type Node = Expr;
+
+ fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
+ if let Expr::BinaryExpr(binary) = node {
+ match (&*binary.left, &binary.op, &*binary.right) {
+ // process simple expressions (involving a column, operator and literal)
+ (Expr::Column(col), op, Expr::Literal(lit)) => {
+ let col_pred = self.convert_column_expr(col, op, lit);
+ self.stack.push_back(col_pred);
+ }
+ // process compound expressions (involving AND or OR and children)
+ (_left, op, _right) if matches!(op, Operator::And | Operator::Or) => {
+ let right_pred = self.stack.pop_back().flatten();
+ let left_pred = self.stack.pop_back().flatten();
+ let children: Vec<_> = [left_pred, right_pred].into_iter().flatten().collect();
+ let compound_pred = self.convert_compound_expr(&children, op);
+ self.stack.push_back(compound_pred);
+ }
+ _ => {}
Review Comment:
If there sth not supported, I think the correct behavior should be `TreeNodeRecursion::Stop` and clear the stack?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate [iceberg-rust]
Posted by "a-agmon (via GitHub)" <gi...@apache.org>.
a-agmon commented on code in PR #588:
URL: https://github.com/apache/iceberg-rust/pull/588#discussion_r1740096736
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,3 +147,223 @@ async fn get_batch_stream(
Ok(Box::pin(stream))
}
+
+/// Converts DataFusion filters ([`Expr`]) to an iceberg [`Predicate`].
+/// If none of the filters could be converted, return `None` which adds no predicates to the scan operation.
+/// If the conversion was successful, return the converted predicates combined with an AND operator.
+fn convert_filters_to_predicate(filters: &[Expr]) -> Option<Predicate> {
+ PredicateConverter.visit_many(filters)
+}
+
+#[cfg(test)]
+mod tests {
Review Comment:
Sure. Will fix.
Thanks
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org