feat(planner): Example 1 completion — identical-expression CSE, whole-expression top-k heaps, counter-sample weights, per-second plan-selection cost

crates/asap-aware-mapping/src/accuracy/composition.rs

@@ -48,6 +48,23 @@ pub(crate) fn source_norm_statistic(norm: &str, source: &str) -> String {
     format!("{norm}[{source}]")
 }
 
+/// Whether every value folded is non-negative, for values that are raw
+/// samples of `metric` or plain sums of them, e.g.
+/// `values_non_negative[m]`: an indicator, 1 when proven. Plan selection
+/// binds it to 1 when the data workload declares `metric` a counter, and
+/// leaves it unbound otherwise.
+pub(crate) fn samples_sign_statistic(metric: &str) -> String {
+    format!("{VALUES_NON_NEGATIVE}[{metric}]")
+}
+
+/// The metric a [`samples_sign_statistic`] names.
+pub(crate) fn parse_samples_sign_statistic(statistic: &str) -> Option<&str> {
+    statistic
+        .strip_prefix(VALUES_NON_NEGATIVE)?
+        .strip_prefix('[')?
+        .strip_suffix(']')
+}
+
 /// The norm and source a [`source_norm_statistic`] names.
 pub(crate) fn parse_source_norm_statistic(statistic: &str) -> Option<(&'static str, &str)> {
     [STREAM_L1_NORM, STREAM_L2_NORM]
@@ -352,11 +369,31 @@ impl DefaultAccuracyModel {
         let mut provenance = Vec::new();
         let count = row_count(stats, &mut provenance);
         let bound = if needs_sign && stats.values_non_negative.is_none() {
-            provenance.push(GuaranteeSource::UnavailableStatistic {
-                statistic: VALUES_NON_NEGATIVE.into(),
-            });
-            BoundExpr::Unknown {
-                statistic: VALUES_NON_NEGATIVE.into(),
+            match &stats.sign_statistic {
+                // The sign of one metric's samples: the bound, gated on it.
+                Some(statistic) => {
+                    provenance.push(GuaranteeSource::UnavailableStatistic {
+                        statistic: statistic.clone(),
+                    });
+                    BoundExpr::Product {
+                        factors: vec![
+                            BoundExpr::Max {
+                                terms: inputs.iter().map(|g| g.bound.clone()).collect(),
+                            },
+                            BoundExpr::Unknown {
+                                statistic: statistic.clone(),
+                            },
+                        ],
+                    }
+                }
+                None => {
+                    provenance.push(GuaranteeSource::UnavailableStatistic {
+                        statistic: VALUES_NON_NEGATIVE.into(),
+                    });
+                    BoundExpr::Unknown {
+                        statistic: VALUES_NON_NEGATIVE.into(),
+                    }
+                }
             }
         } else {
             BoundExpr::Max {
@@ -1292,6 +1329,39 @@ mod tests {
             assert!(with(Some(false)).is_err(), "{op:?}");
         }
     }
+    /// Over the samples of one metric, an unknown sign is named by that
+    /// metric ([`samples_sign_statistic`]) and gates the bound: binding it
+    /// to 1 (proven) gives the relative bound, and left unbound it stays
+    /// unknown.
+    #[test]
+    fn a_sign_over_one_metrics_samples_gates_the_relative_bound() {
+        let statistic = samples_sign_statistic("m");
+        assert_eq!(parse_samples_sign_statistic(&statistic), Some("m"));
+        assert_eq!(parse_samples_sign_statistic(VALUES_NON_NEGATIVE), None);
+        for op in [
+            CompositionOperator::ExactSum,
+            CompositionOperator::ExactAverage,
+        ] {
+            let out = DefaultAccuracyModel
+                .propagate(
+                    &op,
+                    &[rel(0.01)],
+                    None,
+                    &PropagationStats {
+                        sign_statistic: Some(statistic.clone()),
+                        input_row_count: Some(3),
+                        ..Default::default()
+                    },
+                )
+                .unwrap();
+            assert_eq!(out.bound.evaluate(), None, "{op:?}");
+            let proven = out.bind(&|s| (s == statistic).then_some(1.0));
+            assert_eq!(proven.bound.evaluate(), Some(0.01), "{op:?}");
+            let other = out.bind(&|s| (s == samples_sign_statistic("n")).then_some(1.0));
+            assert_eq!(other.bound.evaluate(), None, "{op:?}");
+        }
+    }
+
     #[test]
     fn exact_average_has_its_own_absolute_error_rule() {
         let out = DefaultAccuracyModel

crates/asap-aware-mapping/src/accuracy/evidence.rs

@@ -57,6 +57,13 @@ pub struct PropagationStats {
     /// the multiplicative relative-error rule, which is unsound across a
     /// sign change.
     pub values_non_negative: Option<bool>,
+    /// The statistic an unknown sign is named by when the folded values are
+    /// one metric's raw samples or plain sums of them
+    /// ([`samples_sign_statistic`](crate::accuracy::composition::samples_sign_statistic)):
+    /// the relative bound is then gated on it, and plan selection binds it
+    /// to 1 when the metric is declared a counter. `None`: the bare
+    /// `values_non_negative`, which nothing binds.
+    pub sign_statistic: Option<String>,
     /// Number of input rows an exact aggregation consumes (e.g. the number
     /// of groups a function folds), for exact aggregate union bounds
     /// bound over per-input failures.

crates/asap-aware-mapping/src/analytical_cost.rs

@@ -151,25 +151,51 @@ pub fn cache_hit_ratios(
 /// Conversion from physical dimensions to one deployment-specific objective.
 /// Memory's coefficient means cost units per retained byte over this model's
 /// explicit comparison scope; it is not mixed with a rate implicitly.
+///
+/// Plan selection prices per second of wall time and also reads
+/// `cost_per_retained_byte_second`, the price of state kept across
+/// evaluations per byte per second, and `horizon_s`, the seconds over which
+/// it spreads one-off, scheduled and unknown queries.
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct ResourceCalibration {
     pub cost_per_cpu_op: f64,
     pub cost_per_scan_byte: f64,
     pub cost_per_retained_byte: f64,
+    #[serde(default)]
+    pub cost_per_retained_byte_second: f64,
+    #[serde(default = "default_horizon_s")]
+    pub horizon_s: f64,
     pub version: String,
 }
 
+/// One hour.
+pub const DEFAULT_HORIZON_S: f64 = 3_600.0;
+
+fn default_horizon_s() -> f64 {
+    DEFAULT_HORIZON_S
+}
+
 impl ResourceCalibration {
     pub fn validate(&self) -> Result<(), AnalyticalCostError> {
         for (name, value) in [
             ("cost_per_cpu_op", self.cost_per_cpu_op),
             ("cost_per_scan_byte", self.cost_per_scan_byte),
             ("cost_per_retained_byte", self.cost_per_retained_byte),
+            (
+                "cost_per_retained_byte_second",
+                self.cost_per_retained_byte_second,
+            ),
         ] {
             if !value.is_finite() || value < 0.0 {
                 return Err(AnalyticalCostError::InvalidCalibration(name, value));
             }
         }
+        if !self.horizon_s.is_finite() || self.horizon_s <= 0.0 {
+            return Err(AnalyticalCostError::InvalidCalibration(
+                "horizon_s",
+                self.horizon_s,
+            ));
+        }
         if self.cost_per_cpu_op == 0.0
             && self.cost_per_scan_byte == 0.0
             && self.cost_per_retained_byte == 0.0