[zzylol]

Closes #894

Why

Hierarchical labels are over-counted. In the RollupHeavy evaluation workload, each of 625 endpoints belongs to one of 25 services. The data workload could only declare per-label counts, so the planner priced GROUP BY service, endpoint at min(25 x 625, series_count) = 10,000 groups instead of 625. At that size the cheapest plan for the 15-minute distinct-users query was a query-time raw scan: 1.8e9 rows per evaluation, 21.2 s predicted latency and 158 GB of raw data on disk.

What

• A data workload can declare how many distinct combinations a set of labels (PromQL) or columns (SQL) takes together, through the new field SourceFacts.distinct_values_per_label_set:

  "distinct_values_per_label_set": [{"labels": ["service", "endpoint"], "distinct_values": 625}]

• A grouping's count is the minimum of every bound the declared facts give:
  • the smallest product of declared set counts and per-label counts whose labels cover the grouping's keys;
  • series_count, rows_held and the rows read, as before.
• The rule applies everywhere a group count is used:
  • cost model 1 group counts: by groupings, partitioned sort/limit, sketch regroup and SQL GROUP BY;
  • the G that a whole-result accuracy target splits its failure probability δ over.
• Only declared facts are used. No functional dependency is inferred. A set count applies only when the keys are one source row's unchanged labels: not across a join or a binary operator, and not after a relabeling.
• Validation, as input errors:
  • a set count above the product of its labels' declared counts (names the source);
  • a set with fewer than two distinct labels (names the source);
  • a zero count (the existing `ZeroDistinctValues`).
• The viewer's data-workload panel has a new "distinct values per label set" column, e.g. (service, endpoint) 625. The PlanningWorkload docs (input-output workflow, library API, accuracy models, the analytical-cost proposal) and the stage-viewer guide describe the field.

How

• SourceFacts::declared_distinct_combinations(labels) (types crate) computes the smallest covering product. It is a dynamic program over subsets of the keys that some declared set names; every other key contributes its own per-label count.
• Cost path: group_count_bound calls it when the grouping reads one source and every key is in that input's unchanged_source_columns. That is a new per-node trace in the physical DAG of the output columns that still hold one source row's labels unchanged. Otherwise it uses the previous per-label product.
• G path: declared_group_bound now traces all keys together (declared_distinct_combinations(node, columns, data), which replaces the per-column declared_distinct_values) down to the scans, and asks the source there.

Key code interfaces added / modified

crates/types/src/workload/mod.rs

/// What is known about one source (a PromQL metric or a SQL table). Stays.
pub struct SourceFacts {
    /// Series in the source.
    pub series_count: Option<u64>,                                   // unchanged
    /// Declared Prometheus type.
    pub metric_type: Option<MetricType>,                             // unchanged
    /// Every sample is non-negative.
    pub values_non_negative: bool,                                   // unchanged
    /// How often each series is sampled.
    pub sample_interval: Option<DurationMs>,                         // unchanged
    /// Bounds on the stream norms.
    pub stream_norms: Option<StreamNorms>,                           // unchanged
    /// Distinct items one distinct-count evaluation counts, at most.
    pub max_distinct_per_evaluation: Option<u64>,                    // unchanged
    /// Distinct values per label or column.
    pub distinct_values_per_label: BTreeMap<String, u64>,            // unchanged
    /// Distinct combinations per set of dependent labels; default empty, omitted from JSON when empty.
    pub distinct_values_per_label_set: Vec<LabelSetDistinctValues>,  // new
    /// Rows the source holds at rest.
    pub rows_held: Option<u64>,                                      // unchanged
}

/// One declared label-set count: an upper bound on the combinations of values its labels take together. New.
pub struct LabelSetDistinctValues {
    /// Two or more distinct labels (or columns), in any order.
    pub labels: Vec<String>,                                         // new
    /// Distinct combinations of their values in the source.
    pub distinct_values: u64,                                        // new
}

impl SourceFacts {
    /// The smallest product of declared set and per-label counts covering `labels`;
    /// `None` when some label is in no declared count. Plan selection calls it for every grouping.
    pub fn declared_distinct_combinations(&self, labels: &[&str]) -> Option<u64>; // new
}

/// Why a planning workload is rejected. Every variant is bad input. Stays; no longer `Copy`.
pub enum WorkloadError {
    // ... every existing variant unchanged ...
    /// A zero count in `distinct_values_per_label` or `distinct_values_per_label_set`.
    ZeroDistinctValues,                                              // message now names both fields
    /// A label set naming fewer than two distinct labels. Bad input.
    LabelSetNeedsTwoLabels { source_name: String, labels: Vec<String> },          // new
    /// A label set count above the product of its labels' declared counts. Bad input.
    LabelSetCountAboveLabelProduct {
        source_name: String, labels: Vec<String>, distinct_values: u64, label_product: u64,
    },                                                               // new
}

(WorkloadError lists every variant in the source; only the changed ones are shown above, and none is removed.)

