Distribute UNION operations

[gabotechs]

This PR builds on top of #259 and uses its tools for distributing UNION operations.

The way unions are distributed is with a new ChildrenIsolatorUnionExec operator, that will selectively execute its children depending on the distributed task index in which its running.

The best way of imagining how this works is by looking at the simplest case, when the union operation runs in the same amount of tasks than children it has. In that case, each tasks executes one child respectively:

┌─────────────────────────────┐┌─────────────────────────────┐┌─────────────────────────────┐
│           Task 1            ││           Task 2            ││           Task 3            │
│┌───────────────────────────┐││┌───────────────────────────┐││┌───────────────────────────┐│
││ ChildrenIsolatorUnionExec ││││ ChildrenIsolatorUnionExec ││││ ChildrenIsolatorUnionExec ││
│└───▲─────────▲─────────▲───┘││└───▲─────────▲─────────▲───┘││└───▲─────────▲─────────▲───┘│
│    │                        ││              │              ││                        │    │
│┌───┴───┐ ┌  ─│ ─   ┌  ─│ ─  ││┌  ─│ ─   ┌───┴───┐ ┌  ─│ ─  ││┌  ─│ ─   ┌  ─│ ─   ┌───┴───┐│
││Child 1│  Child 2│  Child 3│││ Child 1│ │Child 2│  Child 3│││ Child 1│  Child 2│ │Child 3││
│└───────┘ └  ─  ─   └  ─  ─  ││└  ─  ─   └───────┘ └  ─  ─  ││└  ─  ─   └  ─  ─   └───────┘│
└─────────────────────────────┘└─────────────────────────────┘└─────────────────────────────┘

It can get complicated if there are either more tasks than children, ore more children than tasks.

For example if there are more children than tasks, their execution will be spread as evenly as possible across distributed tasks:

┌─────────────────────────────┐┌─────────────────────────────┐
│           Task 1            ││           Task 2            │
│┌───────────────────────────┐││┌───────────────────────────┐│
││ ChildrenIsolatorUnionExec ││││ ChildrenIsolatorUnionExec ││
│└───▲─────────▲─────────▲───┘││└───▲─────────▲─────────▲───┘│
│    │         │              ││                        │    │
│┌───┴───┐ ┌───┴───┐ ┌  ─│ ─  ││┌  ─│ ─   ┌ ─ ┴ ─ ┐ ┌───┴───┐│
││Child 1│ │Child 2│  Child 3│││ Child 1│  Child 2  │Child 3││
│└───────┘ └───────┘ └  ─  ─  ││└  ─  ─   └ ─ ─ ─ ┘ └───────┘│
└─────────────────────────────┘└─────────────────────────────┘

If there are more tasks than children, then some children might not be able to act as if they were running in single-node, and they might need to be executed in a "reduced" distributed context:

┌─────────────────────────────┐┌─────────────────────────────┐┌─────────────────────────────┐┌─────────────────────────────┐
│           Task 1            ││           Task 2            ││           Task 3            ││           Task 4            │
│┌───────────────────────────┐││┌───────────────────────────┐││┌───────────────────────────┐││┌───────────────────────────┐│
││ ChildrenIsolatorUnionExec ││││ ChildrenIsolatorUnionExec ││││ ChildrenIsolatorUnionExec ││││ ChildrenIsolatorUnionExec ││
│└───▲─────────▲─────────▲───┘││└───▲─────────▲─────────▲───┘││└───▲─────────▲─────────▲───┘││└───▲─────────▲─────────▲───┘│
│    │                        ││    │                        ││              │              ││                        │    │
│┌───┴───┐ ┌  ─│ ─   ┌  ─│ ─  ││┌───┴───┐ ┌  ─│ ─   ┌  ─│ ─  ││┌  ─│ ─   ┌───┴───┐ ┌  ─│ ─  ││┌  ─│ ─   ┌  ─│ ─   ┌───┴───┐│
││Child 1│  Child 2│  Child 3││││Child 1│  Child 2│  Child 3│││ Child 1│ │Child 2│  Child 3│││ Child 1│  Child 2│ │Child 3││
││ (1/2) │ └  ─  ─   └  ─  ─  │││ (2/2) │ └  ─  ─   └  ─  ─  ││└  ─  ─   └───────┘ └  ─  ─  ││└  ─  ─   └  ─  ─   └───────┘│
│└───────┘                    ││└───────┘                    ││                             ││                             │
└─────────────────────────────┘└─────────────────────────────┘└─────────────────────────────┘└─────────────────────────────┘

