Distributed Aggregation in Data Federation

A federated query engine can reduce network traffic dramatically by asking each source to compute a partial aggregate and merging the partial states at a coordinator. The appealing plan is:

1. push GROUP BY and aggregates to every source;
2. transfer one row per group instead of every input row;
3. combine the source results.

That plan is correct only when the aggregate has a mergeable state and every source agrees on filters, grouping, types, nulls, and snapshot boundaries. Performance is the last step; first we must define what result the system promises.

This article uses Apache DataFusion as the Rust execution framework, but the reasoning applies to any federation layer.

An aggregate is a state machine

An aggregate can be modeled with three functions:

• accumulate(state, row) -> state;
• merge(left_state, right_state) -> state;
• finalize(state) -> result.

Distributed execution is safe when partitioning the input and repeatedly applying merge produces the same result as a single logical aggregation, subject to the documented numeric behavior.

Some familiar aggregates have compact mergeable states:

An exact distinct count is mergeable if every source sends its full set, but that state can be as large as the input. A fixed-size sketch such as HyperLogLog trades exactness for bounded state and a stated error model.

Median and arbitrary percentiles are not derived by taking the median of source medians. They require a mergeable distribution summary, an ordered-set protocol, or access to the underlying values.

Why averages fail so easily

Suppose one source contains one order worth 100 units and another contains nine orders worth 10 units each. The two source averages are 100 and 10. Their unweighted average is 55, while the global average is:

(100 + 9 × 10) / (1 + 9) = 19
      

Every source must return SUM(value) and COUNT(value). The coordinator sums both fields and divides once.

For a fixed-point domain, a small Rust state makes that contract explicit:

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
      struct AverageState {
          // Values use a source-agreed fixed-point scale.
          scaled_sum: i128,
          non_null_count: u64,
      }
      
      #[derive(Debug, PartialEq, Eq)]
      enum MergeError {
          SumOverflow,
          CountOverflow,
      }
      
      impl AverageState {
          fn merge(self, other: Self) -> Result<Self, MergeError> {
              Ok(Self {
                  scaled_sum: self
                      .scaled_sum
                      .checked_add(other.scaled_sum)
                      .ok_or(MergeError::SumOverflow)?,
                  non_null_count: self
                      .non_null_count
                      .checked_add(other.non_null_count)
                      .ok_or(MergeError::CountOverflow)?,
              })
          }
      
          fn finalize(self) -> Option<f64> {
              (self.non_null_count != 0).then(|| {
                  self.scaled_sum as f64 / self.non_null_count as f64
              })
          }
      }
      

The state is part of the execution contract. A source returning only a finalized average has thrown away information the coordinator needs.

Algebraic classes are useful, but implementation state matters more

Database literature often groups aggregates into distributive, algebraic, and holistic classes.

• Distributive: a partial result can be merged with the same operation, such as SUM, COUNT, MIN, and MAX.
• Algebraic: a fixed-size tuple of partial values is sufficient, such as AVG represented by (sum, count).
• Holistic: no bounded exact partial state exists in general, such as exact median or exact distinct count.

This classification is a good planning shortcut, but a real engine still needs the concrete state representation. Decimal sums need a scale and overflow policy. Floating-point sums have order-dependent rounding. Variance needs a numerically stable merge formula. Sketches need compatible parameters.

“Mergeable” is a typed protocol, not a boolean property attached only to the SQL function name.

GROUP BY keys must be semantically identical

Partial aggregation also assumes that every source partitions rows into groups the same way.

Consider:

SELECT country, COUNT(*)
      FROM events
      GROUP BY country
      

If one source compares strings case-sensitively and another applies a case-insensitive collation, their partial rows do not represent the same grouping domain.

The coordinator can safely merge source groups only if the canonical key semantics agree on issues such as:

• string collation and normalization;
• time-zone interpretation for derived date keys;
• decimal scale and coercion;
• signed versus unsigned integer conversions;
• floating-point NaN handling;
• null grouping behavior;
• expression semantics for computed keys.

