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