Let’s look at the plan of the following -

SELECT sum(cast(json_extract(:column, '$.:num') as int) ) FROM :table 
WHERE cast(json_extract(:column, '$.:num') as int) > 1 AND
WHERE cast(json_extract(:column, '$.:num') as int) < 9

The output is -

[35]

And the plan (base case, no union) -

Output[columnNames = [_col0]]
│   Layout: [sum:bigint]
│   _col0 := sum
└─ Aggregate[type = FINAL]
   │   Layout: [sum:bigint]
   │   sum := sum(sum_1)
   └─ LocalExchange[partitioning = SINGLE]
      │   Layout: [sum_1:bigint]
      └─ RemoteExchange[type = GATHER]
         │   Layout: [sum_1:bigint]
         └─ Aggregate[type = PARTIAL]
            │   Layout: [sum_1:bigint]
            │   sum_1 := sum(projected)
            └─ TableScan[table = splunk:schema=default, table=main, timestampField=timestamp, exprNode=And[children=[Domain[name=name='condition', condition={CommonColumnHandle{name=num, type=integer}=[ SortedRangeSet[type=integer, ranges=1, {(1,<max>)}] ]}], Domain[name=name='condition', condition={CommonColumnHandle{name=num, type=integer}=[ SortedRangeSet[type=integer, ranges=1, {(<min>,9)}] ]}]]], sortOrder=[], queryId=20250420_174939_00000_zm7sd, aggregationColumnHandles=[], extraHandle=null]
                   Layout: [projected:bigint]
                   projected := CommonColumnHandle{name=num, type=bigint}

Exchange - do we care?

We at Vega mainly use Trino as a ‘predicate pushdown machine’, and we don’t give much attention to the exchanges. Do we really need them? Let’s remove the LocalExchange and find out.

Exercise - remove LocalExchange 

diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/AddLocalExchanges.java b/core/trino-main/src/main/jav
a/io/trino/sql/planner/optimizations/AddLocalExchanges.java
index 3f56971182..7f34a9c8d2 100644
--- a/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/AddLocalExchanges.java
+++ b/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/AddLocalExchanges.java
@@ -1008,9 +1008,9 @@ public class AddLocalExchanges
             PlanWithProperties result = node.accept(this, preferredProperties);
 
             // enforce the required properties
-            result = enforce(result, requiredProperties);
-
-            checkState(requiredProperties.isSatisfiedBy(result.getProperties()), "required properties not enforced");
             return result;
         }

The output is -

[35]
[null]
[null]
[null]

With the plan -


Output[columnNames = [_col0]]
│   Layout: [sum:bigint]
│   _col0 := sum
└─ Aggregate[type = FINAL]
   │   Layout: [sum:bigint]
   │   sum := sum(sum_1)
   └─ RemoteExchange[type = GATHER]
      │   Layout: [sum_1:bigint]
      └─ Aggregate[type = PARTIAL]
         │   Layout: [sum_1:bigint]
         │   sum_1 := sum(projected)
         └─ TableScan[table = splunk:schema=default, table=main, timestampField=timestamp, exprNode=And[children=[Domain[name=name='condition', condition={CommonColumnHandle{name=num, type=integer}=[ SortedRangeSet[type=integer, ranges=1, {(1,<max>)}] ]}], Domain[name=name='condition', condition={CommonColumnHandle{name=num, type=integer}=[ SortedRangeSet[type=integer, ranges=1, {(<min>,9)}] ]}]]], sortOrder=[], queryId=20250420_175739_00000_62apy, aggregationColumnHandles=[], extraHandle=null]
                Layout: [projected:bigint]
                projected := CommonColumnHandle{name=num, type=bigint}

Aggregation pipeline

So what is happening?

The plan of the original query separates Aggregation into PARTIAL and FINAL. This allows calculating the SUM in a distributed fashion: each Partial performs sum over part of the table scan, and the Aggregation Final combines their results -

Table Scan -> Aggregation Partial
                                  -> Aggregation Final
Table Scan -> Aggregation Partial

Implemented here -

@AggregationFunction(value = "sum", windowAccumulator = LongSumAggregation.LongSumWindowAccumulator.class)
public final class LongSumAggregation {
    @InputFunction
    public static void sum(@AggregationState NullableLongState state, @SqlType(StandardTypes.BIGINT) long value) {
        state.setNull(false);
        state.setValue(BigintOperators.add(state.getValue(), value));
    }

    @CombineFunction
    public static void combine(@AggregationState NullableLongState state, @AggregationState NullableLongState otherState) {
        if (state.isNull()) {
            state.set(otherState);
            return;
        }

        state.setValue(BigintOperators.add(state.getValue(), otherState.getValue()));
    }

    @OutputFunction(StandardTypes.BIGINT)
    public static void output(@AggregationState NullableLongState state, BlockBuilder out) {
        NullableLongState.write(BIGINT, state, out);
    }

And they all output their result into the next stage in the pipeline. However, when we removed the LocalExchange[SINGLE], the pipeline actually becomes something like this -

Table Scan -> Aggregation Partial -> Aggregation Final
Table Scan -> Aggregation Partial -> Aggregation Final

We have a small test table, and for some reason (?) also 4 SplitRunners working. So this is the plan, each row representing a SplitRunner thread -

Table Scan -> Aggregation Partial -> Aggregation Final
 all rows       sum(), output()=[35]      output()=[null]

Table Scan -> Aggregation Partial -> Aggregation Final
   /             output()=[null]         output()=[null]

Table Scan -> Aggregation Partial -> Aggregation Final
   /             output()=[null]         output()=[null]

Table Scan -> Aggregation Partial -> Aggregation Final
   /             output()=[null]         output()=[null]

Usually, when stages in the plan (e.g. Project) have no input - they produce no output. However, aggregations of the form COUNT(*) and SUM(*) are special, and that’s because by SQL standard, they should produce a single row even when there’s no data - [0] and [null], respectively. This is handled by NullableLongState.write(BIGINT, state, out); which writes data regardless of whether the state is null.

How the exchange solves the [null]s?

By having LocalExchange[partitioning = SINGLE] below the Aggregation[FINAL], it says that the aggregation should be performed on a single thread. Meaning, there’s only a single Aggregation[FINAL]. By default, trino wants to go distributed, but this exchange prevents it. This is a unique case where we can actually see the exchanges in action impact the result.

Table Scan -> Aggregation Partial -> 
 all rows       sum(), output()=[35] 

Table Scan -> Aggregation Partial -> 
   /             output()=[null] 

                                       -> Aggregation Final

Table Scan -> Aggregation Partial -> 
   /             output()=[null]         

Table Scan -> Aggregation Partial -> 
   /             output()=[null]

Furhter reading

Go to the file ValidateAggregationsWithDefaultValues, having some nice documentation -

/**
 * When an aggregation has an empty grouping set then a default value needs to be returned in the output (e.g: 0 for COUNT(*)).
 * In case if the aggregation is split into FINAL and PARTIAL, then default values are produced by PARTIAL
 * aggregations. In order for the default values not to be duplicated, FINAL aggregation needs to be
 * separated from PARTIAL aggregation by a remote repartition exchange or the FINAL aggregation needs to be executed
 * on a single node. In case both FINAL and PARTIAL aggregations are executed on a single node, then those need to separated
 * by a local repartition exchange or the FINAL aggregation needs to be executed in a single thread.
 */