HeartSaVioR opened a new pull request, #60019: URL: https://github.com/apache/spark/pull/60019
### What changes were proposed in this pull request? This PR adds the `EXTERNAL_LOOKUP` join hint 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-60691](https://issues.apache.org/jira/browse/SPARK-60691), part of [SPARK-60031](https://issues.apache.org/jira/browse/SPARK-60031) (SPIP: External Lookup Join). It depends on apache/spark#59910 (SPARK-60690, basic execution), and through it on apache/spark#59909 (SPARK-60689, planning rules) and apache/spark#59908 (SPARK-60688, connector API). The hint opts a stream-static join over a table that supports both `EXTERNAL_LOOKUP` and `BATCH_READ` into an external lookup join. A lookup-only table (`EXTERNAL_LOOKUP` without `BATCH_READ`) needs no hint. For example: ```scala val profiles = spark.read.format("kv-store").load() // supports lookups and batch reads orders.join(profiles.hint("external_lookup"), orders("userId") === profiles("id")) // lookup join orders.join(profiles, orders("userId") === profiles("id")) // ordinary join ``` ```sql SELECT /*+ EXTERNAL_LOOKUP(p) */ o.*, p.tier FROM orders o JOIN profiles p ON o.userId = p.id ``` In scope: - The `EXTERNAL_LOOKUP` hint in DataFrame and SQL forms. Without parameters it marks the plan it annotates, as join strategy hints do; with relation names it marks those relations. - A dual-capability table (`EXTERNAL_LOOKUP` and `BATCH_READ`) becomes a lookup join only with the hint; without it, it keeps the ordinary join. - The lookup side is found through hint markers, aliases and views, so SQL and catalog table references run. - A join that must be a lookup join but reaches its table through any other operator, such as a projection or a filter, fails with `LOOKUP_PLAN` when the DataFrame is built. - A user join hint is never silently overridden: - a physical strategy hint on a join over a lookup-only table fails with `CONFLICTING_JOIN_STRATEGY`; - a physical strategy hint together with the `EXTERNAL_LOOKUP` hint on a dual-capability table fails with `INVALID_EXTERNAL_LOOKUP_HINT.CONFLICTING_JOIN_STRATEGY`; - a hinted join must also be stream-static (`STATIC_LEFT_SIDE`). - Error reporting: - ineligible hinted joins fail with `JOIN_TYPE`, `MISSING_JOIN_CONDITION` or `STREAMING_LOOKUP_SIDE`, and a hinted table without the `EXTERNAL_LOOKUP` capability fails with `TABLE`; - misplaced or invalid hints fail with `INVALID_EXTERNAL_LOOKUP_HINT` (`NOT_ON_LOOKUP_SIDE`, `MULTIPLE_HINTS`, `RELATION_NOT_FOUND`, `DUPLICATE_RELATION`, `UNSUPPORTED_PARAMETER`); - a hint attached to no join is dropped with a warning. - USING joins (`df.join(lookup.hint("external_lookup"), "id")`) run. **The hint.** `ResolveExternalLookupHint`, in the analyzer's Hints batch, resolves the hint to an `ExternalLookupHint` marker. - Without parameters, it marks the plan it is attached to, as a join strategy hint without parameters applies to its whole subtree. `lookup.hint("external_lookup")` therefore marks the lookup side, while a bare `/*+ EXTERNAL_LOOKUP */` on a SQL join query marks the whole query, which is the lookup side of no join, and is dropped with a warning. In SQL, name the relation. - With relation names, it uses the target scoping of join strategy hints: unaliased or multipart tables and aliases, but not names inside a view or an aliased subquery. Unlike join strategy hints, which only warn about a name that matches nothing, a name that matches nothing or is given twice is an error, because a hint the user wrote is never silently dropped. - The marker is not a `ResolvedHint` with a join strategy, as join strategy hints are. `ResolveJoinStrategyHints` merges a strategy hint on an already hinted relation into one `HintInfo`, which keeps one strategy with only a warning, but a conflicting strategy hint must fail. And `EliminateResolvedHint` moves `ResolvedHint`s into `Join.hint` for physical join selection, while this hint selects how the lookup side is read. **Which joins are lookup joins.** `ExternalLookupJoinEligibility`, shared by `CheckExternalLookupJoin` (analysis) and `FormExternalLookupJoin` (start of optimization), now also classifies hinted joins. A join must run as an external lookup join when its right side reads a lookup-only table or carries the hint. - The table is found only through hint markers, `ResolvedHint`s, aliases and views. Any other operator between the table and the join, such as `lookup.select($"id")`, `lookup.where(…)` or a SQL subquery `(SELECT * FROM t) s`, fails with `LOOKUP_PLAN` when the DataFrame is built. The outcome never depends on what the optimizer later removes. - The hint is found through any unary operator above the join input, as join strategy hints are, so a hint beneath a projection is reported, not dropped. - `FormExternalLookupJoin` removes the markers from the lookup side of each join it forms, and removes any other marker with a warning. No existing rule changes; only `ResolveExternalLookupHint` is added to the Hints batch. **Errors.** Every error below is raised when the DataFrame is built, except `MISSING_JOIN_CONDITION`, which is raised at the start of optimization because the classic Dataset API analyzes a USING join first as a join without a condition. | Shape | Error | |---|---| | A physical strategy hint (`BROADCAST`, `MERGE`, `SHUFFLE_HASH`, `SHUFFLE_REPLICATE_NL`) on a join over a lookup-only table, on either side, e.g. `broadcast(lookup)` | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.CONFLICTING_JOIN_STRATEGY` | | A physical strategy hint together with the `EXTERNAL_LOOKUP` hint on a dual-capability table, including both naming the same relation in SQL, in either order | `INVALID_EXTERNAL_LOOKUP_HINT.CONFLICTING_JOIN_STRATEGY` | | A hinted join whose left side is not a streaming query | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.STATIC_LEFT_SIDE` | | A lookup-only table read by the left side, including through an alias or a view | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.LOOKUP_TABLE_ON_LEFT` | | A hinted join of a type other than inner or left outer, including a cross join | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.JOIN_TYPE` | | A hinted inner or left outer join without a condition, including `join().where(…)` | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.MISSING_JOIN_CONDITION` | | A hinted lookup side that is a streaming relation | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.STREAMING_LOOKUP_SIDE` | | A lookup side that reaches its table through an operator other than a hint, an alias or a view, or that reads no DSv2 table, e.g. a hinted local relation | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.LOOKUP_PLAN` | | A hinted DSv2 table without the `EXTERNAL_LOOKUP` capability | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.TABLE` | | The hint on the left side of a join, also beneath an alias | `INVALID_EXTERNAL_LOOKUP_HINT.NOT_ON_LOOKUP_SIDE` | | Two `EXTERNAL_LOOKUP` hints on one lookup side | `INVALID_EXTERNAL_LOOKUP_HINT.MULTIPLE_HINTS` | | A relation name that matches no relation, or is given twice | `INVALID_EXTERNAL_LOOKUP_HINT.RELATION_NOT_FOUND`, `DUPLICATE_RELATION` | | A parameter that is not a relation name | `INVALID_EXTERNAL_LOOKUP_HINT.UNSUPPORTED_PARAMETER` | | A hint that is on the lookup side of no join | Dropped with a warning | A strategy hint on a nested join belongs to that join, and a strategy hint on a dual-capability table without `EXTERNAL_LOOKUP` plans the ordinary join it asks for. Out of scope, tracked under [SPARK-60031](https://issues.apache.org/jira/browse/SPARK-60031): - Lookup sides with projections or filters, including ones that optimization would fold away (they fail with `LOOKUP_PLAN`). - Left semi/anti joins, `<=>` and range predicates. - Hint options, such as cache options. ### Why are the changes needed? A table that can both scan and look up needs an explicit, per-join opt-in. The SPIP (Q8) states that with an explicit `EXTERNAL_LOOKUP` hint, Spark uses an external lookup join or reports an error rather than falling back to another join strategy. This PR applies that to every explicit hint on a lookup join, which is stricter than Spark's usual warning for a join hint that cannot be honored. The dual-capability default, an ordinary join unless hinted, follows the proposal on the dev@ thread. ### Does this PR introduce _any_ user-facing change? Yes: - the `EXTERNAL_LOOKUP` join hint; - SQL and catalog table references, aliases and views over lookup-only tables now run as lookup joins; before, they failed with `UNSUPPORTED_FEATURE.TABLE_OPERATION`; - a lookup-only table read through a projection or filter fails with `LOOKUP_PLAN` when the DataFrame is built, instead of with `TABLE_OPERATION` during planning; - a strategy hint on a join over a lookup-only table also fails when it is on the lookup side. ### How was this patch tested? `ExternalLookupJoinSuite` runs stream-static joins end to end. 21 tests are new, covering: - **Opt-in:** a dual-capability table keeps an ordinary join without the hint and becomes a lookup join with it, also in a running micro-batch query; a hinted dual-capability join whose only equality the connector does not select fails with `NO_LOOKUP_PREDICATE`. - **Stream-static contract:** a hint on a batch join gives `STATIC_LEFT_SIDE`; a lookup-only table beneath an alias on the left gives `LOOKUP_TABLE_ON_LEFT`. - **Strategy hints:** with the `EXTERNAL_LOOKUP` hint on a dual-capability table, all four strategies on the lookup side and on the stream side, and `EXTERNAL_LOOKUP(r), MERGE(r)` in both SQL orders, give `INVALID_EXTERNAL_LOOKUP_HINT.CONFLICTING_JOIN_STRATEGY`. On a lookup-only table, `lookup.hint("broadcast")`, `broadcast(lookup)`, a stream-side `merge` and `shuffle_hash` with the lookup hint give `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.CONFLICTING_JOIN_STRATEGY`. A dual-capability table with only `merge` plans an ordinary join with no connector request, and a strategy hint on a nested join does not stop an outer lookup join. - **Placement and targets:** `NOT_ON_LOOKUP_SIDE`, `MULTIPLE_HINTS`, `RELATION_NOT_FOUND`, `DUPLICATE_RELATION` and `STREAMING_LOOKUP_SIDE`. Named targets cover unaliased, multipart, catalog-qualified and aliased relations; a name inside an aliased subquery is not matched, and naming the subquery fails with `LOOKUP_PLAN`. A hint beneath an alias is found, and a hint without names marks the query or join it annotates, so it does not opt a dual-capability table in. - **Ineligible hinted joins:** a right outer join and a cross join give `JOIN_TYPE` when the DataFrame is built; a catalog-resolved lookup-only table in a cross join or a SQL full outer join gives `JOIN_TYPE`. - **USING joins:** with the hint, over dual-capability and lookup-only tables; without a condition, `MISSING_JOIN_CONDITION`. - **Catalog and SQL references:** they follow the same selection; a catalog table composes into DataFrame joins; an ordinary join over an enriched lookup-join result stays ordinary. - **Lookup-side operators:** `where(lit(false))`, `select(…)`, a hint above or below a projection or filter, and SQL `WHERE 1 = 0` fail with `LOOKUP_PLAN` when the DataFrame is built, without calling the connector. - **Other:** a hint outside any join is ignored; a hinted DataFrame can be reused by several joins; a hinted local relation and a hinted batch-only table fail with `LOOKUP_PLAN` and `TABLE` when the DataFrame is built; connector-side planning failures give `SCAN_BUILDER` and `NO_LOOKUP_PREDICATE`. `FormExternalLookupJoinSuite` (2 new tests) covers the hint on analyzed plans: it opts a dual-capability table in, the lookup side is found beneath hints and aliases, markers are removed from formed joins, and a stray marker is dropped. A lookup-only or hinted lookup side with a projection or filter fails with `LOOKUP_PLAN` in both the check and the conversion. A dual-capability table beneath a projection, without the hint, stays an ordinary join. On this commit, JDK 17: ``` build/sbt hive/Test/compile \ 'catalyst/testOnly *FormExternalLookupJoinSuite *ExternalLookupApiSuite' \ 'sql/testOnly *PrepareExternalLookupJoinSuite *.ExternalLookupJoinSuite \ *ExternalLookupJoinRealTimeModeSuite *TableCapabilityCheckSuite' \ 'core/testOnly *SparkThrowableSuite' \ catalyst/scalastyle catalyst/Test/scalastyle sql/scalastyle sql/Test/scalastyle # JUnit XML: ExternalLookupJoinSuite 53/53, ExternalLookupJoinRealTimeModeSuite 1/1, # TableCapabilityCheckSuite 14/14, FormExternalLookupJoinSuite 9/9, # PrepareExternalLookupJoinSuite 7/7, ExternalLookupApiSuite 5/5, 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]
