[FEAT] Filter predicates in SQL join

[kevinzwang]

Adds support for things like:

SELECT * FROM a JOIN b ON a.x = b.x AND a.y > 0

Enables TPC-H q13

[codspeed-hq]

CodSpeed Performance Report

Merging #3371 will not alter performance

_{Comparing kevin/sql-join-on-filter (c018163) with main (5fee192)}

Summary

✅ 17 untouched benchmarks

[kevinzwang]

decided revert my new_with_context here because SQLPlanner.plan_relation does not mutate the plan nor does it use the alias map. Subquery relations are already handled in plan_relation with a new_with_context so this is redundant.

[kevinzwang]

I use a macro here because a function would have to own the error object to return it, and I would like to use the error in later parts of the code if it is not returned.

[universalmind303]

@kevinzwang wouldn't this be better to handle directly in the logical plan? That would allow us to support non equi joins in the dataframe api as well.

ex

df1.join(df2, on=(df1["a"] == df2["a"] & df2["b"] > 0))

[codecov]

Codecov Report

Attention: Patch coverage is 87.80488% with 15 lines in your changes missing coverage. Please review.

Project coverage is 77.45%. Comparing base (b6695eb) to head (c018163).
Report is 4 commits behind head on main.

Files with missing lines	Patch %	Lines
src/daft-sql/src/planner.rs	87.70%	15 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3371      +/-   ##
==========================================
+ Coverage   77.39%   77.45%   +0.05%     
==========================================
  Files         678      678              
  Lines       83300    83278      -22     
==========================================
+ Hits        64469    64501      +32     
+ Misses      18831    18777      -54     

Files with missing lines	Coverage Δ
src/daft-sql/src/lib.rs	100.00% <100.00%> (ø)
src/daft-sql/src/planner.rs	71.81% <87.70%> (+1.17%)	⬆️

... and 3 files with indirect coverage changes

---- 🚨 Try these New Features:

• Flaky Tests Detection - Detect and resolve failed and flaky tests

[graphite-app]

Graphite Automations

"Request reviewers once CI passes" took an action on this PR • (11/21/24)

1 reviewer was added to this PR based on Andrew Gazelka's automation.

[kevinzwang]

@kevinzwang wouldn't this be better to handle directly in the logical plan? That would allow us to support non equi joins in the dataframe api as well.

ex

df1.join(df2, on=(df1["a"] == df2["a"] & df2["b"] > 0))

I agree. However, we would potentially need to have a concept of table-associated columns in our logical plan as well as make changes to our join op struct.

I think we should definitely do it in the future, but I didn't want to broaden the scope of this PR. Additionally, I think we would still need to have SQL-specific logic to identify the join keys for each side in a query, so a lot of the work in this PR is still relevant.

Let me know what you think!

[universalmind303]

@kevinzwang wouldn't this be better to handle directly in the logical plan? That would allow us to support non equi joins in the dataframe api as well.
ex

df1.join(df2, on=(df1["a"] == df2["a"] & df2["b"] > 0))

I agree. However, we would potentially need to have a concept of table-associated columns in our logical plan as well as make changes to our join op struct.

I think we should definitely do it in the future, but I didn't want to broaden the scope of this PR. Additionally, I think we would still need to have SQL-specific logic to identify the join keys for each side in a query, so a lot of the work in this PR is still relevant.

Let me know what you think!

I think we would still need to have SQL-specific logic to identify the join keys for each side in a query, so a lot of the work in this PR is still relevant.

that sounds good for now. I think an interesting dataframe case is df1.join(df2, on=(df1["a"] == df2["a"] & df2["b"] > 0)). How would the dataframe side identify df1.a vs df2.a? Wouldn't we need similar logic to handle this case?

Additionally, how would you express this using the dsl col syntax instead of bracket notation. col("df1.a")? or df1.col("a")

I think we should definitely do it in the future, but I didn't want to broaden the scope of this PR.

I'll open up an issue for dataframe non-equi joins just so we don't lose this!

[kevinzwang]

that sounds good for now. I think an interesting dataframe case is df1.join(df2, on=(df1["a"] == df2["a"] & df2["b"] > 0)). How would the dataframe side identify df1.a vs df2.a? Wouldn't we need similar logic to handle this case?

DataFusion has a concept of a table reference in a column expression and I think we could something similar if you use df["a"] notation