HeartSaVioR opened a new pull request, #60198: URL: https://github.com/apache/spark/pull/60198
### What changes were proposed in this pull request? This PR pushes predicates through external lookup joins. A filter on stream columns then runs before the lookup, so fewer rows are looked up, and a condition on lookup columns can reach the connector as a filter. Query results do not change. This PR implements [SPARK-60694](https://issues.apache.org/jira/browse/SPARK-60694), part of [SPARK-60031](https://issues.apache.org/jira/browse/SPARK-60031) (SPIP: External Lookup Join; see the [SPIP](https://docs.google.com/document/d/1fBPM5cku8UJXacG2fw9df1zqALJ-WFm8E9-b9oiFP5M/edit?tab=t.0) and the [dev@ discussion](https://lists.apache.org/thread/7s6tgy9cm18jnzykr8htrtrw4nxyrhp1)). It builds on apache/spark#60087 (SPARK-60693, the lookup cache). Worth a reviewer's attention, because existing optimizer rules change and no new rule is added: - **`PushPredicateThroughJoin`** handles `ExternalLookupJoin` as it handles `AsOfJoin`, reusing its `split`. `PushDownPredicates` also visits plans with a lookup join. - **`EliminateOuterJoin`** turns a left outer lookup join into an inner one when a filter above it rejects null lookup columns. It also removes a left outer lookup join under an aggregate that reads only stream columns and ignores duplicates, so no lookup runs. Its code is shared between joins and lookup joins. What moves, for deterministic conjuncts that cannot throw, as for an ordinary join: | Where the conjunct is | Inner | Left outer | Left semi | Left anti | |---|---|---|---|---| | Filter above, stream side only | stream side | stream side | stream side | stream side | | Filter above, lookup side only | lookup side | stays above (1) | (not possible) | (not possible) | | Filter above, both sides | join condition | stays above (1) | (not possible) | (not possible) | | Join condition, lookup side only | lookup side | lookup side | lookup side | lookup side | | Join condition, stream side only | stream side | stays | stream side | stays | (1) Unless the filter rejects null lookup columns. Then `EliminateOuterJoin` makes the join inner, and the filter moves. Three deliberate exceptions, each for a reason specific to lookup joins: - A comparison between a bare lookup column and a constant, such as `t.dept = 'x'`, stays in the join condition when its type can be a lookup key, so the connector can still use it. On other types, such as a double or a collated string, it moves like any other lookup-side conjunct. - A conjunct over lookup columns with a Python UDF, an external UDF or a subquery never moves, because the lookup side cannot evaluate it. - A conjunct without references stays in the join condition, so a condition that folds to `false` still needs no lookup. Each rule keeps the result on its own. Excluding `EliminateOuterJoin` or `PushDownPredicates` with `spark.sql.optimizer.excludedRules` only loses the pushdown. ### Why are the changes needed? Without this PR, a filter on the stream side of a lookup join applies only after every stream row has been looked up. Conditions on lookup columns reach the connector only when it selects them as lookup keys. ### Does this PR introduce _any_ user-facing change? There is no new API. Queries with filters on stream columns send fewer lookup requests, and filters on lookup columns can reach the connector. A left outer lookup join under a duplicate-agnostic aggregate over stream columns no longer runs. ### How was this patch tested? `PushPredicateThroughExternalLookupJoinSuite` (new plan tests) covers each join type, the three exceptions, and `EliminateOuterJoin`. An end-to-end test in `ExternalLookupJoinSuite` checks the requested keys, the filter that reaches the connector, and the same results with either rule excluded. The suites of the shared rules still pass: `OuterJoinEliminationSuite`, `FilterPushdownSuite` and `LeftSemiAntiJoinPushDownSuite`. All pass on JDK 21 at the top of the PR stack, which includes this PR. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Opus 5.5 This pull request and its description were written by Isaac. -- 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]
