Thanks Jerry for the input! Replied inline. On Thu, Oct 15, 2026 at 2:15 PM Jerry Peng <[email protected]> wrote:
> 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? > > External lookup join is a high level API assuming the data source would handle the requests for scanning/pointing with some simple expressions (mostly subset of the join condition), e.g. equality, range, etc. Beyond those expressions, Spark does not have an idea to build a request. One example is dynamic request - say, if they want to prompt based on the data, this wouldn't be easy to generalize on the connector itself. It's most likely that every user would have their own business logic to cover, which warrants lower level API so that users can customize on their own. The dynamic prompt may not even simply rely on a single column of the record; it could construct a complex routing leveraging multiple columns. So I think this proposal does not cover the demand of async UDF (I can update the doc if SPIP doc didn't explicit call this out) - I think TWS isn't even a proper fit to cover async UDF, not because the current offering does not help building the async requests, but because being stateful operator requires a lot of downsides and the case doesn't even need to be grouped. There might be potential that AI integration could be generalized, but at least it's beyond this project. > > 1. > > *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. > > I'd love to see this in phased development. I'd say the final goal should be something Spark reasons about the cost of the strategy and choose it. The existing strategies (broadcast, sort merge, hash) also have meaningfully different performance and execution characteristics and there are always questions/complaints why Spark chooses one strategy over another. But we certainly don't want users to specify the join hint every time they join. I think external lookup should be just included, not considered as a whole different join strategy being differentiated with them. That said, I see the concern that a new join strategy is chosen without a reasonable cost model when the data source supports both. I think the preference of the join strategy for that case could be a knob (internal) which the default value is non-external lookup (to keep the behavior the same), while users can set this to external-lookup or even fail the query to request the join hint explicitly. I just don't want to require users to always opt in with the join hint even if the data source only supports external lookup - I assume the case is not uncommon; e.g. Redis doesn't work well with scanning data (though this is > 5 years old memory when Redis was single threaded). > > 1. > > *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? > > Built-in support for batching will be one of requirements for MVP (initial version) since the model is straightforward. Building the general framework to help data source do async requests easily and correctly would justify its own design, which I'd love to defer to the next version. We can consider including this if we realize batching cannot satisfy the performance, but I have some preliminary benchmark that batching is sufficient to keep the join low latency. 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] >> >>
