Stage 1 whole-expression top-k heap keeps outer partition keys indexed against the absorbed aggregate #639

Problem

For a nested PromQL top-k such as

topk by(job)(2, sum by(service, job)(rate(m[1m])))

Stage 1 offers a whole-expression heap realization (Count-Min + heap and CountSketch + heap): one heap sketch over the raw rate rows that absorbs the inner grouped sum. That SummaryAgg keeps the outer top-k's Reduction, whose key indices point into the inner aggregate's output schema. The heap actually reads the raw rate rows ([ts, value, job, service]), so by(job) is read as key 0, which is ts. The result is partitioned by timestamp and loses job.

Repro

This goes through stage1_candidates (enumerate_local_logical_candidates + compose_logical_candidate), compile_physical_asap_dag, the executor's compile/instantiate, and then execution at both ingestion and query phases. It is the test planner_weighted_topk_binds_at_either_deployment_phase in crates/executor/tests/weighted_topk_binding.rs, which is currently #[ignore]d with this reason:

cargo test -p asap-executor --test weighted_topk_binding planner_weighted -- --ignored

Input: 7 rate rows, 4 with job="api" and 3 with job="batch". Expected output: 2 items per job, 4 rows, scores [0.3125, 0.375, 80, 100].

Actual:

panicked at crates/executor/tests/weighted_topk_binding.rs:127:9:
assertion `left == right` failed
  left: 2
 right: 4

All rows share one ts, so a single partition comes back with the global top 2, and job is not in the result.

A Pass 1 check of the same plan shows the heap's partition key resolving to ["ts"] against the heap's child schema when it should be ["job"].

Root cause

crates/logical-optimizer/src/pass1/logical_candidates.rs (base stack/cleanup-11-docs, 64e122d):

• whole_expression_input (L318–366) swaps in the inner aggregate's input (realized.child) as the rows the heap reads, but returns no partitioning for those rows.
• realize uses that input as the SummaryAgg child (L650) but still builds the node with reduction: reduction.clone() (L686). That is the outer target's reduction, which is indexed against the absorbed aggregate's output.

Before it was removed in #635, the legacy search (construct_summary_agg in pass1/replacement.rs, see origin/stack/cleanup-8-port-legacy-callers) remapped the top-k partition keys. It resolved each key to its column in the aggregate's child, matched that column against the realized input's schema, and failed on a missing or ambiguous match.

Fix

Move the outer partition keys onto the heap's input by column identity. If they cannot be moved (a key is missing or ambiguous in the input, or the keys are given by without), don't offer the realization. Un-ignore the executor test and have it assert the per-job grouping. Example 1 (topk by (job) (10, sum_over_time(...))) must stay unchanged.