feat(asap-aware-mapping): make CSE costing recurrence-aware

[zzylol]

Summary

Closes #287.

This PR makes CSE share-vs-recompute costing aware of how often workload roots execute, rather than relying only on structural consumer counts.

It carries repeating-query intervals and workload ingest characteristics into ASAP-aware cost context, aggregates recurrence across shared sub-DAGs, and exposes recurrence-aware variants of both local candidate ranking and whole-plan selection.

Scheduling, Prometheus rule-group behavior, runtime execution loops, temporal panes, retention, and watermark semantics remain out of scope.

Before vs. after: recurrence-aware E2E selection

The workload associates each DAG root with recurrence metadata: Repeating(interval) or OneShot. The relational DAG structure itself does not repeat or change. The planner propagates each root annotation through its reachable sub-DAG and builds a separate RecurrenceProfile for costing.

The E2E test intentionally uses a deterministic test cost model. Its numbers are fixed test inputs, not production measurements:

• workload ingest rate: 10 updates/second;
• maintaining the shared summary for one update: 1 cost unit/update;
• reading the maintained summary for one query evaluation: 1 cost unit/evaluation; and
• recomputing the result from raw input for one query evaluation: 50 cost units/evaluation.

The compared rates are therefore:

maintained cost/second =
    10 updates/second * 1 maintenance unit/update
  + evaluation_rate * 1 summary-read unit/evaluation

recompute cost/second =
    evaluation_rate * 50 raw-recompute units/evaluation

Before this PR, both workloads below have the same DAG and the same structural consumer count of two. Structural costing cannot distinguish them, so it gives both the same ranking. After this PR, only the root recurrence metadata differs:

Workload recurrence	Concrete after calculation	Selected CSE alternative
Two roots, each evaluated every 10 ms	Each root runs 1 / 0.010 s = 100 times/second, so the shared sub-DAG is evaluated 100 + 100 = 200 times/second. Maintaining it costs 10 updates/s * 1 = 10 units/s, and reading it costs 200 evaluations/s * 1 = 200 units/s, for 210 units/s total. Recomputing costs 200 evaluations/s * 50 = 10,000 units/s.	Share, because 210 < 10,000.
Two roots, each evaluated every 100 s	Each root runs 1 / 100 s = 0.01 times/second, so the shared sub-DAG is evaluated 0.01 + 0.01 = 0.02 times/second. Maintaining it still costs 10 updates/s * 1 = 10 units/s because ingest continues even when queries are infrequent; reading it costs 0.02 evaluations/s * 1 = 0.02 units/s, for 10.02 units/s total. Recomputing costs 0.02 evaluations/s * 50 = 1 unit/s.	Recompute independently, because 1 < 10.02.

The important crossover is that maintained state pays an ingest-driven cost continuously, whereas recomputation pays only when a query evaluates. Frequent queries amortize maintenance; infrequent queries may be cheaper to recompute.

plan_selection_uses_recurrence_profiles_for_cse_choices verifies both local candidate ordering and whole-plan selection end to end: the high-frequency workload selects CseShare, while the structurally identical low-frequency workload selects CseRecompute.

Cost model

The new recurrence model keeps steady-state rates and one-time costs in distinct types:

• UpdateRate: raw-data updates per second;
• EvaluationRate: consumer evaluations per second;
• CostRate: cost units per second;
• Horizon: seconds used to combine recurring and one-shot costs.

Maintained and recomputed alternatives are compared as:

maintained_cost_rate =
    update_rate * maintenance_cost_per_update
  + evaluation_rate * summary_read_cost

recompute_cost_rate =
    evaluation_rate * raw_recompute_cost

Mixed repeating and one-shot workloads require an explicit horizon:

maintained_total(H) =
    maintained_cost_rate * H
  + summary_build_cost
  + one_shot_count * summary_read_cost

recompute_total(H) =
    recompute_cost_rate * H
  + one_shot_count * raw_recompute_cost

Here maintained_cost_rate and recompute_cost_rate are the two recurring-rate expressions defined immediately above; there is no separate undefined recurring_cost_rate input.

CostModel gains overridable hooks for maintenance updates, summary reads, raw recomputation, and initial summary construction. Missing recurrence metadata delegates to the existing structural cse_share_decision, preserving previous behavior.

Core algorithm

Recurrence handling is not a new ReplacementStrategy and does not generate replacements. It augments the cost context after the existing workload-wide replacement search has discovered the legal alternatives.

workload roots
    |
    v
share_common_subtrees
    |  intern structurally identical legal sub-DAGs as shared Rcs
    v
discover_targets
    |  discover every sub-DAG and structural consumer_count
    v
run ReplacementStrategy implementations
    |  SketchAlgorithmStrategy, HydraGroupingStrategy,
    |  SharedSubtreeStrategy, AvgToSumOverCountStrategy,
    |  RollupStrategy, and TopKLimitReuseStrategy
    v
PlanSpace / MemoGroups
    |  retain every legal replacement candidate for each target
    v