A federation layer may need to normalize keys at the coordinator or decline a pushdown when it cannot prove semantic equivalence.

Nulls are part of the algebra

SQL aggregate behavior around nulls is not optional detail.

COUNT(*)   -- counts rows
      COUNT(x)   -- counts non-null x values
      SUM(x)     -- ignores nulls, returns null for an empty input in SQL semantics
      AVG(x)     -- sum and count over non-null values
      

If a source API returns zero for an empty sum while the federation layer expects SQL NULL, finalization must repair that difference or the aggregate cannot be pushed safely.

Likewise, a partial state must distinguish “no values observed” from “values observed whose sum happens to be zero.”

Filters must be equivalent before aggregation is pushed

The biggest network savings occur when a predicate can be pushed below the aggregate:

source scan
         -> source filter
         -> source partial aggregate
         -> network
         -> coordinator merge
      

But this is correct only if the remote filter implements the same predicate.

Differences can come from:

• timestamp parsing;
• time zones;
• string comparison;
• regular-expression dialects;
• implicit casts;
• null comparison;
• overflow behavior;
• source-specific functions.

A planner should have a capability contract for expressions rather than assume that two SQL-looking syntaxes mean the same thing.

Pushdown should be a proof obligation

Instead of asking “can this source run SUM?”, ask a sequence of narrower questions:

1. Can the source evaluate the pushed filter with equivalent semantics?
2. Can it produce grouping keys in a compatible representation?
3. Can it return a mergeable partial state for every aggregate?
4. Can the coordinator decode that state without losing type information?
5. Are the source snapshots compatible with the query's consistency contract?
6. Does partial failure have a defined result policy?

Only then is the optimization valid.

This turns source capabilities into planner metadata rather than ad-hoc SQL string generation.

A capability model

A simplified Rust interface might look like:

#[derive(Clone, Debug)]
      enum AggregateCapability {
          Sum { input_type: String, state_type: String },
          Count,
          Average { sum_type: String },
          Min,
          Max,
          HllDistinct { precision: u8, hash_family: String },
      }
      
      trait FederationSource {
          fn supports_filter(&self, expr: &str) -> bool;
          fn supports_group_key(&self, expr: &str) -> bool;
          fn aggregate_capability(&self, function: &str) -> Option<AggregateCapability>;
      }
      

A production implementation should use typed expressions rather than strings. The point is that capability describes semantics and state shape, not merely the presence of a function name.

Planning partial and final aggregation

For a query such as:

SELECT region, AVG(latency_ms)
      FROM requests
      WHERE status = 200
      GROUP BY region
      

a correct federated rewrite is conceptually:

logical global aggregate
              |
              v
      rewrite AVG -> SUM + COUNT state
              |
              +-------------------+
              |                   |
              v                   v
      source A                 source B
      filter status=200        filter status=200
      GROUP BY region          GROUP BY region
      SUM(latency), COUNT      SUM(latency), COUNT
              |                   |
              +---------+---------+
                        |
                        v
              merge by canonical region
              add sums and counts
                        |
                        v
                 finalize AVG once
      

The coordinator is still performing an aggregate. It is aggregating states, not raw rows.

DataFusion's logical and physical-plan extension points make this model natural: rewrite a global aggregate into source-local partial plans plus a merge/finalize operator when source capabilities allow it.

Variance needs a real merge formula

A common mistake is to ask every source for VARIANCE(x) and average the results. That is not correct.

A parallel variance algorithm can maintain state such as:

count
      mean
      M2 = sum of squared deviations from the mean
      

Two states can then be merged using the difference between their means. This preserves the information required to account for between-partition variation and is more numerically stable than naïve global sum(x^2) - sum(x)^2/n formulas.

The broader lesson is that final SQL results are often not valid partial states.

Approximate distinct has a compatibility contract

HyperLogLog is attractive in federation because every source can send a compact sketch instead of a large set of distinct values.

