Support limit pushdown through left right outer join

[Ted-Jiang]

Which issue does this PR close?

run

 explain select * from order left  join item  on  order.o_orderkey = item.l_orderkey  limit 1;

before

| logical_plan  | Limit: 1                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
|               |   Projection: #order.o_orderkey, #order.o_custkey, #order.o_orderstatus, #order.o_totalprice, #order.o_orderdate, #order.o_orderpriority, #order.o_clerk, #order.o_shippriority, #order.o_comment, #item.l_orderkey, #item.l_partkey, #item.l_suppkey, #item.l_linenumber, #item.l_quantity, #item.l_extendedprice, #item.l_discount, #item.l_tax, #item.l_returnflag, #item.l_linestatus, #item.l_shipdate, #item.l_commitdate, #item.l_receiptdate, #item.l_shipinstruct, #item.l_shipmode, #item.l_comment                                                                                                                                                                                                                                                                                                      |
|               |     Left Join: #order.o_orderkey = #item.l_orderkey                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
|               |       TableScan: order projection=Some([0, 1, 2, 3, 4, 5, 6, 7, 8])                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
|               |       TableScan: item projection=Some([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15])

after:

| logical_plan  | Limit: 1                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
|               |   Projection: #order.o_orderkey, #order.o_custkey, #order.o_orderstatus, #order.o_totalprice, #order.o_orderdate, #order.o_orderpriority, #order.o_clerk, #order.o_shippriority, #order.o_comment, #item.l_orderkey, #item.l_partkey, #item.l_suppkey, #item.l_linenumber, #item.l_quantity, #item.l_extendedprice, #item.l_discount, #item.l_tax, #item.l_returnflag, #item.l_linestatus, #item.l_shipdate, #item.l_commitdate, #item.l_receiptdate, #item.l_shipinstruct, #item.l_shipmode, #item.l_comment                                                                                                                                                                                                                                                                                                      |
|               |     Left Join: #order.o_orderkey = #item.l_orderkey                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
|               |       TableScan: order projection=Some([0, 1, 2, 3, 4, 5, 6, 7, 8]), limit=1                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       |
|               |       TableScan: item projection=Some([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15])

Closes #2579.

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

[alamb]

Thank you for the contribution @Ted-Jiang , howe I don't think this is a valid optimization.

Specifically, because a join can filter out rows, if you limit the input you may actually end up with fewer output rows than the limit.

Consider this input:

left:

l
1
2
...
100

right:

r
99
100

The output of select * from left LEFT JOIN right ON (l = r) should be:

l	r
99	99
100	100

However, if you push the limit down to the scan on left it would only send this into the JOIN

left:

l
1
2

And thus would produce no output. 

If we want to optimize limits in Joins, I think it would have to be done in the Join Operator itself (to stop producing rows once the limit is hit). However, as long as the Join operator is producing rows in batches, the effect of implementing an internal limit will likely be small (because it would save only a part of one output batch, when the limit on the output of a join is hit)

[Ted-Jiang]

Thank you for the contribution @Ted-Jiang , howe I don't think this is a valid optimization.

Specifically, because a join can filter out rows, if you limit the input you may actually end up with fewer output rows than the limit.

Consider this input:

left:

l
1
2
...
100
right:

r
99
100
The output of select * from left LEFT JOIN right ON (l = r) should be:

l r
99 99
100 100
However, if you push the limit down to the scan on left it would only send this into the JOIN

left:

l
1
2

And thus would produce no output. 

If we want to optimize limits in Joins, I think it would have to be done in the Join Operator itself (to stop producing rows once the limit is hit). However, as long as the Join operator is producing rows in batches, the effect of implementing an internal limit will likely be small (because it would save only a part of one output batch, when the limit on the output of a join is hit)

I think this situation is select * from left LEFT JOIN right ON (l = r) without limit. There will no limit in left table_scan
will still produce

left:

l
1

...
100

But this rule will apply select * from left LEFT JOIN right ON (l = r) limit 2
which left table will get the limit will send two values to join, i think this is right
and the result will be

l	r
1       Null
2	Null

@alamb is this situation make sense

[Ted-Jiang]

One question is df LEFT JOIN is equal to LEFT OUTER JOIN？

[alamb]

One question is df LEFT JOIN is equal to LEFT OUTER JOIN？

Yes

[alamb]

Yes, you are right @Ted-Jiang -- I was confused about the side of the join 🤦

Upon more thought I think this is correct.

Thank you again.

I'll leave it open for another day or so before merging in case anyone else has thoughts

[Ted-Jiang]

Yes, you are right @Ted-Jiang -- I was confused about the side of the join 🤦

Upon more thought I think this is correct.

Thank you again.

I'll leave it open for another day or so before merging in case anyone else has thoughts

Thanks @alamb ❤️.
I find this rule in spark
https://github.com/apache/spark/blob/efe43306fcab18f076f755c81c0406ebc1a5fee9/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala#L707

[jackwener]

LGTM.
It can reduce data source in the side of join.
Great job❤️.

[alamb]

it seems as if there is a clippy error now 😢

[alamb]

And now there is a conflicts 🤦

[alamb]

reopened as part of #2596

[alamb]

Strangely github won't let this PR be reopened 🤔

[alamb]

#2596 is now merged 👏