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]
