feat: add optimizer config param to avoid grouping partitions `prefer_existing_union` #10259

NGA-TRAN · 2024-04-26T21:47:45Z

Which issue does this PR close?

Closes ##10257

Rationale for this change

What changes are included in this PR?

Add an option to avoid converting Union to Interleave

Are these changes tested?

Yes

Are there any user-facing changes?

No, because by the fault, nothing is changed

NGA-TRAN · 2024-04-26T21:49:24Z

datafusion/core/src/physical_optimizer/enforce_distribution.rs

+            plan,
+            !first_enforce_distribution,
+            prefer_existing_sort,
+            prefer_existing_union


Observing the test union_to_interleave above it. If it covers all needed cases, these 2 tests will cover all needed cases for the not-convert, too. Let me know if it is not the case

NGA-TRAN · 2024-04-26T21:50:13Z

@alamb Is this PR is good enough? We do not need to set prefer_existing_union in DF, right? It will be set in IOx?

datafusion/core/src/physical_optimizer/enforce_distribution.rs

phillipleblanc · 2024-04-27T06:54:53Z

datafusion/core/src/physical_optimizer/enforce_distribution.rs

-    plan = if plan.as_any().is::<UnionExec>() && can_interleave(children_plans.iter()) {
+
+    plan = if plan.as_any().is::<UnionExec>()
+        && !config.optimizer.prefer_existing_union


Given that one of the motivations (influxdata#4) for this new flag is to preserve the sorting of the Union - would re-using the existing flag prefer_existing_sort make sense here?

One argument against that would be if you wanted prefer_existing_sort: false and prefer_existing_union: true.

Given that one of the motivations (influxdata#4) for this new flag is to preserve the sorting of the Union - would re-using the existing flag prefer_existing_sort make sense here?

One argument against that would be if you wanted prefer_existing_sort: false and prefer_existing_union: true.

This is an excellent point @phillipleblanc . I just double checked and we actually have prefer_existing_sort set to true in IOx already (code ref, not public)

What do you think about using the existing flag @NGA-TRAN ?

@phillipleblanc and @alamb
@mustafasrepo 's comment #10259 (comment) suggest to use the current approach. Do I need to do anything more here?

Nope, I think this approach looks good! Thanks!

alamb

Thanks @NGA-TRAN -- this is looking like it is pretty close in my opinion

I think @phillipleblanc 's suggestions are quite good and we should consider them carefully

alamb · 2024-04-27T12:06:51Z

datafusion/core/src/physical_optimizer/enforce_distribution.rs

-    plan = if plan.as_any().is::<UnionExec>() && can_interleave(children_plans.iter()) {
+
+    plan = if plan.as_any().is::<UnionExec>()
+        && !config.optimizer.prefer_existing_union


Given that one of the motivations (influxdata#4) for this new flag is to preserve the sorting of the Union - would re-using the existing flag prefer_existing_sort make sense here?

One argument against that would be if you wanted prefer_existing_sort: false and prefer_existing_union: true.

This is an excellent point @phillipleblanc . I just double checked and we actually have prefer_existing_sort set to true in IOx already (code ref, not public)

What do you think about using the existing flag @NGA-TRAN ?

alamb · 2024-04-27T12:08:04Z

datafusion/core/src/physical_optimizer/enforce_distribution.rs

        };

        ($EXPECTED_LINES: expr, $PLAN: expr, $FIRST_ENFORCE_DIST: expr, $PREFER_EXISTING_SORT: expr) => {
-            assert_optimized!($EXPECTED_LINES, $PLAN, $FIRST_ENFORCE_DIST, $PREFER_EXISTING_SORT, 10, false, 1024);
+            assert_optimized!($EXPECTED_LINES, $PLAN, $FIRST_ENFORCE_DIST, $PREFER_EXISTING_SORT, 10, false, 1024, false);


these macros are getting a bit hairy -- maybe we can clean them up (convert to using functions rather than macros) in a subsequent PR

alamb · 2024-04-27T12:08:47Z

cc @mustafasrepo and @ozankabak for your thoughts

ozankabak · 2024-04-27T13:10:21Z

Thanks @NGA-TRAN and @alamb, we will take a look and circle back on Monday

mustafasrepo · 2024-04-29T09:56:28Z

I think having dedicated config setting is more verbose and clear (as in prefer_existing_union). If we were to use prefer_existing_sort that might also work. However, if the condition to replace UnionExec to InterleaveExec is changed to

plan.as_any().is::<UnionExec>() 
&& !config.optimizer.prefer_existing_sort 
&& can_interleave(children_plans.iter())

this will prefer UnionExec instead of InterleaveExec even if inputs of the UnionExec is unordered when the config.optimizer.prefer_existing_sort flag is true. Which might be counter intuitive given there is no ordering to preserve. However, config.optimizer.prefer_existing_union does exactly what it says. Hence, it is a bit clearer to me. Hence, I think it is better to proceed with current approach in this PR.

In the future, if we add support for OrderPreservingInterleaveExec (this might be accomplished by replacing CombinedRecordBatchStream with streaming_merge in the fn execute method of the InterleaveExec.)
using the flag config.optimizer.prefer_existing_sort to decide between InterleaveExec and OrderPreservingInterleaveExec might solve the issue. This approach may invalidate the requirement for prefer_existing_union setting. However, until we have this support current approach is much more clear.

alamb · 2024-04-29T15:46:08Z

Hence, I think it is better to proceed with current approach in this PR. (as in prefer_existing_union)

Sounds good to me.

In the future, if we add support for OrderPreservingInterleaveExec (this might be accomplished by replacing CombinedRecordBatchStream with streaming_merge in the fn execute method of the InterleaveExec.)

This is an excellent idea, I will file a ticket to track the idea

@NGA-TRAN can you get the tests passing on this PR and we'll give it another review?

NGA-TRAN · 2024-04-29T18:35:28Z

@alamb @phillipleblanc and @mustafasrepo
all the tests have passed

alamb

This PR looks good to me.

Thank you for the reviews @phillipleblanc @mustafasrepo and @ozankabak

I plan to file the follow on tickets (e.g. about sort preserving interleave) tomorrow

mustafasrepo

This PR is LGTM! Thanks @NGA-TRAN

phillipleblanc

LGTM!

phillipleblanc · 2024-04-30T12:31:12Z

datafusion/core/src/physical_optimizer/enforce_distribution.rs

-    plan = if plan.as_any().is::<UnionExec>() && can_interleave(children_plans.iter()) {
+
+    plan = if plan.as_any().is::<UnionExec>()
+        && !config.optimizer.prefer_existing_union


Nope, I think this approach looks good! Thanks!

alamb · 2024-04-30T15:45:32Z

Since this PR is good to go, merging it in. I am also in the process of filing the follow on work

…_existing_union` (apache#10259) * feat: add a config param to avoid converting union to interleave * chore: update config for the tests * chore: update configs.md

alamb · 2024-04-30T16:40:15Z

Filed #10314 to track order preserving UnionExec

alamb · 2024-04-30T16:40:52Z

I also started collecting sorted related optimizations in #10313 as well

…_existing_union` (apache#10259) * feat: add a config param to avoid converting union to interleave * chore: update config for the tests * chore: update configs.md

feat: add a config param to avoid converting union to interleave

bab026e

github-actions bot added the core Core DataFusion crate label Apr 26, 2024

NGA-TRAN commented Apr 26, 2024

View reviewed changes

phillipleblanc reviewed Apr 27, 2024

View reviewed changes

alamb reviewed Apr 27, 2024

View reviewed changes

chore: update config for the tests

e7e48ff

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

chore: update configs.md

35560ac

NGA-TRAN mentioned this pull request Apr 29, 2024

feat: avoid converting union into interleave influxdata/arrow-datafusion#4

Closed

alamb changed the title ~~feat: add a config param to avoid grouping partitions~~ feat: add optimizer config param to avoid grouping partitions prefer_existing_union Apr 29, 2024

alamb approved these changes Apr 29, 2024

View reviewed changes

mustafasrepo approved these changes Apr 30, 2024

View reviewed changes

phillipleblanc approved these changes Apr 30, 2024

View reviewed changes

alamb merged commit 2231183 into apache:main Apr 30, 2024
25 checks passed

appletreeisyellow mentioned this pull request Apr 30, 2024

WIP: apply upstream commit for "optimizer config param" influxdata/arrow-datafusion#16

Closed

This was referenced Apr 30, 2024

Add an option to avoid grouping hash-partitioning #10257

Closed

Implement a way to preserve partitioning through UnionExec without losing ordering #10314

Open

appletreeisyellow mentioned this pull request Apr 30, 2024

WIP(iox-10578): patched df upgrade 202-04-27 influxdata/arrow-datafusion#17

Closed

NGA-TRAN mentioned this pull request Apr 30, 2024

chore: add function to set prefer_existing_union #10322

Merged

appletreeisyellow mentioned this pull request May 1, 2024

WIP(iox-10578): patched df upgrade 202-04-30 influxdata/arrow-datafusion#18

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add optimizer config param to avoid grouping partitions `prefer_existing_union` #10259

feat: add optimizer config param to avoid grouping partitions `prefer_existing_union` #10259

NGA-TRAN commented Apr 26, 2024

NGA-TRAN Apr 26, 2024

NGA-TRAN commented Apr 26, 2024

phillipleblanc Apr 27, 2024

alamb Apr 27, 2024

NGA-TRAN Apr 29, 2024

phillipleblanc Apr 30, 2024 •

edited

Loading

alamb left a comment

alamb Apr 27, 2024

alamb Apr 27, 2024

alamb commented Apr 27, 2024

ozankabak commented Apr 27, 2024

mustafasrepo commented Apr 29, 2024

alamb commented Apr 29, 2024 •

edited

Loading

NGA-TRAN commented Apr 29, 2024

alamb left a comment

mustafasrepo left a comment

phillipleblanc left a comment

phillipleblanc Apr 30, 2024 •

edited

Loading

alamb commented Apr 30, 2024

alamb commented Apr 30, 2024

alamb commented Apr 30, 2024

feat: add optimizer config param to avoid grouping partitions prefer_existing_union #10259

feat: add optimizer config param to avoid grouping partitions prefer_existing_union #10259

Conversation

NGA-TRAN commented Apr 26, 2024

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?

NGA-TRAN Apr 26, 2024

Choose a reason for hiding this comment

NGA-TRAN commented Apr 26, 2024

phillipleblanc Apr 27, 2024

Choose a reason for hiding this comment

alamb Apr 27, 2024

Choose a reason for hiding this comment

NGA-TRAN Apr 29, 2024

Choose a reason for hiding this comment

phillipleblanc Apr 30, 2024 • edited Loading

Choose a reason for hiding this comment

alamb left a comment

Choose a reason for hiding this comment

alamb Apr 27, 2024

Choose a reason for hiding this comment

alamb Apr 27, 2024

Choose a reason for hiding this comment

alamb commented Apr 27, 2024

ozankabak commented Apr 27, 2024

mustafasrepo commented Apr 29, 2024

alamb commented Apr 29, 2024 • edited Loading

NGA-TRAN commented Apr 29, 2024

alamb left a comment

Choose a reason for hiding this comment

mustafasrepo left a comment

Choose a reason for hiding this comment

phillipleblanc left a comment

Choose a reason for hiding this comment

phillipleblanc Apr 30, 2024 • edited Loading

Choose a reason for hiding this comment

alamb commented Apr 30, 2024

alamb commented Apr 30, 2024

alamb commented Apr 30, 2024

feat: add optimizer config param to avoid grouping partitions `prefer_existing_union` #10259

feat: add optimizer config param to avoid grouping partitions `prefer_existing_union` #10259

phillipleblanc Apr 30, 2024 •

edited

Loading

alamb commented Apr 29, 2024 •

edited

Loading

phillipleblanc Apr 30, 2024 •

edited

Loading