OOM when nested join + limit

[mmooyyii]

Is your feature request related to a problem or challenge?

# make testdata
import pandas as pd

n_rows = 100000

data = {'id': range(1, n_rows + 1)}
df = pd.DataFrame(data)
df.to_parquet('test1.parquet')
df.to_parquet('test2.parquet')

yimo@YideMacBook-Pro PythonProject % datafusion-cli
DataFusion CLI v46.0.1
> select * from 'test1.parquet' as t1 join 'test2.parquet' as t2 on t1.id < t2.id limit 10;
zsh: killed     datafusion-cli

Describe the solution you'd like

same sql in postgresql

create table t1
(
    id integer
);

create table t2
(
    id integer
);
truncate t1;
truncate t2;
insert into t1(select *
               from generate_series(0, 100000));
insert into t2(select *
               from generate_series(0, 100000));

explain analyse select * from t1 join t2 on t1.id <= t2.id limit 10;

Limit  (cost=0.00..0.45 rows=10 width=8) (actual time=0.021..36.926 rows=10 loops=1)
  ->  Nested Loop  (cost=0.00..169452460.14 rows=3765537655 width=8) (actual time=0.021..36.924 rows=10 loops=1)
        Join Filter: (t1.id <= t2.id)
        Rows Removed by Join Filter: 299997
        ->  Seq Scan on t2  (cost=0.00..1572.65 rows=112965 width=4) (actual time=0.007..0.011 rows=4 loops=1)
        ->  Materialize  (cost=0.00..1943.02 rows=100001 width=4) (actual time=0.008..6.338 rows=75002 loops=4)
              ->  Seq Scan on t1  (cost=0.00..1443.01 rows=100001 width=4) (actual time=0.003..4.574 rows=100001 loops=1)
Planning Time: 0.044 ms
Execution Time: 37.334 ms

Describe alternatives you've considered

I think should merge limit and join to one physical plan;

Additional context

No response

[kosiew]

take

[kosiew]

@alamb

I think a SortMergeJoin derivative that handles inequality could be helpful here but there are comments that
SortMergeJoin is not used in plans yet.

Is there a reason for this?

[2010YOUY01]

@alamb

I think a SortMergeJoin derivative that handles inequality could be helpful here but there are comments that SortMergeJoin is not used in plans yet.

Is there a reason for this?

The optimization to execute join conditions like t1.id <= t2.id efficiently with SMJ is not implemented yet, now SMJ can only handle an equal join condition (plus additional join conditions to be processed after the equal condition is evaluated)

[2010YOUY01]

I think #15760 is needed to fix the issue.

[kosiew]

@2010YOUY01

What do you think of using sorted inputs, just like SortMergeJoinExec

Instead of ==, allow conditions like:

t1.id < t2.id
t1.timestamp <= t2.event_time

For each row in left, scan right from current pointer and stop afer emitting LIMIT rows

[Dandandan]

There is also a now closed PR for IEJoin

#12754 (review)

That allows joining on some other conditions.

[alamb]

TLDR while this is straight forward bug report, I think fixing it is not not something we are going to make a patch for -- it will require a more serious implementation effort for joins in DataFusion

Joins implementations in general are a complex topic. Maybe it is time to mount and organize a project to improve the situation in DataFusion

As @Dandandan alluded to, I don't think we need to reinvent this feature -- there is lots of prior academic work on this topic.

What I suggest is that someone updates our documentation with the current state of joins in DataFusion (namely what operators are implemented and what types of joins they are used for any limitations

Then we can figure out which of the many many exotic join algorithms / implementations exist would be good to move on.

[xudong963]

What I suggest is that someone updates our documentation with the current state of joins in DataFusion (namely what operators are implemented and what types of joins they are used for any limitations

Then we can figure out which of the many many exotic join algorithms / implementations exist would be good to move on.

Makes a lot of sense. For new contributors/users who are interested in the Join topic, this proposal would definitely be helpful and clear

[alamb]

I can try and help this effort in a few weeks, but I need to finish up the filter pushdown work in parquet and topk work first (I have too many things outstanding at the moment and am spread too thin)

[mmooyyii]

It fixed after datafusion-cli 50.2.0, @kosiew

> select * from 'test1.parquet' as t1 join 'test2.parquet' as t2 on t1.id < t2.id limit 10;
+----+------+
| id | id   |
+----+------+
| 1  | 8193 |
| 1  | 8194 |
| 1  | 8195 |
| 1  | 8196 |
| 1  | 8197 |
| 1  | 8198 |
| 1  | 8199 |
| 1  | 8200 |
| 1  | 8201 |
| 1  | 8202 |
+----+------+
10 row(s) fetched. 
Elapsed 0.009 seconds.

[2010YOUY01]

I think should merge limit and join to one physical plan;

We haven't implement this idea yet, the reproducer query is working I think is due to NLJ is rewritten to limit intermediate data size in #16996

I think pushing limit to joins is a good idea and it can speed up certain workloads, filed #18295 to track it.