A federated query engine reads data from independent systems without first copying every row into one store. Aggregation is where federation can produce its largest win: compute a small partial state at each source, transfer those states, and merge them into the final result.

The attractive slogan is “push SUM to the sources.” The difficult work is proving that the pushed computation has the same semantics as the original query under nulls, types, retries, filters, snapshots, and non-decomposable aggregates.

Begin with an algebra, not an API

Suppose the query is:

SELECT SUM(value) / NULLIF(SUM(quantity), 0) AS ratio
FROM all_sources
WHERE event_time >= :start;

For source i, compute a state:

Sᵢ = (sum_valueᵢ, sum_quantityᵢ)

Define a merge operation:

(a, b) ⊕ (c, d) = (a + c, b + d)

The operation is associative and commutative, and (0, 0) is an identity. Therefore a tree of workers can merge states in any grouping and obtain:

finalize(S₁ ⊕ S₂ ⊕ ... ⊕ Sₙ) = total_value / total_quantity

The finalize step is applied exactly once after merging. Dividing at each source and averaging the ratios would answer a different question.

Which aggregates are decomposable?

Many SQL aggregates can be represented by mergeable state:

Aggregate Partial state Merge Finalize
COUNT(*) count add count
SUM(x) sum, seen-non-null add null if none seen
MIN(x) optional minimum min value/null
AVG(x) sum, count pairwise add sum / count
variance count, mean, second moment stable parallel merge derive variance
approximate distinct sketch registers/state sketch union estimate

AVG illustrates why final scalar results are usually not mergeable. The average of averages is correct only when every partial average represents the same count. The state must contain both sum and count.

Exact COUNT(DISTINCT x), arbitrary percentiles, and ordered string aggregation may require a state proportional to the input or a more specialized algorithm. An optimizer must know whether an aggregate supports partial evaluation; it cannot assume every function does.

A typed merge state in Rust

If value and quantity are integer minor units, an exact accumulator can delay floating-point conversion until finalization:

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct RatioState {
    value_sum: i128,
    quantity_sum: i128,
    rows: u64,
}

#[derive(Debug, thiserror::Error)]
enum AggregateError {
    #[error("partial aggregate overflow")]
    Overflow,
    #[error("denominator is zero")]
    ZeroDenominator,
}

impl RatioState {
    fn merge(self, other: Self) -> Result<Self, AggregateError> {
        Ok(Self {
            value_sum: self
                .value_sum
                .checked_add(other.value_sum)
                .ok_or(AggregateError::Overflow)?,
            quantity_sum: self
                .quantity_sum
                .checked_add(other.quantity_sum)
                .ok_or(AggregateError::Overflow)?,
            rows: self
                .rows
                .checked_add(other.rows)
                .ok_or(AggregateError::Overflow)?,
        })
    }

    fn ratio(self) -> Result<f64, AggregateError> {
        if self.quantity_sum == 0 {
            return Err(AggregateError::ZeroDenominator);
        }
        Ok(self.value_sum as f64 / self.quantity_sum as f64)
    }
}

i128 is not automatically the correct SQL type. The federation layer must map each source’s decimal precision and scale to a common representation and define overflow behavior. Converting database Decimal(38, 9) values to f64 at the source can silently change results.

The rows field is not needed for this ratio, but it is useful for validation and observability. A richer state might include non_null_values, non_null_quantities, and a source snapshot token.

SQL null semantics must survive pushdown

SQL SUM(x) ignores null input and returns NULL when no non-null rows exist. Replacing an empty partial result with numeric zero can make an empty global input indistinguishable from rows that sum to zero.

For a rigorous partial state, include a presence bit or count:

PartialSum = { value, non_null_count }

Merge both fields. Finalize to NULL when the total count is zero.

The same care applies to filters:

  • three-valued boolean logic must match;
  • collations and case-folding may differ;
  • timestamps need a shared instant/time-zone interpretation;
  • floating-point NaN ordering varies across systems;
  • string-to-number casts may reject or coerce different inputs.

Pushdown is safe only for an expression whose semantics the source can reproduce.

A source contract for partial aggregation

Instead of pretending every source is just a table scan, define an explicit request and response:

PartialAggregateRequest
  relation
  filters
  group_keys
  aggregate_states
  required_snapshot
  request_id

PartialAggregateResponse
  schema_version
  source_id
  snapshot_token
  zero_or_more_state_rows
  statistics

The response schema should describe intermediate states, not final user-facing columns. For the ratio query:

source_id: Utf8
snapshot_token: Utf8
sum_value: Decimal128(p, s)
sum_quantity: Decimal128(p, s)
row_count: UInt64

