Aggregates in Datalog: Group, Reduce, Stratify

Pure Datalog derives facts by recursive joining: if ancestor(X,Y) holds whenever parent(X,Y) or parent(X,Z) and ancestor(Z,Y), every answer falls out of the fixpoint. But many real questions are not about what is true — they are about how many or how much. "How many employees per department?" "What is the largest score in each group?" "What is the total salary paid to each team?"

These are aggregate queries. Classical Datalog has no aggregates: the least-fixpoint semantics is defined over sets of facts, and adding "count" looks innocuous but breaks monotonicity — later derivations can change a count that an earlier rule already consumed. ExDatalog introduces aggregates as a constrained first-class feature: four integer-only operators, one aggregate per rule, no recursion through the aggregate, and a stratification pass that forces the aggregate's head relation to live strictly above every relation that feeds it.

Why Aggregates Are Not Just Another Constraint

Comparison and arithmetic constraints are per-binding. Given a binding %{"X" => 10, "Y" => 3}, evaluating gt(X, 5) or add(X, Y, Z) is a local, stateless operation — the answer depends only on that one binding. The engine can fold them into the per-row constraint pipeline (ConstraintEval.apply/3) without caring about other bindings.

Aggregates break that model. count(E, N) needs to see all employees in a department before it can pick N. There is no single binding that yields 2 — the value 2 only exists relative to a group of bindings. That is why ExDatalog.Constraints.Aggregate.evaluate/3 does not compute anything:

@impl ExDatalog.Constraint
def evaluate(_constraint, _binding, _context) do
  raise "aggregate constraints are not evaluated per-binding; " <>
          "use ExDatalog.Engine.Evaluator group-and-reduce path"
end

The behaviour callback exists only so the aggregate module fits the same dispatch table as comparisons and arithmetic. Real evaluation happens later, in the engine, after the per-binding pipeline is done. This split is the central design decision behind aggregates in ExDatalog: a constraint declares the intent ("count E into N"), and the engine decides when to honor it.

The Four Operators

ExDatalog ships four aggregate operators. They are integer-only — there is no avg yet — and live in the same @aggregate_ops list as the rest of the constraint categories:

@aggregate_ops [:count, :sum, :min, :max]

The reducer itself is a four-clause function in ExDatalog.Constraints.Aggregate:

defp compute(:count, values), do: length(values)
defp compute(:sum, values), do: Enum.sum(values)
defp compute(:min, values), do: Enum.min(values)
defp compute(:max, values), do: Enum.max(values)

count works on any group; count of employees is the size of the group regardless of the input variable's type. sum, min, and max require integer inputs, and this is guarded at runtime: a non-integer input raises a descriptive ArgumentError (for example, "sum aggregate requires integer inputs, got: :x") from the reducer. Build-time type enforcement of aggregate inputs is future work. Because aggregates are stratified above their inputs (see below), the input list is fully materialized and final before the reducer runs, so a type error during reduction is a property of the data, not of evaluation order.

The DSL Syntax

In the Schema DSL, aggregates appear in rule bodies, never in heads. The DSL expands count(X, N) in a body position into a %Constraint{op: :count, left: {:var, "X"}, right: nil, result: {:var, "N"}}. A typical rule:

rule dept_count(D, N) do
  emp(E, D)
  count(E, N)
end

Two compile-time safeguards prevent misuse:

1. Aggregates are rejected in the rule head. rule dept_count(D, count(E)) raises:

   ** (ExDatalog.DSL.CompileError) aggregate :count/2 cannot appear in a rule head
   or as a term; place it in the rule body (e.g. `count(X, Result)`) and use
   Result in the head

2. The legacy agg(:count, X) form is rejected with a hint to use the named operator:

   ** (ExDatalog.DSL.CompileError) use count/sum/min/max for aggregates
   (e.g. `count(X, N)`)

Both checks live in parse_term/1 and parse_body_call/1. The body form is generated from the same [:count, :sum, :min, :max] list used by the builder, so the DSL and the programmatic API cannot drift:

