Other Physical Operators

Every time you ask a database "How many customers do we have?" or "What's the average order value by region?", you invoke one of the most powerful yet computationally demanding categories of operators: aggregation operators. These operators transform potentially billions of individual rows into summarized insights—counts, sums, averages, minimums, maximums, and statistical measures that drive business intelligence across the modern enterprise.

Aggregation is fundamentally different from relational selection or projection. While selection filters rows and projection removes columns, aggregation collapses entire groups of tuples into single summary values. This collapse operation introduces unique challenges: How do we efficiently compute aggregates over massive datasets? How do we handle grouping when groups don't fit in memory? How do we pipeline aggregate results when the aggregate itself requires seeing all inputs before producing output?

Before diving into physical implementations, we must establish a precise understanding of what aggregation means in relational algebra and SQL. Aggregation operators perform computations across sets of rows to produce summary results.

The SQL Aggregation Model:

In SQL, aggregation involves two conceptually distinct phases:

1. Grouping Phase: Rows are partitioned into groups based on the GROUP BY clause. Rows with identical values in the grouping columns belong to the same group.

2. Aggregation Phase: Within each group, aggregate functions compute summary values over the group members.

When no GROUP BY clause is present, the entire input relation forms a single group—this is called scalar aggregation. When GROUP BY is present, we perform grouped aggregation.

The Physical Challenge:

From a physical operator perspective, aggregation presents several challenges:

1. Memory Pressure: When there are many distinct groups, the aggregation state may not fit in memory.

2. Blocking Nature: Most aggregate implementations must see all input tuples before producing any output—this blocks pipelining.

3. Data Movement: Grouping requires bringing together tuples with matching group keys, potentially from different disk pages.

4. State Accumulation: Each group maintains running state (counters, sums) that must be updated efficiently.

5. Partitioning Complexity: For distributed or parallel aggregation, we must correctly partition work while maintaining correctness.

Hash-based aggregation is the most common strategy in modern database systems due to its excellent average-case performance and natural fit with main-memory processing. The fundamental idea is elegant: use a hash table where keys are the grouping columns and values are the aggregation state (running sums, counts, etc.).

Basic Algorithm (In-Memory):

HASH_AGGREGATE(input_relation, group_keys, aggregate_functions):
    hash_table = empty hash table
    
    for each tuple t in input_relation:
        key = extract group_keys from t
        
        if key exists in hash_table:
            entry = hash_table[key]
            update entry's aggregate state with t's values
        else:
            create new entry with initial aggregate state
            hash_table[key] = entry
    
    for each entry in hash_table:
        output (entry.key, finalize(entry.aggregate_state))

This algorithm scans the input exactly once, making it O(n) for n input tuples, assuming hash table operations are O(1) amortized.

Memory Layout and Hash Table Design:

Efficient hash aggregation requires careful attention to memory layout. Modern implementations typically use:

1. Open Addressing with Linear Probing: Better cache locality than chaining, important when hash table fits in L2/L3 cache.

2. Pre-allocated Buckets: Each bucket contains space for all aggregate state, avoiding pointer chasing.

3. Key Normalization: Fixed-size keys enable faster comparison and hashing.

4. Aggregation State Compaction: Store only the minimum state needed (e.g., for AVG, store sum and count, not all values).

When the number of distinct groups exceeds available memory, hash-based aggregation must spill to disk. This is similar to the partitioning approach used in grace hash join, adapted for aggregation.

Partition-Based External Aggregation:

The algorithm proceeds in two phases:

Phase 1: Partition

• Use a hash function h₁ to partition input tuples into P partitions based on group keys
• Write each partition to a separate temporary file on disk
• Tuples with the same group key always go to the same partition

Phase 2: Aggregate Each Partition

• For each partition:
  • Read partition into memory
  • Build in-memory hash table using a different hash function h₂
  • Aggregate as normal
  • Output results for this partition
  • Clear hash table for next partition

The key insight is that if we have G distinct groups and memory M sufficient for M/B groups (where B is bytes per group), we need ceiling(G × B / M) partitions.

I/O Cost Analysis:

Let's analyze the I/O cost of external hash aggregation:

• Phase 1 (Partitioning):

  • Read input: N pages (one scan of input relation)
  • Write partitions: N pages (each tuple written once)

• Phase 2 (Aggregation):

  • Read partitions: N pages (each partition read once)

• Total I/O Cost: 3N page operations

This is significantly more expensive than in-memory aggregation (just N reads), but unavoidable when groups don't fit in memory.

Sort-based aggregation takes a fundamentally different approach: instead of hashing to group tuples, we sort the input by the grouping columns, then scan sequentially to aggregate consecutive tuples belonging to the same group.

Algorithm:

SORT_AGGREGATE(input_relation, group_keys, aggregate_functions):
    // Phase 1: Sort by group keys
    sorted_input = external_sort(input_relation, group_keys)
    
    // Phase 2: Sequential scan with aggregation
    current_key = null
    current_state = null
    
    for each tuple t in sorted_input:
        key = extract group_keys from t
        
        if key ≠ current_key:
            if current_key ≠ null:
                output (current_key, finalize(current_state))
            current_key = key
            current_state = initialize aggregate state
        
        update current_state with t's values
    
    // Output last group
    if current_key ≠ null:
        output (current_key, finalize(current_state))

