morristai opened a new issue, #26632: URL: https://github.com/apache/datafusion/issues/26632
### Describe the bug `ProjectionPushdown` moves single-side parts of a `NestedLoopJoinExec` filter into a `ProjectionExec` below the join. When a filter expression reads both sides, [`JoinFilterRewriter::rewrite`](https://github.com/apache/datafusion/blob/f43f67b844189af217e252f9894fe66d02546124/datafusion/physical-optimizer/src/projection_pushdown.rs#L342-L379) recurses into all of its children ([lines 357-362](https://github.com/apache/datafusion/blob/f43f67b844189af217e252f9894fe66d02546124/datafusion/physical-optimizer/src/projection_pushdown.rs#L357-L362)). That includes children the expression evaluates only for some rows: the THEN and ELSE branches of a `CASE`, the right operand of `OR`, the right operand of an `AND` nested below the top-level conjuncts, and the lazy arguments of a function such as `coalesce`. The pushed-down projection evaluates them on every row of that side. When such a child can fail, a query that has no error as written fails after the rewrite, for example a `CASE` that guards a cast. ### To Reproduce `datafusion-cli` built from `main` at f43f67b8: ```sql SET datafusion.execution.target_partitions = 1; CREATE TABLE l(s VARCHAR) AS VALUES ('1'), ('x'); CREATE TABLE r(v INT) AS VALUES (0), (5); CREATE TABLE t AS SELECT l.s, r.v FROM l CROSS JOIN r; -- The same predicate as a filter over the materialized cross product: 0 rows SELECT * FROM t WHERE CASE WHEN v > 100 THEN CAST(s AS INT) ELSE 0 END > 1; -- As a join filter: SELECT l.s, r.v FROM l JOIN r ON CASE WHEN r.v > 100 THEN CAST(l.s AS INT) ELSE 0 END > 1; -- Failed to cast field 's' from Utf8View to Int32 -- caused by -- Arrow error: Cast error: Cannot cast string 'x' to value of Int32 type ``` The THEN branch was moved under the join: ``` NestedLoopJoinExec: join_type=Inner, filter=CASE WHEN join_proj_push_down_1@0 THEN join_proj_push_down_2@1 ELSE 0 END > 1, projection=[v@0, s@2] ProjectionExec: expr=[v@0 as v, v@0 > 100 as join_proj_push_down_1] DataSourceExec: partitions=1, partition_sizes=[1] ProjectionExec: expr=[s@0 as s, CAST(CAST(s@0 AS Int32) AS Int64) as join_proj_push_down_2] DataSourceExec: partitions=1, partition_sizes=[1] ``` `OR`, `AND` and `coalesce` behave the same way. The `AND` row uses `r2`, one row more than `r`. The optimizer derives `v > 100 OR v = -1` from that predicate and pushes it below the join, and `(0), (5)` alone would not pass it, leaving the join filter nothing to evaluate. ```sql CREATE TABLE r2(v INT) AS VALUES (0), (5), (-1); ``` | Predicate | Filter over the cross product | Join filter | | --- | --- | --- | | `r.v < 100 OR CAST(l.s AS INT) > 1` | 4 rows | cast error | | `coalesce(CAST(r.v AS VARCHAR), CAST(CAST(l.s AS INT) AS VARCHAR)) = '0'` | 2 rows | cast error | | `(r2.v > 100 AND CAST(l.s AS INT) > 1) OR r2.v = -1` | 2 rows | cast error | datafusion-python 55.0.0 gives the same results for the `CASE`, `OR` and `coalesce` queries. ### Expected behavior The join returns the same rows as the filter over the cross product: 0 rows for the `CASE` query, 4 for `OR`, 2 for `coalesce` and 2 for `AND`. ### Additional context The logical common subexpression eliminator already defines which children are evaluated conditionally, in [`conditional_children`](https://github.com/apache/datafusion/blob/f43f67b844189af217e252f9894fe66d02546124/datafusion/optimizer/src/common_subexpr_eliminate.rs#L654-L757): for `CASE`, only the base expression and the first `WHEN` are always evaluated; for `AND` and `OR`, only the left operand; for a scalar function, what [`ScalarUDFImpl::conditional_arguments`](https://github.com/apache/datafusion/blob/f43f67b844189af217e252f9894fe66d02546124/datafusion/expr/src/udf.rs#L885-L894) reports. The fix I have in mind applies the same split to the physical `CaseExpr`, `BinaryExpr` and `ScalarFunctionExpr` in `JoinFilterRewriter::rewrite`, below the top-level conjuncts of the filter: recurse into the always-evaluated children and leave the conditional ones in the join filter. The top-level conjuncts stay searchable, since `PushDownFilter` already splits, reorders and pushes them. #2 6399 added the lambda guard at the top of the same function for a related reason. A weaker variant, which I am not counting as a bug: when the other side is empty the join never evaluates its filter, but the pushed-down projection still runs. Take `r0` as `CREATE TABLE r0(v INT) AS SELECT v FROM r WHERE v > 100`, which is empty. `SELECT * FROM l JOIN r0 ON CAST(l.s AS INT) > r0.v` fails with the same cast error. SQL does not fix the evaluation order outside `CASE`, so this one may be acceptable. We ran into the same question in [Algeon](https://github.com/algeon-dev/algeon), which runs on DataFusion and lowers join filters to the GPU. Its rule is that an expression that can fail is not evaluated before the join. ### Willingness to contribute I plan to contribute a fix for this bug independently -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
