Trino Aggregation System Overview

This document explains the core components of Trino’s aggregation system, including the Aggregation object, AggregationNode, and the different execution steps (PARTIAL, FINAL, SINGLE).

Aggregation Object

The Aggregation class represents a single aggregation operation within a query. Each aggregation function call gets its own Aggregation object.

Key Components 

class Aggregation {
    ResolvedFunction resolvedFunction;  // Function metadata (COUNT, SUM, etc.)
    List<Expression> arguments;         // Input expressions 
    boolean distinct;                   // Whether it's DISTINCT aggregation
    Optional<Symbol> filter;            // FILTER clause symbol
    Optional<OrderingScheme> orderingScheme; // ORDER BY for aggregations
    Optional<Symbol> mask;              // Row masking symbol
}

Purpose

  • Encapsulates aggregation metadata: Each Aggregation object contains everything needed to execute one aggregation function
  • Supports complex aggregations: Handles DISTINCT, filtered aggregations, and ordered aggregations
  • Enables optimization: Provides decomposability information for distributed execution

Example

For the SQL query:

SELECT COUNT(*), SUM(DISTINCT price) FILTER (WHERE region = 'US')
FROM orders 
GROUP BY category

This creates two Aggregation objects:

  1. Aggregation(resolvedFunction=COUNT, arguments=[], distinct=false)
  2. Aggregation(resolvedFunction=SUM, arguments=[price], distinct=true, filter=region_filter_symbol)

AggregationNode

AggregationNode is a plan node that represents aggregation operations in Trino’s query execution tree.

Key Components 

class AggregationNode extends PlanNode {
    PlanNode source;                           // Input data source
    Map<Symbol, Aggregation> aggregations;     // Output symbols → Aggregation objects
    GroupingSetDescriptor groupingSets;        // GROUP BY structure
    List<Symbol> preGroupedSymbols;            // Already-grouped input symbols
    Step step;                                 // Execution step (PARTIAL/FINAL/SINGLE)
    Optional<Symbol> hashSymbol;               // Hash partitioning symbol
    Optional<Symbol> groupIdSymbol;            // Grouping set identifier
}

Key Methods

  • isDecomposable(): Determines if aggregations can be split into partial/final phases
  • hasDistinct(): Checks if any aggregations use DISTINCT
  • hasSingleNodeExecutionPreference(): Indicates preference for single-node execution

Plan Tree Structure 

AggregationNode(step=FINAL, groupBy=[category])
├── aggregations: {count_symbol → COUNT(*), sum_symbol → SUM(price)}
└── Exchange(REPARTITION on [category])
    └── AggregationNode(step=PARTIAL, groupBy=[category])
        ├── aggregations: {partial_count → PARTIAL_COUNT(*), partial_sum → PARTIAL_SUM(price)}
        └── TableScan(orders)

Aggregation Step Types

The Step enum defines different phases of aggregation execution for distributed processing.

SINGLE Step

  • Input: Raw data from tables/sources
  • Output: Final aggregation results
  • Purpose: Complete aggregation in one operation
  • Usage: Small datasets or single-node queries
TableScan(orders: id, price, category)
└── AggregationNode(step=SINGLE, groupBy=[category])
    └── Output: category, COUNT(*), SUM(price)

PARTIAL Step

  • Input: Raw data from tables/sources
  • Output: Intermediate aggregation state
  • Purpose: First phase of distributed aggregation
  • Usage: Runs on workers close to the data
TableScan(orders: id, price, category)
└── AggregationNode(step=PARTIAL, groupBy=[category])
    └── Output: category, partial_count_state, partial_sum_state

FINAL Step

  • Input: Intermediate states from PARTIAL/INTERMEDIATE steps
  • Output: Final aggregation results
  • Purpose: Final phase of distributed aggregation
  • Usage: Combines partial results after data shuffle
Exchange(REPARTITION on [category])
└── AggregationNode(step=FINAL, groupBy=[category])
    └── Output: category, COUNT(*), SUM(price)

INTERMEDIATE Step

  • Input: Intermediate states from other phases
  • Output: Intermediate aggregation state (still partial)
  • Purpose: Middle phase in multi-stage distributed aggregation
  • Usage: Rarely used, for complex multi-level aggregations

Distributed Aggregation Flow

Decomposable vs Non-Decomposable Functions

