[zzylol]

Why

plan_installable returns plans its own executor cannot compile. For avg by (region) (quantile_over_time(0.99, data[100s])) with raw data retained, Stage 3 selects the exact plan, which reduces each series' raw samples with a per-entity Aggregate. The executor runs a per-entity aggregate only for distinct, L2 and entropy, so compiling the plan fails with per-entity aggregate requires an explicit entity binding. A runtime using this plan then has to forward every evaluation of these queries elsewhere.

What

• The executor runs an exact per-entity Aggregate of quantile/sum/avg/min/max/count_over_time, as it already did for distinct/l2/entropy_over_time.
• RawSampleSelector::window_batch drops Prometheus stale markers, which Remote Write carries when a series ends, so no range function counts one as a sample. An instant selector whose latest sample is a stale marker returns no row for that series, as in PromQL. An ordinary NaN is still a sample.
• Stage 3 rejects any candidate holding a per-entity Aggregate that the executor cannot run (for example exact stddev_over_time), with the new cause PerEntityAggregateNotExecutable. Such a candidate is never selected.
• A grouped PromQL avg built from exact sum and count now compiles. The issue's query needs this fix too: its avg by (region) failed in the executor's PromQL binary operator, because the count was Int64.

How

• One list of kernels. agg_has_exact_per_entity_kernel (in asap-types) lists the intents the executor's per-entity path runs. The executor refuses every other intent with this predicate, and Stage 3 rejects with it. Plan selection does not depend on the executor, so the list sits in the shared types crate, and the two sides cannot drift.
• Executor. The per-entity branch maps Count/Sum/Avg/Min/Max/Quantile to the existing columnar Operator::aggregate reductions. It groups by every column except the value and time columns (the series identity and labels), as the distinct path does. No new operator is added: grouping uses chunked_group_ids, memory is charged per chunk, and cooperative checkpoints are already in columnar.rs. The existing projection casts an Int64 count to the Float64 sample, and range functions drop __name__.
• Stage 3. per_entity_aggregate_violation scans the candidate's physical DAG. It accepts a per-entity Aggregate only with one measure that has a kernel, no measure filter and no HAVING, which is exactly what the executor's branch accepts. The check runs before the accuracy check.
• Stale markers. They are dropped once, where samples enter a plan (RawSampleSelector::window_batch), so every consumer of a window gets clean samples: the new per-series path, exact accumulators and sketch builds. The instant selector picks each series' latest sample first and then drops it if it is a stale marker, so an ended series is absent rather than falling back to an older sample. This matches the precompute boundary's contract that stale markers are not samples. STALE_MARKER keeps a single definition.
• Grouped avg. The PromQL sum/count avg casts a grouped count to Float64 (last_as_float64), as it already did for a non-Float64 sum.

Key code interfaces added / modified

crates/types/src/ir/operator/agg_intent.rs

/// Whether the executor reduces each series' raw samples exactly for an
/// `Aggregate` of `op` with `Reduction::PerEntity`. Read by the executor
/// (to compile) and by Stage 3 (to reject); true for Count, Sum, Avg, Min,
/// Max, Quantile, Cardinality, FrequencyL2 and FrequencyEntropy.
pub fn agg_has_exact_per_entity_kernel(op: &AggIntent) -> bool; // new

crates/executor/src/samples.rs

/// The window's input rows. Its signature is unchanged; its contract now says
/// that a stale marker is never a row (a range window drops it, and an instant
/// selector whose latest sample is one has no row) and that an ordinary NaN is kept.
pub fn window_batch(&self, window_end_ms: i64, samples: &[Sample]) -> Result<Batch, Error>; // behavior changed

crates/plan-selection/src/lib.rs

/// Why Stage 3 did not select a candidate. The type stays; one variant is added.
pub enum RejectionCause {
    /// Its summaries, resized for the failure probabilities, do not rebuild.
    FailureProbabilityResizeFailed, // unchanged
    /// A query's accuracy target is not met.
    AccuracyTargetMissed, // unchanged
    /// A Count-Min reads update weights not proven non-negative.
    CountMinWeightsNotProvenNonNegative, // unchanged
    /// It reduces each series' raw samples with an exact aggregate the
    /// executor has no kernel for (e.g. `stddev_over_time`). A failed check:
    /// the request is unsatisfiable on this executor, not bad input.
    PerEntityAggregateNotExecutable, // new; JSON "per_entity_aggregate_not_executable"
    /// A query's output has no accuracy guarantee.
    NoAccuracyGuarantee, // unchanged
    /// It scans raw data the deployment does not retain.
    RawDataNotRetained, // unchanged
    /// A query's latency is over its bound.
    LatencyBoundExceeded, // unchanged
    /// Under cost model 3, its usage is over a cluster limit.
    OverClusterCapacity, // unchanged
    /// Costlier than the selected candidate.
    Costlier, // unchanged
    /// Slower than the selected candidate.
    Slower, // unchanged
    /// Uses more memory than the selected candidate.
    MoreMemory, // unchanged
    /// On the Pareto front but not its lowest-latency member.
    NotLowestLatencyOnParetoFront, // unchanged
    /// Dominated by another candidate.
    Dominated, // unchanged
    /// Ties with the selected candidate and comes later.
    TiedLostOnEnumerationOrder, // unchanged
}