The elegance of this approach is that after sorting, aggregation is trivially streaming—we only need to track state for one group at a time, regardless of how many total groups exist.

I/O Cost Analysis:

The total cost depends on the sorting phase:

• External Sort Cost: Typically 2N × (1 + ⌈logₘ₋₁(N/M)⌉) where:

  • N = number of pages in the input
  • M = number of memory buffer pages
  • The first term (2N) is for creating sorted runs
  • The logarithmic term accounts for merge passes

• Aggregation Scan: N pages (one sequential scan of sorted data)

Total: O(N log N) I/O for sorting plus O(N) for aggregation

This seems worse than hash aggregation's O(N) I/O for in-memory case and O(3N) for spilling case. So why would we ever use sort-based aggregation?

In distributed and parallel database systems, aggregation presents unique challenges: data is spread across multiple nodes, and network transfer is expensive. The solution is multi-phase aggregation that minimizes data movement.

Two-Phase Aggregation Pattern:

1. Local (Partial) Aggregation: Each node computes partial aggregates over its local data
2. Global (Final) Aggregation: Partial results are shuffled/merged to produce final output

This pattern exploits the distributive and algebraic properties of aggregates:

• SUM(global) = SUM(local_SUM₁, local_SUM₂, ...)
• COUNT(global) = SUM(local_COUNT₁, local_COUNT₂, ...)
• AVG(global) = SUM(local_SUM) / SUM(local_COUNT)
• MIN(global) = MIN(local_MIN₁, local_MIN₂, ...)

Data Reduction Benefits:

The power of partial aggregation lies in data reduction. Consider aggregating 1 billion rows across 100 nodes with 1,000 distinct groups:

• Without partial aggregation: Ship 1 billion rows across the network
• With partial aggregation: Ship at most 100 × 1,000 = 100,000 partial aggregate records

The reduction factor can be 10,000x or more, dramatically reducing network I/O which is often the bottleneck in distributed queries.

Push-Down Optimization:

Smart query optimizers push partial aggregation as close to the data source as possible, even into storage engines or remote data sources. This is particularly important in:

• Federated databases: Aggregate remotely before transfer
• Column stores: Leverage vectorized partial aggregation
• Data lakes: Push aggregation into Parquet/ORC readers

Production database systems employ numerous optimizations beyond the basic algorithms. Understanding these details illuminates how modern query engines achieve exceptional aggregation performance.

Aggregate Function State Management:

Each aggregate function defines three operations:

1. Initialize: Create the initial state (e.g., count=0, sum=0)
2. Accumulate: Update state with a new value (e.g., count++, sum+=value)
3. Finalize: Compute final result from accumulated state (e.g., for AVG: return sum/count)

For distributive aggregates, we also need:

4. Merge: Combine two partial states (e.g., sum1 + sum2, count1 + count2)

These operations are typically implemented as specialized code per aggregate type, often JIT-compiled for maximum efficiency.

NULL Handling:

SQL semantics for NULLs in aggregates are precise but often overlooked:

• COUNT(*): Counts all rows, including those with NULL values
• COUNT(column): Counts only non-NULL values
• SUM, AVG, MIN, MAX: Ignore NULL values entirely
• All aggregates over empty set: Return NULL (except COUNT returns 0)

Physical implementations must track NULL state separately, adding a NULL bitmap or flag to the aggregate state.

DISTINCT Aggregates:

Aggregates like COUNT(DISTINCT column) or SUM(DISTINCT column) require deduplication before aggregation. This is typically implemented by:

1. Building a hash set of values within each group
2. Aggregating over the deduplicated set

This significantly increases memory consumption because we must store all distinct values, not just aggregate state.

Modern analytical database systems leverage SIMD (Single Instruction, Multiple Data) instructions to dramatically accelerate aggregation. Rather than processing one tuple at a time, vectorized aggregation operates on batches of tuples simultaneously.

Vectorized Hash Aggregation:

The key insight is to separate the expensive per-tuple hash table operations from the cheap arithmetic accumulation:

1. Batch Hashing: Compute hashes for a batch of group keys using SIMD
2. Batch Lookup: Probe hash table for all batch entries, recording bucket addresses
3. Batch Update: Use gathered bucket addresses to update aggregate states

Each step can be pipelined and vectorized, achieving 4-16x speedup over scalar code.

Column-Oriented Aggregation:

Column stores are particularly well-suited for aggregation because:

1. Sequential Access: Reading a single column is sequential I/O
2. Compression: Columns compress well; aggregation operates on compressed data
3. Cache Efficiency: Only aggregate columns loaded into cache, not entire rows
4. SIMD Compatibility: Fixed-width columns align to SIMD registers

Late Materialization:

A powerful optimization called late materialization defers constructing result tuples until necessary:

1. During aggregation, track only column positions (not full values)
2. After grouping complete, fetch actual values only for output columns
3. Avoid copying non-aggregate columns that aren't needed until final output

Aggregation operators are fundamental to analytical query processing, transforming raw data into actionable insights. Let's consolidate the key concepts:

What's Next:

Aggregation operators often work alongside duplicate elimination, which shares many implementation strategies. The next page explores how DISTINCT queries and duplicate handling are physically implemented, building on the hash and sort techniques we've learned here.