`RewriteSetComparison` is overly aggressive rewriting some set comparison into multiple mark joins
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
Currently `RewriteSetComparison` turns set comparisons into `CASE` with multiple `EXISTS`, which ends up as multiple mark joins.
For example, from https://github.com/apache/datafusion/pull/25273/ (and using that branch to get a correct plan) we get:
```
explain select id from emp WHERE salary = ANY (SELECT salary FROM emp WHERE id <= 2);
+---------------+-------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+-------------------------------------------------------------------------------------------------------------------+
| logical_plan | Projection: emp.id |
| | LeftMark Join: Filter: emp.salary = __correlated_sq_3.salary IS TRUE |
| | Projection: emp.id, emp.salary |
| | LeftMark Join: Filter: emp.salary = __correlated_sq_2.salary IS NULL |
| | Projection: emp.id, emp.salary |
| | Filter: __correlated_sq_1.mark |
| | LeftMark Join: Filter: emp.salary = __correlated_sq_1.salary IS TRUE |
| | TableScan: emp projection=[id, salary] |
| | SubqueryAlias: __correlated_sq_1 |
| | Projection: emp.salary |
| | Filter: emp.id <= Int32(2) |
| | TableScan: emp projection=[id, salary] |
| | SubqueryAlias: __correlated_sq_2 |
| | Projection: emp.salary |
| | Filter: emp.id <= Int32(2) |
| | TableScan: emp projection=[id, salary] |
| | SubqueryAlias: __correlated_sq_3 |
| | Projection: emp.salary |
| | Filter: emp.id <= Int32(2) |
| | TableScan: emp projection=[id, salary] |
| physical_plan | NestedLoopJoinExec: join_type=LeftMark, filter=(salary@0 = salary@1) IS NOT DISTINCT FROM true, projection=[id@0] |
| | CoalescePartitionsExec |
| | NestedLoopJoinExec: join_type=LeftMark, filter=salary@0 = salary@1 IS NULL, projection=[id@0, salary@1] |
| | CoalescePartitionsExec |
| | FilterExec: mark@2, projection=[id@0, salary@1] |
| | RepartitionExec: partitioning=RoundRobinBatch(14), input_partitions=1 |
| | NestedLoopJoinExec: join_type=RightMark, filter=(salary@0 = salary@1) IS NOT DISTINCT FROM true |
| | FilterExec: id@0 <= 2, projection=[salary@1] |
| | DataSourceExec: partitions=1, partition_sizes=[1] |
| | DataSourceExec: partitions=1, partition_sizes=[1] |
| | RepartitionExec: partitioning=RoundRobinBatch(14), input_partitions=1 |
| | FilterExec: id@0 <= 2, projection=[salary@1] |
| | DataSourceExec: partitions=1, partition_sizes=[1] |
| | RepartitionExec: partitioning=RoundRobinBatch(14), input_partitions=1 |
| | FilterExec: id@0 <= 2, projection=[salary@1] |
| | DataSourceExec: partitions=1, partition_sizes=[1] |
| | |
+---------------+-------------------------------------------------------------------------------------------------------------------+
```
But for the equivalent `IN` query we get the much simpler (and similar to DuckDB):
```
+---------------+---------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+---------------------------------------------------------------------------------------------------+
| logical_plan | Projection: emp.id |
| | LeftSemi Join: emp.salary = __correlated_sq_1.salary |
| | TableScan: emp projection=[id, salary] |
| | SubqueryAlias: __correlated_sq_1 |
| | Projection: emp.salary |
| | Filter: emp.id <= Int32(2) |
| | TableScan: emp projection=[id, salary] |
| physical_plan | HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(salary@0, salary@1)], projection=[id@0] |
| | FilterExec: id@0 <= 2, projection=[salary@1] |
| | DataSourceExec: partitions=1, partition_sizes=[1] |
| | DataSourceExec: partitions=1, partition_sizes=[1] |
| | |
+---------------+---------------------------------------------------------------------------------------------------+
```
Contributor guide
Assessment
This issue has not been assessed yet.