Cleanup the usage of round-robin repartitioning #8794

viirya · 2024-01-08T23:26:47Z

Which issue does this PR close?

Closes #.

Rationale for this change

Currently the logic of adding roundrobin partitioning is somehow confusing. This patch tries to clean it up.

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

viirya · 2024-01-08T23:28:13Z

datafusion/core/src/physical_optimizer/enforce_distribution.rs

-        if repartition_beneficial_stats {
-            // Since hashing benefits from partitioning, add a round-robin repartition
-            // before it:
-            input = add_roundrobin_on_top(input, n_target)?;
-        }
-


This add_roundrobin_on_top currently is not controlled by enable_round_robin.

viirya · 2024-01-08T23:29:09Z

datafusion/sqllogictest/test_files/repartition.slt

+----RepartitionExec: partitioning=Hash([column1@0], 4), input_partitions=1
+------AggregateExec: mode=Partial, gby=[column1@0 as column1], aggr=[SUM(parquet_table.column2)]


Should not see RoundRobinBatch partitioning as enable_round_robin_repartition is disabled.

Currently roundrobin partitioning is still added even it is disabled. We should make the config behavior consistent.

viirya · 2024-01-08T23:33:45Z

datafusion/core/src/physical_optimizer/enforce_distribution.rs

-                // count.
-                child = add_roundrobin_on_top(child, target_partitions)?;
-            }
-
            // Satisfy the distribution requirement if it is unmet.
            match requirement {
                Distribution::SinglePartition => {
                    child = add_spm_on_top(child);


Although due to would_benefit, single partition won't satisfy the condition to add roundrobin partitioning. This change makes it explicitly clear that single partition cases don't add roundrobin.

viirya · 2024-01-09T06:26:09Z

datafusion/physical-plan/src/joins/symmetric_hash_join.rs

-    fn benefits_from_input_partitioning(&self) -> Vec<bool> {
-        vec![false, false]
-    }


I don't see why this sets all false for SymmetricHashJoinExec. When the mode is StreamJoinPartitionMode::Partitioned, it requires input distributions to be HashPartitioneds.

I think that we don't need to override default function.

Current test case also proves that.

Roundbound and hashpartition together remove the input order of the SHJ. This is why we added this guard, but it was long ago. I will send a fixing PR for this purpose. If these changes do not affect your tests, can you please keep it as it is and let me delete it?

If I don't have this change, current test will fail because roundrobin will not be added.

Roundbound and hashpartition together remove the input order of the SHJ.

I don't see it requires input ordering (i.e., required_input_ordering).

It can run without order but may run out of memory. I am planning to add an optional input requirement for it, thus we can keep the order if it is present in the plan before optimization. I think you can delete it, then I'll prepare the PR after this merged.

Okay. Sounds good.

viirya · 2024-01-09T06:26:52Z

datafusion/core/src/physical_optimizer/enforce_distribution.rs

@@ -1908,7 +1900,7 @@ pub(crate) mod tests {
        let distribution_context = DistributionContext::new(plan);
        let mut config = ConfigOptions::new();
        config.execution.target_partitions = target_partitions;
-        config.optimizer.enable_round_robin_repartition = false;
+        config.optimizer.enable_round_robin_repartition = true;


These tests expect roundrobin partitioning is added, so this config should be enabled.

alamb · 2024-01-09T20:32:27Z

datafusion/sqllogictest/test_files/repartition_scan.slt

@@ -61,7 +61,7 @@ Filter: parquet_table.column1 != Int32(42)
 physical_plan
 CoalesceBatchesExec: target_batch_size=8192
 --FilterExec: column1@0 != 42
----ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..101], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:101..202], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:202..303], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:303..403]]}, projection=[column1], predicate=column1@0 != 42, pruning_predicate=column1_min@0 != 42 OR 42 != column1_max@1
+----ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..180], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:180..360], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:360..540], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:540..717]]}, projection=[column1], predicate=column1@0 != 42, pruning_predicate=column1_min@0 != 42 OR 42 != column1_max@1


I think the reason this test changed is that the repartition.slt test is writing files into the wrong temp space (see note above)

Ah, good catch, I didn't notice that.

datafusion/sqllogictest/test_files/repartition.slt

Co-authored-by: Andrew Lamb <[email protected]>

datafusion/core/src/physical_optimizer/enforce_distribution.rs