Decomposable functions are aggregations that can be mathematically split into partial computation phases and then combined to produce the correct final result. This property enables distributed execution.

Requirements for decomposability:

  • The function must support creating intermediate state that captures partial results
  • Combining multiple intermediate states must produce the same result as processing all data together
  • The intermediate type must be serializable for network transfer

Mathematical property: For decomposable function f, combining partial results equals processing all data:

f(data1 ∪ data2) = combine(f(data1), f(data2))

Examples:

Decomposable functions (can be split into partial/final phases):

  • ✅ COUNT(*): Sum of partial counts equals total count
  • ✅ SUM(x): Sum of partial sums equals total sum
  • ✅ AVG(x): Combine (count, sum) pairs to calculate final average
  • ✅ MIN/MAX(x): Minimum/maximum of partial min/max values
  • ✅ Most statistical functions with combinable intermediate states

Non-decomposable functions (require all data in single location):

  • ❌ MEDIAN, PERCENTILE_CONT: Need complete dataset to find middle values
  • ❌ RANK, ROW_NUMBER: Require global ordering
  • ❌ Complex analytics functions that depend on data distribution

Implementation Examples

This section shows how common aggregation functions implement partial and final execution phases.

COUNT Implementation

COUNT(*) uses simple state accumulation:

// PARTIAL phase: Count rows locally
state.setValue(state.getValue() + 1);

// FINAL phase: Sum partial counts  
state.setValue(state.getValue() + otherState.getValue());

Execution flow:

  • PARTIAL: Each worker counts local rows (e.g., 1000, 500)
  • FINAL: Coordinator sums partial counts (1000 + 500 = 1500)
  • Intermediate type: BIGINT

SUM Implementation

SUM(column) accumulates values with overflow protection:

// PARTIAL phase: Accumulate values locally
state.setValue(BigintOperators.add(state.getValue(), value));

// FINAL phase: Sum partial results
state.setValue(BigintOperators.add(state.getValue(), otherState.getValue()));

Intermediate type: Same as input type (BIGINT, DOUBLE, etc.)

AVG Implementation

AVG(column) requires tracking both count and sum:

// PARTIAL phase: Track count and sum
state.setLong(state.getLong() + 1);         // count++
state.setDouble(state.getDouble() + value); // sum += value

// FINAL phase: Combine counts and sums separately
state.setLong(state.getLong() + otherState.getLong());
state.setDouble(state.getDouble() + otherState.getDouble());

// OUTPUT phase: Calculate final average
return state.getDouble() / state.getLong();

Intermediate type: ROW(BIGINT count, DOUBLE sum)

Function Decomposability

Trino determines if a function can be decomposed using metadata:

public boolean isDecomposable() {
    return !intermediateTypes.isEmpty();
}

Functions with defined intermediate types can be distributed across partial/final phases.

Complete Example: Distributed Aggregation

Example query demonstrating partial/final execution:

SELECT region, COUNT(*), AVG(order_value) 
FROM orders 
GROUP BY region

Query Plan Transformation

Optimizer transforms single-step aggregation into distributed execution:

AggregationNode(step=FINAL)
├── COUNT: sum partial counts
├── AVG: combine (count,sum) pairs and divide
└── Exchange(REPARTITION on region)
    └── AggregationNode(step=PARTIAL)
        ├── COUNT: count rows per partition
        ├── AVG: track (count,sum) per partition  
        └── TableScan(orders)

Execution Flow

PARTIAL Phase (workers process local data):

  • Worker 1: {US: (1000 rows, $50000), EU: (500 rows, $25000)}
  • Worker 2: {US: (800 rows, $40000), CA: (300 rows, $15000)}
  • Worker 3: {CA: (200 rows, $10000), EU: (600 rows, $30000)}

Exchange Phase: Redistribute intermediate states by region

FINAL Phase (combine partial results):

  • US: COUNT = 1000+800 = 1800, AVG = (50000+40000)/(1000+800) = 50.00
  • EU: COUNT = 500+600 = 1100, AVG = (25000+30000)/(500+600) = 50.00
  • CA: COUNT = 300+200 = 500, AVG = (15000+10000)/(300+200) = 50.00

Benefits

This architecture provides:

  • Scalability: Processes data across multiple nodes in parallel
  • Network efficiency: Transfers aggregated intermediate results instead of raw data
  • Automatic optimization: Transparent to SQL users
  • Fault tolerance: Built-in retry and recovery mechanisms