Price groupings from declared distinct counts of label sets (hierarchical labels)

crates/integration-tests/tests/sql_group_bound.rs

@@ -194,3 +194,30 @@ fn an_undeclared_sql_grouping_is_priced_at_the_groups_of_rows_or_the_rows_read()
         priced_groups_and_scan_rows(workload(DISTINCT_USERS, "relative", every_column()));
     assert_eq!(groups, std::collections::BTreeSet::from([625]));
 }
+
+/// Each endpoint is in one service: declaring `(service, endpoint)`'s 625
+/// combinations prices `GROUP BY service, endpoint` at 625 groups, where
+/// the per-column product (25 x 625) is capped only by the 10,000 groups of
+/// rows.
+#[test]
+fn a_declared_column_set_prices_a_sql_grouping_at_its_combinations() {
+    let hierarchical =
+        serde_json::json!({"region": 4, "service": 25, "endpoint": 625, "status": 4});
+    let (groups, _) =
+        priced_groups_and_scan_rows(workload(DISTINCT_USERS, "relative", hierarchical.clone()));
+    assert_eq!(groups, std::collections::BTreeSet::from([10_000]));
+    let mut declared = workload(DISTINCT_USERS, "relative", hierarchical);
+    declared
+        .data_workload
+        .as_mut()
+        .unwrap()
+        .sources
+        .get_mut("http")
+        .unwrap()
+        .distinct_values_per_label_set = vec![asap_types::workload::LabelSetDistinctValues {
+        labels: vec!["service".to_string(), "endpoint".to_string()],
+        distinct_values: 625,
+    }];
+    let (groups, _) = priced_groups_and_scan_rows(declared);
+    assert_eq!(groups, std::collections::BTreeSet::from([625]));
+}

docs/design_docs/architecture/input-output-workflow.md

@@ -201,8 +201,13 @@ struct SourceFacts {
     stream_norms: Option<StreamNorms>,
     max_distinct_per_evaluation: Option<u64>,
     distinct_values_per_label: BTreeMap<String, u64>,
+    distinct_values_per_label_set: Vec<LabelSetDistinctValues>,
     rows_held: Option<u64>,
 }