Enum.each(aggregate_ops, fn op ->
  defp parse_body_call({unquote(op), _, [input, result]}) do
    {:constraint, build_aggregate(unquote(op), input, result)}
  end
end)

The result variable (N in count(E, N)) is the one that appears in the rule head. The input variable (E) must be bound by a positive body atom. Everything else about the rule looks ordinary — and that is the point: aggregates are constraints from the DSL's perspective, even though the engine treats them specially.

The Builder API

Programs that bypass the DSL construct aggregate constraints directly. ExDatalog.Constraint exposes one constructor per operator:

iex> ExDatalog.Constraint.count({:var, "Emp"}, {:var, "N"})
%ExDatalog.Constraint{op: :count, left: {:var, "Emp"}, right: nil, result: {:var, "N"}}

iex> ExDatalog.Constraint.sum({:var, "Amount"}, {:var, "Total"})
%ExDatalog.Constraint{op: :sum, left: {:var, "Amount"}, right: nil, result: {:var, "Total"}}

iex> ExDatalog.Constraint.min({:var, "Score"}, {:var, "Lowest"})
%ExDatalog.Constraint{op: :min, left: {:var, "Score"}, right: nil, result: {:var, "Lowest"}}

iex> ExDatalog.Constraint.max({:var, "Score"}, {:var, "Highest"})
%ExDatalog.Constraint{op: :max, left: {:var, "Score"}, right: nil, result: {:var, "Highest"}}

All four constructors route through the private aggregate/3 helper, which is what distinguishes an aggregate structurally from a comparison: right is always nil, and result is always a {:var, name} tuple. The validity check enforces this:

defp valid_right?(op, nil) when op in @type_ops or op in @aggregate_ops, do: true

defp valid_result?(op, {:var, name}) when op in @arithmetic_ops or op in @aggregate_ops,
  do: is_binary(name) and byte_size(name) > 0

The same shape is reachable through Constraint.from_tuple/1, which is what the DSL uses internally and what tests use to check the wire format:

iex> ExDatalog.Constraint.from_tuple({:count, {:var, "X"}, {:var, "N"}})
%ExDatalog.Constraint{op: :count, left: {:var, "X"}, right: nil, result: {:var, "N"}}

iex> ExDatalog.Constraint.from_tuple({:sum, :A, :T})
%ExDatalog.Constraint{op: :sum, left: {:var, "A"}, right: nil, result: {:var, "T"}}

from_tuple/1 reuses Term.from/1, so the Prolog convention applies: uppercase atoms become variables, lowercase atoms and other values become constants.

A full aggregate rule built by hand — the same rule the DSL dept_count example compiles to — looks like:

Rule.new(
  Atom.new("dept_count", [Term.var("D"), Term.var("N")]),
  [{:positive, Atom.new("emp", [Term.var("E"), Term.var("D")])}],
  [Constraint.count(Term.var("E"), Term.var("N"))]
)

The aggregate lives in the constraints list alongside any comparisons; the engine distinguishes it by op and routes it to the group-and-reduce path.

Grouping Semantics

The hard question for any aggregate is "group by what?" ExDatalog's answer is mechanical and unambiguous: the group key is the set of head variables other than the aggregate result. Concretely, for a rule with head dept_count(D, N) and aggregate count(E, N), the result variable is N, so the group key is [D]. Every surviving binding for the rule is bucketed by its D value, and each non-empty bucket is reduced to a single output binding.

The engine computes the group key in aggregate_group_vars/2:

defp aggregate_group_vars(%IR.Atom{terms: terms}, result_var) do
  terms
  |> Enum.flat_map(fn
    {:var, name} -> [name]
    _ -> []
  end)
  |> Enum.reject(fn name -> name == result_var end)
  |> Enum.uniq()
end

Constants in the head are ignored (they are not variables), and result_var is dropped because it is the output of the reduction, not a grouping column. If the head is total(D, T) and the constraint is sum(A, T), the groups are keyed by D; if the head is lowest(D, V) with min(S, V), the groups are also keyed by D.

The reduction itself is a one-liner using Enum.group_by/2:

