[SPARK-46707][SQL] Added throwable field to expressions to improve predicate pushdown

[kelvinjian-db]

What changes were proposed in this pull request?

This PR adds the field throwable to Expression. If an expression is marked as throwable, we will avoid pushing filters containing these expressions through joins, filters, and aggregations (i.e. operators that filter input).

Why are the changes needed?

For predicate pushdown, currently it is possible that we push down a filter that ends up being evaluated on more rows than before it was pushed down (e.g. if we push the filter through a selective join). In this case, it is possible that we now evaluate the filter on a row that will cause a runtime error to be thrown, when prior to pushing this would not have happened.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Added UTs.

Was this patch authored or co-authored using generative AI tooling?

No.

[HyukjinKwon]

Can we leverage NoThrow trait?

[kelvinjian-db]

the problem with NoThrow is that it doesn't mark that many expressions right now (i.e. if we only push down NoThrow expressions then we would be not pushing down most things). we want the opposite here (marking only certain expressions not to push down). defining it this way also allows us to be more precise with defining what is throwable, compared to just setting it for the class

[cloud-fan]

These two are similar but be conservative in two opposite directions. Filter pushdown is a very important feature and we don't want to suddenly disable it for many predicates. We only mark sequence as throwable for now.

NoThrow was added for a new optimization and only a few expressions are marked as NoThrow that can be optimized.

Since runtime error is rare, I think the current solution is better to produce more optimized plans for most cases.

[HyukjinKwon]

If it's rare, why do we add a property to Expression?

[cloud-fan]

Because it's indeed an expression property. NoThrow is not the right way to add properties as the caller side needs to do recursion: https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala#L53-L59 , which is not good for code reuse (you need to call a util function)

[HyukjinKwon]

Should probably remove NoThrow then ... I approved it for now.

[beliefer]

I'm a bit confused. Why not pushdown predicate through join will not throw exception ?

[kelvinjian-db]

@beliefer if we have a plan like

Filter [isnotnull(sequence(col1, col2, 1)]
+- Join …
  +- LocalRelation [col0, col1, col2]
  +- …

it is possible to have a scenario where:

• if we push down the filter through the join, and then there is a row where col1 > col2 in the LocalRelation, causing an exception to be thrown
• but if we didn't push down the filter, the join would have filtered out all the rows where col1 > col2, so the exception would not have been thrown

[cloud-fan]

thanks, merging to master!

[beliefer]

@kelvinjian-db I guess it causes an exception if the join key is not related to col1 or col2.