refactor(types)!: DataWorkload as arrival plus per-source facts; G bound per source

[zzylol]

Why

DataWorkload mixed facts about how data arrives with facts about each source, and gave every field its own Evidence<T>. One global input_cardinality stood for the series of every source, so a workload over two metrics could not say how many series each has, and Stage 3 bound a query's number of values G from data the query may not read. One global data_ingestion_interval stood for the sampling cadence of every metric, although metrics can be scraped at different intervals. Per-source facts (stream_norms, max_distinct_per_evaluation, metric_types) were already keyed by source, but in three separate maps.

What

DataWorkload now has two parts:

• arrival: DataArrival: AtRest, Ingesting { rows_per_second }, Mixed { rows_per_second } or Unknown.
• sources: BTreeMap<String, SourceFacts>, keyed by PromQL metric or SQL table. Each entry holds series_count (was the global input_cardinality, and must be positive), metric_type, sample_interval (was the global data_ingestion_interval), stream_norms and max_distinct_per_evaluation.

Every fact is declared. The planner has no clock, so freshness evidence is gone: a caller drops facts that no longer hold before planning.

• The PromQL and MetricsQL frontends give each bare instant selector the sample_interval of the metric it reads.
• A PromQL selector with no metric name, such as {job="a"}, is now rejected with selector without one exact metric name, as MetricsQL already did. Before, it scanned a source named "".
• Plan selection bounds G and the input row count by the series_count of the sources each query reads. Each branch of an or or union now counts, which fixes an unsound G for or.

Selected plans and costs for Examples 1, 2, 3a, 3b and 4 are unchanged: the stage_pipeline documents are byte-identical to main (see Verification).

How

• Mapping.

  • data_ingestion_interval → sources[metric].sample_interval.
  • ingestion_rate → arrival.{Ingesting,Mixed}.rows_per_second.
  • input_cardinality → sources[source].series_count.
  • stream_norms, max_distinct_per_evaluation and metric_types → the matching SourceFacts fields.

• Validation. DataWorkload::validate checks that the rate is finite and non-negative. For each source, it checks that sample_interval and max_distinct_per_evaluation are positive, and that the stream norms are finite and non-negative. It also rejects series_count: 0 with the new WorkloadError::ZeroSeriesCount. A source with zero series has nothing to plan for, and absent(m) over it still returns one row, which the bound would not count.

• Sample interval per metric (frontends). The PromQL frontend collects the metric names that a query's selectors read. It then looks up each one's interval with PlanningWorkload::sample_interval_of, and instant_source gives each bare instant selector its own metric's interval. MetricsQL does the same per MetricExpression. WorkloadError::MissingSampleInterval fires only when a bare instant selector's metric declares no interval, whatever the arrival. Range-only queries need no interval, so Example 4's at-rest Pattern A still plans.

  • The single-query lower_metricsql(query, accuracy, sample_interval) keeps its signature: its one interval holds for every metric.

• Selectors with no metric name (PromQL). vs_parts rejects a selector that has neither a metric name nor an equality __name__ matcher. A Source::TimeSeries names exactly one metric, so such a selector has no representation. collect_metric_names, and with it promql_metric_names, skips the selectors that lowering rejects.

  • 29 queries in the Prometheus testdata corpus use such selectors and lowered before. They are now counted as expected-unsupported (no_metric_name) in promql_corpus, which asserts that there are exactly 29. Examples: sum({idx="0"}), absent_over_time({job="grok"}[20m]), rate({env="1"}[10m]), {job="app-server"} == 80. The testdata floor goes from 1121 to 1097 lowered, which is main's 1126 minus these 29. The docs corpus and the other three corpora have none.

