HeartSaVioR opened a new pull request, #59908: URL: https://github.com/apache/spark/pull/59908
### What changes were proposed in this pull request? An external lookup join enriches a stream with a static table in a key-value or OLTP store: each micro-batch looks up only the join keys it needs instead of scanning the table ([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-60688](https://issues.apache.org/jira/browse/SPARK-60688), part of [SPARK-60031](https://issues.apache.org/jira/browse/SPARK-60031) (SPIP: External Lookup Join): it adds only the connector-facing DSv2 API, one error condition and one `ToStringSQLBuilder` case. No engine code calls them yet, and no behavior changes. All `@Evolving`, `@since 4.4.0`, in `connector.read` unless noted: | Piece | Role | |---|---| | `TableCapability.EXTERNAL_LOOKUP` (`connector.catalog`) | The table has a parameterized lookup access path | | `SupportsExternalLookup` | `ScanBuilder` mix-in: `Predicate[] selectLookupPredicates(Predicate[])` | | `LookupParameter` | The typed `?` in an offered predicate | | `Scan.toExternalLookup()` | Like `toBatch()`; the default throws `DATA_SOURCE_EXTERNAL_LOOKUP_NOT_SUPPORTED` | | `ExternalLookup` | `maxParameterRowsPerRequest()`, `createReaderFactory()` | | `ExternalLookupReaderFactory` | `Serializable`; creates readers on executors | | `ExternalLookupReader` | `lookupBatch(InternalRow[])` returns a cursor; `currentMetricsValues()` | | `ExternalLookupResultCursor` | `next()`, `inputOrdinal()`, `get()` | **Why the contracts look this way** - **Identity matching.** Same-shaped candidates must stay distinct. Identity, rather than `equals()` or stable ids, does that with the standard `Predicate` type: ``` ON t.id = s.a AND t.id = s.b + 1 -> offered [id = ? (s.a), id = ? (s.b + 1)] returning the second binds s.b + 1; Spark evaluates t.id = s.a itself ``` - **Positional parameter rows.** There is one `LookupParameter` per offered predicate, so field i of a parameter row is the i-th returned predicate's value; no per-row maps. - **Complete, parameter-only results.** Spark does not re-evaluate selected predicates, and it may reuse a result for equal parameter rows. - **Distinct rows under Spark equality.** Spark sends each equal value once, carrying the first value seen: ``` -0.0, NaN(bits ..01), 0.0, NaN(bits ..02) -> [(-0.0), (NaN ..01)] 'LOOKUP', 'lookup' under UTF8_LCASE -> [('LOOKUP')] ``` - **The selection survives pruning.** `SELECT s.*, t.name FROM s JOIN t ON t.id = s.id` selects `[id = ?]`, then prunes `id`, so the read schema is `(name)`; lookups are still by `id`. - **Reader lifecycle.** Readers are created lazily, at most one per input partition; a task may hold several (two lookup joins, or after `coalesce`). Calls are sequential, never concurrent. The cursor closes before the reader, at input end or task end, possibly on another thread. - **Request size.** At most `maxParameterRowsPerRequest()` rows; Spark may send fewer. - **Failures.** Exceptions fail the task and never count as misses; retries and speculation can repeat a request. - **Empty selection.** No fallback to a scan join exists, so connectors return `[]` rather than throw; Spark then fails the query with `EXTERNAL_LOOKUP_JOIN_UNSUPPORTED.NO_LOOKUP_PREDICATE`. - **Joins and capability ([SPIP](https://docs.google.com/document/d/1fBPM5cku8UJXacG2fw9df1zqALJ-WFm8E9-b9oiFP5M/edit?tab=t.0) Q4, Q6, Q8).** Only stream-static joins: the left side is a streaming query and the lookup table is on the right side. A table without `BATCH_READ` is joined by lookup automatically and cannot be scanned; a table with both needs the `EXTERNAL_LOOKUP` hint. **Differences from [SPIP](https://docs.google.com/document/d/1fBPM5cku8UJXacG2fw9df1zqALJ-WFm8E9-b9oiFP5M/edit?tab=t.0) Appendix A** - The sketch's prose said `selectLookupPredicates` returns "indexes", contradicting its own `Predicate[]` signature; this API returns the offered instances, matched by identity. - `LookupParameter` is `Serializable`, so a reader factory can keep the selected predicates. - Added contracts the sketch omits: nulls, distinctness, input shape, failure, threading and the read schema. - A `ToStringSQLBuilder` case renders `LookupParameter` as `?`; without it, `describe()` on a candidate throws `UNEXPECTED_V2_EXPRESSION`. - The names `LookupParameter` and `lookupBatch` are the sketch's. Out of scope, tracked under [SPARK-60031](https://issues.apache.org/jira/browse/SPARK-60031): - Planning rules. - Execution. - The join hint. - Lookup-side operators and more predicate shapes. - Cache. - Predicate pushdown. - Python Data Source support. ### Why are the changes needed? A stream-static join loads the static side for every micro-batch, even when only a few keys match ([SPIP](https://docs.google.com/document/d/1fBPM5cku8UJXacG2fw9df1zqALJ-WFm8E9-b9oiFP5M/edit?tab=t.0) Q3). No existing read API sends per-key requests with join-side values. ### Does this PR introduce _any_ user-facing change? New public DSv2 interfaces only; no behavior change. ### How was this patch tested? On this commit, JDK 17: ``` build/sbt 'catalyst/testOnly *ExternalLookupApiSuite' 'core/testOnly *SparkThrowableSuite' \ catalyst/scalastyle catalyst/Test/scalastyle sql/Test/compile # JUnit XML: ExternalLookupApiSuite 3/3, SparkThrowableSuite 38/38 passed # scalastyle: 0 errors (main and test); sql/Test/compile: success javac -Xdoclint:html,syntax,reference,accessibility # 0 issues (8 API files) build/mvn -pl sql/catalyst checkstyle:check # checkstyle 11.0.1, dev/checkstyle.xml: 0 violations ``` MiMa was not run. ### 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]
