feat (datafusion integration): convert datafusion expr filters to Iceberg Predicate

[a-agmon]

This PR 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() thereby enabling data files pruning by data fusion to boost scan.
• Currently supports binary expressions (e.g., (X > 1 AND Y = 'something') OR Z > 100), and any combination of them.

Update:

• added support in schema to convert Date32 to correct arrow type
• refactored scan to use new predicate converter as visitor and separated it to a new mod
• added support for simple predicates

[FANNG1]

Maybe we could use visitor pattern to make code more cleaner to avoid code bomb if supporting more expressions.

[sdd]

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

[a-agmon]

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 -

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

[a-agmon]

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:

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.

[sdd]

Thanks for the work you've put into this! I think there are some structural changes that could be made, by aligning more closely to the visitor style that we already have, but I also think that perfect is the enemy of done, and since this datafusion integration is still quite experimental, and that Rust makes refactorings quite easy, I'm happy to proceed here with something working that we can evolve further as we go forwards.

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.

[a-agmon]

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 we 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.

[a-agmon]

you can see what I mean in this test case:

    #[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

[sdd]

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 👍

[a-agmon]

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)

[a-agmon]

Hey @liurenjie1024 ,
I think this one is ready to go :)

[liurenjie1024]

Thanks @a-agmon for this pr! This helps to improve datafusio integration, but there are still some missing part.

[a-agmon]

Thanks much for the review and comments @liurenjie1024
I refactored the code to be based on TreeNodeVisitor now, and all unit tests moved there.
I initially thought that TreeNodeVisitor will add an unnecessary layer of complexity but it actually seems to simplify it in some sense. Thanks

cc @sdd @FANNG1

[liurenjie1024]

Suggested change

	fn f_down(&mut self, _node: &'n Self::Node) -> Result<TreeNodeRecursion, DataFusionError> {
	fn f_down(&mut self, _node: &Expr) -> Result<TreeNodeRecursion, DataFusionError> {

I would suggest to use concrete type to make it more readable.

[a-agmon]

Done

[liurenjie1024]

Ditto.

[a-agmon]

Done

[liurenjie1024]

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.

[a-agmon]

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.

[liurenjie1024]

I'm a little confusing about this case, when will this be valid?

[a-agmon]

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.

[liurenjie1024]

If there sth not supported, I think the correct behavior should be TreeNodeRecursion::Stop and clear the stack?

[a-agmon]

Done

[liurenjie1024]

Thanks for all these tests, really appreciate it!

[a-agmon]

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)

[a-agmon]

Hi @liurenjie1024 ,
just fixed a few conflicts due to #594 (@FANNG1)
Should be good now

[sdd]

Thanks for being so patient on this and explaining the rationale, @a-agmon. I think that there has been a lot of confusion on this because the approach of ignoring unsupported predicates (because they will be handled downstream by DataFusion) was not quite what we're used to seeing, but I think your approach makes sense. And, as I said before, perfect is the enemy of done and I'm happy to move forward with what we have here as there is nothing preventing us from further iterating on this.

Thanks again!

[Xuanwo]

Thank you @a-agmon for working this and thanks @liurenjie1024, @sdd, @FANNG1's review. Let's move!