Enforce sorting handle fetchable operators, add option to repartition based on row count estimates #11875

mustafasrepo · 2024-08-07T14:51:03Z

Which issue does this PR close?

Closes #.

Rationale for this change

What changes are included in this PR?

Recently @alihandroid added limit pushdown support for physical plan in the PR. After this PR, I recognized that EnforceSorting rule has some problems in handling operators with fetch. It sometimes loses fetch number during sort pushdown (Since LimitPushdown rule is after EnforceSorting we do not hit erroneous cases currently.). Hence, I used unit tests to trigger erroneous handlings.

Are these changes tested?

Yes, unit tests are added.

Are there any user-facing changes?

mustafasrepo · 2024-08-07T15:12:26Z

datafusion/core/src/dataframe/mod.rs

@@ -3019,11 +3019,11 @@ mod tests {

        assert_batches_sorted_eq!(
            [
-                "+-----+-----+----+-------+",


Result of this test changes with this PR. I have analyzed the change, previously this tes was generating the following plan:

"ProjectionExec: expr=[c1@0 as c1, c2@1 as c2, c3@2 as c3, CAST(c2@1 AS Int8) + c3@2 as sum]", " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", " SortExec: expr=[c1@0 ASC,c2@1 ASC,c3@2 ASC], preserve_partitioning=[false]", " GlobalLimitExec: skip=0, fetch=1", " CoalescePartitionsExec", " CoalesceBatchesExec: target_batch_size=8192, fetch=1", " FilterExec: c2@1 = 3 AND c1@0 = a", " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", " CsvExec: file_groups={1 group: [[<PATH>]]}, projection=[c1, c2, c3], has_header=true",

After the changes in this PR, following plan is generated

"ProjectionExec: expr=[c1@0 as one, c2@1 as two, c3@2 as c3, CAST(c2@1 AS Int8) + c3@2 as total]", " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", " SortExec: TopK(fetch=1), expr=[c3@2 ASC], preserve_partitioning=[false]", " CoalescePartitionsExec", " CoalesceBatchesExec: target_batch_size=8192", " FilterExec: c2@1 = 3 AND c1@0 = a", " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", " CsvExec: file_groups={1 group: [[<PATH>]]}, projection=[c1, c2, c3], has_header=true",

I think the second plan generates a deterministic result. However, the query (dataframe query) is not deterministic as is.
With this observation, I have updated the place of the limit to make sure the query is deterministic after execution. With the change of the place of the limit, the following plan is generated:

"ProjectionExec: expr=[c1@0 as one, c2@1 as two, c3@2 as c3, CAST(c2@1 AS Int8) + c3@2 as total]", " GlobalLimitExec: skip=0, fetch=1", " SortPreservingMergeExec: [c1@0 ASC,c2@1 ASC,c3@2 ASC], fetch=1", " SortExec: TopK(fetch=1), expr=[c3@2 ASC], preserve_partitioning=[true]", " CoalesceBatchesExec: target_batch_size=8192", " FilterExec: c2@1 = 3 AND c1@0 = a", " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", " CsvExec: file_groups={1 group: [[<PATH>]]}, projection=[c1, c2, c3], has_header=true",

I agree it also makes sense that the previous test did a sort right after a select + filter which will not produce a deterministic result. Doing the limit after the sort makes sense

ozankabak

LGTM

ozankabak · 2024-08-07T15:19:13Z

datafusion/sqllogictest/test_files/window.slt

+05)--------ProjectionExec: expr=[]
+06)----------AggregateExec: mode=FinalPartitioned, gby=[c1@0 as c1], aggr=[]
+07)------------CoalesceBatchesExec: target_batch_size=4096
+08)--------------RepartitionExec: partitioning=Hash([c1@0], 2), input_partitions=2
+09)----------------AggregateExec: mode=Partial, gby=[c1@0 as c1], aggr=[]
+10)------------------ProjectionExec: expr=[c1@0 as c1]
+11)--------------------CoalesceBatchesExec: target_batch_size=4096
+12)----------------------FilterExec: c13@1 != C2GT5KVyOPZpgKVl110TyZO0NcJ434
+13)------------------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+14)--------------------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1, c13], has_header=true
+


A better plan 🚀

datafusion/core/src/physical_optimizer/enforce_sorting.rs

datafusion/core/src/physical_optimizer/sort_pushdown.rs

ozankabak

This is looking very good!

For other reviewers: The optimizer removes the RR in some SLT tests because we estimate to have a single batch (the RR would be pointless). We are getting very smart 🚀

@alamb it would be great if you could take a look

alamb · 2024-08-08T12:46:59Z

@alamb it would be great if you could take a look

Will put it on my list for today

alamb

Thanks @mustafasrepo and @ozankabak -- I went through this PR carefully and I think it looks good to me.

I had some improvement suggestions but I don't think any are ncessary prior to merge

datafusion/common/src/config.rs

alamb · 2024-08-09T13:46:20Z

datafusion/core/src/dataframe/mod.rs

@@ -3019,11 +3019,11 @@ mod tests {

        assert_batches_sorted_eq!(
            [
-                "+-----+-----+----+-------+",


I agree it also makes sense that the previous test did a sort right after a select + filter which will not produce a deterministic result. Doing the limit after the sort makes sense

datafusion/core/src/physical_optimizer/enforce_distribution.rs

datafusion/physical-plan/src/limit.rs

datafusion/physical-plan/src/sorts/sort.rs

datafusion/sqllogictest/test_files/count_star_rule.slt

datafusion/sqllogictest/test_files/order.slt

ozankabak · 2024-08-09T17:56:33Z

Thanks for the review @alamb -- I will send one more commit and then this will be good to go.

ozankabak

After a careful study of the code, I have one issue in mind (for which I left an inline comment). We can merge the code after we make sure the plan change in question is not due to a regression.

ozankabak · 2024-08-09T20:38:57Z

datafusion/sqllogictest/test_files/limit.slt

-05)--------MemoryExec: partitions=4
+04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+05)--------AggregateExec: mode=Partial, gby=[i@0 as i], aggr=[]
+06)----------MemoryExec: partitions=1


