fix(cost): distinguish PromQL binary operand modes

Author: zzylol

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

@@ -13,7 +13,7 @@ use serde::{Deserialize, Serialize};
 
 use crate::analytical_statistics::{
     validate_comparison_scopes, ComparisonScope, OperatorStatistics, OperatorStatisticsProvider,
-    SourceCoverage,
+    PromqlBinaryOperandMode, SourceCoverage,
 };
 
 pub const ANALYTICAL_MODEL_VERSION: &str = "analytical-resource-at-rest-v1";
@@ -465,9 +465,9 @@ pub fn estimate_operator(
     let output = statistics.output;
     if matches!(operator, PhysicalOperator::PromqlScalarLeaf) {
         let promql = require_promql_statistics(&statistics, 0)?;
-        if promql.output_series > output.rows && output.rows > 0 {
+        if promql.output_series != 0 || output.rows != promql.evaluation_steps {
             return Err(AnalyticalCostError::InconsistentOperatorStatistics(
-                "output series exceed output rows",
+                "PromQL scalar leaf must emit one scalar row per evaluation step",
             ));
         }
         return Ok(ResourceEstimate {
@@ -601,27 +601,29 @@ pub fn estimate_operator(
         PhysicalOperator::PromqlVectorBinary => {
             let promql = require_promql_statistics(&statistics, 2)?;
             let right = input(1)?;
-            let scalar_vector = matches!(operator, PhysicalOperator::PromqlVectorBinary)
-                && promql.input_series.contains(&0);
-            let matching_bytes = if scalar_vector {
-                per_row_width(output.rows, output.bytes)?
-            } else {
-                let build_side = statistics
-                    .hash_join_build_side
-                    .ok_or(AnalyticalCostError::MissingOrZero("hash_join_build_side"))?;
-                let key_bytes = statistics
-                    .key_bytes
-                    .ok_or(AnalyticalCostError::MissingOrZero("key_bytes"))?;
-                let build_series = match build_side {
-                    HashJoinBuildSide::Left => promql.input_series[0],
-                    HashJoinBuildSide::Right => promql.input_series[1],
-                };
-                checked_bytes(&[
-                    build_series,
-                    key_bytes
-                        .checked_add(16)
-                        .ok_or(AnalyticalCostError::Overflow)?,
-                ])?
+            let mode = validate_promql_binary(&statistics)?;
+            let matching_bytes = match mode {
+                PromqlBinaryOperandMode::VectorScalar | PromqlBinaryOperandMode::ScalarVector => {
+                    per_row_width(output.rows, output.bytes)?
+                }
+                PromqlBinaryOperandMode::VectorVector => {
+                    let build_side = statistics
+                        .hash_join_build_side
+                        .ok_or(AnalyticalCostError::MissingOrZero("hash_join_build_side"))?;
+                    let key_bytes = statistics
+                        .key_bytes
+                        .ok_or(AnalyticalCostError::MissingOrZero("key_bytes"))?;
+                    let build_series = match build_side {
+                        HashJoinBuildSide::Left => promql.input_series[0],
+                        HashJoinBuildSide::Right => promql.input_series[1],
+                    };
+                    checked_bytes(&[
+                        build_series,
+                        key_bytes
+                            .checked_add(16)
+                            .ok_or(AnalyticalCostError::Overflow)?,
+                    ])?
+                }
             };
             ResourceEstimate {
                 cpu_ops: left.rows as f64 + right.rows as f64 + output.rows as f64,
@@ -756,6 +758,49 @@ fn validate_promql_bridge(
     Ok(())
 }
 
+fn validate_promql_binary(
+    statistics: &OperatorStatistics,
+) -> Result<PromqlBinaryOperandMode, AnalyticalCostError> {
+    let promql = require_promql_statistics(statistics, 2)?;
+    let mode = promql
+        .binary_operand_mode
+        .ok_or(AnalyticalCostError::MissingOrStale(
+            "promql_binary_operand_mode",
+        ))?;
+    match mode {
+        PromqlBinaryOperandMode::VectorVector => {
+            if statistics.hash_join_build_side.is_none() || statistics.key_bytes.is_none() {
+                return Err(AnalyticalCostError::MissingOrStale(
+                    "vector_binary_label_match_statistics",
+                ));
+            }
+        }
+        PromqlBinaryOperandMode::VectorScalar => {
+            if promql.input_series[1] != 0
+                || statistics.inputs[1].rows != promql.evaluation_steps
+                || statistics.hash_join_build_side.is_some()
+                || statistics.key_bytes.is_some()
+            {
+                return Err(AnalyticalCostError::InconsistentOperatorStatistics(
+                    "vector/scalar binary evidence has an invalid scalar edge or label-match state",
+                ));
+            }
+        }
+        PromqlBinaryOperandMode::ScalarVector => {
+            if promql.input_series[0] != 0
+                || statistics.inputs[0].rows != promql.evaluation_steps
+                || statistics.hash_join_build_side.is_some()
+                || statistics.key_bytes.is_some()
+            {
+                return Err(AnalyticalCostError::InconsistentOperatorStatistics(
+                    "scalar/vector binary evidence has an invalid scalar edge or label-match state",
+                ));
+            }
+        }
+    }
+    Ok(mode)
+}
+
 fn require_promql_statistics(
     statistics: &OperatorStatistics,
     inputs: usize,
@@ -1202,6 +1247,7 @@ mod tests {
                 window_samples_per_series: None,
                 subquery_steps: None,
                 scalar_ops_per_row: None,
+                binary_operand_mode: None,
             });
             statistics
         };
@@ -1246,6 +1292,58 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn promql_binary_mode_distinguishes_empty_vectors_from_scalars() {
+        use crate::analytical_statistics::{PromqlBinaryOperandMode, PromqlOperatorStatistics};
+
+        let empty = EdgeStatistics { rows: 0, bytes: 0 };
+        let mut vector_vector = OperatorStatistics {
+            key_bytes: Some(16),
+            hash_join_build_side: Some(HashJoinBuildSide::Right),
+            promql: Some(PromqlOperatorStatistics {
+                input_series: vec![0, 0],
+                output_series: 0,
+                evaluation_steps: 10,
+                window_samples_per_series: None,
+                subquery_steps: None,
+                scalar_ops_per_row: None,
+                binary_operand_mode: Some(PromqlBinaryOperandMode::VectorVector),
+            }),
+            ..statistics(vec![empty, empty], empty)
+        };
+        assert!(
+            estimate_operator(PhysicalOperator::PromqlVectorBinary, vector_vector.clone()).is_ok()
+        );
+
+        vector_vector.hash_join_build_side = None;
+        assert_eq!(
+            estimate_operator(PhysicalOperator::PromqlVectorBinary, vector_vector),
+            Err(AnalyticalCostError::MissingOrStale(
+                "vector_binary_label_match_statistics"
+            ))
+        );
+    }
+
+    #[test]
+    fn promql_scalar_leaf_requires_one_row_per_evaluation_step() {
+        use crate::analytical_statistics::PromqlOperatorStatistics;
+
+        let mut scalar = statistics(vec![], EdgeStatistics { rows: 9, bytes: 72 });
+        scalar.promql = Some(PromqlOperatorStatistics {
+            input_series: vec![],
+            output_series: 0,
+            evaluation_steps: 10,
+            window_samples_per_series: None,
+            subquery_steps: None,
+            scalar_ops_per_row: None,
+            binary_operand_mode: None,
+        });
+        assert!(matches!(
+            estimate_operator(PhysicalOperator::PromqlScalarLeaf, scalar),
+            Err(AnalyticalCostError::InconsistentOperatorStatistics(_))
+        ));
+    }
+
     #[test]
     fn comparison_rejects_different_snapshot_predicate_time_or_horizon() {
         use std::rc::Rc;

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

@@ -8,7 +8,8 @@ use crate::analytical_cost::{
     AnalyticalCostError, ExecutionMultiplicity, PhysicalDagNode, PhysicalOperator,
 };
 use crate::analytical_statistics::{
-    ComparisonScope, EdgeStatistics, OperatorStatistics, OperatorStatisticsProvider, SourceCoverage,
+    ComparisonScope, EdgeStatistics, OperatorStatistics, OperatorStatisticsProvider,
+    PromqlBinaryOperandMode, SourceCoverage,
 };
 
 /// A lowered physical DAG and the node whose output is the query result.
@@ -605,6 +606,19 @@ pub fn lower_query_physical_dag(
                             "PromQL binary inputs do not match child outputs",
                         ));
                     }
+                    let operand_mode = match (left_scalar, right_scalar) {
+                        (false, false) => PromqlBinaryOperandMode::VectorVector,
+                        (false, true) => PromqlBinaryOperandMode::VectorScalar,
+                        (true, false) => PromqlBinaryOperandMode::ScalarVector,
+                        (true, true) => unreachable!("scalar/scalar was rejected above"),
+                    };
+                    if require_promql_statistics(statistics, 2)?.binary_operand_mode
+                        != Some(operand_mode)
+                    {
+                        return Err(AnalyticalCostError::InconsistentOperatorStatistics(
+                            "PromQL binary operand mode disagrees with its logical operands",
+                        ));
+                    }
                     if left_scalar || right_scalar {
                         if vector_match.is_some()
                             || statistics.hash_join_build_side.is_some()
@@ -895,7 +909,10 @@ fn require_operator_statistics(
         return invalid("only Scan may charge source bytes");
     }
     if matches!(operator, PhysicalOperator::PromqlScalarLeaf) {
-        require_promql_statistics(statistics, 0)?;
+        let promql = require_promql_statistics(statistics, 0)?;
+        if promql.output_series != 0 || statistics.output.rows != promql.evaluation_steps {
+            return invalid("PromQL scalar leaf must emit one scalar row per evaluation step");
+        }
         return Ok(());
     }
     let input = statistics.inputs.first().copied().ok_or(
@@ -982,7 +999,43 @@ fn require_operator_statistics(
             }
         }
         PhysicalOperator::PromqlVectorBinary => {
-            require_promql_statistics(statistics, 2)?;
+            let promql = require_promql_statistics(statistics, 2)?;
+            let mode = promql
+                .binary_operand_mode
+                .ok_or(AnalyticalCostError::MissingOrStale(
+                    "promql_binary_operand_mode",
+                ))?;
+            match mode {
+                PromqlBinaryOperandMode::VectorVector => {
+                    if statistics.hash_join_build_side.is_none() || statistics.key_bytes.is_none() {
+                        return Err(AnalyticalCostError::MissingOrStale(
+                            "vector_binary_label_match_statistics",
+                        ));
+                    }
+                }
+                PromqlBinaryOperandMode::VectorScalar => {
+                    if promql.input_series[1] != 0
+                        || statistics.inputs[1].rows != promql.evaluation_steps
+                        || statistics.hash_join_build_side.is_some()
+                        || statistics.key_bytes.is_some()
+                    {
+                        return invalid(
+                            "vector/scalar binary evidence has an invalid scalar edge or label-match state",
+                        );
+                    }
+                }
+                PromqlBinaryOperandMode::ScalarVector => {
+                    if promql.input_series[0] != 0
+                        || statistics.inputs[0].rows != promql.evaluation_steps
+                        || statistics.hash_join_build_side.is_some()
+                        || statistics.key_bytes.is_some()
+                    {
+                        return invalid(
+                            "scalar/vector binary evidence has an invalid scalar edge or label-match state",
+                        );
+                    }
+                }
+            }
         }
         PhysicalOperator::PromqlInfoEnrich => {
             let promql = require_promql_statistics(statistics, 2)?;
@@ -1441,6 +1494,7 @@ mod tests {
             window_samples_per_series: None,
             subquery_steps: None,
             scalar_ops_per_row: None,
+            binary_operand_mode: None,
         });
         statistics
     }
@@ -2333,10 +2387,15 @@ mod tests {
         let empty_scope = scope(vec![]);
         let scalar = Rc::new(QueryExpr::promql_scalar(5.0));
         let scalar_statistics = with_promql(statistics(vec![], edge(10, 80)), vec![], 0, 10);
-        let provided = HashMap::from([("query-0".into(), evidence(scalar_statistics))]);
+        let mut provided = HashMap::from([("query-0".into(), evidence(scalar_statistics))]);
         let dag = lower_query_physical_dag(&scalar, &empty_scope, &scripted(&provided)).unwrap();
         assert_eq!(dag.nodes[0].operator, PhysicalOperator::PromqlScalarLeaf);
         assert!(estimate_physical_dag(&dag.nodes, &dag.root, &empty_scope, &dag.evidence).is_ok());
+        provided.get_mut("query-0").unwrap().statistics.output = edge(9, 72);
+        assert!(matches!(
+            lower_query_physical_dag(&scalar, &empty_scope, &scripted(&provided)),
+            Err(AnalyticalCostError::InconsistentOperatorStatistics(_))
+        ));
 
         let timestamp = Rc::new(QueryExpr::EvalTimestamp);
         let timestamp_statistics = with_promql(statistics(vec![], edge(10, 80)), vec![], 0, 10);
@@ -2549,12 +2608,14 @@ mod tests {
         );
         vector_statistics.source_scan_bytes = 8_000;
         let scalar = with_promql(statistics(vec![], edge(10, 80)), vec![], 0, 10);
-        let binary = with_promql(
+        let mut binary = with_promql(
             statistics(vec![edge(1_000, 16_000), edge(10, 80)], edge(1_000, 16_000)),
             vec![100, 0],
             100,
             10,
         );
+        binary.promql.as_mut().unwrap().binary_operand_mode =
+            Some(PromqlBinaryOperandMode::VectorScalar);
         let provided = HashMap::from([
             ("query-1".into(), evidence(vector_statistics)),
             ("query-2".into(), evidence(scalar)),
@@ -2649,6 +2710,8 @@ mod tests {
         );
         binary.key_bytes = Some(32);
         binary.hash_join_build_side = Some(HashJoinBuildSide::Right);
+        binary.promql.as_mut().unwrap().binary_operand_mode =
+            Some(PromqlBinaryOperandMode::VectorVector);
         let provided = HashMap::from([
             ("query-1".into(), evidence(left)),
             ("query-2".into(), evidence(right)),
@@ -2659,6 +2722,32 @@ mod tests {
         assert_eq!(estimate.cpu_ops, 3_400.0);
         assert_eq!(estimate.scan_bytes, 12_000);
 
+        let mut empty = provided.clone();
+        for id in ["query-1", "query-2"] {
+            let evidence = empty.get_mut(id).unwrap();
+            evidence.statistics.inputs[0] = edge(0, 0);
+            evidence.statistics.output = edge(0, 0);
+            let promql = evidence.statistics.promql.as_mut().unwrap();
+            promql.input_series[0] = 0;
+            promql.output_series = 0;
+            evidence.output_buffer_bytes = 0;
+        }
+        let evidence = empty.get_mut("query-0").unwrap();
+        evidence.statistics.inputs = vec![edge(0, 0), edge(0, 0)];
+        evidence.statistics.output = edge(0, 0);
+        let promql = evidence.statistics.promql.as_mut().unwrap();
+        promql.input_series = vec![0, 0];
+        promql.output_series = 0;
+        evidence.output_buffer_bytes = 0;
+        let empty_dag = lower_query_physical_dag(&root, &scope, &scripted(&empty)).unwrap();
+        assert!(estimate_physical_dag(
+            &empty_dag.nodes,
+            &empty_dag.root,
+            &scope,
+            &empty_dag.evidence
+        )
+        .is_ok());
+
         let mut missing = provided;
         missing
             .get_mut("query-0")

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

@@ -229,6 +229,17 @@ pub struct PromqlOperatorStatistics {
     pub window_samples_per_series: Option<u64>,
     pub subquery_steps: Option<u64>,
     pub scalar_ops_per_row: Option<u64>,
+    /// Physical operand shape for a PromQL binary operator. Series cardinality
+    /// cannot encode this because a vector may legitimately contain zero series.
+    #[serde(default)]
+    pub binary_operand_mode: Option<PromqlBinaryOperandMode>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub enum PromqlBinaryOperandMode {
+    VectorVector,
+    VectorScalar,
+    ScalarVector,
 }
 
 /// Resolves physical statistics and owns their catalog/observation freshness.

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

@@ -161,6 +161,7 @@ OperatorStatistics {
     promql: PromqlOperatorStatistics {
         input_series, output_series, evaluation_steps,
         window_samples_per_series, subquery_steps, scalar_ops_per_row,
+        binary_operand_mode,
     },
 }
 ```
@@ -353,6 +354,14 @@ both input and output contain one row per evaluation step. A vector-to-scalar
 bridge emits zero series and exactly one scalar row per evaluation step; its
 input-series count must equal the vector child's output-series count.
 
+PromQL binary evidence also records an explicit physical operand mode:
+vector/vector, vector/scalar, or scalar/vector. Cardinality does not determine
+the mode because a vector may legitimately contain zero series. Lowering
+checks the mode against the canonical logical operands. Vector/vector always
+requires label-match state, including for an empty vector; only the two
+scalar/vector modes omit it. A scalar leaf has no inputs or series and emits
+exactly one scalar row per evaluation step.
+
 `info()` resolves its default `target_info` metric, or one exact `__name__`
 matcher, to a concrete existing `Source::TimeSeries` coverage. Its right side
 is an ordinary physical Scan with its own statistics, buffer, snapshot, and