---

Impact of this PR in the TPC-DS benchmarks that use UNION operations (queries 2,5,14,26,36,49,54,71,75,76,77,80):

$ WORKERS=8 ./benchmarks/run.sh --threads 2 --files-per-task 2 --path benchmarks/data/tpcds_sf1 --query 2,5,14,26,36,49,54,71,75,76,77,80
VS
$ WORKERS=8 ./benchmarks/run.sh --threads 2 --files-per-task 2 --path benchmarks/data/tpcds_sf1 --query 2,5,14,26,36,49,54,71,75,76,77,80 --children-isolator-unions

==== Comparison with the previous benchmark from 2025-12-26 12:58:34 UTC ====
os:        macos
arch:      aarch64
cpu cores: 16
threads:   2 -> 2
workers:   8 -> 8
=============================================================================
 Tpcds 2: prev=  82 ms, new=  70 ms, diff=1.17 faster ✔
 Tpcds 5: prev=  83 ms, new=  77 ms, diff=1.08 faster ✔
Tpcds 14: prev= 563 ms, new= 443 ms, diff=1.27 faster ✅
Tpcds 26: prev=  49 ms, new=  50 ms, diff=1.02 slower ✖
Tpcds 36: prev= 169 ms, new=  98 ms, diff=1.72 faster ✅
Tpcds 49: prev= 111 ms, new=  72 ms, diff=1.54 faster ✅
Tpcds 54: prev= 102 ms, new=  62 ms, diff=1.65 faster ✅
Tpcds 71: prev=  50 ms, new=  42 ms, diff=1.19 faster ✔
Tpcds 75: prev= 265 ms, new= 217 ms, diff=1.22 faster ✅
Tpcds 76: prev=  56 ms, new=  48 ms, diff=1.17 faster ✔
Tpcds 77: prev=  68 ms, new=  63 ms, diff=1.08 faster ✔
Tpcds 80: prev= 173 ms, new= 143 ms, diff=1.21 faster ✅

[NGA-TRAN]

Very nice. I just want to know if you have verified whether the a child is splittable before splitting

[NGA-TRAN]

This looks good

[NGA-TRAN]

This looks good

[NGA-TRAN]

This needs more thinking. You can only do this if the child is splittable.

• DataSource and (Sub)Union : are splittable
• Other operators that can handle individual partitions are splittable. Example:
  • Partitioned hash join is splittable. CollectLeft is not
  • Aggregation with preserved partitioning is splittable

Have you considered them? Maybe in the first round, you only split those you think safe to split first and add more as we go

[gabotechs]

This will in fact only be done if "Child 1" is splittable. If the child contains a CollectLeft HashJoin at any point in the subtree, it will not be split.

The doc comment just shows how it works in case it can be split, and in case it makes sense to split it.

[NGA-TRAN]

I think if you add a line or something between the children for us to know they will go to different tasks, it will be easier to read

[NGA-TRAN]

And if you split one of them into many tasks, you may want to add something for us to know that, too

[NGA-TRAN]

Yeah, here we need to verify if the tasks are splittable

[gabotechs]

It's not really this piece of code the one that checks if the child can be split. For deciding if a child can be split or not, there is not really any extra mechanism shipped in this PR, the mechanism is the one from Rework distributed planning logic:

• A CollectLeft HashJoin (or any other non-distributable node) gets a task count annotation of Maximum(1)
• Nodes upstream see that, and they are also marked as Maximum(1)
• The union will not try to scale the CollectLeft HashJoin (or any other non-distributable node) further, respecting its original task count annotation, therefore not distributing it.

[NGA-TRAN]

LGTM. Let’s start using it for the obvious use cases first — I think that will naturally lead us to more improvements and better support for the more complex ones.