GROUP BY

Status: Native for non-windowed GROUP BY, both as the single-phase (immediate) plan and the two-phase mini-batch plan, over the aggregate/value-type combinations in Type support below — with a real, enumerated gap list, see Still falls back.

Flink picks between the two plan shapes itself, based on table.exec.mini-batch.enabled; this page covers both, since a query only accelerates when whichever shape Flink chose is fully native.

Single-phase

The immediate plan applies every input row to the keyed accumulator state and emits on every change — no batching. SUM/MIN/MAX/COUNT are native over DECIMAL (SUM → DECIMAL(38, s) with overflow → NULL; MIN/MAX → DECIMAL(p, s); carried as an i128 at scale s, matching Flink).

AVG is native: a running sum — widened to bigint for any integer input, double for float/double — plus the non-null count, emitting count == 0 ? NULL : sum / count cast back to the input type, with integer division truncating toward zero. This is a direct port of Flink's AvgAggFunction, over bigint/int/smallint/tinyint/float/double, and is retract-aware. Decimal AVG is native too: the sum reuses SUM's DECIMAL(38, s) accumulator, and the emit divides by the non-null count using Flink's exact decimal division — a 38-significant-digit quotient then HALF_UP rescale — reporting DECIMAL(38, max(6, s)), findAvgAggType's result type.

Idle-state TTL. table.exec.state.ttl runs natively here (and on the two-phase global merge below — the local half is transient and holds no TTL-eligible state). Semantics match Flink exactly: every stored value carries its last-write wall-clock timestamp (reads never refresh it), expiry happens at last_write + ttl inclusive, and expired state reads as absent and is deleted on read. The STATE_TTL hint overrides the job-wide retention on aggregates specifically.

Two-phase / mini-batch

The mini-batch plan splits the aggregate into four cooperating operators, all of them native:

1. MiniBatchAssigner emits the batching marker.
2. Local — a transient in-memory bundle, flushed on that marker, on a mini-batch.size trigger, or before each checkpoint. It holds no checkpointed state, mirroring Flink's own MapBundleOperator.
3. A keyed shuffle — a native columnar exchange — repartitions bundled partials by key.
4. Global reuses the single-phase group-aggregate operator to merge partials (COUNT merges as a SUM over partial counts).

Scope. SUM/MIN/MAX/COUNT over bigint/int/double value columns (Flink's SUM partial keeps the value's own type, so nothing is lost to widening in the split), and AVG over the full single-phase numeric set — bigint/int/smallint/tinyint/float/double. An AVG spans two positional partials: the widened running sum (bigint for integer inputs, double for float/double) plus the bigint non-null count. The local runs these as a widened-sum state and a COUNT over the same column; the global folds the pre-summed pair into the ordinary AVG state (the count partial bumps the non-null count), so the final divide/truncate/cast-back — including the cast back to a narrow integer or float result — is byte-identical to the single-phase AVG.

Decimal SUM/MIN/MAX/AVG carry through the split too: SUM's partial is the i128 running sum as DECIMAL(38, s) (a bundle overflow emits NULL and latches the merged AVG NULL, skipped by the SUM merge — the host's own null-propagation), MIN/MAX partials keep DECIMAL(p, s) through the extremes multiset, and AVG merges the (DECIMAL(38, s), bigint) pair into the exact division emit.

Both mini-batch assigner modes are native: proc-time (markers generated from the clock) and row-time (upstream event-time watermarks filtered to the mini-batch interval — a pure function of the input watermarks, so results stay deterministic).

Distinct aggregates ride the split natively in the default (no-split) plan: the local's bundle set travels as a trailing view column — its distinct (value, count) entries as a list of structs, the Arrow form of Flink's serialized MapView partial — and the global folds the entries into its per-key distinct state with multiplicities, so a value repeating across bundles counts once. Scope: COUNT(DISTINCT) over bigint/int/smallint/tinyint/float/double/string/decimal, SUM(DISTINCT) over bigint/int (the merge folds in set-iteration order, so order-sensitive float/double sums stay on the host).

Per-aggregate FILTER (WHERE …) rides the split too, on plain and distinct aggregates alike: the predicate is a boolean column the local gates every fold on, so the merge stays filter-blind. Filtered distinct instances each get their own native view/set per (args, filter) pair — the same final output as Flink's shared bitmask view, since a filtered distinct is an unfiltered distinct over the filtered row subset.

