feat(planner): Pass 1 offers HydraCms for grouped approximate counts

[zzylol]

Stack: Wave 2 chain: #599 → #601 → #600 → #604 → #606 → #603 → #605

Rebased on main d4869a7 (DF 54). At this head cargo fmt --all --check, cargo clippy --workspace --all-targets -- -D warnings and cargo test --workspace pass (1642 passed, 0 failed, 23 ignored).

Why

#580 item F / W7: Hydra for count and point queries first. Part of #509. #600 added the executor's HydraCms kernel and stated the IR the planner must emit, but Stage 1 never offered Hydra. Stage 3's analytical check also ignored grouping, so it would have certified a shared grid with a per-group sketch's guarantee.

What

SELECT src_ip, COUNT(*) FROM flows GROUP BY src_ip at ε = 0.1, through plan_stages:

	candidates	Hydra
Before this PR	5: pass-through, Count acc, CMS, CountSketch, UnivMon	none
After this PR	6	P6: SummaryAgg{Sketch(Cms), SharedMultiSubpopulation{HydraCms}} → SummaryEstimate{PointCount{SampleValue, None}}. It passes Stage 3's accuracy check, is priced at 0.0157 (the exact accumulator costs 0.0073), compiles, and executes to a=3, b=2, c=1.

Selection does not change in these examples:

• Examples 1, 3a and 3b have no grouped count; their fixtures and DP tests are unchanged.
• Example 2's design queries go from 400 to 480 candidates; P2, the exact Count accumulator, is still selected.
• I did not tune costs.

How

• Pass 1 (logical_candidates::add_hydra_alternatives): for a single Count with an approximate target and by groups (not without), it emits executor: HydraCms kernel for count and point queries #600's contract:
  • item: a non-null Utf8/Int64/Bool column. A grouping column comes first; PromQL labels are nullable, so PromQL uses the series identity.
  • weight: Constant(1.0), with the UnitCount proof.
  • width/depth equal across the family and the grid.
• New axis: LocalLogicalTarget.groupings is parallel to alternatives, like windows and absorbs. Tumbling forms copy it, since HydraCms merges per executor: HydraCms kernel for count and point queries #600. The capability rule resets it; count targets are never keyed.
• Sizing: hydra_guarantee adds the grid's collision term to the inner sketch's error. So the inner sketch and the grid are each sized for ε/2 and δ/2 (default_size_params + default_hydra_params). At ε = δ = 0.01 that gives a 544 × 6 inner sketch in a 6 × 544 grid.
• Stage 3 accuracy: estimators::local_guarantee is now grouping-aware. For HydraCms it returns the existing pass1::grouping::hydra_guarantee over the inner Count-Min, with collision bound e/shared_columns and failure probability e^-shared_rows. HydraCountSketch and HydraKll get None, which means uncertified.
• Devtool: stage_pipeline labels these candidates "HydraCms".
• Legacy origin: pass1/grouping.rs HydraGroupingStrategy, whose guarantee composition is reused.

Review points

• Error basis: the collision term is relative to the whole input's weight, not one group's. This matches executor: HydraCms kernel for count and point queries #600's e·N·(1/shared_columns + 1/width). A small group's count can therefore be dominated by the error, whereas a per-group sketch's bare count is exact. The target is read the same way for both.
• Memory: the grid holds shared_rows × shared_columns cells, each the size of one per-group sketch. At ε = 0.01 that exceeds the executor's default memory limit, so the execution test uses ε = 0.1.

Tests

All four failed or did not exist on the parent:

• pass1_sql_coverage::grouped_count_offers_a_priced_executable_hydra_plan (written first; it failed with "a Hydra candidate is generated") checks the contract, Stage 3 pricing, compilation and execution.
• logical_candidates::grouped_approximate_count_offers_hydra checks that Hydra is offered only for by plus an approximate target, and checks the sizes.
• estimators::cms::hydra_guarantee_adds_the_shared_grid_term: a grid sized like one per-group sketch misses ε, and one sized for the ε/2 split meets it.
• stage_pipeline_selection::sql_hydra_count_dp_equals_exhaustive: the DP picks the exhaustive minimum over the 18 combinations.

Not in this PR

• Point frequency (PointCount{Named(item), Some(v)}): no AggIntent asks for one. A filtered count, COUNT(*) FILTER (WHERE service = 'checkout') GROUP BY job, could use it, so it belongs with Pass 2 sharing and Stage 1 coverage parity with the retired MajorPass #580 J (filtered aggregates).
• Top-k (W7 defers it), HydraCountSketch, and grids sized smaller than the inner sketch.

Gates

• fmt and clippy (-D warnings): pass.
• cargo test --workspace: 1,642 passed, 23 ignored. fix(planner): SQL COUNT(*) candidates count rows, not a sample value #603 had 1,638, and this PR adds 4 tests.
• Viewer tests: 29 ran, 6 skipped.

Stacked on #603. Integration with #604: fix: integrate with #604 sets latency_ms: None in the Hydra test's demand and checks the all-query-time physical candidate, since a logical candidate now has several. The numbers above are unchanged on the rebased chain.

🤖 Generated with Claude Code