HeartSaVioR opened a new pull request, #59910: URL: https://github.com/apache/spark/pull/59910
### What changes were proposed in this pull request? This PR registers the external lookup join rules and adds the physical operator that runs them: batched, synchronous key lookups against a DSv2 connector in each micro-batch ([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-60690](https://issues.apache.org/jira/browse/SPARK-60690), part of [SPARK-60031](https://issues.apache.org/jira/browse/SPARK-60031) (SPIP: External Lookup Join). A stream-static inner or left outer equi-join whose right side is a lookup-only DSv2 table now runs as an external lookup join. Depends on #59909 ([SPARK-60689](https://issues.apache.org/jira/browse/SPARK-60689)), and through it on #59908 ([SPARK-60688](https://issues.apache.org/jira/browse/SPARK-60688)); until they merge, this PR's diff includes their commits. Please review only the last commit. A **lookup-only** table supports `EXTERNAL_LOOKUP` but not `BATCH_READ`. For example: ```scala val customers = spark.read.format("kv-store").load() // lookup-only val enriched = orders // a streaming DataFrame .join(customers, orders("customerId") === customers("id"), "left_outer") // Each micro-batch sends only the distinct customer ids of its rows to the connector. ``` **Wiring** - The analyzer runs `ResolveExternalLookupJoin` in a new batch after `Resolution`. - `SparkOptimizer` runs `PrepareExternalLookupJoin` after operator optimization and before `V2ScanRelationPushDown`, and the rule cannot be excluded. Code comments state why each position is required. - `ExternalLookupJoinStrategy` plans `ExternalLookupJoinExec`. - `DataFrameReader` loads a lookup-only source as `ExternalLookupRelation`, and `TableCapabilityCheck` lets it pass analysis, so the DataFrame can be joined later. Read outside a lookup join, it fails with `UNSUPPORTED_FEATURE.TABLE_OPERATION` ("does not support batch scan"). - `ApplyCharTypePadding` pads `ExternalLookupRelation` as it pads a batch read. **Execution.** `ExternalLookupJoinExec` needs no shuffle: each task joins its input partition through `ExternalLookupJoinEvaluatorFactory`, Spark's evaluator pattern. - **Requests.** It batches stream rows into requests of at most `min(1000, maxParameterRowsPerRequest())` parameter rows. 1000 bounds the stream rows a task buffers while a request is in flight. - **Deduplication.** The parameter rows of a request are distinct under Spark equality: ``` -0.0, 0.0, NaN, NaN -> 2 parameter rows 'a', 'A' under UTF8_LCASE -> 1 parameter row ``` A null key never matches under `=`, so it is not sent. - **Results.** One key may return many rows, and they are streamed; under a `LIMIT`, reading stops early. - **Reader lifecycle.** At most one reader per input partition, created lazily; there is none if no key can match. Readers are closed on success, failure and cancellation, the cursor before its reader. - **Failures.** A connector failure fails the task. It is never treated as a miss. - **Metrics.** The operator reports input rows, input batches, requests, parameter rows and lookup time. The connector's custom metrics are registered under `source.<name>`, so they never collide with an operator metric. The values of all readers in a task are merged, for example under `coalesce`. A join condition that the optimizer folds to `false`, such as `s.id = t.id AND false`, calls no connector: an inner join returns no rows, and a left outer join returns the stream rows padded with nulls. **Supported shapes and errors.** The lookup-only table must be the direct right child of the join, and the connector must select at least one `=` comparison between a lookup column and a stream-side expression or a constant. Composite keys are supported, and other conjuncts are evaluated on the joined rows. Other shapes fail with a named error: | Shape | Error | |---|---| | A batch (static) left side | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.STATIC_LEFT_SIDE` | | The lookup-only table on the left side | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.LOOKUP_TABLE_ON_LEFT` | | Left semi/anti, right or full outer joins | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.JOIN_TYPE` | | A cross join, or a join without a condition | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.MISSING_JOIN_CONDITION` | | A join strategy hint on the stream side, e.g. `stream.hint("merge")` | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.CONFLICTING_JOIN_STRATEGY` | | Only `<=>` or range comparisons | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.NO_LOOKUP_PREDICATE` | | CHAR lookup columns, whose read padding is a projection on the lookup side | `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.LOOKUP_PLAN` | | A projection, filter, alias or hint above the table, including SQL and catalog table references | `UNSUPPORTED_FEATURE.TABLE_OPERATION` | | Real-time mode | `STREAMING_REAL_TIME_MODE.OPERATOR_OR_SINK_NOT_IN_ALLOWLIST` | A join hint the user wrote is never silently dropped: a strategy hint that attaches to a lookup join fails the query. A hint on another join of the same query does not conflict. VARCHAR lookup columns, and nested fields whose nullability differs from the table, run. Out of scope, tracked under [SPARK-60031](https://issues.apache.org/jira/browse/SPARK-60031): - SQL and catalog table references, and aliases, views, projections or filters above the lookup table. - The `EXTERNAL_LOOKUP` hint, and strategy hints on the lookup side. - Left semi/anti joins, `<=>` and range predicates. - CHAR lookup columns. - Cache. - Pushing stream-side filters below the join. ### Why are the changes needed? A stream-static join loads its static side for every micro-batch, even when only a few keys match. With this PR, a connector that serves key lookups can run such joins by sending only the keys each micro-batch needs. ### Does this PR introduce _any_ user-facing change? Yes. A micro-batch stream-static join over a lookup-only DSv2 table now runs as an external lookup join, or fails with a named error. Before, it failed as a plain scan. Tables without `EXTERNAL_LOOKUP` are unaffected. ### How was this patch tested? `ExternalLookupJoinSuite` (new) runs every end-to-end test as a stream-static join: the left side is a memory stream, and the query runs on a memory sink. Test connectors count requests, parameter rows and readers. - **Planning:** a lookup-only table becomes a single external lookup join; stream-static only, with the table on the right; every connector contract error of `DATA_SOURCE_INVALID_EXTERNAL_LOOKUP`, `SCAN_BUILDER`, `TABLE`, `NO_LOOKUP_PREDICATE`, and `DATA_SOURCE_EXTERNAL_LOOKUP_NOT_SUPPORTED` for a scan without `toExternalLookup()`; stream-side expressions; a selected predicate binds its own occurrence, in the connector's order; a projection of some lookup columns still plans a lookup join. - **Results:** inner and left outer joins, residual conditions; null, missing, duplicate and partial batches; composite keys; a condition that folds to `false` calls no connector; VARCHAR and nested read schemas. - **Requests:** capped at 1000 parameter rows when the connector allows 5000; floating-point and collation deduplication; an upstream limit stops reading a one-to-many result. - **Hints:** a strategy hint on the stream side fails; one on a nested join does not. - **Reads outside a lookup join:** `DataFrameReader` and catalog sources fail with `TABLE_OPERATION`, a cross join with `MISSING_JOIN_CONDITION`, and a full outer join with `JOIN_TYPE`. - **Lifecycle and metrics:** readers are closed after success, failure and cancellation; cleanup attempts every resource and keeps the first failure; no reader without a key that can match; metric sampling during a large response; connector metrics merged across the readers of a task, under `source.<name>`; micro-batch streaming. - `ExternalLookupJoinRealTimeModeSuite` (new): real-time mode rejects the operator and opens no reader. - `TableCapabilityCheckSuite`: three new cases for the lookup-only capability checks. On this commit, JDK 17: ``` build/sbt 'catalyst/testOnly *ResolveExternalLookupJoinSuite' \ 'sql/testOnly *PrepareExternalLookupJoinSuite *.ExternalLookupJoinSuite \ *ExternalLookupJoinRealTimeModeSuite *TableCapabilityCheckSuite' \ 'core/testOnly *SparkThrowableSuite' \ catalyst/scalastyle catalyst/Test/scalastyle sql/scalastyle sql/Test/scalastyle # JUnit XML: ExternalLookupJoinSuite 26/26, ExternalLookupJoinRealTimeModeSuite 1/1, # TableCapabilityCheckSuite 14/14, 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]