alamb · 2024-01-10T19:47:19Z

datafusion/sqllogictest/test_files/repartition.slt

@@ -0,0 +1,76 @@
+# Licensed to the Apache Software Foundation (ASF) under one


I ran this test locally without the code changes in this PR and this test still passes

Hm, I ran the test locally before and it failed. Let me re-verify it.

Fixed it. I added a few lines before which broken it.

alamb

This PR makes sense to me and all tests pass (and there is even added coverage). Thus I think it is a nice improvement.

Thank you @viirya @Dandandan @metesynnada

Co-authored-by: Andrew Lamb <[email protected]>

viirya · 2024-01-11T17:34:28Z

Thanks @alamb @Dandandan @metesynnada

viirya added 2 commits January 4, 2024 12:03

Clean up

d6aad99

Merge remote-tracking branch 'upstream/main' into cleanup_roundrobin

04215c5

github-actions bot added the core Core DataFusion crate label Jan 8, 2024

Add sqllogictest

83c326f

github-actions bot added the sqllogictest SQL Logic Tests (.slt) label Jan 8, 2024

viirya commented Jan 8, 2024

View reviewed changes

viirya added 2 commits January 8, 2024 15:30

fix

97af772

Fix

5f5d639

viirya commented Jan 8, 2024

View reviewed changes

viirya added 2 commits January 8, 2024 15:44

Enable enable_round_robin_repartition for enforce_distribution tests

bc72f91

Fix

acbb296

viirya commented Jan 9, 2024

View reviewed changes

metesynnada mentioned this pull request Jan 9, 2024

Make benefits_from_input_partitioning Default in SHJ #8801

Merged

Try to stable test

5062eb7

alamb mentioned this pull request Jan 9, 2024

DataFusion weekly project plan (Andrew Lamb) - Jan 8, 2024 #8786

Closed

7 tasks

alamb reviewed Jan 9, 2024

View reviewed changes

viirya and others added 3 commits January 9, 2024 12:35

Update datafusion/sqllogictest/test_files/repartition.slt

fb40db8

Co-authored-by: Andrew Lamb <[email protected]>

Update datafusion/sqllogictest/test_files/repartition.slt

8fdb9f1

Co-authored-by: Andrew Lamb <[email protected]>

Fix scratch space

f33bef9

alamb reviewed Jan 10, 2024

View reviewed changes

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

alamb reviewed Jan 10, 2024

View reviewed changes

alamb approved these changes Jan 10, 2024

View reviewed changes

viirya and others added 2 commits January 10, 2024 12:27

Fix test

3d7d4ab

Update datafusion/core/src/physical_optimizer/enforce_distribution.rs

e303a85

Co-authored-by: Andrew Lamb <[email protected]>

viirya merged commit 1f13607 into apache:main Jan 11, 2024
22 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Cleanup the usage of round-robin repartitioning #8794

Cleanup the usage of round-robin repartitioning #8794

viirya commented Jan 8, 2024 •

edited

Loading

viirya Jan 8, 2024

viirya Jan 8, 2024

viirya Jan 8, 2024

viirya Jan 8, 2024

viirya Jan 9, 2024 •

edited

Loading

metesynnada Jan 9, 2024

viirya Jan 9, 2024

viirya Jan 9, 2024

metesynnada Jan 9, 2024

viirya Jan 9, 2024

viirya Jan 9, 2024

alamb Jan 9, 2024

viirya Jan 9, 2024

alamb Jan 10, 2024

viirya Jan 10, 2024

viirya Jan 10, 2024

alamb left a comment

viirya commented Jan 11, 2024

		----RepartitionExec: partitioning=Hash([column1@0], 4), input_partitions=1
		------AggregateExec: mode=Partial, gby=[column1@0 as column1], aggr=[SUM(parquet_table.column2)]

		@@ -0,0 +1,76 @@
		# Licensed to the Apache Software Foundation (ASF) under one

Cleanup the usage of round-robin repartitioning #8794

Cleanup the usage of round-robin repartitioning #8794

Conversation

viirya commented Jan 8, 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?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya Jan 9, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb left a comment

Choose a reason for hiding this comment

viirya commented Jan 11, 2024

viirya commented Jan 8, 2024 •

edited

Loading

viirya Jan 9, 2024 •

edited

Loading