But two HLL states are unionable only when their representation agrees. At minimum the federation layer needs compatible:

• precision or register count;
• hash function/family;
• seed or hashing convention;
• serialization version;
• treatment of nulls and input encoding.

The result also needs an explicit approximation contract. A planner must not silently substitute a sketch for exact COUNT(DISTINCT x) unless the query semantics or API explicitly allow approximation.

Snapshot boundaries are part of correctness

Even perfect aggregate algebra cannot fix inconsistent source time.

Suppose a federated query reads:

US source at snapshot 120
      EU source at snapshot 118
      APAC source at snapshot 121
      

What does the global result mean?

There are several legitimate contracts:

• best-effort current reads;
• a common event-time watermark;
• a coordinator-selected snapshot per source recorded in the result;
• only data through the minimum complete watermark;
• fail unless all required sources can satisfy a requested logical snapshot.

The engine must choose one deliberately.

“Push aggregation down” is not an excuse to hide inconsistent coverage. The coordinator should carry snapshot and coverage metadata beside the partial states.

Partial failure needs a query policy

If three sources are required and one times out, returning the sum of the other two without warning is usually worse than failing.

Useful policies include:

Strict

All required sources must succeed. Otherwise the query fails.

Partial with explicit coverage

Return available results but attach metadata that identifies missing sources or time ranges.

Best effort for explicitly approximate workflows

Allow partial participation only when the caller selected a semantics that makes the incompleteness acceptable.

The choice should be visible in the result contract, not hidden in a retry loop.

Retries can duplicate partial results

Distributed execution retries create another correctness trap.

Suppose source A returns a partial state, the coordinator loses the acknowledgement, and a task is retried. If both completions are merged, source A is counted twice.

A distributed aggregate needs an identity model for partial states, such as:

query_id
      source_id
      snapshot_id
      partition_id
      attempt_id
      

The coordinator should accept one logical partial result per expected partition, even if multiple attempts execute.

Exactly-once execution is not required. Exactly-once logical contribution is.

Floating point is mergeable but not bitwise associative

For real numbers, addition is associative. IEEE floating-point addition is not.

A distributed SUM(f64) can therefore produce slightly different low-order bits depending on partitioning and merge order.

That does not necessarily make the aggregate invalid. It means the query engine must document its numeric guarantees. Reproducible analytics may require deterministic merge trees, compensated summation, decimals, or tolerances instead of pretending distributed floating-point reduction is bitwise invariant.

Cost comes after correctness

Once a pushdown is proven valid, the planner can ask whether it is worthwhile.

Useful signals include:

• estimated input rows versus output groups;
• network bandwidth and latency;
• remote CPU cost;
• source concurrency limits;
• cardinality of grouping keys;
• partial-state size;
• coordinator memory pressure;
• whether the source already has useful indexes or pre-aggregated data.

A group-by with nearly one unique group per input row may save little network and add remote CPU. A global COUNT(*) over billions of rows can save enormous transfer cost.

A safer optimizer rule

A practical planner rule can be summarized as:

for each aggregate query:
          validate source snapshot/coverage contract
          validate pushed filter semantics
          validate group-key semantics
          derive mergeable partial state for every aggregate
          validate state compatibility across sources
      
          if any proof fails:
              keep the operation above the federation boundary
          else if cost model predicts benefit:
              push partial aggregation down
              merge states at coordinator
              finalize once
      

The fallback is important. A federation system remains correct when an optimization is unavailable.

Correctness is the optimization boundary

Distributed aggregation is powerful because it moves computation toward data and shrinks network traffic. It is also a place where superficially reasonable plans produce subtly wrong answers.

The reliable mental model is not “each database runs the same SQL.” It is:

Every source produces a typed partial state under equivalent row and grouping semantics, and the coordinator merges those states under an explicit snapshot and failure contract.

Once that invariant is true, pushdown becomes a performance optimization. Before it is true, pushdown is a semantic change.