+struct LabelSetDistinctValues {
+    labels: Vec<String>,
+    distinct_values: u64,
+}
 ```
 
 | Field | Required | Purpose and behavior when unavailable |
@@ -215,6 +220,7 @@ struct SourceFacts {
 | `sample_interval` | PromQL and MetricsQL; nonzero | How often each series of the metric is sampled. Bare instant selectors over it use it as their selection horizon; without it, such a query fails to lower. |
 | `stream_norms`, `max_distinct_per_evaluation` | Optional | Bounds that Count-Min/Count Sketch and HLL/UnivMon guarantees are bound with. |
 | `distinct_values_per_label` | Optional; default empty; each count nonzero | Distinct values each label (PromQL) or column (SQL) takes, e.g. `{"host": 1000, "job": 10}`. Plan selection prices a `by (k1, k2, ...)` aggregate or summary, and a partitioned sort or limit, at the product of its keys' counts, at most the `series_count` (series, or groups of rows) of the sources it reads and the rows it reads per evaluation. Over several sources a key's counts add. No count is assumed for a key some source read does not declare: the grouping is priced at the smaller of that `series_count` and those rows. Each count must be an upper bound, counting a missing label (or a NULL) as one more value: when every key of a grouped result declares one, their product (at most the series read) also bounds `G`, which sizes per-group sketches for a whole-result target; otherwise `G` keeps its series bound, or none for SQL. |
+| `distinct_values_per_label_set` | Optional; default empty; each set two or more distinct labels, its count nonzero and at most the product of its labels' declared counts | Distinct combinations a set of labels (or columns) takes together, for labels that depend on each other, e.g. `[{"labels": ["service", "endpoint"], "distinct_values": 625}]` when each of 625 endpoints is in one of 25 services. For a grouping over one source whose keys are one row's unchanged labels, the per-label product above becomes the smallest product of declared set and label counts whose labels cover the keys: `by (service, endpoint)` is 625 groups, not 25 x 625, and `by (region, service, endpoint)` at most 4 x 625. A set covering more labels than the keys also bounds them. Nothing is inferred from the per-label counts (no functional dependency is assumed). Over several sources, or for keys a join or binary operator pairs from different rows, only the per-label counts apply; a key a relabeling rewrites is not bounded by any declared set. The same bound gives `G` for a whole-result target. A count above its labels' product is an input error naming the source. |
 | `rows_held` | Optional; nonzero | Rows the source holds at rest. A query-time scan reads at most these; a scan with no window above it (a SQL query without a time predicate) reads all of them. |
 
 Every fact is declared: the planner has no clock, so a caller drops facts

docs/design_docs/concepts/accuracy-models.md

@@ -84,8 +84,12 @@ strictest: every group (`G`, when `G` is above 100), then a 1% share (100),
 then a 5% share (20), then a single value (1). A whole-result target with an unbound `G`
 enlarges nothing and stays unmet. A grouped result's `G` is bound by the
 distinct values its sources declare for its group keys
-(`SourceFacts.distinct_values_per_label`): the product over the keys, at most
-the series read. A key counts only when it is a source column passed through
+(`SourceFacts.distinct_values_per_label`, and
+`distinct_values_per_label_set` for labels that depend on each other): the
+smallest product of declared label and label-set counts covering the keys, at
+most the series read. With each of 625 endpoints in one of 25 services and
+`(service, endpoint)` declared at 625, `by (service, endpoint)` returns at
+most 625 values, not 25 x 625. A key counts only when it is a source column passed through
 unchanged; a relabeled or computed key leaves `G` unbound. So a sketch meets a whole-result target over SQL groups when
 every grouping column's distinct values are declared, and over 25 services
 and 25 endpoints sizes each group's sketch for `δ / 625`. Nothing declared

docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md

@@ -231,7 +231,7 @@ normalized workload, lowered query IR, and declared statistics:
 | Cost concept | Canonical source |
 |---|---|
 | Source row cardinality | `DataWorkload.sources[source].series_count`, and `rows_held` for data at rest or a scan with no window. No default: a missing fact fails planning, naming the source and the fact. |
-| Groups of a `by (...)` grouping or partition | The smallest fact-based bound: the product of `DataWorkload.sources[source].distinct_values_per_label` over its keys (all declared), the declared `series_count` (series, or groups of rows for a SQL table) of the sources it reads, and the rows it reads per evaluation (at ingestion time, the arriving rows of the window it maintains; a grouping with no window is an error). No count is assumed for an undeclared key. |
+| Groups of a `by (...)` grouping or partition | The smallest fact-based bound: the product of `DataWorkload.sources[source].distinct_values_per_label` over its keys (all declared), or over one source whose keys are one row's unchanged labels (no join, binary operator or relabeling below), the smallest product of its declared label and `distinct_values_per_label_set` counts covering the keys, the declared `series_count` (series, or groups of rows for a SQL table) of the sources it reads, and the rows it reads per evaluation (at ingestion time, the arriving rows of the window it maintains; a grouping with no window is an error). No count is assumed for an undeclared key. |
 | Arrival and update rate | `DataWorkload.arrival` and its `rows_per_second`. |
 | Query demand | `QueryWorkloadEntry.recurrence`, normalized over an explicit horizon. |
 | Event-time range | The query text (its range and offset). |

docs/develop_docs/library-api.md

@@ -298,7 +298,12 @@ plan selection prices a `by (...)` grouping's groups at: the product of its
 keys' counts, at most the declared `series_count` (series or groups of
 rows) and the rows it reads; with a key undeclared, no count is assumed and
 the grouping is priced at the smaller of those two; over several sources a
-key's counts add), and `rows_held` (the rows the source holds at rest,
+key's counts add), `distinct_values_per_label_set` (distinct combinations
+a set of dependent labels takes together, e.g.
+`[{"labels": ["service", "endpoint"], "distinct_values": 625}]` when each
+endpoint is in one service: a grouping over one source is priced at the
+smallest product of declared set and label counts covering its keys, so
+`by (service, endpoint)` is 625 groups rather than 25 x 625), and `rows_held` (the rows the source holds at rest,
 which bound a scan with no window and data at rest). Every fact is declared: the planner has no
 clock, so a caller drops facts that no longer hold before planning, and
 pricing assumes no default: a scan whose source declares no arrival rate

docs/user_guide_docs/stage-viewer.md

@@ -127,12 +127,15 @@ planner must meet:
 Beside the queries, the facts the planner was given about the data: how it
 arrives (`ingesting · 200000 rows/s`, `at rest`, `mixed`) and, per source
 (a PromQL metric or a SQL table), its series count, metric type, whether
-its values are non-negative, its sample interval, its stream-norm bounds and
-its distinct-count bound, `—` for a fact the workload does not declare. The
+its values are non-negative, its sample interval, its stream-norm bounds,
+its distinct-count bound, the distinct values per label (`service 25,
+endpoint 625`) and per label set (`(service, endpoint) 625`, for labels that
+depend on each other, such as endpoints that each belong to one service),
+`—` for a fact the workload does not declare. The planner prices a grouping
+at the smallest product of these counts covering its keys. The
 **read by** column lists the queries whose text names the source. The line
 above the table shows the deployment capabilities the plan was made for
-(for example `raw data not retained`). A workload cannot declare labels,
-their hierarchy or their cardinalities, so the panel shows none.
+(for example `raw data not retained`).
 
 The line under the queries counts the candidates at each stage, for example,
 in Example 1, `Candidates per stage: 1 logical DAG → 27 logical ASAP → 72

