HeartSaVioR opened a new pull request, #60087: URL: https://github.com/apache/spark/pull/60087
### What changes were proposed in this pull request? This PR adds a lookup cache 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-60693](https://issues.apache.org/jira/browse/SPARK-60693), part of [SPARK-60031](https://issues.apache.org/jira/browse/SPARK-60031) (SPIP: External Lookup Join). Repeated keys then don't cost repeated remote lookups. It depends on apache/spark#60055 (lookup-side projections and filters, semi/anti joins and range predicates, SPARK-60692), and through it on apache/spark#60019 (the `EXTERNAL_LOOKUP` hint, SPARK-60691), apache/spark#59910 (basic execution, SPARK-60690) and apache/spark#59909 (planning rules, SPARK-60689). The connector API (SPARK-60688) is already in master. **Cache** - **Scope.** The cache holds the complete lookup result of each key, including keys that returned no rows. It lives within a task, per input partition: a task that reads several input partitions, e.g. after `coalesce`, has one cache per partition. It is discarded when the task ends, so a cached result never outlives the micro-batch that read it. - **Bounds.** It is an LRU over `maxEntries` keys, with an optional TTL. It is a Guava cache with one segment, so eviction is exactly least recently used, and expiry is measured from the write. A key whose result has more than `maxRowsPerKey` rows is streamed to the join but not cached. - **Hits and requests.** Each distinct key of a batch is classified once, either as a cache hit or as a key to request. So a TTL expiry between duplicate keys of one batch cannot send the key again. Requests are still deduplicated within a batch, and the cache extends that across the batches of an input partition. - **What is cached.** The cache stores the rows the join consumes. These are the connector's value rows, in read-schema order and nullability, after the lookup-side projections and filters. A hit therefore skips those stages, and it reuses the values of any non-deterministic lookup-side expression, as a broadcast build side does. `maxRowsPerKey` counts rows after the stages. - **Complete results.** A key's result is cached once its request's cursor is read to the end, when the result is complete. A failure fails the task, which discards its cache, and the join never retries a request. - **Metric.** `numCacheHits` counts the distinct keys served from the cache, once per batch. The description is "external lookup cache hits". **Configuration** | Conf | Default | Meaning | |---|---|---| | `spark.sql.join.externalLookup.cache.enabled` | `true` | Cache lookup results | | `spark.sql.join.externalLookup.cache.maxEntries` | `1000` | Keys per cache, i.e. per input partition | | `spark.sql.join.externalLookup.cache.ttl` | `0` (no expiry) | Time-to-live of a cached result | | `spark.sql.join.externalLookup.cache.maxRowsPerKey` | `100` | Rows cached per key, after lookup-side stages | Per-join options are quoted `key=value` strings on the `EXTERNAL_LOOKUP` hint, in both DataFrame and SQL form: `cache_enabled`, `cache_max_entries`, `cache_ttl`, `cache_max_rows_per_key`. For example: ```scala stream.join(customers.hint("external_lookup", "cache_ttl=30s", "cache_max_entries=500"), cond) ``` ```sql SELECT /*+ EXTERNAL_LOOKUP(c, 'cache_enabled=false') */ ... ``` - Options are validated while the query is analyzed. Invalid ones fail with the new `INVALID_EXTERNAL_LOOKUP_OPTION` condition: - `DUPLICATE`: the same option given twice. - `INVALID_FORMAT`: not in `key=value` form. - `INVALID_VALUE`: a value of the wrong kind, e.g. a negative duration. - `UNSUPPORTED`: an unknown option; the error lists the supported ones. - `INVALID_EXTERNAL_LOOKUP_HINT.UNSUPPORTED_PARAMETER` now says that the hint accepts relation names and string options. - A hint's options apply to the join whose lookup side it annotates, including when lookup-side projections or filters sit between the hint and the join. A join without the hint uses the session confs. **For discussion** - The cache is enabled by default. The SPIP's examples set `cache_enabled=true`, but the SPIP states no default. - Memory is bounded by key and row counts, not bytes. The worst case is `maxEntries × maxRowsPerKey` rows per input partition, which is 100k rows at the defaults. `ExternalLookupHint`, `ExternalLookupJoin`, `PreparedExternalLookupJoin` and `ExternalLookupJoinExec` each gain a `cacheOptions` field. Out of scope, tracked under [SPARK-60031](https://issues.apache.org/jira/browse/SPARK-60031): - Caches shared across tasks or queries. - Pushing stream-side filters and join-condition conjuncts through the join. ### Why are the changes needed? A stream often repeats lookup keys. Without a cache, every batch requests its keys again, even when an earlier batch of the same task just read them. ### Does this PR introduce _any_ user-facing change? Yes. There are four new SQL configs and four `EXTERNAL_LOOKUP` hint options. The cache is on by default for external lookup joins, so repeated keys within an input partition send fewer requests. ### How was this patch tested? New tests in `ExternalLookupJoinSuite`, all run as stream-static joins: - **Options.** Duplicate, malformed, invalid and unsupported options are rejected, and so is a non-string hint parameter. A SQL hint's cache options reach the prepared join. - **Basic behavior.** - Disabling the cache keeps request deduplication within a batch. - SQL defaults apply, and hint options override them. - The cache trades freshness, and caches misses. - TTL expiry, including an expiry between duplicate keys of one batch. - `numCacheHits`. - A result over `maxRowsPerKey` is streamed but not cached. - Floating-point keys that are equal under Spark equality share a cache entry. - **LRU.** With capacity 2, the keys `1, 2, 1, 3, 1, 2` in one-row requests must send `1, 2, 3, 2`. An insertion-order map would send `1, 2, 3, 1, 2`. - **Scope.** With two input partitions and one-row requests, each task can avoid its second request only through its own cache. - **Hint placement.** A hint's `cache_enabled=false` applies through a lookup-side projection between the hint and the join. - **With the lower layers.** - Hits through a reordered read schema and through a read schema more nullable than the table return the same rows as requests. - Hits through lookup-side projections and filters work in inner, semi and anti joins. Earlier tests that count connector requests, or check non-cache behavior, opt out with `cache_enabled=false`, on six lines. The read-schema test fixture now sends one-row requests, so its repeated keys hit the cache. `TableCapabilityCheckSuite` passes the new `cacheOptions` argument. Tested on JDK 21, with this commit and a planned optimizer-only change on top: ``` build/sbt \ 'catalyst/testOnly *FormExternalLookupJoinSuite *PushPredicateThroughExternalLookupJoinSuite' \ 'sql/testOnly *PrepareExternalLookupJoinSuite *.ExternalLookupJoinSuite \ *ExternalLookupJoinRealTimeModeSuite' \ catalyst/scalastyle catalyst/Test/scalastyle sql/scalastyle sql/Test/scalastyle # JUnit XML: all tests passed, with no failures or errors # 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]
