HeartSaVioR opened a new pull request, #59909: URL: https://github.com/apache/spark/pull/59909
### What changes were proposed in this pull request? This PR adds the logical nodes and the analyzer and optimizer rules that plan an external lookup join: a stream-static join whose right side is a table that the connector serves by key lookups ([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-60689](https://issues.apache.org/jira/browse/SPARK-60689), part of [SPARK-60031](https://issues.apache.org/jira/browse/SPARK-60031) (SPIP: External Lookup Join). The rules are not registered yet, so no query reaches them and nothing changes for users. Depends on #59908 ([SPARK-60688](https://issues.apache.org/jira/browse/SPARK-60688)); until it merges, this PR's diff includes that commit. Please review only the last commit. A **lookup-only** table supports `EXTERNAL_LOOKUP` but not `BATCH_READ`: only a lookup join can read it. **Logical nodes** | Node | Role | |---|---| | `ExternalLookupRelation` | Placeholder for a lookup-only table, which has no batch scan | | `ExternalLookupJoin` | The analyzed lookup join, kept through operator optimization. Its `requiredLookupOutput` is part of its references, so `ColumnPruning` cannot put a `Project` between the join and the lookup relation | | `PreparedExternalLookupJoin` | The join after negotiation with the connector: scan, reader factory, parameter expressions, read schema and residual condition | Both join nodes support `Inner` and `LeftOuter` (lookup columns become nullable); any other join type is an internal error. **`ResolveExternalLookupJoin`** (analyzer) turns a join whose right side is directly a lookup-only relation into an `ExternalLookupJoin`: ``` stream.join(lookupOnly, stream("id") === lookupOnly("id")) -> ExternalLookupJoin stream.join(lookupOnly, $"s.tenant" === $"t.tenant" && $"s.id" === $"t.id", "left_outer") -> ExternalLookupJoin ``` A join over a lookup-only table that cannot run as a lookup join fails with the reason, `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.*` (sqlState `0A000`), naming the table: | Join | Error | |---|---| | The left side is not a streaming query | `STATIC_LEFT_SIDE` | | The lookup-only table is on the left side (the sides are never swapped) | `LOOKUP_TABLE_ON_LEFT` | | A join type other than inner or left outer, e.g. `FULL OUTER` | `JOIN_TYPE` | | A join strategy hint attaches to the join, e.g. `stream.hint("merge").join(lookupOnly, …)` | `CONFLICTING_JOIN_STRATEGY` | | No join condition | `MISSING_JOIN_CONDITION`, reported on the final plan by `PrepareExternalLookupJoin`, because the Dataset API analyzes a USING join first as a join without a condition | A join hint the user wrote is never silently overridden: a lookup join cannot run a physical join strategy, and a lookup-only table has no batch scan for one, so the query fails instead of dropping the hint. Only hints that attach to this join count; a hint on a nested join belongs to that join. Every other join is unchanged, including joins over tables that also support `BATCH_READ`. The lookup-only table must be the direct right child of the join; in other shapes, such as beneath a projection, filter or alias, the join stays an ordinary join and fails as a plain scan of the table (`UNSUPPORTED_FEATURE.TABLE_OPERATION`). **`PrepareExternalLookupJoin`** (optimizer) negotiates each lookup join with the connector: - It offers every `lookupColumn = <stream-side expression>` conjunct as a candidate `column = ?`, with a typed `LookupParameter`. - It maps the connector's selection back to the candidates by identity, in the connector's order, which is the field order of each parameter row. - Unselected candidates and all other conjuncts stay as a residual condition, evaluated on the joined rows: ``` ON s.id = t.id AND s.region = t.region AND t.score > s.minScore offered: [id = ?, region = ?] selected: [region = ?, id = ?] -> parameter rows (s.region, s.id) residual: t.score > s.minScore ``` - It validates the connector contract. The data source errors are `DATA_SOURCE_INVALID_EXTERNAL_LOOKUP.*` (sqlState `KD000`): `NULL_PREDICATE_SELECTION`, `DUPLICATE_SELECTED_PREDICATE`, `UNOFFERED_SELECTED_PREDICATE`, `NON_POSITIVE_MAX_PARAMETER_ROWS`, `UNKNOWN_SCAN_FIELD` and `DUPLICATE_SCAN_FIELD`. A table that is not a `SupportsRead` with `EXTERNAL_LOOKUP`, a scan builder without `SupportsExternalLookup`, and an empty selection fail with `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.TABLE`, `SCAN_BUILDER` and `NO_LOOKUP_PREDICATE`. - It matches the scan's read schema to the relation the way ordinary V2 scans do (`PushDownUtils.toOutputAttrs`): a connector that reports `CHAR(5)` or `VARCHAR(5)`, or `s STRUCT<a INT>` for a table's `s STRUCT<a INT NOT NULL>`, plans; `name INT` for a `STRING` column fails with `UNKNOWN_SCAN_FIELD`. - A lookup-only relation still read outside a lookup join afterwards fails with `UNSUPPORTED_FEATURE.TABLE_OPERATION`. `ExternalLookupJoin` has its own tree pattern, `EXTERNAL_LOOKUP_JOIN`, and both rules prune their traversals by pattern. Out of scope, tracked under [SPARK-60031](https://issues.apache.org/jira/browse/SPARK-60031): - Registering the rules and executing joins. - Lookup sides beneath hints, aliases, views, projections or filters. - The `EXTERNAL_LOOKUP` hint and dual-capability opt-in. - Left semi/anti joins, `<=>` and range predicates. - Cache. - Predicate pushdown through the join. ### Why are the changes needed? An external lookup join needs planning that existing joins do not have: deciding which joins read a lookup-only table, and negotiating with the connector which join conditions become lookup parameters. Joins that cannot run this way must fail with a reason, not later with a generic batch-read error. ### Does this PR introduce _any_ user-facing change? No. The rules are not registered. ### How was this patch tested? Plan-level tests, as Spark tests catalyst rules: - `ResolveExternalLookupJoinSuite` (catalyst, new), on hand-built plans with a streaming left side over a table stub: - inner and left outer joins with single and composite `=` keys become an `ExternalLookupJoin` with the join's output (lookup columns nullable under left outer) and no missing input; the rule is idempotent; - unchanged: a table that also supports `BATCH_READ`, a batch-only table, a `LocalRelation`, a nested join on the right side, and a join without a condition; - rejected, each naming the table: `STATIC_LEFT_SIDE`, `LOOKUP_TABLE_ON_LEFT`, `JOIN_TYPE` for a right outer join, and `CONFLICTING_JOIN_STRATEGY` for `BROADCAST`, `MERGE`, `SHUFFLE_HASH` and `SHUFFLE_REPLICATE_NL` hints; - `output` of a `FULL OUTER` `ExternalLookupJoin` is an `INTERNAL_ERROR`. - `PrepareExternalLookupJoinSuite` (sql/core, new), over a lookup-only table whose scan selects `id = ?`: - a lookup join is prepared, and a lookup-only relation outside a lookup join fails with `TABLE_OPERATION`; - a left outer join binds the selected parameter, keeps the unselected candidate as residual and keeps the join's output; preparing twice is a no-op; `output` of a `FULL OUTER` `PreparedExternalLookupJoin` is an `INTERNAL_ERROR`; - a join without a condition fails with `MISSING_JOIN_CONDITION`. The connector-contract checks and read-schema matching run end to end once the rules are registered ([SPARK-60690](https://issues.apache.org/jira/browse/SPARK-60690)). On this commit, JDK 17: ``` build/sbt 'catalyst/testOnly *ResolveExternalLookupJoinSuite' \ 'sql/testOnly *PrepareExternalLookupJoinSuite' 'core/testOnly *SparkThrowableSuite' \ catalyst/scalastyle catalyst/Test/scalastyle sql/scalastyle sql/Test/scalastyle # JUnit XML: ResolveExternalLookupJoinSuite 4/4, PrepareExternalLookupJoinSuite 3/3, # SparkThrowableSuite 38/38 passed # scalastyle: 0 errors (catalyst and sql, main and test) ``` ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Isaac (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]
