feat(core)!: collapse L3 grouping — remove Partition node, unify on GroupKeys (#12, #13)

[zzylol]

Summary

Collapses L3 grouping to one representation, closing #12 and tying off the #13 discussion. Grouping had two problems: a Partition node that duplicated Aggregate.by, and — once that was resolved — grouping keys living in three unrelated Vec<ColumnId> fields. This PR fixes both.

End state — grouping has one home per concept, all sharing one type:

concept	home	type
reducing GROUP BY	Aggregate.by	GroupKeys
per-group ranking (topk by, bottomk)	Sort.partition_by	GroupKeys
per-group window (OVER (PARTITION BY …))	WindowFunc.partition_by	GroupKeys
parallel / sharding split	L5 stage allocator	— (physical, not in the IR)

Commits

1. docs(core) — reframe: Partition is split-without-reduce (PARTITION BY), not a 2nd GROUP BY; reconcile the contradictory design.md text; add a regression test pinning that reducing GROUP BY → Aggregate.by.
2. feat(core)! — remove the L3 (and L2) Partition node + PartitionKeys. The sole producer was generic topk by (…); its grouping is semantic (changes which rows rank within which group), so it moves onto the operator that consumes it — a new Sort.partition_by — rather than an ambiguous standalone node. The windowed reduction beneath stays label-preserving (by: []).
3. refactor(core) — unify the three grouping-key fields into a shared GroupKeys newtype over Vec<ColumnId>.

Algebra changes

New type (L3, query_expr.rs):

#[serde(transparent)]
pub struct GroupKeys(pub Vec<ColumnId>);   // + Deref<[ColumnId]>, IntoIterator,
                                           //   From<Vec<…>>, FromIterator, PartialEq<Vec<…>>

L3 QueryExpr nodes (query_expr.rs):

// changed — field type Vec<ColumnId> → GroupKeys
Aggregate  { by: GroupKeys, aggs: Vec<AggIntent>, output_names, having, child }
// changed — NEW field `partition_by` (per-group ranking; row-preserving)
Sort       { keys: Vec<SortKey>, partition_by: GroupKeys, child }
// changed — field type Vec<ColumnId> → GroupKeys
WindowFunc { func, args, partition_by: GroupKeys, order_by, output_name, child }

// REMOVED
- Partition { keys: PartitionKeys, child }
- enum PartitionKeys { By(Vec<String>), Without(Vec<String>) }

L2 relational::QueryExpr (relational.rs):

// changed — NEW field `partition_by` (name-based; resolved to GroupKeys at L3)
Sort { keys: Vec<L2SortKey>, partition_by: Vec<ColumnRef>, input }
// REMOVED
- Partition { keys: PartitionKeys, input }

New lowering error (lower.rs):

// guards the now-unreachable "group keys on a per-series windowed reduction"
// shape (generic topk routes its grouping to Sort.partition_by instead)
ConvertError::WindowedReductionKeys

Why a node was the wrong fix (and a shared type is the right one)

A standalone group/partition node is exactly the Partition we removed: with nothing consuming it, it's a row-preserving pass-through whose meaning (semantic grouping vs physical sharding) is ambiguous — and grouping's semantics are defined by the consumer (reduce vs rank vs window), so the keys belong on the operator. The real duplication (keys in three places) is fixed by a shared parameter type, not a node.

GroupKeys and positional ColumnId

GroupKeys(pub Vec<ColumnId>) is a transparent newtype — grouping stays positional (an index into the Scan schema), unchanged. #[serde(transparent)] keeps it wire-compatible with the bare arrays it replaces, and Deref<[ColumnId]> + IntoIterator for &GroupKeys + From<Vec<…>> + PartialEq<Vec<…>> keep all read/iterate/compare/construct sites working — hence no behavior change.

Ties off #13

AggIntent::TopK stays a first-class intent (it's the sketchable heavy-hitter operator, distinct from exact Sort + Limit). Its grouping rides on Aggregate.by; generic topk/bottomk grouping rides on Sort.partition_by — both now GroupKeys. So every grouping concept has one spelling, and the lowering remains one-canonical-form-per-query.

Compatibility

Breaking IR change (feat!): QueryExpr::Partition / PartitionKeys are gone, Sort gains partition_by, and the three grouping fields are now GroupKeys. partition_by is #[serde(default)] and GroupKeys is #[serde(transparent)], so older serialized forms still deserialize. No front end emitted Partition directly (converter-only fallback), so PromQL/SQL surface behavior is unchanged — topk by now lowers to Limit{Sort{partition_by,…}}.

Test plan

• cargo build — clean, no warnings
• cargo test -p asap-control-core -p asap-control-lower -p asap-e2e — 193 tests pass, incl. the rewritten topk_over_avg_is_generic_sort_limit, new generic_topk_grouping_lowers_to_sort_partition_by, reducing_group_by_lowers_to_aggregate_by, and the SQL count_ranked_topk_is_heavy_hitter
• swept the workspace — no remaining Partition / PartitionKeys references

Closes #12. Resolves the #13 discussion (closed: keep TopK).

🤖 Generated with Claude Code

[zzylol]

Answers to the review questions on this PR:

1. "Partition is removed, and now Sort has an optional partition key?" Yes. The L3 Partition node is gone; Sort now carries partition_by: GroupKeys (crates/core/src/intent_algebra/query_expr.rs) — empty = global order-by, non-empty = rank-within-group. GroupKeys is the single shared spelling for Aggregate.by, Sort.partition_by, and WindowFunc.partition_by.

2. "What about aggregate and group by?" Reducing GROUP BY stays on Aggregate.by — the γ+α operator that collapses N→1-per-group and rewrites/freezes the schema. That's structurally different from Sort.partition_by / WindowFunc.partition_by, which are row-preserving (rank/window without reducing). All three share GroupKeys. Heavy-hitter AggIntent::TopK carries no keys of its own — its grouping rides on the enclosing Aggregate.by.

3. "Can we represent topk(3, http_requests_total) by (job) (top-3 per job)?" The L3 shape — Sort { keys:[value desc], partition_by:[job] } → Limit{3} — is exactly what Sort.partition_by was designed for, no Partition node needed. But the lowering had a bug for the bare-selector case: it inserted an implicit cross-series Sum that collapsed the job label before partition_by could resolve it (and PromQL topk ranks raw samples, it doesn't sum).

Filed as #30 and fixed in #31 — topk(3, http_requests_total) by (job) now lowers to Limit{3} → Sort{value desc, partition_by:[job]} → Scan. (Range-function args like topk(3, avg_over_time(m[5m])) by (job) already worked.)