I’ve iterated on the DSv2 APIs offline with Jungtaek and reviewed the SQL APIs. Both look good to me. On-demand lookups against external stores should be particularly useful. +1 on moving this SPIP forward.
On Sun, Oct 11, 2026 at 6:50 PM Jungtaek Lim <[email protected]> wrote: > I've updated the doc to contain the sketched version of DSv2 interface > change in Appendix A, and the justification of not requiring join hint in > stream-static join in Q6. I think other points do not seem to fit with the > SPIP doc, but I can update the doc if we think otherwise. > > Bumping this to see more feedback from broader people. > > On Thu, Oct 8, 2026 at 1:21 PM Jungtaek Lim <[email protected]> > wrote: > >> Thanks Dongjoon for giving detailed feedback! >> >> I intended to keep the SPIP doc short, as the SPIP guide doc states "A >> SPIP document is a short document with a few questions, inspired by the >> Heilmeier Catechism:". >> https://spark.apache.org/improvement-proposals.html >> >> If I read the SPIP guide doc properly (and the set of questions state), >> SPIP should only describe the design at 30000-ft view; it is to make >> consensus whether the project is useful enough to add in Apache Spark (it's >> closer to a business justification, like much simplified version of PRD), >> not to hold a design review for the feature/project. Appendix B also states >> that the justification of the feasibility is the point if we were to have >> the content of design. >> >> > Appendix B. Optional Design Sketch: How are the goals going to be >> accomplished? Give sufficient technical detail to allow a contributor to >> judge whether it’s likely to be feasible. Note that this is not a full >> design document. >> >> That said, I'm deferring the answer/clarification of the detailed design >> for now. Also it sounds like some questions are biased with Flink's ability >> to lookup join, which I think we don't need to catch up at once but make >> incremental improvements. It doesn't mean the functionalities Flink >> provides would be all useful - we need our own assessment, and that can be >> covered by the progress of the project (or even from the feedback of the >> initial version). >> >> Other than that, I answered inline. Updating the doc (e.g. sketched DSv2 >> interface) may need some more time but I'll update it sooner than later. >> >> On Thu, Oct 8, 2026 at 2:12 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) >>> >> >> I can provide the initial version of this - I just hope this doesn't go >> down into the code review. The change of DSv2 interface is tightly related >> to the feature, so it is expected that the interface is subject to change >> on implementing the new join. If this is just to have an outlook of a >> sketch, I'm happy to add it. >> >> >>> 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) >>> >> >> This depends on the established semantic for existing join. >> >> I see static-static join is done with a consistent view on the right side >> and this would be the baseline of expectation, so I agree this is risky >> for batch join. That is why I've mentioned it in the risk section. >> >> stream-static join is different. In practice, the batch cut isn't really >> predictable, so the existing semantic of stream-static join is not >> technically providing a consistent view over changing static side which is >> not the same with static-static join. >> >> For example, let's say there is data 'A' 'B' 'C' 'D' 'C' in batch 1 and >> the static side has changed in the middle of execution on batch 1 to have >> different matches with 'C'. The existing stream-static join provides the >> same match for the two 'C's which seems nice. But what if batches were cut >> in a way to have 'A' 'B' 'C' 'D' in batch 1 and have another 'C' in batch >> 2, and the static side was changed to have different matches with 'C' >> before batch 2? We get different outcomes with the "same" status on stream >> and static. >> >> That said, it sounds like they are different in a batch level, but when >> we look at it over batches, it is hard to reason about the difference, >> because really there is no guarantee we provide a consistent view on >> stream-static join. It is a false expectation if someone reasons about that >> consistency. >> >> That said, I don't think we should require explicit opt-in on >> stream-static join. I think I have heard the similar concern that join hint >> should not change the semantic (if we ever define this as "semantic >> change"), so explicit opt-in on join hint will get into the same concern >> and make this be in a loop. Rather, I argue we are not breaking the >> semantics since we never guarantee consistent output over the static side >> "changing over time". I can drop the support of static-static join for this >> if this would address the concern. (Later if we see the demand then we can >> reconsider.) >> >> We should really solve the issue in the other way around e.g. ASOF JOIN >> with the event time if we want to be consistent on which row to match in >> stream-static join. >> >> >>> 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. >>> >> >> This is a bit ambiguous whether we need to have this in SPIP or not. This >> does not seem to be necessary to justify the project's value, but I >> understand everyone can have different opinions. This is an important >> design question so I'll answer it. >> >> Let's say the new capability is "EXTERNAL_LOOKUP", then we can consider >> the fallback on the data source which supports both. >> >> * stream-static join: EXTERNAL_LOOKUP -> fall back to BATCH_READ >> * static-static join: BATCH_READ (EXTERNAL_LOOKUP is ignored, due to >> semantic in 2) >> >> Worth noting that real-time mode only supports broadcast hash join; users >> can rely on the optimizer to make a decision, but an extremely safe >> approach is to give an explicit hint "broadcast". We can't have two opt-ins >> with fallback, so users may want to provide a join hint to avoid falling >> back. >> >> I can add the above in the SPIP doc if we want to make this explicit. >> >> >>> 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. >>> >> >> I admit I had to either include the sketch of the DSv2 interface or at >> least describe what would be supported. My bad. >> >> Spark will request to evaluate multiple filter expressions to the data >> source, and the data source will report back the chosen (or not chosen) >> filter expressions to Spark, so that the operator will only pass the column >> values over the expressions. The data source is responsible for failing the >> query (or asking Spark to trigger the fallback, depending on the >> complexity) if the given filter expressions aren't making the efficient >> lookup. >> >> I understand the concern of moving responsibility to evaluate the >> condition to the external store. But I think the same thing happens with >> filter pushdown; with JDBC data source, we construct the query to retrieve >> the data with pruning in Spark's semantic but how it would be handled is >> very dependent on the RDBMS. I assume this has been deferred to the >> connector's responsibility, which I don't see a reason not to apply the >> same in here? >> >> >>> 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. >>> >> >> I'm not an expert on this subject, but I don't think lookup join >> operation itself can be deterministic by nature upon task retry. (It is not >> practical to track all the join results so far to produce the same output >> on task retry, and even this is hard to apply when retried tasks are >> assigned to different executors.) We can bake this fact in, if it's >> sufficient to report its deterministic level. >> >> > In addition, for left outer joins, if a lookup error is not >> distinguished from "not found", errors will silently turn into NULL-padded >> rows. >> >> A lookup error won't be swallowed, but the responsibility goes to the >> connector to handle; the connector can retry if the request is idempotent >> (it is true for most of the time on retrieval), but the connector should >> end up with triggering task failure rather than marking it as "null result" >> if it cannot be retried further. >> >> >>> 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. >>> >> >> I don't think the comparison against existing join types is valid here. >> The existing join strategy is high latency by nature because the processing >> part on stream side must wait for the static side to be loaded and be built >> (either broadcast or shuffle). Spark may be able to load and broadcast some >> tiny tables in a sub-second, but this heavily depends on the storage, size, >> how the prune is effective, Spark cluster's capability, etcetc. It is not >> surprising that it took multiple seconds for the "broadcast" to be built, >> which impacts the tail latency. We don't really need to talk about >> non-broadcastable tables. >> >> For low latency processing, it is very important to eliminate the parts >> which incur an outstanding latency at some point, and a one of the valid >> strategies is to smoothen the latency over multiple inputs. (The latency >> problem is mostly happening with tail, P90/P99/P100, not P50 or so. If we >> see the latency problem in P50, it is a proxy that we may have a throughput >> issue.) >> >> That said, having a benchmark to quantify success criteria is a good >> point. We can aim for P99 < 300ms with RDBMS/Redis in real-time mode in >> some benchmark workload (may be artificial, or borrow some query on NexMark >> - it contains some queries for join). >> >> The requirements on batching, key dedup, async IO with concurrency >> control, timeouts, backoff retry, connection pooling & rate limiting are >> just methods to achieve the goal. This isn't even a high level design and I >> think SPIP is not a proper place to have this kind of discussion. Btw, >> supporting batching and/or async IO is not a one way door - batching is >> mostly mandatory, but async IO may not. We may end up not considering async >> IO to MVP and leave it to be a follow-up work if batching solely works >> pretty well. >> (IMHO, async IO is the main caveat to making a ton of requests against >> external storage since this assumes there are multiple requests on the fly. >> I wouldn't start with this in the initial version, unless I couldn't >> achieve the goal without it.) >> >> >>> 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. >>> >> >> I see this as a detailed design, not a high level one. Only the high >> level design decision would be the lifecycle of the cache - I don't >> consider the cache to live beyond the task lifecycle at this point, to >> simplify the thing. e.g. this won't end up with more stale data than today. >> An initial version does not need to be perfect, and taking a simpler model >> is better for incremental improvement over taking a complicated model and >> struggling with it. (In line with YAGNI) >> >> > The SPIP has both hint options and SQL configs for the cache, but >> their precedence is not defined. >> >> I missed to add it but the scope is natural to how they work; SQL configs >> are session level, and hint options will override it to tune it to an >> operator level. >> >> >>> 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] >>> >>>