• G counts what the root can return (Stage 3, soundness). On main, G was the one global input_cardinality. Now series_read walks the root and computes a RowBound for each node: copies per source, plus constant rows. It memoizes shared nodes. Per-source counting needs this walk because of or:

  • rate(x[5m]) or sum without (instance) (rate(x[5m])) returns up to series(x) + groups values;
  • rate(x[5m]) or vector(0) returns one more.

  The RowBound of each node:

  • A Scan counts its source once.
  • A union (PromQL or, SQL UNION, Concat) adds its branches.
  • vector(s) adds one constant row and keeps the copies of the sources below it, so the bound never shrinks going up the tree. at_most_one_row now gives a vector(s) root G = 1.
  • VALUES contributes its rows.
  • Every other operator (fold, join, filter) takes the per-source maximum of its inputs, so a join or fold of one source still counts it once.

  The result is Σ copies × series_count + constant rows. It is None when the root reads no source, or when a source it reads declares no series_count. The same bound serves the input_row_count statistic.

• Pricing (Stage 3). On main, every node was priced with the global input_cardinality. Now the new priced_series prices each physical node with the sum of series_count over the sources it reads. Undeclared sources use the 1,000 default, as before. price() builds each node's producer list once from dag.edges, and reuses it for the inputs, the sources and SummaryMerge's states.

• One bound per candidate. accuracy_check computes series_read once per root, then passes it to values_per_result(root, series_read) and to bound_root_guarantee(root, series, data, models).

• No planning clock. The frontends' now_ms argument, Evidence<T> with its value_at, and Stage 3's freshness filter (value_at(ANY_TIME)) are gone. "Unbound" now means only that series_count is not declared, or that no source is read. DemandEstimate::is_fresh_at stays: recurrence and statistics read it with their own planning time.

• Deleted because no non-test code reads them:

  • WorkloadAccuracyEvidence (only re-exported).
  • Evidence<T> (with value, source, observed_at_ms, valid_for_ms and value_at).
  • EvidenceSource and DemandEstimate.source.
  • DemandEstimate.max_concurrency (no reader), and DemandEstimate.peak_rate, confidence and observation_window (only validated).
  • The types Confidence and ObservationWindow, which only those fields used.
  • WorkloadError::{UnavailableDataIngestionInterval, MissingPromqlDataWorkload, MissingDataIngestionInterval, AtRestWithPositiveIngestionRate}, and EmptyObservationWindow and InvalidConfidence, which checked the deleted fields.
  • devtools' declared<T>(value) -> Evidence<T>.

• devtools.

  • DataWorkload::sampled_every(metrics, interval) and promql_metric_names(query) let lower_promql_with_sample_interval and stage_pipeline --promql declare the sample interval on each metric that the queries read.
  • stage_pipeline's --interval-ms becomes --sample-interval-ms, the name that sketch_coverage, variant_coverage, analyze_corpora and show_logical_dag now use (they had --data-ingestion-interval-ms). The stage viewer's tools/dag-viewer/server.py passes the new flag.
  • The false "ingesting at 1,000 rows/s" claim is gone: the arrival stays Unknown, as on main.
  • shared_data_workload(arrival, source) gains the source parameter. It declares that source's series_count (1,000,000) and its 15 s sample_interval. Callers pass the new SHARED_ROWS_PER_SECOND (1,000,000 / 15) in an arriving DataArrival.
  • pattern_a(recurrence, arrival: DataArrival) keeps its signature. The arrival now carries the rate, so callers pass DataArrival::Mixed { rows_per_second: SHARED_ROWS_PER_SECOND } where they passed DataArrival::Mixed.

• Docs. Edited only where they describe the changed API:

  • docs/develop_docs/library-api.md: the PromQL lowering row and rules, the DataWorkload paragraph, its JSON, the defaults table, and the four examples' data workloads.
  • docs/design_docs/architecture/input-output-workflow.md: the data_workload rows, the flowchart node, and the DataWorkload section.
  • docs/develop_docs/pre-asap-ir.md: the TimeRangeKind::Instant line.
  • docs/design_docs/architecture/updated_interface_with_pluggable_optimization.md: the FrontendInput row.
  • docs/user_guide_docs/run-a-query.md: every --data-ingestion-interval-ms becomes --sample-interval-ms. The note on the flag now says that it is declared for every metric a query reads.
  • docs/user_guide_docs/stage-viewer.md: the --sample-interval-ms row.

Key code interfaces added / modified

crates/types/src/workload.rs