crates/plan-selection/src/lib.rs (private functions, listed because they carry the algorithm)

/// Output columns of a physical node that still hold one source row's labels unchanged.
fn unchanged_source_columns<'a>(node, producers, nodes, unchanged_columns_of) -> BTreeSet<&'a str>; // new
/// Groups a `by (keys)` is priced at.
fn group_count_bound(keys, input, sources, unchanged_input_columns: &BTreeSet<&str>, data, rows_read) -> u64; // new parameter
/// Combinations of several columns traced to the scans (replaces per-column `declared_distinct_values`).
fn declared_distinct_combinations(node: &OperatorNode, columns: &[usize], data: &DataWorkload) -> Option<u64>; // was declared_distinct_values(node, column, data)

tools/dag-viewer/stages.js, app.js: dataWorkloadView rows gain distinctValuesPerLabelSet; the panel gains a column.

Usage

This is the end-to-end run on the RollupHeavy component L5/http-distinct-service_endpoint-15m, with a release stage_pipeline at this PR's head, raw data retained.

# The workload: three copies of the query below, every minute, accuracy enforced.
#   SELECT service, endpoint, COUNT(DISTINCT user_id) AS value FROM http
#   WHERE ts >= now() - INTERVAL '15 minutes' GROUP BY service, endpoint
# Its data workload for http (the eval generator now adds the set):
#   "series_count": 10000,
#   "distinct_values_per_label": {"region": 4, "service": 25, "endpoint": 625, "status": 4},
#   "distinct_values_per_label_set": [{"labels": ["service", "endpoint"], "distinct_values": 625}]
stage_pipeline --workload http-distinct-service_endpoint-15m.json --table '<http schema>' \
  --deployment-capabilities '{"raw_data_retained":true,"query_time_results_retained":false}' \
  --cost-model aws-fargate-us-east-1 --enforce-accuracy-guarantees --out after.stages.json

1. PlanningWorkload::validate checks the set: 625 ≤ 25 x 625.
2. In Stage 3, each candidate's SummaryAgg/PaneChain over http reaches group_count_bound. Its keys service, endpoint are unchanged scan columns of one source, so declared_distinct_combinations(["service", "endpoint"]) returns 625 (before this PR: 25 x 625 capped at 10,000).
3. The whole-result δ of each KMV is split over G = 625 (declared_group_bound).

The selected plan:

plan: Kmv for Q1, Q2, Q3 in 15 x 1m panes · ingestion time: Kmv x15 panes
n1 build Kmv in 15 panes of 60000 ms for 625 groups: 2000000 rows x (1 + depth 1); keeps 666720000 bytes
n0 scan 2000000 samples
cost 2.68 USD/30 d, disk 0, predicted latency 0.011 ms

Before this PR, the same workload (without the set) selected a query-time scan: scan 1800000000 samples; keeps 158400000000 bytes of raw data on disk, then build Kmv into 10000 states, at 11.22 USD/30 d and 21,240 ms.

Major algorithm changes

Group count from declared label sets (cost model 1 and the whole-result δ split)

Problem. A grouping's count was the product of its keys' per-label counts. With service 25 and endpoint 625, where each endpoint is in one service, by (service, endpoint) was priced at 10,000 groups (the series cap) instead of 625. The planner then selected a raw-scan plan (above).

Before.

1. group_count_bound: declared = Π_key Σ_source distinct_values_per_label[key].
2. Count = min(declared, Π series_count, rows_read).
3. declared_group_bound (G): product over keys of declared_distinct_values(node, key), each traced to the scans independently.

After.

1. unchanged_source_columns (per physical node, in topological order) gives the output columns still holding one source row's labels:
   • a scan: every column;
   • filter, sort, limit, dedup, time range and time shift: their input's columns;
   • a projection or grouping key: a column passed on under its own name;
   • a per-series reduction: every column except the reduced and named ones;
   • a union: the columns unchanged in every branch;
   • summary operators: their grouping columns;
   • anything else (join, binary operator, relabel, info enrich, computed columns): none.
2. group_count_bound checks whether the keys read one source and are all unchanged columns.
   • If so, declared = SourceFacts::declared_distinct_combinations(keys).
   • Otherwise it uses the per-label product as before.
3. declared_distinct_combinations(keys):
   • Split the keys into those some declared set names (in_sets) and the rest (alone).
   • smallest[∅] = 1, and smallest[m] = min over counts c with c ∩ m ≠ ∅ of smallest[m \ c] · count(c). The counts c are the declared sets and the single labels of in_sets.
   • Result: smallest[in_sets] · Π_{alone} distinct_values_per_label.
4. G: declared_distinct_combinations(node, columns) traces all keys together through the same unary operators to a scan. Union and concat branches add.

Correctness.

