Jungtaek, Thank you for sharing the SPIP! I’m generally very supportive of this work. Efficient external lookups are an important capability for streaming queries, especially for queries running in Real-Time Mode (RTM).
A few questions and thoughts: 1. *External APIs / RPCs beyond OLTP lookups* This proposal focuses primarily on joins with OLTP systems. What are your thoughts on supporting use cases where users simply need to invoke an external API or RPC? For example, a streaming query might need to call an external LLM service for inference or evaluation. Would you consider this a separate use case, where we should introduce something like an *async UDF*, rather than model it as an external lookup join? 2. *How should users select the join strategy?* Today we have broadcast joins and shuffle joins, and this proposal would introduce external lookup joins. A connector could potentially support all three, so how should Spark decide which strategy to use? Since an external lookup join has meaningfully different performance and execution characteristics, I would recommend requiring users to explicitly opt into it through the API. For example: streamDf.join(lookup(staticDf), "dept_id") This makes the execution semantics clear rather than having Spark implicitly choose an external lookup strategy. 3. *Built-in support for batching and asynchronous execution* To perform external lookups efficiently, users will typically need to batch requests—potentially based on both time and row count—and execute multiple lookups asynchronously. While connector implementations could build these mechanisms themselves, doing so correctly and efficiently can be fairly involved. Could we provide some built-in abstractions or APIs to help connector authors implement performant batching, concurrency, backpressure, and asynchronous lookup execution? Again, I’m very glad we’re looking at adding this capability. I think it will be particularly valuable for low-latency streaming workloads. Best, Jerry 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] > >
