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] >> >>