/// Facts about the data being queried: how it arrives, and what is known
/// about each source. Read by the frontends (sample intervals) and by
/// Stage 3 (rate, series counts, bounds). Every fact is declared. Type stays.
pub struct DataWorkload {
    /// How the data arrives (default: unknown).                    // was a unit enum
    pub arrival: DataArrival,
    /// Facts per source, keyed by PromQL metric or SQL table name
    /// (default: none); a source without an entry has no known facts. // new
    pub sources: BTreeMap<String, SourceFacts>,
    // removed: data_ingestion_interval, ingestion_rate, input_cardinality,
    //          stream_norms, max_distinct_per_evaluation, metric_types
}
impl DataWorkload {
    /// Data of unknown arrival whose `metrics` are each sampled every
    /// `sample_interval`, with no other facts: what PromQL/MetricsQL
    /// queries over them need to lower; used by devtools and tests.    // new
    pub fn sampled_every(metrics: impl IntoIterator<Item = String>, sample_interval: DurationMs) -> Self;
    /// Validates the rate and each source's series count, interval,
    /// norms and distinct bound.                                      // changed
    pub fn validate(&self) -> Result<(), WorkloadError>;
}

/// How the queried data arrives. JSON: "at_rest" |
/// {"ingesting": {"rows_per_second": ..}} | {"mixed": {..}} | "unknown".
/// Type stays; the arriving variants now carry the rate.
pub enum DataArrival {
    /// Every row is already stored; no ingestion-time builds.      // unchanged
    AtRest,
    /// Rows keep arriving while the queries run.                   // was ContinuouslyIngesting
    Ingesting {
        /// Rows (PromQL samples) per second over all sources; Stage 3
        /// prices scans and ingestion-time work with it.            // was DataWorkload::ingestion_rate
        rows_per_second: f64,
    },
    /// Stored history plus arriving rows.                           // was a unit variant
    Mixed {
        /// Rows per second over all sources, as for `Ingesting`.    // was DataWorkload::ingestion_rate
        rows_per_second: f64,
    },
    /// Not declared (default): no ingestion-time builds.            // unchanged
    Unknown,
}
impl DataArrival {
    /// Rows per second when ingesting or mixed; read by recurrence
    /// profiles, the replacement cost model and Stage 3 pricing.   // new
    pub fn rows_per_second(&self) -> Option<f64>;
}

/// What is known about one source (a PromQL metric or a SQL table).
/// Every fact is optional; unknown is never read as zero. Type is new.
pub struct SourceFacts {
    /// Series (PromQL) or groups of rows (SQL) the source holds
    /// (default: unknown; must be positive). Bounds G and the input row
    /// count of the queries that read it; pricing uses it (1,000 when
    /// unknown).                                                    // was DataWorkload::input_cardinality (global)
    pub series_count: Option<u64>,
    /// Declared Prometheus type (default: unknown); a counter proves
    /// samples non-negative. Never inferred from a name.            // was DataWorkload::metric_types[source]
    pub metric_type: Option<MetricType>,
    /// PromQL/MetricsQL only: how often each series of the metric is
    /// sampled (default: unknown). A bare instant selector over the metric
    /// looks back this far; without it, lowering one fails.         // was DataWorkload::data_ingestion_interval (global)
    pub sample_interval: Option<DurationMs>,
    /// Bounds on ‖f‖₁/‖f‖₂ (default: none) that turn Count-Min and Count
    /// Sketch bounds into absolute errors.                          // was DataWorkload::stream_norms[source]
    pub stream_norms: Option<StreamNorms>,
    /// Deployment-enforced bound on the distinct items per evaluation
    /// (default: none), for classic HLL and distinct-count UnivMon.  // was DataWorkload::max_distinct_per_evaluation[source]
    pub max_distinct_per_evaluation: Option<u64>,
}

// removed: pub struct Evidence<T> { value, source, observed_at_ms, valid_for_ms } and value_at
// removed: pub enum EvidenceSource { Declared, Observed, Derived, Unknown }