def group_and_reduce(bindings, group_vars, op, input_var, result_var) do
  bindings
  |> Enum.group_by(fn binding -> Map.take(binding, group_vars) end)
  |> Enum.map(fn {_key, group} ->
    values = Enum.map(group, fn b -> Map.fetch!(b, input_var) end)
    Map.put(hd(group), result_var, compute(op, values))
  end)
end

A group only exists because at least one binding produced its key — so Enum.min/1 and Enum.max/1 are never called on an empty list. The output binding is the group's first binding extended with result_var => computed_value; the other columns of that binding are the surviving group key, which Join.project/2 then reorganizes into the head tuple.

Safety Rules

Aggregates are non-monotone: a department whose head count is 2 today might be 3 after more facts arrive, and a rule that already consumed "2" would be wrong. The safety validator (ExDatalog.Validator.Safety) rejects any program that could exhibit this. The aggregate-specific checks are layered on top of the ordinary variable-safety rules:

1. At most one aggregate per rule. A rule may not mix count and sum:

# Rejected: two aggregates in one rule
Rule.new(
  Atom.new("stats", [Term.var("D"), Term.var("N"), Term.var("T")]),
  [{:positive, Atom.new("emp", [Term.var("E"), Term.var("D"), Term.var("A")])}],
  [Constraint.count(Term.var("E"), Term.var("N")),
   Constraint.sum(Term.var("A"), Term.var("T"))]
)
# => {:error, [%Error{kind: :multiple_aggregates, ...}]}

The check is a single length test:

defp check_single_aggregate(errors, aggregates, _rule, rule_index) do
  if length(aggregates) > 1 do
    [Error.new(:multiple_aggregates, ..., "a rule may contain at most one " <>
       "aggregate (found #{length(aggregates)}); split into separate rules") | errors]
  else
    errors
  end
end

The fix is mechanical: split into two rules, one deriving dept_count(D, N) and one deriving dept_total(D, T), both reading from the same emp relation.

2. No aggregate through a self-recursive relation. A rule whose head relation also appears in its own body — positively or negatively — cannot aggregate:

# Rejected: path counts itself
Rule.new(
  Atom.new("path", [Term.var("X"), Term.var("N")]),
  [{:positive, Atom.new("path", [Term.var("X"), Term.var("Y")])}],
  [Constraint.count(Term.var("Y"), Term.var("N"))]
)
# => {:error, [%Error{kind: :aggregate_in_recursion, ...}]}

The check inspects every body literal for a relation that matches the head:

self_recursive? =
  Enum.any?(rule.body, fn
    {:positive, %Atom{relation: ^head_rel}} -> true
    {:negative, %Atom{relation: ^head_rel}} -> true
    _ -> false
  end)

This is a conservative, syntactic recursion check. Mutual recursion through a helper relation is not blocked here, but is caught by the stratification pass described next — the head relation will still be forced into a stratum that depends on itself, which the greedy assigner cannot satisfy without violating the aggregate-stratification invariant.

3. The aggregate input must be bound. count(Z, N) where Z is never bound by a positive body atom is rejected by the ordinary "unbound constraint variable" check, the same path that catches gt(W, 5) with an unbound W. The aggregate is treated as a constraint with input variables Constraint.input_variables/1 returns, which for an aggregate is just [left]:

def input_variables(%__MODULE__{left: left, right: nil}), do: Term.variables([left])

4. The result variable must appear in the head. Because the result variable is computed by the reduction, it must be projected into the head to be observable at all. The head-safety check already allows an arithmetic or aggregate result variable to satisfy head safety:

result_vars =
  Enum.flat_map(constraints, fn c ->
    if Constraint.arithmetic?(c) or Constraint.aggregate?(c),
      do: [Constraint.result_variable(c)],
      else: []
  end)

So Rule.head_variables(rule) including N is satisfied by the count(E, N) constraint without N appearing in any positive body atom. If the head omits the result variable, the rule is rejected at validation time with an :aggregate_result_not_in_head error — the safety checker refuses to compile a rule whose aggregate result would be silently discarded:

# rule dept_count(D) do      # head omits N
#   emp(E, D)
#   count(E, N)
# end
#=> {:error, [%ExDatalog.Validator.Error{kind: :aggregate_result_not_in_head, ...}]}

Stratification: Forcing the Aggregate Above Its Inputs

Aggregates must be evaluated after every relation they read from is fully materialized. The stratification validator enforces this by running a fixpoint that bumps each aggregate rule's head relation until it sits strictly above all of its positive body relations:

defp stabilize_aggregate_strata(strata, aggregate_rules, graph, fuel) do
  bumped =
    Enum.reduce(aggregate_rules, strata, fn rule, acc ->
      head_rel = rule.head.relation
      body_rels =
        for {:positive, %Atom{relation: rel}} <- rule.body, rel != head_rel, do: rel

      required =
        case body_rels do
          [] -> Map.get(acc, head_rel, 0)
          _ -> (body_rels |> Enum.map(&Map.get(acc, &1, 0)) |> Enum.max()) + 1
        end
      ...
    end)

  bumped = propagate_strata(bumped, graph)
  if bumped == strata, do: strata, else: stabilize_aggregate_strata(bumped, ...)
end

For dept_count(D, N) :- emp(E, D), count(E, N), suppose emp is at stratum 0 (it is an EDB relation). The fixpoint bumps dept_count to max(stratum(emp)) + 1 = 1. If instead the aggregate reads a derived relation, such as

mid(K, V)       :- base(K, V)
mid_count(K, N) :- mid(K, V), count(V, N)

then mid is at stratum 0 (it depends only on EDB base), and mid_count is bumped to stratum 1. The aggregate test suite asserts exactly this ordering:

mid_stratum = Enum.find(ir.rules, fn r -> r.head.relation == "mid" end).stratum
count_stratum = Enum.find(ir.rules, fn r -> r.head.relation == "mid_count" end).stratum
assert count_stratum > mid_stratum

Because the aggregate is forced above its inputs, by the time the engine's eval_aggregate_rule/3 runs against the full view, every tuple it will join is already final. There is no half-materialized relation to worry about; the aggregate sees the answer set, not a work-in-progress.

propagate_strata/2 then re-runs the normal rule so that any relation depending on the aggregate inherits the bumped stratum. The outer fixpoint repeats until a full pass makes no change, with a fuel parameter to guarantee termination.

The Evaluation Pipeline

Aggregates take a different path through the engine. Engine.Evaluator.eval_rule_iteration/5 first asks aggregate_rule?/1 — a body that contains any {:constraint, %IR.Constraint{op: op}} with op in [:count, :sum, :min, :max] makes the rule an aggregate rule:

if aggregate_rule?(rule) do
  rule
  |> eval_aggregate_rule(full, ctx)
  |> MapSet.new()
  |> MapSet.difference(existing)
  |> MapSet.to_list()
else
  eval_normal_rule(rule, full, delta, old, ctx, existing)
end

Notice what aggregate evaluation does not see: no delta, no old. Aggregates stratify above their inputs, so all source facts are already in full. There is no semi-naive delta to consume — the rule fires once, against the final view of its inputs, and the result is whatever the reduction produces.

eval_aggregate_rule/3 then runs the engine's full pre-grouping pipeline, in this order:

defp eval_aggregate_rule(%IR.Rule{} = rule, full, ctx) do
  agg_constraint = find_aggregate(rule.body)

  joined     = join_positive_body([%{}], rule.body, full)
  filtered   = apply_constraints(rule.body, joined, ctx)
  with_cbs   = apply_callbacks(rule.body, filtered, ctx)
  bindings   = apply_negation(rule.body, with_cbs, full)

  %IR.Constraint{op: op, left: {:var, input_var}, result: {:var, result_var}} =
    agg_constraint

  group_vars = aggregate_group_vars(rule.head, result_var)

  bindings
  |> Aggregate.group_and_reduce(group_vars, op, input_var, result_var)
  |> Enum.map(&Join.project(rule.head, &1))
end

The pipeline is:

