Hi Jungtaek, +1 from my side on the proposal as well.
I had a couple of questions around the proposal: - Whats our proposed test coverage strategy for this feature in Spark ? - Do we plan to support executor level caches in v1 as well ? If so, could we add some detail around this section ? Thanks, Anish On Wed, Oct 7, 2026 at 10:13 AM Dongjoon Hyun <[email protected]> wrote: > Hi, Jungtaek. > > Thank you for sharing the SPIP. > > 1. The DSv2 interface is missing > > Appendix A covers the new hint and SQL configs, but not the DSv2 > interfaces and Python Data Source APIs, which are the main new public > contracts. The DSv2 interface is the most long-lived decision because > third-party connectors will implement it, but it is omitted, and Appendix B > is skipped. I'd like to see at least a sketch of: > > - how the lookup capability is expressed (a new `TableCapability` or a > mix-in interface) > - how a connector declares which key sets can be looked up (primary key, > indexes) > - the batched lookup signature (a batch of keys in, matching rows out) > - projection and filter pushdown into the lookup > - the lifecycle of lookup readers and connections (per task or per > executor) > > 2. Automatic selection changes the semantics, not just the plan > > Today, the choice of a join strategy does not change the query result. A > lookup join, however, changes the semantics: rows with the same key in the > same query (or micro-batch) can see different values, as Q6 acknowledges. > With automatic selection, upgrading a connector that adds this capability > would silently change both the result semantics of existing queries and the > load on the external database. In addition, the selection is based only on > the capability, without any cost consideration. For example, for a 100-row > dimension table, a broadcast hash join is likely cheaper than remote > lookups. > > Note that Flink's lookup join has the same processing-time semantics. The > difference is that Flink makes it explicit in the syntax (`FOR SYSTEM_TIME > AS OF` with a processing-time attribute). Since the SPIP reuses the > existing join syntax without such a marker, I'd suggest: > > - scoping the first version to stream-static joins (Real-Time Mode first) > - requiring an explicit opt-in via a hint or a SQL config (disabled by > default) > > 3. Fall back instead of failing > > Q6 says that queries will fail when the static side has operators other > than project and filter. However, the existing join hints are best-effort > ("if doable") and only log a warning when they cannot be applied > (`HintErrorLogger`). Failing in the automatic path would break queries that > work today. So, sources that also support `BATCH_READ` should fall back to > the existing join strategies. On the other hand, a lookup-only source > without `BATCH_READ` (e.g., a store that cannot scan) cannot even be used > as the static side today: `DataSourceV2Utils.loadV2Source` creates a > `DataSourceV2Relation` only for tables with `BATCH_READ`, and > `TableCapabilityCheck` rejects a `DataSourceV2Relation` without it. We need > a capability model that covers both READ+LOOKUP and LOOKUP-only sources. > > 4. Equality is evaluated by the external store > > With a regular join, Spark evaluates `=`. With a lookup join, the external > store evaluates it, and the semantics can differ: collations (Spark's > `UTF8_LCASE`, MySQL's case-insensitive default collation), CHAR padding, > type conversions, NULL and `<=>`, NaN and -0.0. Like the post-scan filters > of DSv2 filter pushdown, Spark should re-evaluate the full join condition > on the returned rows to drop extra matches. However, this cannot recover > missing matches (e.g., a `UTF8_LCASE` key against a case-sensitive store). > In addition, a key wrapped in an expression on the lookup side (e.g., a > cast, which Spark's type coercion may add when the key types differ) may > not be usable for lookups, and a key without an index can turn each lookup > into a full table scan. This is why a connector should declare which keys > can be looked up and with which equality semantics, and the planner should > match against them. > > 5. Determinism on retries > > A retried task can see different lookup results. In batch and micro-batch > queries, the order-independent shuffle checksum > (`spark.sql.shuffle.orderIndependentChecksum.enabled`, enabled by default > since 4.2.0) detects such differences and retries all tasks of the consumer > stages, which protects downstream shuffles but is costly. In Real-Time > Mode, this checksum-based retry does not apply to pipelined shuffles, and > `DAGScheduler.checkPipelinedProducerSupported` rejects a statically > INDETERMINATE producer of a pipelined shuffle, so we need to design how the > lookup join reports its deterministic level there. In addition, for left > outer joins, if a lookup error is not distinguished from "not found", > errors will silently turn into NULL-padded rows. > > 6. Protecting the external store, and evidence for "low latency" > > A remote round trip per record is slower than probing an in-memory > broadcast hash table. So, the benefits come from (a) no waiting for the > broadcast at the start of a batch, (b) fresher data, and (c) joining > dimension tables too large to broadcast without a shuffle. I'd suggest > adding a benchmark to the success criteria to quantify them. The design > also needs batching (multi-get), key deduplication, async I/O with bounded > concurrency, timeouts, retries with backoff, and per-executor connection > pooling or rate limiting. Otherwise, thousands of tasks may hit a single > PostgreSQL instance at the same time. For reference, Flink's LOOKUP hint > covers `async`, `output-mode`, `capacity`, `timeout`, `retry-predicate` > (`lookup_miss`), and `shuffle`, while the hint options proposed so far > cover only the cache. > > 7. Cache design > > The SPIP doesn't define where the cache lives (task or executor), how its > memory is accounted (ideally by bytes with the memory manager rather than > by the number of entries), negative caching, or invalidation. In > particular, if `cache_max_rows_per_key` truncates the rows of a key, the > join would produce wrong results. A TTL cache can also serve staler data > than today's per-batch reload, so it should be opt-in on its own. Flink > puts the cache settings into the connector (table) options (`lookup.cache`, > `lookup.partial-cache.*`) and uses hints for execution behavior. The SPIP > has both hint options and SQL configs for the cache, but their precedence > is not defined. > > Thank you again for driving this. > > Dongjoon. > > > On 2026/10/07 08:30:22 Jungtaek Lim wrote: > > Hi devs, > > > > I'm coming up with a proposal for a new join strategy, named "External > > Lookup Join". > > > > tl;dr: > > All existing join strategies require the static side (DataFrame/Relation) > > to be materialized before the join operator. This doesn't allow us to > > achieve low latency join where the storage of the static side is expected > > to produce low latency response over on-demand request. (e.g. OLTP > > optimized like RDBMS, in-memory storage, NoSQL, etc.) The new join > strategy > > will issue the query to the external storage "on demand", when processing > > the data on the left side. > > > > The most benefit will come with real-time mode, but it is not limited to > > real-time mode. This join strategy would be useful as long as the > external > > storage of the static side is better suited for on-demand requests > (rather > > than bulk load with pruning). Both static-static and stream-static can be > > benefited by this, though static-static join needs to be careful on doing > > this. > > > > More details are included in the SPIP doc. > > > > Issue: https://issues.apache.org/jira/browse/SPARK-60031 > > SPIP doc: > > > https://docs.google.com/document/d/1fBPM5cku8UJXacG2fw9df1zqALJ-WFm8E9-b9oiFP5M/edit?tab=t.0 > > > > Looking forward to hearing from you. > > > > Thanks! > > Jungtaek Lim (HeartSaVioR) > > > > --------------------------------------------------------------------- > To unsubscribe e-mail: [email protected] > >