recurrence_profiles
    |  propagate per-root Repeating(interval) or OneShot metadata
    |  through every reachable sub-DAG
    v
global_selection_with_recurrence
    |  select candidates using recurrence-aware CSE costing

In short, replacement strategies determine which alternatives are legal, while recurrence profiles help determine whether sharing or recomputing a common sub-DAG is cheaper for this workload. The recurrence-aware logic currently participates specifically in SharedSubtreeStrategy CSE selection between CseShare and CseRecompute; it does not uniformly rerank candidates produced by every other replacement strategy. Those non-CSE candidates retain their existing ranking behavior.

The recurrence-aware CSE decision then has two stages.

1. Build a recurrence profile for each candidate target

For every workload root r:

root contribution =
    Repeating(interval) -> evaluation rate 1000 / interval_ms
    OneShot             -> one-shot count 1

That contribution is propagated through the root's reachable query DAG. Edge and ancestor path multiplicities are preserved, so if an independently evaluated parent reaches a descendant twice, the descendant receives twice the root's rate or one-shot contribution.

For each target v, the contributions are folded into:

profile[v].evaluation_rate  = sum(root_rate * path_multiplicity)
profile[v].one_shot_count   = sum(root_one_shot * path_multiplicity)
profile[v].update_rate      = workload ingest rate, when v is root-reachable

This supports shared sub-DAGs consumed by roots with different intervals without exposing or constraining caller query identifiers.

2. Compare maintained state with raw recomputation

For a CSE share/recompute pair at target v:

maintained_cost_rate(v) =
    profile[v].update_rate     * cost_model.maintenance_cost_per_update(v)
  + profile[v].evaluation_rate * cost_model.summary_read_cost(v)

recompute_cost_rate(v) =
    profile[v].evaluation_rate * cost_model.raw_recompute_cost(v)

For a purely repeating workload, the planner selects:

Share                   if maintained_cost_rate <= recompute_cost_rate
RecomputeIndependently  otherwise

If one-shot consumers are present, the comparison uses an explicit horizon H whenever recurring work is also present:

maintained_total =
    maintained_cost_rate * H
  + summary_build_cost
  + one_shot_count * summary_read_cost

recompute_total =
    recompute_cost_rate * H
  + one_shot_count * raw_recompute_cost

The lower total wins. A one-shot-only profile uses the same formula with a zero recurring-rate term, so no caller-supplied horizon is needed. An empty profile falls back to the original structural consumer-count decision.

cost_sorted_with_recurrence uses this comparison to order the local CSE candidates. global_selection_with_recurrence applies it during whole-plan traversal while retaining the existing effective structural consumer accounting and selected-rewrite dependency propagation.

Workload integration

PlanSpace::recurrence_profiles associates each workload root with either a repeating interval or a one-shot execution and folds those contributions into every reachable target.

The implementation:

• combines different root intervals using sum(1 / interval_i);
• tracks one-shot consumers separately;
• carries structural path multiplicity transitively to descendants;
• applies workload update rate only to sites reachable from real roots;
• validates intervals, update rates, horizons, and root/profile cardinality; and
• retains target Rcs inside RecurrenceProfileMap, preventing stale pointer identity.

Recurrence is connected to real planning through:

• PlanSpace::cost_sorted_with_recurrence; and
• PlanSpace::global_selection_with_recurrence.

Both use recurrence-aware CSE decisions while preserving existing ranking behavior for non-CSE candidates. The original cost_sorted and global_selection APIs remain unchanged for callers without recurrence metadata.

Explanation output

RecurrenceCostExplanation reports:

• the selected alternative;
• maintained and recompute cost rates;
• absolute totals when a horizon applies;
• update and evaluation rates;
• one-shot consumer count;
• maintenance, read, recompute, and summary-build costs; and
• model/path provenance.

Structural fallback values are represented as None, rather than misleading numeric zeroes.

Tests

Coverage includes:

• mixed query intervals and shared sub-DAGs;
• high-frequency share versus low-frequency recompute using identical IR;
• recurrence-aware cost_sorted and global_selection end to end;
• one-shot-only and mixed one-shot/repeating workloads;
• transitive descendant multiplicity;
• unreachable rewrite-generated sites;
• invalid intervals, rates, horizons, and root counts;
• explicit rate/one-shot unit separation; and
• complete recurrence cost explanations, including summary build cost.

Verification:

cargo test --workspace
cargo clippy -p asap-aware-mapping --all-targets -- -D warnings
cargo fmt --all -- --check

All workspace tests pass; asap-aware-mapping has 174 passing tests.

[zzylol]

Question: Is subDAG recurrence considered?

[zzylol]

Yes. Each workload root’s recurrence (Repeating(interval) or OneShot) is propagated through every reachable sub-DAG. If a sub-DAG is shared by multiple roots, their recurrence contributions are aggregated—including path multiplicity—and used by the cost model.

The sub-DAG structure itself remains unchanged; recurrence is stored in a separate RecurrenceProfile, not directly in the DAG node.

[milindsrivastava1997]

Thanks, LGTM!