2010YOUY01 commented on PR #23738: URL: https://github.com/apache/datafusion/pull/23738#issuecomment-5056713964
okay I got some idea how to implement it. Here are some thoughts: ### Semantics to implement > > * What are the precise semantics of an ASOF join? I suspect that different systems, such as DuckDB and Snowflake, may differ slightly, so we should first define the semantics rigorously. > > I propose Snowflake-style semantics: each left row selects the closest eligible right row within an optional equality-key group. `>/>=` selects the greatest right key; `</<=` selects the smallest. Left rows are always preserved, unmatched rows are NULL-padded, NULL keys do not match, and ties are nondeterministic. I think it's a good idea. DuckDB distinguishes between inner and outer ASOF joins, while Snowflake supports only one variant: left outer. I don't fully understand why Snowflake made this decision, but perhaps it is the most common pattern in practice. In any case, it would be preferable to start simple. Regarding ASOF join conditions, both DuckDB and Snowflake support a conjunction consisting of: - Exactly one inequality condition - Zero or more equality conditions For example: ```sql SELECT t.*, p.price FROM trades t ASOF JOIN prices p ON t.symbol = p.symbol AND t.when >= p.when; -- 1 ie cond + 1 eq cond ``` We probably want to support the same join conditions. ### Algorithm to implement > > * What is the high-level algorithm used by the ASOF join operator? > > The initial operator uses an ordered merge: co-partition and order both inputs, then advance the right cursor monotonically while retaining the latest eligible candidate. This is O(L + R) after ordering with bounded join state. Other implementations could later share the same logical semantics. Let's call this idea the repartition-based ASOF join. An alternative is the broadcast-based ASOF join, described below. I tend to think it's better to implement the broadcast-based algorithm in the first version, here are the reasons: (TLDR: assuming its common to have workloads with only inequality condition, but no equality ASOF join condition, the broadcast approach can fully utilize all CPUs, while the repartition based approach can't) I think both approaches will be useful in the long term, as they target different workload patterns. - repartition-based: More efficient for large number of equality groups, no skew - broadcast-based: If there are no equality condition, or equality keys have very low cardinality, or equality key are heavily skewed, it performs better than repartition-based approach. (it seems the no-equality workloads are quite common) For example, consider the join condition `(t1.v1 < t2.v1) AND (t1.tag = t2.tag)`. The repartition-based algorithm requires both inputs to be repartitioned by `tag` before sorting and joining. If there are only two distinct tags, there will be only two effective partitions, leaving most CPU cores unused in a multicore environment. While the broadcast-based approach can fully utilize all the available CPU cores for the joining phase. For workloads that can be perfectly repartitioned, the broadcast approach should not be much slower. Its main additional cost is maintaining and advancing more cursors over the broadcast side, which we could probably vectorize relatively easily. #### Broadcast ASOF Join - Broadcast and sort the build side; repartition the probe side using batch-level round-robin partitioning, then sort each resulting partition. - Perform a monotonic scan independently within each probe partition. ### Implementation plan 1. Question: Are the current split PRs independently mergeable? 🤔 In other words, would merging the first one leave the main branch in a consistent and functional state, or are the PRs split primarily to make the implementation easier to read and review? 2. Suggestion: Another useful principle might be to keep the first patch simple and defer aggressive low-level optimizations when they make the code substantially harder to follow. 3. Suggestion: It would be great to write a top-level design document explaining the approach to readers without much background knowledge. It could start with an example query, explain the semantics, show the query-plan shape, and finally describe the algorithm inside the ASOF join operator. It should also summarize the major design decisions. This could be written as a top-level comment. Here is a good reference: https://github.com/apache/datafusion/blob/1c8295c992c597a53dcfa01f6dcf1cdc51268adb/datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs#L67 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