/// A demand forecast for a repeating query. Type stays.
pub struct DemandEstimate {
    /// Expected invocations per second; recurrence and statistics read it. // unchanged
    pub expected_rate: Rate,
    /// When the estimate was made; read by `is_fresh_at`.           // unchanged
    pub observed_at: Option<TimestampMs>,
    /// How long after `observed_at` it holds; read by `is_fresh_at`. // unchanged
    pub valid_for: Option<DurationMs>,
    // removed: source: EvidenceSource (no reader)
    // removed: peak_rate: Option<Rate> (only validated)
    // removed: max_concurrency: Option<u64> (no reader)
    // removed: confidence: Confidence (only validated)
    // removed: observation_window: ObservationWindow (only validated)
}

// removed: pub struct Confidence(pub f64) (only DemandEstimate.confidence used it)
// removed: pub struct ObservationWindow { start, end } (only DemandEstimate.observation_window used it)

impl PlanningWorkload {
    /// The sample interval the data workload declares for `metric`; read
    /// by the PromQL and MetricsQL frontends per instant selector.  // new
    pub fn sample_interval_of(&self, metric: &str) -> Option<DurationMs>;
    /// Validates the query and data workloads; a sample interval is
    /// checked only when a bare instant selector needs it.          // changed
    pub fn validate(&self) -> Result<(), WorkloadError>;
}

/// Why a planning workload is rejected. Every variant is bad input. Type stays.
pub enum WorkloadError {
    /// A PromQL/MetricsQL bare instant selector whose metric declares no
    /// `sample_interval`.                                           // new; was MissingDataIngestionInterval
    MissingSampleInterval,
    /// A source declares a sample interval of zero.                 // was ZeroDataIngestionInterval
    ZeroSampleInterval,
    /// A one-time query with zero invocations.                      // unchanged
    ZeroInvocations,
    /// A repeating query whose fixed interval is zero.              // unchanged
    ZeroRepetitionInterval,
    /// A scheduled repeating query with no timestamps.              // unchanged
    EmptySchedule,
    /// A negative or non-finite rate (demand or `rows_per_second`). // unchanged
    InvalidRate(f64),
    /// A negative or non-finite `max_latency_ms`.                   // unchanged
    InvalidLatency(f64),
    /// A negative or non-finite stream norm bound.                  // unchanged
    InvalidStreamNorm(f64),
    /// A `max_distinct_per_evaluation` of zero.                     // unchanged
    ZeroMaxDistinct,
    /// A source's `series_count` of zero.                           // new
    ZeroSeriesCount,
    // removed: UnavailableDataIngestionInterval (interval evidence expired; no clock)
    // removed: MissingPromqlDataWorkload (only selectors need an interval)
    // removed: AtRestWithPositiveIngestionRate (AtRest has no rate)
    // removed: EmptyObservationWindow, InvalidConfidence(f64) (their fields are gone)
}

crates/frontend-promql/src/lib.rs, crates/frontend-metricsql/src/lib.rs, crates/planner/src/lib.rs

/// Lower every PromQL entry; a bare instant selector reads the latest
/// sample within its metric's sample_interval, else MissingSampleInterval;
/// a selector with no metric name is an unsupported feature.
pub fn lower_promql_workload(workload: &PlanningWorkload)                      // was (workload, now_ms)
    -> Result<Vec<Rc<OperatorNode>>, PromqlError>;
/// As above, with a histogram catalog.
pub fn lower_promql_workload_with_histograms(workload: &PlanningWorkload, histograms: HistogramCatalog)
    -> Result<Vec<Rc<OperatorNode>>, PromqlError>;                              // was (.., now_ms)
/// As above, keeping scalar roots.
pub fn lower_promql_query_workload(workload: &PlanningWorkload)
    -> Result<Vec<QueryRoot>, PromqlError>;                                     // was (.., now_ms)
/// As above, with a histogram catalog.
pub fn lower_promql_query_workload_with_histograms(workload: &PlanningWorkload, histograms: HistogramCatalog)
    -> Result<Vec<QueryRoot>, PromqlError>;                                     // was (.., now_ms)