This is the only thing I don't understand here. I studied the rule logic but it is not clear to me why we don't use source output multi-partitioning but a RR later on.

Once we are sure this is not due to some regression, we can merge this PR

OK, I figured out what is going on here. With the optimizations we now do, the CREATE TABLE ... SELECT ... query doesn't create a multi-partition table (because it is not helpful). Therefore we see the RR in the downstream test. Reducing the batch size just before the test gives us the old plan. I updated the comment above accordingly.

So all is good, ready to go.

…artition based on row count estimates (apache#11875)" This reverts commit 79fa6f9.

mustafasrepo added 9 commits August 7, 2024 09:17

Tmp

510b16c

Minor changes

6ef4369

Minor changes

c3efafc

Minor changes

2bf220d

Implement top down recursion with delete check

eb83917

Minor changes

0b66b15

Minor changes

c769f9f

Address reviews

0ad7063

Update comments

3661f06

github-actions bot added core Core DataFusion crate sqllogictest SQL Logic Tests (.slt) labels Aug 7, 2024

Minor changes

60967c1

mustafasrepo commented Aug 7, 2024

View reviewed changes

Make test deterministic

6b87c4c

ozankabak approved these changes Aug 7, 2024

View reviewed changes

Add fetch info to the statistics

8dd7e0a

ozankabak reviewed Aug 8, 2024

View reviewed changes

datafusion/core/src/physical_optimizer/enforce_sorting.rs Outdated Show resolved Hide resolved

ozankabak reviewed Aug 8, 2024

View reviewed changes

datafusion/core/src/physical_optimizer/sort_pushdown.rs Outdated Show resolved Hide resolved

ozankabak reviewed Aug 8, 2024

View reviewed changes

datafusion/core/src/physical_optimizer/sort_pushdown.rs Outdated Show resolved Hide resolved

mustafasrepo added 4 commits August 8, 2024 14:37

Enforce distribution use inexact count estimate also.

15423ae

Minor changes

94fb83d

Minor changes

9053b9f

Minor changes

1171584

ozankabak approved these changes Aug 8, 2024

View reviewed changes

mustafasrepo added 4 commits August 9, 2024 15:37

Do not add unnecessary hash partitioning

711038d

Minor changes

7e598e5

Add config option to use inexact row number estimates during planning

12ad2c2

Update config

2e3cc5d

github-actions bot added the documentation Improvements or additions to documentation label Aug 9, 2024

alamb changed the title ~~Enforce sorting handle fetchable operators.~~ Enforce sorting handle fetchable operators, add option to repartition based on row count estimates Aug 9, 2024

mustafasrepo added 2 commits August 9, 2024 17:19

Minor changes

34af8ba

Minor changes

98760bc

alamb approved these changes Aug 9, 2024

View reviewed changes

ozankabak and others added 6 commits August 9, 2024 17:27

Final review

1e4dada

Address reviews

9fc4f3d

Add handling for sort removal with fetch

1116058

Fix linter errors

44dc292

Minor changes

c6d2de6

Update config

c7c85f4

Cleanup stats under fetch

7c8967d

ozankabak reviewed Aug 9, 2024

View reviewed changes

Update SLT comment

ed35660

ozankabak merged commit 79fa6f9 into apache:main Aug 10, 2024
25 checks passed

wiedld added a commit to influxdata/arrow-datafusion that referenced this pull request Aug 15, 2024

Revert "Enforce sorting handle fetchable operators, add option to rep…

fa9b680

…artition based on row count estimates (apache#11875)" This reverts commit 79fa6f9.

This pull request was closed.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Enforce sorting handle fetchable operators, add option to repartition based on row count estimates #11875

Enforce sorting handle fetchable operators, add option to repartition based on row count estimates #11875

mustafasrepo commented Aug 7, 2024 •

edited

Loading

mustafasrepo Aug 7, 2024 •

edited

Loading

alamb Aug 9, 2024

ozankabak left a comment

ozankabak Aug 7, 2024

ozankabak left a comment

alamb commented Aug 8, 2024

alamb left a comment

alamb Aug 9, 2024

ozankabak commented Aug 9, 2024

ozankabak left a comment

ozankabak Aug 9, 2024

ozankabak Aug 10, 2024

Enforce sorting handle fetchable operators, add option to repartition based on row count estimates #11875

Enforce sorting handle fetchable operators, add option to repartition based on row count estimates #11875

Conversation

mustafasrepo commented Aug 7, 2024 • edited Loading

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

mustafasrepo Aug 7, 2024 • edited Loading

Choose a reason for hiding this comment

alamb Aug 9, 2024

Choose a reason for hiding this comment

ozankabak left a comment

Choose a reason for hiding this comment

ozankabak Aug 7, 2024

Choose a reason for hiding this comment

ozankabak left a comment

Choose a reason for hiding this comment

alamb commented Aug 8, 2024

alamb left a comment

Choose a reason for hiding this comment

alamb Aug 9, 2024

Choose a reason for hiding this comment

ozankabak commented Aug 9, 2024

ozankabak left a comment

Choose a reason for hiding this comment

ozankabak Aug 9, 2024

Choose a reason for hiding this comment

ozankabak Aug 10, 2024

Choose a reason for hiding this comment

mustafasrepo commented Aug 7, 2024 •

edited

Loading

mustafasrepo Aug 7, 2024 •

edited

Loading