Usage (the end-to-end test an_exact_per_series_quantile_runs_from_raw_samples):

// The user's workload: one exact query every minute over 3 series of
// `latency_ms` sampled every 10 s; `requirements: {}` asks for an exact answer.
let workload: PlanningWorkload = serde_json::from_str(r#"{"query_workload": {"language": "promql",
    "repeating_queries": [{"query": "avg by (region) (quantile_over_time(0.99, latency_ms[100s]))",
      "demand": {"fixed_interval": 60000}, "requirements": {},
      "predictability": {"predictable": {"known_at": null}}}]},
  "data_workload": {"arrival": {"ingesting": {"rows_per_second": 3.0}},
    "sources": {"latency_ms": {"series_count": 3, "sample_interval": 10000}}}}"#)?;
// A deployment that keeps raw samples (the default capabilities).
let models = PlanningModels::builtin().with_deployment_capabilities(&DeploymentCapabilities::default());
// Stages 1-3. Stage 3 checks each candidate with per_entity_aggregate_violation:
// this one's per-entity Aggregate is [Quantile { q: 0.99 }], which has a kernel,
// so it is kept and selected as the exact plan: no stored outputs, one raw window.
let plan = plan_installable(UserInput::new(&workload, FrontendInput::Promql { histograms: None }, models)).await?;
// The runtime feeds the window's raw samples (a: 1..=10, b: 10..=100 in east;
// c: constantly 5 in west) and compiles and runs the plan with the executor.
// It returns east = (9.91 + 99.1) / 2 = 54.505 and west = 5, the hand-computed
// exact answers.

For stddev_over_time(x[5m]) at an exact target, each candidate holding the per-entity StdDev aggregate is rejected with cause: per_entity_aggregate_not_executable and a reason such as node 2: the executor has no exact per-series aggregate [StdDev { col: None, population: true }] over raw samples.

Before this PR

plan_installable(avg by (region) (quantile_over_time(0.99, latency_ms[100s])))  -> Ok(exact plan)
executor compile                                                                -> Err("node 2: per-entity aggregate requires an explicit entity binding")

With the quantile fixed alone, the grouped avg then failed:

executor compile -> Err("node 7: invalid DAG: PromQL vector requires labels, time and one value")

Exact stddev_over_time and the other per-series reducers without a kernel were selected and then failed in the executor in the same way.

After this PR

• The plan compiles and runs, and returns {region="east"} 54.505 and {region="west"} 5, equal to the query-time reference.
• A plan the executor cannot run is never returned. When no runnable candidate is left, Stage 3 reports NoValidCandidate with the named cause, so a runtime learns at planning time rather than at compile time.

Verification

Regression tests. Each new test failed on the code before its fix and passes now.

• planner/tests/runtime_install.rs::an_exact_per_series_quantile_runs_from_raw_samples: the end-to-end case above. Before the fix it failed with both errors shown, one after the other.
• integration-tests/tests/per_series_exact_execution.rs, with rows fed through RawSampleSelector:
  • exact_per_series_range_functions_execute: sum/avg/min/max/count_over_time and quantile_over_time(0.5, …) over two series, against hand-computed values, with no __name__ in the output.
  • a_stale_marker_is_not_a_sample: over 3, stale, 5 it expects count 2, sum 8, avg 4, min 3, max 5 and quantile(0) 3. Before the stale fix it got count 3.
  • a_nan_sample_is_a_sample: over 1, NaN, 3 it expects count 3, sum and avg NaN, min 1, max 3 and quantile(0.5) 1.
  • quantile_bounds_follow_promql: φ of 0, 1, −1 and 2 give 10, 40, −Inf and +Inf.
  • window_edges_and_an_empty_window: a sample exactly on the left edge is excluded and one on the right edge included. A series with no sample in the window has no answer.
• executor samples::tests::stale_markers_are_not_rows: a range window drops a stale marker and keeps an ordinary NaN. An instant selector whose latest sample is a stale marker returns no row.
• plan-selection: a_per_entity_aggregate_the_executor_cannot_run_is_rejected checks that stddev_over_time is rejected with the named cause and never selected. an_executable_per_entity_aggregate_is_not_rejected checks that the exact quantile is not rejected.

Commands run at the head commit. Each was run with CARGO_PROFILE_DEV_DEBUG=0 CARGO_PROFILE_TEST_DEBUG=0 CARGO_INCREMENTAL=0, which only reduces disk use. All passed.

• cargo test --workspace --locked: exit 0, 150 test binaries, 0 failures.
• cargo fmt --all -- --check
• cargo clippy -p asap-executor -p asap-integration-tests --all-targets --all-features --locked -- -D warnings, for the crates touched by the stale-marker commits.
• cargo clippy -p asap-types -p asap-executor -p asap-plan-selection -p asap-logical-asap-optimization -p asap-planner -p asap-integration-tests -p asap-devtools --all-targets --all-features --locked -- -D warnings, run at 05a5c34. No later commit touches crates outside the line above.
• cargo check -p asap-planner --tests --locked

Not run:

• Performance measurements. The change reuses the existing columnar aggregate and adds one pass over the samples to drop stale markers.
• The DAG viewer's Python tests. The only viewer change adds one string to a list.

Review. An independent review was done at 05a5c34. Its Medium finding (stale markers) and Low finding (edge-case tests) are addressed in 1eb5928 and 511f010.

Closes #770