/// The metrics a query's selectors read (selectors lowering rejects, such
/// as one with no metric name, are skipped): the sources whose
/// sample_interval lowering looks up; used by devtools and tests to
/// declare them.                                                               // new
pub fn promql_metric_names(query: &str) -> Result<BTreeSet<String>, PromqlError>;

/// MetricsQL counterpart: each bare selector reads its metric's interval.
pub fn lower_metricsql_query_workload(workload: &PlanningWorkload)
    -> Result<Vec<QueryRoot>, MetricsqlError>;                                  // was (workload, now_ms)
/// One query whose metrics are all sampled every `sample_interval`.
pub fn lower_metricsql(query: &str, accuracy: AccuracyTarget, sample_interval: Duration)
    -> Result<Rc<OperatorNode>, MetricsqlError>;                                // unchanged (parameter was ingestion_interval)
/// As above, keeping a scalar root.
pub fn lower_metricsql_query(query: &str, accuracy: AccuracyTarget, sample_interval: Duration)
    -> Result<QueryRoot, MetricsqlError>;                                       // unchanged (parameter was ingestion_interval)

/// The lowering dependencies of one frontend. Type stays.
pub enum FrontendInput<'a> {
    /// The SQL catalog the queries read.                       // unchanged
    Sql { catalog: &'a SqlCatalog },
    /// An optional histogram catalog.                          // removed field: now_ms
    Promql { histograms: Option<HistogramCatalog> },
    /// No inputs.                                              // was Metricsql { now_ms }
    Metricsql,
}

crates/asap-aware-mapping/src/stage3_plan_selection.rs (private)

