feat(types,asap-aware-mapping): add Hydra frequency-sketch grouping (#256)

[zzylol]

Closes #256. Part of #33.

What this adds

This PR makes physical grouping of grouped sketch state an explicit planning axis:

• GroupingStrategy::PerSubpopulationInstance keeps the existing/default behavior: one independent sketch per group key.
• GroupingStrategy::SharedMultiSubpopulation represents one Hydra structure shared across an aggregate's subpopulations.
• HydraCms and HydraCountSketch model the shared-grid frequency-sketch constructions backed by Hydra's collision/error analysis.
• HydraGroupingStrategy is registered in the default and custom-cost-model workload-search paths, so eligible grouped approximate-count aggregates receive Hydra alternatives alongside their independent sketch alternatives.

The grouping choice is recorded in two places for different purposes:

• SummaryExpr::SummaryAgg::grouping, next to reduction, gives planning access to the by/without context that determines legality.
• SummaryFamilyType::Sketch(SketchKind, GroupingStrategy) commits the physical layout to the state carried on the DAG edge, so independent and Hydra-backed states cannot be treated as merge-compatible merely because their inner sketch kind matches.

Implementation remains focused on which summary implementation satisfies an AggIntent; grouping is applied after selecting and constructing that exact implementation.

Candidate generation

For each bindable aggregate target, the strategy:

1. Requires a real subpopulation concept: a non-empty by(...) or a without(...) reduction. Global by [] and per-entity reductions are rejected.
2. Enumerates the normal, accuracy-sized summary implementations.
3. Retains only algorithms for which hydra_kind_for exposes a semantics-preserving Hydra construction. This is currently CMS and Count-Sketch.
4. Constructs the normal candidate without steering or recursively forcing the cost model.
5. Rebuilds its SummaryAgg with SharedMultiSubpopulation and updates the sketch-valued output schema with the same grouping layout.
6. Adds the Hydra candidate to the same memo group; the planner/cost model still decides whether sharing is worthwhile.

When Hydra helps

Hydra is most useful when a grouped workload has a very large and dynamic subpopulation space, so allocating a full independent sketch for every active group would dominate memory.

For example:

SELECT tenant_id, endpoint, COUNT(request_id)
FROM requests
GROUP BY tenant_id, endpoint;

In a multi-tenant service, hundreds of thousands of tenants may access thousands of endpoints. The active (tenant_id, endpoint) combinations form a large, sparse, long-tailed set: a few groups are hot, while many groups receive only occasional requests.

With independent grouping, every active combination needs its own CMS or Count-Sketch state:

memory ~ active_subpopulations x per_sketch_size

With Hydra grouping, all combinations hash into one shared w x r grid of frequency sketches:

memory ~ shared_rows x shared_columns x inner_sketch_size

This trades strict per-group isolation for bounded collision noise while avoiding per-subpopulation allocation overhead. It is attractive when group cardinality is high, the distribution is sparse or long-tailed, memory is constrained, and the query's accuracy target tolerates the modeled collision error.

Hydra is less attractive when there are only a few groups, when strict isolation is required for every group, or when the requested statistic has no modeled Hydra error guarantee. In particular, this PR does not automatically offer Hydra for quantile/KLL workloads.

How subpopulation cardinality is estimated

The core planner does not guess cardinality from column names or multiply table-wide distinct counts. Cardinality statistics are deployment-specific, so this PR adds the following CostModel hook:

fn estimated_subpopulation_count(
    &self,
    target: &QueryExpr,
) -> Option<usize> {
    None
}

A deployment overrides this method and inspects the aggregate target and its grouping keys -- for this example, (tenant_id, endpoint) -- then returns an estimated number of distinct active combinations from its catalog, optimizer statistics, telemetry, or another workload-specific source.

struct CatalogCostModel {
    estimated_tenant_endpoint_groups: usize,
}

impl CostModel for CatalogCostModel {
    // rank_candidates omitted

    fn estimated_subpopulation_count(
        &self,
        target: &QueryExpr,
    ) -> Option<usize> {
        match target {
            QueryExpr::Aggregate { reduction, .. }
                if reduction groups by tenant_id and endpoint =>
            {
                Some(self.estimated_tenant_endpoint_groups)
            }
            _ => None,
        }
    }
}

The example match above is illustrative pseudocode; production implementations resolve the grouping-key column references in target against their statistics catalog.

Once an estimate N is available, the default grouping cost compares:

independent_cost = N x inner_sketch_cells
hydra_cost       = shared_rows x shared_columns x inner_sketch_cells

Hydra ranks ahead when its shared-state estimate is smaller. For the regression fixture, N = 10,000 selects Hydra, while N = 10 selects independent sketches. If the hook returns None, both layouts remain legal and the planner preserves discovery order instead of fabricating a cardinality estimate.

Before / after IR

Example logical query:

SELECT tenant_id, endpoint, COUNT(request_id)
FROM requests
GROUP BY tenant_id, endpoint;

Assume the count intent has an approximate EpsilonDelta accuracy target. Before this PR, search can bind it only to independent per-subpopulation sketches.

Before -- one CMS instance per active (tenant_id, endpoint):

SummaryAgg {
    family: Sketch(
        Cms { width, depth },
        PerSubpopulationInstance,
    ),
    col: request_id,
    reduction: Reduce(by=[tenant_id, endpoint]),
    grouping: PerSubpopulationInstance,
    child: Scan(requests),
}

output state type:
Sketch(Cms { width, depth }, PerSubpopulationInstance)

Count-Sketch is another independent candidate with the same grouping shape.

After -- search retains the independent candidates and additionally offers a shared Hydra candidate:

SummaryAgg {
    family: Sketch(
        Cms { width, depth },
        SharedMultiSubpopulation {
            kind: HydraCms,
            params: HydraCms {
                width,
                depth,
                shared_rows,
                shared_columns,
            },
        },
    ),
    col: request_id,
    reduction: Reduce(by=[tenant_id, endpoint]),
    grouping: SharedMultiSubpopulation {
        kind: HydraCms,
        params: HydraCms {
            width,
            depth,
            shared_rows,
            shared_columns,
        },
    },
    child: Scan(requests),
}

output state type:
Sketch(Cms { width, depth }, SharedMultiSubpopulation { kind: HydraCms, ... })

The equivalent HydraCountSketch candidate is also emitted. The grouping value appears both on SummaryAgg and in the output state type deliberately: planning can inspect the former, while the latter prevents a downstream merge from confusing shared Hydra state with an independent CMS/Count-Sketch state.

This PR enumerates both alternatives and makes final ranking cardinality-aware. A deployment supplies CostModel::estimated_subpopulation_count(target): independent state is costed as estimated_subpopulations x inner_sketch_size, while Hydra state is costed as shared_rows x shared_columns x inner_sketch_size. The lower-cost layout ranks first. If no cardinality estimate is available, the planner preserves discovery order rather than inventing statistics.

Strategy candidates

Names used below:

• S: Scan(requests)
• I_cms: independent CMS summary, one instance per active (tenant_id, endpoint)
• I_cs: independent Count-Sketch summary, one instance per active (tenant_id, endpoint)
• H_cms: shared Hydra-CMS summary across all (tenant_id, endpoint) subpopulations
• H_cs: shared Hydra-Count-Sketch summary across all (tenant_id, endpoint) subpopulations
• E: count estimate/readout

Before this PR:

grouped approximate-count candidates:
  [Summary(I_cms over S), Summary(I_cs over S)]

After this PR:

grouped approximate-count candidates:
  existing independent strategies:
    [Summary(I_cms over S), Summary(I_cs over S)]

  HydraGroupingStrategy:
    [Summary(H_cms over S), Summary(H_cs over S)]

combined candidate space:
  [I_cms, I_cs, H_cms, H_cs]

HydraGroupingStrategy proposes the two shared alternatives. PlanSpace::cost_sorted then selects their order: a supplied group-cardinality estimate compares independent allocation with the fixed shared-grid state, while an absent estimate preserves the existing candidate order.

Complete selected-plan DAG

Every arrow below represents data flow from producer to result. These diagrams show one possible CMS-based selection; Count-Sketch has the same shape.

Before selecting Hydra

flowchart LR
    S["S: Scan(requests)"] --> I["I_cms: SummaryAgg CMS<br/>GROUP BY tenant_id, endpoint<br/>PerSubpopulationInstance"]
    I --> E["E: count estimate"]
    E --> Q["query result"]

    classDef result fill:#eef,stroke:#668;
    classDef summary fill:#efe,stroke:#484;
    classDef scan fill:#fee,stroke:#844;
    class Q result;
    class I,E summary;
    class S scan;

Loading

S -> I_cms -> E -> result

I_cms represents one independent CMS instance for each active (tenant_id, endpoint).

After selecting Hydra

flowchart LR
    S["S: Scan(requests)"] --> H["H_cms: SummaryAgg CMS<br/>GROUP BY tenant_id, endpoint<br/>SharedMultiSubpopulation / HydraCms"]
    H --> E["E: count estimate"]
    E --> Q["query result"]

    classDef result fill:#eef,stroke:#668;
    classDef summary fill:#efe,stroke:#484;
    classDef scan fill:#fee,stroke:#844;
    class Q result;
    class H,E summary;
    class S scan;

Loading

S -> H_cms -> E -> result

The relational DAG topology is unchanged; the selected SummaryAgg's physical state layout changes from many independent per-key sketches to one Hydra grid shared across the aggregate's subpopulations. That layout difference is now explicit in both the node and its output state type.

Accuracy boundary

HydraKll remains representable as an explicit experimental IR value, but it is not returned by hydra_kind_for and is never emitted by semantics-preserving workload search. The Hydra construction used here does not provide a quantile/KLL error guarantee, so grouped quantile workloads retain only their independent KLL and DDSketch alternatives.

HydraUnivMon is not modeled yet. It needs additional intent/category vocabulary for the statistics it serves.

The shared structure multiplexes subpopulations within one aggregate. This PR does not combine distinct query roots into one Hydra instance, and grouping alone does not make a sketch eligible for roll-up; sketch roll-up still requires explicit merge semantics and legality checks.

Validation

• cargo test -p asap-types -p asap-aware-mapping -p asap-integration-tests
• cargo clippy -p asap-types -p asap-aware-mapping -p asap-integration-tests --all-targets --all-features -- -D warnings
• cargo fmt --all -- --check

Coverage includes CMS/Count-Sketch Hydra discovery, exclusion of unproven HydraKll from automatic selection, custom-cost-model behavior, agreement between SummaryAgg::grouping and the sketch state type stored in the output schema, and a tenant/endpoint regression test proving that 10,000 estimated groups rank Hydra first while 10 estimated groups rank independent sketches first.

[zzylol]

Rebased onto feat/rollup-lattice-254 (base retargeted from feat/replacement-strategy-251 to that branch) and wired the TODO(#254) up to #254's now-landed is_legal_rollup_source:

• rollup::bindable_grouped_aggregate is now pub(crate) so grouping.rs reuses the exact same (by, intent, child) extraction rollup.rs itself uses.
• HydraGroupingStrategy gains with_siblings(...) (mirroring RollupStrategy's own sibling-set injection) and a composes_with_rollup check that calls is_legal_rollup_source directly.

Turns out the answer is legality-neutral, not a gate: every condition is_legal_rollup_source checks is a pre-ASAP fact (intent/grouping-keys/schema), decided before any post-ASAP GroupingStrategy is ever chosen — and a roll-up's readout is identical regardless of which GroupingStrategy built the state underneath it. So composes_with_rollup never removes a candidate, it only annotates the rationale when it holds. In practice it never holds today — Hydra is only modeled for SketchKind::Kll (reachable only via Quantile), and rollup_combinator only re-derives Sum/Min/Max/Count/Increase, never Quantile — so the two vocabularies don't overlap yet. Wired and tested anyway (10 new tests, including an end-to-end one pinning today's correct "stays silent" outcome) so it starts firing the moment either vocabulary grows.

All four checks still pass clean: cargo build --workspace --all-targets, cargo test --workspace (80/80 in asap-aware-mapping), cargo fmt --all -- --check, cargo clippy --workspace --all-targets --all-features -- -D warnings.

[milindsrivastava1997]

@zzylol this has changes related to #275 as well? Is that intended?

[zzylol]

Pushed a follow-up commit (4513a61) that scopes GroupingStrategy::SharedMultiSubpopulation to what the Hydra paper (Manousis et al., VLDB 2022) actually proves.

The paper's accuracy bound (Theorem 2) is over one specific construction: hash each subpopulation into a shared w×r grid of linear, mergeable frequency-vector sketches (its own heavy-hitter substrate is Count-Sketch; Count-Min Sketch is the same collision algebra). It's explicit that this construction excludes quantiles entirely (§4.3: "A statistic that cannot directly be estimated by Hydra-sketch is quantiles.").

Changes:

• Added HydraKind::HydraCms / HydraCountSketch, mapped from the existing SketchAlgorithm::Cms/CountSketch — direct instances of the paper's proven grid construction, with HydraParams carrying the paper's own (width, depth, shared_rows, shared_columns) knobs.
• Kept HydraKind::HydraKll, but its doc now says plainly that it's an unproven extension beyond the paper, not an equivalent guarantee over a different sketch.
• Generalized per_subpopulation_k(node) -> Option<u32> into per_subpopulation_sketch_params(node) -> Option<SketchParams>, and default_hydra_params now takes the whole SketchParams and matches per-HydraKind — a Hydra "sketch of sketches" is a framework over any mergeable inner sketch, not just one with a scalar k like KLL.
• HydraUnivMon (the paper's own actual named "Hydra-sketch": L layers of Count-Sketch plus a heavy-hitter heap, estimating entropy/L1/L2-norm/cardinality as one instance) is deliberately not added here — it needs new AggIntent/category vocabulary (entropy, L1-norm, L2-norm) this crate doesn't have yet. Left as a follow-up.

New tests cover the two new HydraKind mappings, default_hydra_params's per-kind dispatch (including a mismatched kind/params pair returning None), and a grouped Count intent now offering both HydraCms and HydraCountSketch candidates.

Verified: cargo build --workspace --all-targets, cargo test (asap-types, asap-aware-mapping, asap-integration-tests), cargo fmt --all -- --check, cargo clippy --all-targets --all-features -- -D warnings — all clean.

[zzylol]

Follow-up fix pushed in a548387 to make the tenant scenario reflected by planner behavior, not just documentation.

• Added CostModel::estimated_subpopulation_count(target) as the statistics hook.
• PlanSpace::cost_sorted now compares independent state (estimated groups × inner sketch size) with Hydra state (shared grid cells × inner sketch size).
• Added a tenant/endpoint regression test: 10,000 estimated groups rank Hydra first, while 10 estimated groups rank independent sketches first.
• If no cardinality estimate is available, candidate discovery order is preserved rather than assuming Hydra is beneficial.

Local validation: 133 asap-aware-mapping tests and 105 asap-types tests pass; formatting and clippy are clean. The PR remains planning/IR work—downstream physical Hydra execution is still outside this PR.