HeartSaVioR opened a new pull request, #60055: URL: https://github.com/apache/spark/pull/60055
### What changes were proposed in this pull request? This PR adds lookup-side projections and filters, left semi and left anti joins, and `<=>` and range predicates to external lookup join ([SPIP](https://docs.google.com/document/d/1fBPM5cku8UJXacG2fw9df1zqALJ-WFm8E9-b9oiFP5M/edit?tab=t.0), [dev@](https://lists.apache.org/thread/7s6tgy9cm18jnzykr8htrtrw4nxyrhp1)). This PR implements [SPARK-60692](https://issues.apache.org/jira/browse/SPARK-60692), part of [SPARK-60031](https://issues.apache.org/jira/browse/SPARK-60031) (SPIP: External Lookup Join). It depends on apache/spark#60019 (the `EXTERNAL_LOOKUP` hint, SPARK-60691), and through it on apache/spark#59910 (basic execution, SPARK-60690), apache/spark#59909 (planning rules, SPARK-60689) and apache/spark#59908 (connector API, SPARK-60688). In that stack: - `ExternalLookupJoinEligibility` decides which joins must be lookup joins. `CheckExternalLookupJoin` reports, when the DataFrame is built, why such a join cannot run. - `FormExternalLookupJoin` converts the join to an `ExternalLookupJoin` at the start of optimization. - `PrepareExternalLookupJoin` negotiates the lookup with the connector, just before scans are pushed down. - The lookup side could reach its table only through hints, aliases and views. For example, with lookup-only tables `profiles` and `rates`: ```scala val active = profiles.where($"active").select($"id", upper($"tier").as("tier")) orders.join(active, orders("userId") === active("id")) // lookup-side stages orders.join(profiles, orders("userId") === profiles("id"), "left_anti") // anti join events.join(rates, events("ccy") === rates("ccy") && events("ts") >= rates("validFrom") && events("ts") < rates("validTo")) // equality + range ``` In scope: - Lookup sides with projections and filters. The lookup side is also found through projections and filters, so views and SQL subqueries such as `(SELECT ... FROM t) s` run: - projections and filters that remain after optimization run as lookup-side stages; - ones that optimization folds away, such as `WHERE 1 = 0`, run without calling the connector; - a projection or filter that contains a Python UDF, an external UDF or a subquery fails with `LOOKUP_PLAN` when the DataFrame is built. - CHAR padding of a lookup-only table, and the CHAR/VARCHAR read-side checks under standard semantics, run as a lookup-side projection. - Filters over the table are pushed to the scan builder, as for a batch scan. - The lookup columns that the join keeps are pruned in `ColumnPruning`. The lookup scan is pruned once the lookup predicates are selected. - Left semi and left anti joins. - `<=>` (null-safe) and `<`, `<=`, `>`, `>=` are offered as lookup candidates; unselected candidates stay residual. **Which lookup sides run.** Eligibility now also walks through projections and filters above the table, as long as each of them can be evaluated on the lookup results. - A Python UDF, an external UDF or a subquery in one cannot be evaluated there. Such UDFs and subqueries are planned as separate operators, and the optimizer would rewrite some subqueries into joins. So the join fails with `LOOKUP_PLAN` when the DataFrame is built; `PrepareExternalLookupJoin` keeps the same check as a fallback. - Any other operator between the table and the join still fails with `LOOKUP_PLAN` when the DataFrame is built. Whether a join runs never depends on what the optimizer later does to its lookup side. - `ExternalLookupJoinShapes` holds the shapes that eligibility and lookup planning share: `lookupComparison`, the lookup-candidate comparison, and `canEvaluateInLookupStage`. **Lookup-side stages.** `PrepareExternalLookupJoin` splits the lookup side as `V2ScanRelationPushDown` splits a scan. - `ScanOperation` collapses the projections and filters over the table, with aliases substituted, into one filter stage and one projection stage. The projection stage is dropped when it only selects columns. - Only operators above that segment that `ScanOperation` cannot collapse, such as a filter above a non-deterministic projection, stay chained stages. - The operator evaluates the stages, in order, on each lookup result row, before the row joins the stream rows of its key. Stages are initialized per partition, so a seeded `rand()` is drawn once per lookup result row. **Negotiation with the connector**, in the order of the `SupportsExternalLookup` contract: 1. **Candidates.** `=`, `<=>`, `<`, `<=`, `>` and `>=` between a lookup column and a value are offered. - A comparison written with the lookup column on the right is rebuilt with it on the left, so `s.ts < t.event_time` is offered as `event_time > ?`. The offered `Predicate` is named after the comparison's SQL operator, as `V2ExpressionBuilder` names predicates. - The value side may be any expression over the stream side, including a literal. A value that reads lookup columns cannot be a parameter. - A column is offered only if the lookup-side projections pass it through unchanged (`c` or `c AS k`), under the name of the table column it reads. - `<=>` passes nulls to the connector. - If no comparison qualifies, the join fails with `NO_LOOKUP_CANDIDATE` before a scan builder is created. 2. **Filter pushdown.** The collapsed segment's filters go to the scan builder through `DataSourceStrategy.normalizeExprs` and `PushDownUtils.pushFilters`, as in `V2ScanRelationPushDown`. Whatever the source does not evaluate stays in the filter stage. 3. **`selectLookupPredicates`**, as in apache/spark#59909. Unselected candidates stay in the residual condition, in their written order. 4. **Column pruning.** Only now, so that a column that only a selected predicate reads is dropped, Spark computes the table columns that the stages, the residual and the parent need. It passes them to `pruneColumns` as whole top-level columns. `PushDownUtils.pruneColumns` is not used, because it can pass nested-pruned columns and binds the read schema by the resolver. The read schema must include every required column, or the join fails with `MISSING_SCAN_FIELD`. 5. **`build()`.** The stages are rebound to the read schema, so its types and nullability carry through the projections. **Required columns.** `ColumnPruning` gets two cases for `ExternalLookupJoin`, next to its case for left existence joins: - below a projection, the join keeps only the lookup columns in `requiredLookupOutput` that the projection reads, and its children are pruned; - a semi or anti join keeps no lookup column, like a left existence join, and its lookup side is pruned to what the condition reads. `ColumnPruning` runs in the operator optimization batches, after `FormExternalLookupJoin` forms the joins and before `PrepareExternalLookupJoin` reads the narrowed columns. The prepared join outputs the analyzed lookup attributes still required above it, in analyzed order. **Left semi and left anti joins.** Both are now supported join types, so the `JOIN_TYPE` message lists them. - A semi join emits a stream row on its first match, and an anti join emits a stream row that has no match. - A Python or external UDF over lookup columns in the condition is rejected for them with `PYTHON_UDF_IN_ON_CLAUSE` or `EXTERNAL_UDF_IN_ON_CLAUSE`, as for a left outer join. Only an inner join can evaluate such a conjunct in a filter above the join. - A provably empty lookup side, such as `where(lit(false))` or a condition `... AND false`, needs no connector call. A semi join returns no rows and an anti join returns every stream row. The stream side is kept, so its stateful operators still run. **CHAR and VARCHAR.** `ApplyCharTypePadding` now applies read-side padding and checks to a lookup-only table as it does for a batch read. The resulting projection runs as a lookup-side stage. The `CHAR_VARCHAR_COLUMN` error condition, which rejected such tables, is removed. **Message changes.** `LOOKUP_PLAN` names the projections and filters a lookup side may have, and what they cannot contain. `NO_LOOKUP_CANDIDATE` lists the offered comparison operators. Out of scope, tracked under [SPARK-60031](https://issues.apache.org/jira/browse/SPARK-60031): - Cache. - Pushing stream-side filters and join-condition conjuncts through the join. ### Why are the changes needed? Real lookup queries project and filter the lookup table, alias its key columns, read CHAR columns, and use semi or anti joins and range conditions, such as a lookup of the rate valid at an event's time. Without this PR these queries fail with `LOOKUP_PLAN`, `JOIN_TYPE`, `NO_LOOKUP_CANDIDATE` or `CHAR_VARCHAR_COLUMN`. ### Does this PR introduce _any_ user-facing change? Yes: - lookup sides with projections and filters, and lookup-only tables with CHAR columns, run as external lookup joins; - left semi and left anti joins are supported; - `<=>` and range comparisons are offered to the connector; - a lookup-side projection or filter with a Python UDF, an external UDF or a subquery fails with `LOOKUP_PLAN` when the DataFrame is built. ### How was this patch tested? `ExternalLookupJoinSuite` runs stream-static joins end to end. 25 tests are new, and one test that expected `CHAR_VARCHAR_COLUMN` is replaced. They cover: - **Lookup-side stages:** - projections and filters applied before joining; - rebinding after column pruning; - a projected key alias, and nested aliases and filters; - pruning of unused throwable expressions. - **Non-deterministic stages:** a seeded `rand(3)` under a non-tautological filter, with duplicate stream keys, must be drawn once per lookup result row. - **Pushdown and pruning:** - filter and required-column pushdown, including filters canonicalized through a relation alias; - request-key pruning for inner and outer joins, and the source asked only for the columns a parent projection reads; - no pushdown above a non-deterministic filter, but pushdown of a filter below a non-deterministic projection. - **Semi and anti joins:** - their results, and their residuals on the pruned layout; - empty lookup sides; - failure propagation for outer and anti joins; - the engine's request size bound. - **Predicates:** - `<=>` keys and their negotiation; - equality plus range access paths, and multiple ranges on one field; - unselected criteria kept as residuals. - **CHAR and VARCHAR:** CHAR padding, and the VARCHAR read-side check under standard semantics. - **Analysis-time errors:** `LOOKUP_PLAN` when the DataFrame is built, for other operators on the lookup side and for a Python UDF, a scalar subquery, `IN` or `EXISTS` in a lookup-side filter. `PYTHON_UDF_IN_ON_CLAUSE` for a semi join. - **Shapes that now run:** empty lookup sides (`where(lit(false))`, `WHERE 1 = 0`) and a hint naming an aliased subquery, which apache/spark#60019 rejects with `LOOKUP_PLAN`. - **Hints:** the `EXTERNAL_LOOKUP` hint with unsupported join shapes. The other suites: - `PrepareExternalLookupJoinSuite` (1 new test) checks that a candidate is offered with the lookup column first, also for a literal value. - `ColumnPruningSuite` (2 new tests) checks the two new cases. - `FormExternalLookupJoinSuite` checks both outcomes of classification and conversion: lookup sides with projections and filters are classified as lookup joins, and limits, distinct and subquery filters fail with `LOOKUP_PLAN`. On this commit, JDK 17: ``` build/sbt hive/Test/compile \ 'catalyst/testOnly *FormExternalLookupJoinSuite *ExternalLookupApiSuite *ColumnPruningSuite' \ 'sql/testOnly *PrepareExternalLookupJoinSuite *.ExternalLookupJoinSuite \ *ExternalLookupJoinRealTimeModeSuite *TableCapabilityCheckSuite' \ 'core/testOnly *SparkThrowableSuite' \ catalyst/scalastyle catalyst/Test/scalastyle sql/scalastyle sql/Test/scalastyle # JUnit XML: ExternalLookupJoinSuite 77/77, ExternalLookupJoinRealTimeModeSuite 1/1, # TableCapabilityCheckSuite 14/14, FormExternalLookupJoinSuite 9/9, # PrepareExternalLookupJoinSuite 8/8, ExternalLookupApiSuite 5/5, ColumnPruningSuite 27/27, # SparkThrowableSuite 38/38 passed # scalastyle: 0 errors (catalyst and sql) ``` ### 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]