/// Upper bound on the rows one node and the operators below it see or
/// return per evaluation: copies of each source's series plus constant rows. // new
struct RowBound {
    /// How many times each source's series count is counted.               // new
    copies_per_source: BTreeMap<String, u64>,
    /// Rows from constant vectors and VALUES.                               // new
    constant_rows: u64,
}
/// RowBound of `node`: a scan counts its source once, a union (`or`,
/// UNION, Concat) adds its branches, `vector(s)` adds one constant row and
/// keeps its input's copies, any other operator takes the per-source
/// maximum of its inputs; memoized over shared nodes.                       // new
fn row_bound(node: &Rc<OperatorNode>, memo: &mut HashMap<*const OperatorNode, RowBound>) -> RowBound;
/// Σ copies × series_count + constant rows over `root`; None when it reads
/// no source or a source it reads has no series_count. Binds G and
/// input_row_count.                                                         // new
fn series_read(root: &QueryRoot, data: &DataWorkload) -> Option<u64>;
/// G from the root and its precomputed `series_read`.                      // was (root, data), read the global input_cardinality
fn values_per_result(root: &QueryRoot, series_read: Option<u64>) -> Option<u64>;
/// Whether `node` returns at most one row; now also true for `vector(s)`.  // changed
fn at_most_one_row(node: &OperatorNode) -> bool;
/// The root's guarantee with Stage 1's symbolic statistics bound.          // was (root, data, models)
fn bound_root_guarantee(root: &QueryRoot, series: Option<u64>, data: &DataWorkload, models: &PlanningModels<'_>)
    -> Option<AccuracyGuarantee>;
/// The series pricing assumes for a node reading `sources`: the sum of
/// their series_count, 1,000 for an undeclared one or for no source.       // new
fn priced_series(sources: &BTreeSet<&str>, data: &DataWorkload) -> u64;
// removed: const ANY_TIME (the freshness filter's "no planning time")

asap_aware_mapping::WorkloadAccuracyEvidence is removed (no non-test reader).

crates/devtools/src/planner_layering_examples.rs, crates/devtools/src/lib.rs, crates/devtools/src/bin/stage_pipeline.rs

/// Rows per second the #509 shared data arrives at while ingesting:
/// 1,000,000 series, each sampled every 15 s.                    // new
pub const SHARED_ROWS_PER_SECOND: f64 = 1_000_000.0 / 15.0;
/// The #509 shared data, arriving as `arrival` says: 1,000,000 series
/// of `source`, each sampled every 15 s.
pub fn shared_data_workload(arrival: DataArrival, source: &str) -> DataWorkload; // was (arrival: DataArrival)
/// Example 3, Pattern A over the shared data arriving as `arrival`.
pub fn pattern_a(recurrence: PatternARecurrence, arrival: DataArrival) -> PlanningWorkload; // unchanged signature; the arrival now carries the rate
/// The metric both Example 1 queries read.                       // new
pub const EXAMPLE1_METRIC: &str = "http_requests_total";
/// The metric Example 3's Patterns A and B read.                 // new
pub const PATTERN_METRIC: &str = "latency_ms";
// removed: pub fn declared<T>(value: T) -> Evidence<T>
/// Lower one PromQL query whose metrics are each sampled every interval_ms.
pub fn lower_promql_with_sample_interval(query: &str, accuracy: AccuracyTarget, interval_ms: u64)
    -> Result<Rc<OperatorNode>, PromqlError>;                      // was lower_promql_with_data_ingestion_interval

stage_pipeline --sample-interval-ms <u64> (was --interval-ms) is the sample interval of every metric the --promql queries read (default 15000).

Usage

Example 1, end to end; then a bare instant selector over two metrics; then a selector with no metric name. I ran this against this head as a scratch asap-devtools example. It printed:

P1 Some(7.94021)
lowering: invalid PromQL workload: a PromQL or MetricsQL instant selector needs its metric's sample_interval in data_workload.sources
[("TimeSeries { metric: \"http_requests_total\" }", 15000), ("TimeSeries { metric: \"up\" }", 30000)]
lowering: unsupported feature: selector without one exact metric name

use asap_devtools::planning_stages::{lower_promql_roots, run_planning_stages};
use asap_types::ir::{NonASAPOp, OperatorNode, QueryRoot};
use asap_types::workload::{DurationMs, PlanningWorkload, WorkloadError};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // json: what the user writes, Example 1's two panels and the data facts.
    // The data arrives at 1,000,000 series / 15 s; `http_requests_total` has
    // 1,000,000 series, is a counter, and is sampled every 15 s.
    let json = r#"{
      "query_workload": {"language": "promql", "query_batch": null, "repeating_queries": [
        {"query": "sum by (job) (rate(http_requests_total[1m]))",
         "demand": {"fixed_interval": 10000}, "requirements": {},
         "predictability": {"predictable": {"known_at": null}}},
        {"query": "topk by (job) (10, sum_over_time(http_requests_total[1m]))",
         "demand": {"fixed_interval": 10000},
         "requirements": {"accuracy": {"error": 0.01, "failure_probability": 0.001},
                          "max_latency_ms": 100.0},
         "predictability": {"predictable": {"known_at": null}}}]},
      "data_workload": {
        "arrival": {"ingesting": {"rows_per_second": 66666.66666666667}},
        "sources": {"http_requests_total":
          {"series_count": 1000000, "metric_type": "counter", "sample_interval": 15000}}}
    }"#;
    // workload: the planner input, parsed strictly (unknown fields are rejected).
    let workload: PlanningWorkload = serde_json::from_str(json)?;
    // Checks the rate is finite and >= 0, and each series_count and
    // sample_interval is > 0.
    workload.validate()?;
    // roots: Stage 0, the query-language frontends, one root per panel. Both
    // panels use explicit `[1m]` ranges, so neither reads a sample_interval.
    let roots = lower_promql_roots(&workload)?;
    // run: Stages 1-3. Stage 3 reads sources["http_requests_total"]: the
    // counter type proves Count-Min's weights non-negative, series_count
    // bounds G (the values each instant vector returns) and prices nodes over
    // the metric, and rows_per_second prices scans.
    let run = run_planning_stages(&workload, roots, None).map_err(String::from)?;
    // selection: the selected candidate and its cost model 1 cost per second.
    let selection = run.stage3_plan_selection?;
    println!(
        "{} {:?}",
        selection.selected, selection.costs[&selection.selected].cost_per_second
    );

    // bare: Example 1's workload, reduced below to one panel whose bare
    // instant selectors read `http_requests_total` and `up`.
    let mut bare = workload.clone();
    // panels: bare's repeating panels; we keep the first and replace its query.
    let panels = bare.query_workload.repeating_queries.as_mut().unwrap();
    panels.truncate(1);
    panels[0].query.0 = "sum by (job) (http_requests_total) or sum by (job) (up)".into();
    // `up` declares no sample_interval yet: lowering fails in Stage 0.
    // error: Stage 0's message, naming the missing sample_interval.
    let error = lower_promql_roots(&bare).unwrap_err();
    println!("{error}");
    assert!(error.contains(&WorkloadError::MissingSampleInterval.to_string()));
    // data: bare's data workload, where we declare `up` sampled every 30 s.
    let data = bare.data_workload.as_mut().unwrap();
    data.sources.entry("up".into()).or_default().sample_interval = Some(DurationMs(30_000));
    // roots: the panel's one root, now lowered; root: its operator DAG.
    let roots = lower_promql_roots(&bare)?;
    let QueryRoot::Operator(root) = &roots[0] else {
        panic!("an operator root")
    };
    // lookbacks: (metric, ms) for each instant selector; each looks back its
    // own metric's sample_interval.
    let mut lookbacks: Vec<_> = OperatorNode::reachable(root)
        .iter()
        .filter_map(|node| match node.non_asap()? {
            NonASAPOp::TimeRange { range, child, .. } => match child.non_asap()? {
                NonASAPOp::Scan { source, .. } => Some((format!("{source:?}"), range.as_millis())),
                _ => None,
            },
            _ => None,
        })
        .collect();
    lookbacks.sort();
    println!("{lookbacks:?}");

    // nameless: a selector with no metric name is rejected in Stage 0.
    let mut nameless = bare.clone();
    nameless.query_workload.repeating_queries.as_mut().unwrap()[0].query.0 =
        r#"sum({job="api"})"#.into();
    println!("{}", lower_promql_roots(&nameless).unwrap_err());
    Ok(())
}