1. Join — join_positive_body/3 starts from [%{}] and folds each positive body atom against the full view, producing every binding consistent with the body.

2. Filter — apply_constraints/3 runs every non-aggregate constraint per binding. Aggregate constraints are explicitly skipped here:

   constraints =
     for {:constraint, %IR.Constraint{op: op} = c} <- body,
         op not in [:count, :sum, :min, :max],
         do: c

   This is why a rule can combine gte(S, 60) with count(E, N) — the comparison filters bindings down to passing scores before the count runs. The aggregate test exercises exactly this:

   Rule.new(
     Atom.new("passing_count", [Term.var("D"), Term.var("N")]),
     [{:positive, Atom.new("score", [Term.var("E"), Term.var("D"), Term.var("S")])}],
     [Constraint.gte(Term.var("S"), Term.const(60)),
      Constraint.count(Term.var("E"), Term.var("N"))]
   )
   # => {:eng, 2}  (alice with 90 and carol with 75 pass; bob with 40 is filtered)

3. Callbacks — apply_callbacks/3 runs any BEAM predicate literals, the same way as for non-aggregate rules.

4. Negation — apply_negation/3 filters bindings whose negated body atoms are contradicted by a tuple in full, exactly as in the negation pipeline.

5. Group and reduce — Aggregate.group_and_reduce/5 buckets the surviving bindings by group_vars and reduces each bucket's input_var values with op, producing one extended binding per non-empty group.

6. Project — Join.project/2 extracts the head's variables from each extended binding into the final head tuple.

The ordering matters: filtering and negation happen before grouping, so the aggregate sees only the bindings that survived all per-row predicates. There is no "HAVING" clause — a filter after the aggregate would need the result variable to be in scope, which means a follow-up rule reading the aggregate relation. Stratification makes that composition natural: the aggregate relation lives in a higher stratum, so a subsequent rule in an even higher stratum can filter on its results.

Putting It Together: A Full Example

The test suite's count_program/0 is a complete, runnable specification of count:

defp count_program do
  Program.new()
  |> Program.add_relation("emp", [:atom, :atom])
  |> Program.add_relation("dept_count", [:atom, :integer])
  |> Program.add_fact("emp", [:alice, :eng])
  |> Program.add_fact("emp", [:bob, :eng])
  |> Program.add_fact("emp", [:carol, :ops])
  |> Program.add_rule(
    Rule.new(
      Atom.new("dept_count", [Term.var("D"), Term.var("N")]),
      [{:positive, Atom.new("emp", [Term.var("E"), Term.var("D")])}],
      [Constraint.count(Term.var("E"), Term.var("N"))]
    )
  )
end

After materialization:

{:ok, knowledge} = ExDatalog.materialize(count_program())
Knowledge.get(knowledge, "dept_count")
# => MapSet containing {:eng, 2} and {:ops, 1}

The same shape, with Constraint.sum/2, Constraint.min/2, or Constraint.max/2, produces dept_total, lowest, or highest respectively — only the reducer changes. The grouping, stratification, and pipeline are identical.

What Is Not Here Yet

• avg — would require a paired count and sum or rational arithmetic; deferred until integer arithmetic with div is deemed stable enough to define avg as div(sum, count).
• Multiple aggregates per rule — explicitly rejected today. Lifting the restriction means defining a multi-reducer grouping protocol; for now, split into separate rules.
• Aggregates in recursive rules — rejected by :aggregate_in_recursion. General "monotone aggregates" (e.g. min over a lattice) are a research topic and not on the near roadmap.
• HAVING-style post-aggregate filters — expressed today by chaining a second rule in a higher stratum that reads the aggregate relation; a dedicated syntax would be a convenience, not a capability gap.

Aggregates are the first feature in ExDatalog that requires the engine to break out of the per-binding loop and look at a whole set of bindings at once. The design keeps that break localized — one new code path in eval_aggregate_rule/3, one new module Constraints.Aggregate, and one new fixpoint in force_aggregate_strata/2 — while the rest of the engine, the validator, and the DSL continue to treat aggregates as just another kind of constraint.