tools/dag-viewer/README.md

@@ -40,7 +40,7 @@ To run it, see [RUNNING.md](RUNNING.md).
 - **Data workload** beside it: the arrival and one row per declared source
   with exactly the facts `SourceFacts` has (series count, metric type,
   non-negative values, sample interval, stream norms, distinct bound,
-  distinct values per label) and
+  distinct values per label, distinct values per label set) and
   the queries whose text names it, plus the deployment capabilities the
   run planned with.
 - **Stage 3 · plans by cost**: every candidate the document shows. First the

tools/dag-viewer/app.js

@@ -297,8 +297,8 @@
     $('dataArrival').textContent = `arrival: ${view.arrival}`;
     $('deployment').textContent = doc.deployment_capabilities ? `Deployment: ${deploymentText(doc.deployment_capabilities)}` : '';
     $('dataSources').innerHTML = view.rows.length
-      ? table(['source', 'series', 'type', 'non-negative', 'sample interval', 'stream norms', 'distinct bound', 'distinct values per label', 'read by'],
-        view.rows.map((r) => `<tr><td>${esc(r.name)}</td><td>${esc(r.series)}</td><td>${esc(r.type)}</td><td>${esc(r.nonNegative)}</td><td>${esc(r.sampleInterval)}</td><td>${esc(r.streamNorms)}</td><td>${esc(r.maxDistinct)}</td><td class="wrap">${esc(r.distinctValuesPerLabel)}</td><td class="wrap">${esc(r.readBy.join(', ') || '—')}</td></tr>`).join(''))
+      ? table(['source', 'series', 'type', 'non-negative', 'sample interval', 'stream norms', 'distinct bound', 'distinct values per label', 'distinct values per label set', 'read by'],
+        view.rows.map((r) => `<tr><td>${esc(r.name)}</td><td>${esc(r.series)}</td><td>${esc(r.type)}</td><td>${esc(r.nonNegative)}</td><td>${esc(r.sampleInterval)}</td><td>${esc(r.streamNorms)}</td><td>${esc(r.maxDistinct)}</td><td class="wrap">${esc(r.distinctValuesPerLabel)}</td><td class="wrap">${esc(r.distinctValuesPerLabelSet)}</td><td class="wrap">${esc(r.readBy.join(', ') || '—')}</td></tr>`).join(''))
       : '<span class="status">No source declared.</span>';
   }
 

tools/dag-viewer/stages.js

@@ -809,6 +809,7 @@ function dataWorkloadView(dataWorkload, queries) {
     streamNorms: norms(facts.stream_norms),
     maxDistinct: typeof facts.max_distinct_per_evaluation === 'number' ? facts.max_distinct_per_evaluation : '—',
     distinctValuesPerLabel: Object.entries(facts.distinct_values_per_label || {}).map(([label, count]) => `${label} ${count}`).join(', ') || '—',
+    distinctValuesPerLabelSet: (Array.isArray(facts.distinct_values_per_label_set) ? facts.distinct_values_per_label_set : []).map((set) => `(${(set.labels || []).join(', ')}) ${set.distinct_values}`).join(', ') || '—',
     readBy: queriesNaming(name, queries || []),
   }));
   return { arrival: `${kind}${rate}`, rows };

tools/dag-viewer/test_viewer.py

@@ -539,6 +539,15 @@ def test_data_workload_rows_show_each_labels_distinct_values(self):
         self.assertEqual(rows["data"]["distinctValuesPerLabel"], "host 1000, job 10")
         self.assertEqual(rows["other"]["distinctValuesPerLabel"], "—")
 
+    def test_data_workload_rows_show_each_label_sets_distinct_values(self):
+        data_workload = {"arrival": "at_rest", "sources": {
+            "http": {"series_count": 10000, "distinct_values_per_label": {"service": 25, "endpoint": 625},
+                     "distinct_values_per_label_set": [{"labels": ["service", "endpoint"], "distinct_values": 625}]},
+            "other": {"series_count": 5}}}
+        rows = {r["name"]: r for r in self.js.call("dataWorkloadView", data_workload, [])["rows"]}
+        self.assertEqual(rows["http"]["distinctValuesPerLabelSet"], "(service, endpoint) 625")
+        self.assertEqual(rows["other"]["distinctValuesPerLabelSet"], "—")
+
     def test_stored_output_rows_map_to_stage2_nodes_and_their_readers(self):
         rows = self.js.call("storedOutputRows", self.doc)
         stored = self.doc["stage3_plan_selection"]["stored_outputs"]