A retracting local input (the aggregate consumes another aggregate's changelog — Nexmark q4's shape) is native for COUNT and AVG only — their accumulators are layout-invariant under retraction. The local subtracts -U/-D rows, and the appended (or reused) count1 COUNT(*) partial drives per-key liveness in the global (-D and state drop when the merged count reaches zero, Flink's RecordCounter semantics).

Checkpointing. The durable global state stays as a Rust hot map but checkpoints through Flink's raw keyed state: each non-empty key group gets its own snapshot payload, and a rescaled task restores exactly the payloads assigned to its new key-group range, using the same BinaryRow hash/key-group calculation as the native exchange. See the Paimon backend for the persistent-state-backend angle on this same raw-keyed-state layout.

Still falling back, specific to the two-phase split: the opt-in distinct-agg.split.enabled incremental chain (a deliberate non-goal — see Unsupported operators), MIN/MAX/AVG over DISTINCT, smallint/tinyint/float SUM/MIN/MAX partials, and — under a retracting input — any aggregate other than COUNT/AVG (Flink's SUM/MIN/MAX retract variants declare extra accumulator fields, and a monotonicity-exempt MIN/MAX ignores retractions in ways the native fold would not) plus DISTINCT (its view value switches to per-filter live counts under retraction).

Type support

The matcher only accelerates (aggregate, value-type) pairs where DataFusion's native arithmetic agrees byte-for-byte with Flink's — this table is that guardrail; anything marked ✗ falls back.

value type	SUM	AVG	MIN	MAX	COUNT
BIGINT	✓	✓ ¹	✓	✓	✓
INT	✓ ²	✓ ¹	✓	✓	✓
SMALLINT / TINYINT	✓ ²	✓ ¹	✓	✓	✓
DOUBLE	✓	✓	✓	✓	✓
FLOAT (REAL)	✓ ³	✓ ³	✓	✓	✓
DECIMAL	✓ ⁴	✓ ⁴	✓	✓	✓
CHAR / VARCHAR	✗	✗	✓ ⁵	✓ ⁵	✓

¹ Integer AVG diverges from DataFusion's native Float64 average; a custom accumulator sums in int64 and truncates the cast back to the input integer type, matching Flink's AvgAggFunction.

² Integer SUM (INT/SMALLINT/TINYINT) uses a custom wrapping accumulator that keeps the narrow input type and wraps at that type's width on every step, instead of DataFusion's widening sum — the host's exact "store the running sum in the input type, cast back each step" semantics, pinned by an overflow-boundary parity test.

³ SUM/AVG over FLOAT use custom accumulators for host-exact precision: SUM accumulates in 4-byte float (rounding every step) rather than DataFusion's widening double sum; AVG sums in double and narrows the quotient to float, as Flink's FloatAvgAggFunction does. Both fold rows in the same order as the host, so results are bit-identical.

⁴ DECIMAL carries type-preserving MIN/MAX/COUNT over the column's own precision/scale, SUM as an i128 running sum reported as DECIMAL(38, s), and AVG as that sum divided by the non-null count with Flink's exact decimal division. Overflow mirrors Flink's buffer shapes exactly: SUM's buffer is the nullable sum alone, so an overflow past DECIMAL(38, s) goes NULL and the next value resets it (no sticky latch) and the merge skips a NULL partial; AVG's (sum, count) buffer null-propagates instead, so its overflow is sticky. Both are pinned at the overflow boundary by parity tests.

⁵ String MIN/MAX compare byte-lexicographically, matching Flink's BinaryStringData common binary comparison path. The one place this can differ from Flink is its separate materialized-Java-object path for supplementary-plane characters, which this native comparison does not replicate.

Grouping keys admit bigint/int/string/boolean/date/timestamp/decimal; multiple value columns of different types are each read independently (e.g. SUM(a), SUM(b) over columns of different types both accelerate). COUNT(*) reads a synthesized non-null column so it counts every row, including alongside value aggregates.

Still falls back

Both the single-phase gate and, for the two-phase plan, both halves independently must clear their own matcher before the query accelerates — one operator staying on the host drags the whole query back via the all-or-nothing island rule.

Single-phase / either two-phase half in common:

• A UDAF (no native path for arbitrary user aggregation logic).
• AVG/SUM/MIN/MAX over a value type outside Type support's ✓ set.
• AVG(DISTINCT) — the only non-native DISTINCT form. (COUNT(DISTINCT x) keeps a per-key value set; SUM(DISTINCT x) adds a running sum folded as values enter/leave it; MIN/MAX (DISTINCT) run as their plain, multiplicity-blind forms.)
• An approximate aggregate.
• An unsupported grouping-key or value column type.

Local group aggregate (two-phase local half) only:

• Any aggregate other than SUM/MIN/MAX/COUNT/AVG.
• A SUM/MIN/MAX value type outside bigint/int/double/decimal (MIN/MAX also admit a string, merged byte-lexicographically on both halves), or an AVG value type outside bigint/int/smallint/tinyint/float/double/decimal.
• A COUNT(DISTINCT) value type outside bigint/int/smallint/tinyint/float/double/string/decimal, or a SUM(DISTINCT) value outside bigint/int; MIN/MAX/AVG over DISTINCT.
• A partial whose declared type differs from what the native side emits — defensive only, not reachable from Flink's own planner.
• A retracting input with any aggregate other than plain COUNT/AVG.

Global group aggregate (two-phase merge) only:

• Any merge other than SUM/MIN/MAX/COUNT/AVG.
• A partial column outside bigint/int/double/decimal (strings allowed under MIN/MAX).
• An AVG whose partial pair isn't (bigint, bigint) for an integer average, (double, bigint) for float/double, or (decimal(38, s), bigint) for decimal.
• A distinct merge outside the local half's COUNT/SUM(DISTINCT) scope.
• A retracting merge with any aggregate other than plain COUNT/AVG (those merge natively, the count1 partial driving per-key liveness).
• An unsupported grouping-key or output column type.