• Each declared count is an upper bound on the distinct projections of the source rows onto its labels.
• The keys' distinct tuples inject into the product of the projections of any cover, so every product is an upper bound and so is their minimum.
• The bound holds only when every key value of a group comes from one source row. The unchanged-column trace enforces that.
• No count is ever derived from the per-label counts (no functional dependency is inferred), and validation rejects a set count above its labels' product.
• Tests:
  • declared_label_sets_bound_the_combinations_of_their_labels: 625; superset 4·625; no set → 25·625; subset of a set; uncovered key → None; 70 keys.
  • a_label_set_count_is_at_most_its_labels_product: validation; the error names the source.
  • a_grouping_is_priced_at_its_declared_label_set_count: by-groupings 625 / 2500 / 10,000; top-k partitions 625; a label_join-relabeled key and a group_left binary operator fall back to 10,000.
  • values_per_result_is_bound_by_a_declared_label_set: G.
  • a_declared_column_set_prices_a_sql_grouping_at_its_combinations: SQL GROUP BY.
  • The existing per-label tests are unchanged.

Complexity.

• Before: O(k · s) per grouping, for k keys and s sources.
• After: O(2^j · (j + S)) time and O(2^j) memory per grouping, where j ≤ 16 is the number of keys some declared set names and S the number of declared sets. With no set naming a key, it costs O(k) as before.
• The unchanged-column trace is O(columns) per node.
• Measured on the 28 RollupHeavy components: planning wall time is unchanged (largest 75.4 s vs 74.2 s; the L5 15-minute component 5.0 s vs 4.7 s, 10,788 candidates checked in both).

Assumptions and limits.

• Sets apply to one source's unchanged columns. Over several sources, or after a join, binary operator or relabeling, only the per-label counts apply.
• without (...) keeps its series bound, because its kept labels are known only at run time.
• With more than 16 keys named by sets, only the per-label product is used.
• Name-based pricing of a relabeled key under the per-label rule is unchanged by this PR; tracked in A relabeled or renamed group key is priced at its old label's declared count #898.

Before this PR

The data workload could not express dependent labels. Every grouping combining endpoint with service was priced at 10,000 groups. On the RollupHeavy components (raw data retained) that selected query-time raw scans: 21.2 s latency, 158 GB raw data on disk.

After this PR

With (service, endpoint) declared at 625, the groupings are priced at their true sizes. The raw scan disappears wherever the keys contain (service, endpoint) and no label without a covering set.

Component (raw data retained)	Before: USD/30 d · raw disk · latency	After
L5 http-distinct-service_endpoint-15m	11.22 · 158.4 GB · 21,240 ms (raw scan)	2.68 · 0 · 0.011 ms (KMV, 15 × 1 min panes, 625 groups)
L5 http-distinct-service_endpoint-5m	3.74 · 52.8 GB · 7,080 ms (raw scan)	1.43 · 0 · 0.005 ms
L4 http-distinct-region_service_endpoint (5 min + 15 min)	14.37 · 158.4 GB · 21,240 ms (raw scan)	5.90 · 0 · 0.045 ms (2,500 groups)
L4 http-distinct-service_endpoint_status	14.37 · 158.4 GB · 21,240 ms (raw scan)	5.90 · 0 · 0.045 ms
L4 http-p99-service_endpoint	3.71 · 0 · 0.18 ms	0.87 · 0 · 0.011 ms
L4 http-p99-region_service_endpoint	3.71 · 0 · 0.18 ms	1.43 · 0 · 0.045 ms
L4 http-distinct-region_endpoint_status	14.37 · 158.4 GB · 21,240 ms	unchanged
L4 http-distinct-region_service_endpoint_status	14.37 · 158.4 GB · 21,240 ms	unchanged
The other 19 components		unchanged
Sum over the 27 planned components	125.36	92.45

• The two unchanged raw scans are correct: (region, endpoint, status) and (region, service, endpoint, status) do have up to 4 x 625 x 4 = 10,000 combinations, the full series count.
• L4/http-distinct-service_endpoint times out at 3 min both before and after, as it did in the earlier evaluation run. Its two windows plan as the two L5 components above.

Verification

• Failing tests first. On main, the new tests do not compile (the field does not exist), and the scenario they cover is priced at 10,000.
• Local: cargo nextest run -p asap-types -p asap-plan-selection (425 passed); cargo nextest run -p asap-integration-tests -E 'binary(sql_group_bound) | binary(declared_group_counts) | binary(reuse_rules)' (17 passed).
• On the remote build machine: cargo test -p asap-planner --test sql_table_scans --test runtime_install (43 passed).
• cargo check --tests on the frontends, planner and devtools.
• cargo fmt, and cargo clippy --all-targets -- -D warnings on types, plan-selection and integration-tests: clean.
• Viewer: python3 -m unittest test_viewer -k data_workload (5 passed).
• End to end: 28 RollupHeavy components planned before and after on a 56-core server with a release stage_pipeline at this head (default search, accuracy enforced, cost model 1 at AWS Fargate prices, 3-min timeout). Results are in the table above.
• An independent review found that a relabeled key could be undercounted. It is fixed by the unchanged-column trace, with regression tests.