If the query groups by customer, each source returns one state row per local customer. The federation layer then repartitions or merges by customer key.

Planning with DataFusion

Apache DataFusion represents queries as logical and physical plans over Arrow batches. Its local aggregation commonly uses partial and final aggregate stages. A federation layer can use the same idea across source boundaries:

FinalProjection: value_sum / quantity_sum
  FinalAggregate: SUM(partial_value), SUM(partial_quantity)
    Union
      RemoteAggregateExec(source = ClickHouse A)
      RemoteAggregateExec(source = PostgreSQL B)
      LocalPartialAggregateExec(Parquet C)

A custom TableProvider is appropriate for exposing source data and advertising supported filter pushdown. Aggregate pushdown may require a custom logical node, planner extension, or source-specific provider design so the remote query receives aggregate expressions rather than only a scan projection.

Do not fabricate a TableProvider::scan implementation that ignores its projection, filters, or limit. That produces correct-looking demos and incorrect optimizers. Define exactly which expressions are accepted and return “unsupported” for the rest so DataFusion retains the operation above the scan.

Generate source SQL safely

Identifiers cannot usually be passed as ordinary prepared-statement parameters. A table or column name supplied by configuration must be resolved against an allow-listed catalog and quoted with the source dialect. Values belong in bound parameters.

For ClickHouse, a generated query might be conceptually:

SELECT
    toDecimal128(sum(value_minor), 0) AS value_sum,
    toDecimal128(sum(quantity), 0) AS quantity_sum,
    count() AS row_count
FROM analytics.events
WHERE event_time >= {start:DateTime64(3)}
  AND event_time < {end:DateTime64(3)}

The federation connector must validate the result types returned by the server instead of trusting aliases. A schema or server upgrade can change coercion.

Snapshot consistency is part of the answer

If source A is read at 10:00:00 and source B at 10:00:30, the combined value may never have existed at one logical time. Distributed aggregation reduces data movement; it does not create a distributed snapshot.

Possible contracts include:

  • best effort: each source is read when contacted;
  • bounded staleness: every source snapshot must be newer than a threshold;
  • as-of time: sources capable of time travel read a specified timestamp;
  • coordinated transaction: connectors participate in a shared snapshot protocol;
  • watermark-aligned: streaming sources report completeness through an event-time boundary.

Return snapshot tokens in the result metadata. If the system cannot provide a global snapshot, say so rather than labeling the result “consistent.”

Retries can duplicate partial states

Suppose a source request succeeds but its response is lost. Retrying and then accepting both the late original and retry response doubles that source’s contribution.

Use a stable (query_id, stage_id, source_id, attempt) identity and make the coordinator accept exactly one successful attempt per logical source partition. Merge functions being associative does not make them idempotent: S ⊕ S usually differs from S.

For speculative execution, first-writer-wins state must be durable enough that coordinator recovery does not count a second attempt later.

Failure policy

A federated aggregate needs an explicit policy for unavailable sources:

  • fail the entire query;
  • return a partial result marked with missing sources;
  • use a cached state with its age disclosed;
  • retry within a deadline;
  • degrade only for metrics declared tolerant of partial data.

Never return a partial numeric value with the same schema and status as a complete result. Completeness is data.

Cost-based pushdown

Pushdown is not always faster. A remote source may be overloaded, poorly indexed, or charge per scanned byte. Local execution may exploit cached columnar files or a more efficient aggregate implementation.

Estimate:

  • source rows and groups after filtering;
  • selectivity and index support;
  • partial-state cardinality;
  • network bytes and round trips;
  • source concurrency limits;
  • coordinator memory;
  • numerical or semantic conversion cost.

A high-cardinality GROUP BY can return almost as many partial rows as input rows. In that case, pushing only filters/projections or using a different partitioning strategy may be better.

Verification strategy

Test pushdown as a semantic rewrite:

  1. generate small adversarial datasets with nulls, empty inputs, negative values, large decimals, and duplicate keys;
  2. run a reference query without aggregate pushdown;
  3. run the federated partial/final plan;
  4. compare values, nullability, types, and error behavior;
  5. inject retries, timeouts, reordered responses, and source schema changes;
  6. inspect EXPLAIN output to ensure only supported expressions were pushed.

Property tests are especially effective: partition one input dataset in many random ways and verify that merging partial states equals a single complete aggregation.

The durable design rule

Distributed aggregation is correct when an aggregate exposes a mergeable state, the source computes that state with compatible semantics, and the coordinator merges each logical source exactly once over a disclosed snapshot.

The speedup comes after that proof. Without it, moving fewer rows only produces the wrong answer more efficiently.

Sources and further reading