If the panel had a whole-result target, Stage 3 would bound its G by 1,000,000 + series_count(up), because each branch of the or counts.

Before this PR

"data_workload": {"arrival": "continuously_ingesting",
  "data_ingestion_interval": {"value": 15000, "source": "declared", "observed_at_ms": null, "valid_for_ms": null},
  "ingestion_rate": {"value": 66666.7, "source": "declared", ...},
  "input_cardinality": {"value": 1000000, "source": "declared", ...},
  "metric_types": {"http_requests_total": "counter"}}

• One cardinality served every query and every node.
• One interval served every metric.
• PromQL workloads needed the interval up front, even when all their queries used explicit ranges.
• The frontends took a planning clock only to check the interval's freshness.
• {job="a"} lowered to a scan of a metric named "".

After this PR

"data_workload": {
  "arrival": {"ingesting": {"rows_per_second": 66666.7}},
  "sources": {"http_requests_total":
    {"series_count": 1000000, "metric_type": "counter", "sample_interval": 15000}}}

• Each query's G and row count come from the sources it reads, and each branch of an or counts.
• Each bare selector reads its own metric's interval, and only when it needs one.
• A source without facts stays unknown and is never read as zero. A series_count of zero is rejected.
• A selector with no metric name is rejected in both PromQL and MetricsQL.

Verification

