ASOF join support / Specialize Range Joins

Is your feature request related to a problem or challenge? Please describe what you are trying to do.
In many timeseries workloads, the need to join one record to another based on recency or ordering is quite prevalent. This can be with two sensors that maintain different sampling rates, in handling market data (such as situations where you want to find the most recent quote for a given trade), or any other scenario where an ordering of events can be applied. In these situations, it is quite frequent that one set of data is at a much larger scale than the other, typically starting at an order of magnitude. It is also quite common that you would want to actually store your data in the appropriate order so as to minimize the effort when performing this kind of join operation (though that is not strictly the problem itself).

Describe the solution you'd like
Essentially the solution implemented by Clickhouse and KDB (please note that the Clickhouse solution allows for the full breadth of ordering conditions for closest match).

Describe alternatives you've considered
There is a general subquery solution that can be used to achieve the desired outcome, but it is typically not performant and can be fairly awkward to express. Provided implementation of #141 and sufficient query optimization of the subquery solution into sort-merge join, it may not be necessary as direct syntax.

Additional context

This likely depends on Implement Sort-Merge Join #141
Related to Range/inequality joins are slow #8393

Related issue solved by duckdb: duckdb/duckdb#3219 and duckdb/duckdb#3342

I haven’t evaluated the polar join_asof solution yet, but I believe it is also pertinent.

There is a blog post about this from duckdbL https://duckdb.org/2022/05/27/iejoin.html

There is a blog post about this from duckdbL https://duckdb.org/2022/05/27/iejoin.htmlduckdbL 有一篇关于此的博客文章 https://duckdb.org/2022/05/27/iejoin.html

We can treat asof join as a special kind of iejoin of course. Furthermore AsOf can stop searching when it finds the first match because there is at most one match.This leaves us room for optimization. I think it's ok to support asofjoin without iejoin (which can be done later). Here is another blog about AsOf join https://duckdb.org/2023/09/15/asof-joins-fuzzy-temporal-lookups.

I am willing to have a try if nobody is working on this.

@Dandandan

I am willing to have a try if nobody is working on this.

I don't know of anyone working on this. Maybe @Dandandan @liukun4515 or @ozankabak know of others who are (or are interested in helping)

IMO the ASOF join syntax is not well-thought out and it is better to have such operators downstream (or in datafusion-contrib for public experimentation, but not a part of Datafusion core). I agree with the many of the points raised against this syntax (although not the tone) when it was brought up in PostgreSQL mailing list.

At Synnada, we handle time-series use cases like the ones ASOF tries to address -- but we stay within the realm of standard/consistent SQL. I would be happy to share some info on our approach after the holidays.

BTW I think "Range Join" is pretty similar in spirit and does not require special SQL syntax

https://www.vertica.com/blog/what-is-a-range-join-and-why-is-it-so-fastba-p223413/

https://www.vertica.com/docs/9.2.x/HTML/Content/Authoring/AnalyzingData/Queries/Joins/RangeJoins.htm#:~:text=Vertica%20provides%20performance%20optimizations%20for,two%20columns%20of%20another%20table.

I think there is some non trivial overlap with the notion of symmetric hash join that we already have (maybe we could extend the analysis used there)

@ozankabak

At Synnada, we handle time-series use cases like the ones ASOF tries to address -- but we stay within the realm of standard/consistent SQL. I would be happy to share some info on our approach after the holidays.

Looking forward to your sharing.
Thanks @ozankabak

This is on my mind, this week is very busy but I will circle back as soon as possible.

If we plan to add ASOF JOIN /IEJOIN, I would like to keep an eye on the issue and be involved in its implementation.

I would like to take part in this one.

@ozankabak

IMO the ASOF join syntax is not well-thought out and it is better to have such operators downstream

@ozankabak Can you please elaborate on that a bit? Which of the ~10 existing ASOF join implementations are you considering as not well-thought out and why?

At Synnada, we handle time-series use cases like the ones ASOF tries to address.

It doesn't "try" to address anything, it solves a pain point, that is joining two datasets without matching timestamps efficiently. I'd love to see your "standard/consistent" SQL version though.

This topic came up at the recent DataFusion meetup in San Franciso as something of interest. Is there anyone willing to help drive / lead this project?

I can't offer to do it myself, but I can certainly offer to help review designs and implementations

Thinking about how to break down the work, I think we will need:

Syntax support (I thinks sqlparser-rs already has this, but we may need to pick a dialect, etc)
Figure out how we will represent ASOF join (is it a new JoinType? Is it a new LogicalPlan::AsOfJoin variant? Something else? -- maybe someone could research how other systems handle this)
SQL Planner support (take the AST and make a LogicalPlan)
Optimizer support (not sure what is needed here)
An AsOfJoin ExecutionPlan

Reading https://duckdb.org/docs/guides/sql_features/asof_join.html

SELECT *
FROM holdings h
ASOF JOIN prices p USING (ticker, "when");

Given the limited semantics of this join (match the row with the largest value of "when") I suspect we could use a modified hash join and modify the "does the row match" clause to also check for "is when greater than the currently matched row"

Another way to implement this would be with a slightly modified merge join

AsOfJoinExec(on=ticker, range_match:"when")
  SortExec(expr=[ticker, when])
    Scan(holdings)
  SortExec(expr=[ticker, when])
    Scan(prices)

The idea being that you could drastically simplify the join if the input was already sorted by ticker as you just need to hold on to the most recently seen value of when)

This is probably slower (as you have to sort the entire input including many values that will be filtered) but would allow for other behaviors other than just "the most rececnt value)