• Unchanged selections and costs. I ran stage_pipeline from origin/main (5da6a64, which includes refactor(planner)!: Pass 1 candidates are exact or a summary; fused computations become the outer target's FusedComputation #729) and from this head (0c248d3) for these runs:

  • --example planner-layering-1, -2, -3a and -3b, each under --cost-model usage-over-time (cost model 1) and provisioned-capacity (cost model 2);
  • Example 4: planner-layering-3b under --cost-model cluster-capacity with 256 GB and 64 GB of memory and 1 TB of disk, each with the lowest-latency, least-memory and pareto-front objectives;
  • an ad hoc --promql "topk by (job) (10, rate(x[1m]))" --promql "sum(up)" --epsilon 0.01 --delta 0.001 run.

  All 15 documents are byte-identical between the two. main was built in its own target directory, so no artifact was shared. Selected candidates:

  Run	Cost model 1	Cost model 2
  Example 1	P1 (7.94021 per second)	P1 (2002.060136 per second)
  Example 2	P1 (6.0800009)	P36 (3004.310765)
  Example 3a	P485 (39828.8028)	P485 (60545.00006)
  Example 3b	P2 (2.1)	P1 (1024.000008)

  Example 4 (3b, cost model 3)	256 GB	64 GB
  lowest latency	P3-m1	P2
  least memory	P1	P1
  Pareto front	P1, P2, P3-m1	P1, P2

• Tests on the touched crates. I ran cargo nextest run on asap-types, asap-frontend-promql, asap-frontend-metricsql, asap-aware-mapping, asap-physical-operators, asap-devtools, asap-integration-tests and asap-planner: 1667 passed, 35 skipped. python3 -m unittest test_server in tools/dag-viewer passes.

• Checks. These are clean on this head:

  • cargo fmt --all -- --check;
  • cargo clippy --workspace --all-targets --all-features --locked -- -D warnings;
  • cargo test --doc -p asap-types -p asap-frontend-promql -p asap-frontend-metricsql.

  tools/verify_metricsql_external_consumer.sh passes, because lower_metricsql(query, accuracy, Duration) keeps its signature.

• New or rewritten tests. Each fix test was written first, and failed on the previous head:

  • Stage 3:

    • values_per_result_adds_the_branches_of_an_or, with x = 10 series:

      Query	G
      rate(x[5m]) or sum without (instance) (rate(x[5m]))	20
      rate(x[5m]) or vector(0)	11
      rate(x[5m]) / rate(x[5m])	10
      sum without (instance) (rate(x[5m])) * rate(x[5m])	10
      rate(x[5m]) + rate(m[5m])	20

    • a_vector_of_a_scalar_keeps_the_rows_read_below_it (failed with series_read = 1):

      • series_read of vector(scalar(max(rate(x[5m])))) is 11 (10 series + 1 constant row), and its G is 1;
      • G of vector(1) is 1;
      • series_read of rate(x[5m]) or vector(scalar(max(rate(x[5m])))) is 21.

    • stage3_binds_only_the_series_count_of_the_sources_read (was stage3_binds_only_fresh_input_cardinality) checks that another source's series_count leaves the row count unbound.

  • Frontends:
    • selector_without_metric_name_is_rejected__GAP (failed: {job="a"} lowered to a scan of "") covers {job="a"}, rate({job="a"}[5m]), sum by (job) ({job="a"}) and a histogram_quantile over {le!=""}.
    • lowering_is_total_over_the_entire_corpus counts the 29 queries with no metric name apart, and asserts that count.
    • each_instant_selector_uses_its_own_metrics_sample_interval: in sum(data) + sum(other), the selectors look back 1 s and 30 s.
    • instant_selector_needs_its_metrics_sample_interval covers every arrival and no data workload, with and without histograms. Its queries have bare, aggregated, subquery and mixed-metric selectors. A declared metric lowers whatever the arrival.
    • The MetricsQL workload_lowering_uses_the_declared_sample_interval.
  • types:
    • series_count_must_be_positive (failed: zero validated);
    • canonical_workloads_have_a_strict_json_round_trip;
    • the_former_data_workload_shape_is_rejected, which also rejects evidence, series and an arrival sample_interval;
    • arrival_rate_and_sample_interval_are_validated;
    • estimated_demand_validates_its_rate (was estimated_demand_validates_window_rate_and_confidence): a negative expected_rate is InvalidRate;
    • ingesting_payload_without_sample_interval_validates.
  • devtools: sample_interval_ms_sets_the_instant_selector_lookback (failed: unknown argument). With --sample-interval-ms 30000, sum(up) looks back 30 s, and --interval-ms is rejected.

Limitations and follow-ups

• Design proposals under docs/design_docs/proposals/ still name the old fields (tracked in Update the docs after the G1–G11 consolidation groups land #677).
• Queries with selectors that name no metric stay unsupported until the IR has a source that selects across metrics, which a regex __name__ matcher also needs.