Execute query DAGs with consistent SDS snapshots and request limits

control_plane/src/clickhouse.rs

@@ -1413,7 +1413,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn compiles_summary_joined_with_exact_table_into_mixed_dag() {
+    async fn rejects_mixed_snapshots_and_preserves_local_sql_contracts() {
         let config = materialization(
             AggregationType::Sum,
             "value",
@@ -1461,14 +1461,10 @@ mod tests {
                 ("telemetry".into(), timestamped("timestamp_ms", "value")),
                 (
                     "divisors".into(),
-                    Schema::with_time_index(
-                        vec![
-                            Column::new("timestamp", DataType::Int64, false),
-                            Column::new("divisor", DataType::Float64, false),
-                        ],
-                        0,
-                        vec![],
-                    ),
+                    Schema::new(vec![
+                        Column::new("timestamp", DataType::Int64, false),
+                        Column::new("divisor", DataType::Float64, false),
+                    ]),
                 ),
             ]),
             accuracy: AccuracyTarget::Exact,
@@ -1479,6 +1475,16 @@ mod tests {
                 cumulative: true,
             }],
         };
+        let Err(error) = compile_clickhouse_workload(&request).await else {
+            panic!("local summary joined to an external table needs a common snapshot proof");
+        };
+        assert!(
+            error.to_string().contains("common snapshot proof"),
+            "{error}"
+        );
+        request.queries[0].sql =
+            "SELECT sum(value) FROM telemetry WHERE timestamp_ms >= 0 AND timestamp_ms < 2000"
+                .into();
         let publication = compile_clickhouse_workload(&request).await.unwrap();
         // A simple installed aggregate publishes the same key as a refresh.
         let simple_sql =
@@ -1682,31 +1688,11 @@ mod tests {
                 > installed.document.nodes.len()
         );
         let entry = publication.query_plan.entries.values().next().unwrap();
-        // External SQL retains its literal time range until it can be bound.
-        assert!(!entry.canonical_query.starts_with("moving-window-v1:"));
-        assert!(entry
+        assert!(!entry
             .nodes
             .values()
             .any(|node| matches!(node, crate::query_plan::QueryPlanNode::ExternalExact { .. })));
-        assert!(entry.nodes.values().any(|node| matches!(
-            node,
-            crate::query_plan::QueryPlanNode::ReadMaterialization { .. }
-        )));
-        assert!(entry.nodes.values().any(|node| matches!(
-            node,
-            crate::query_plan::QueryPlanNode::RelationalJoin { .. }
-        )));
-        assert!(entry.nodes.values().any(|node| {
-            let crate::query_plan::QueryPlanNode::Relational { operation, .. } = node else {
-                return false;
-            };
-            matches!(
-                serde_json::from_value::<planner_types::post_asap::ValueOperation>(
-                    operation.clone()
-                ),
-                Ok(planner_types::post_asap::ValueOperation::Project { .. })
-            )
-        }));
+        assert!(!entry.materialization_bindings().is_empty());
         let original = request.queries[0].sql.clone();
         let schema = request.tables.get_mut("telemetry").unwrap();
         schema.columns[schema.time_index.unwrap()].name = "other_timestamp".into();

control_plane/src/physical/compiler.rs

@@ -5064,36 +5064,18 @@ pub(crate) mod tests {
     }
 
     #[test]
-    fn metricsql_counter_gate_preserves_an_independent_summary_sibling() {
+    fn metricsql_counter_gate_rejects_mixed_input_snapshots() {
         let mut workload = request("mixed", "max_over_time(m[1m]) + rate(m[1m])");
         workload.allow_mixed_summary_and_exact_execution = true;
         let mut deployment = environment(10_000);
         deployment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
         deployment.target_collector_ids.clear();
-        let plan = DeploymentPlanCompiler
-            .compile_metricsql(workload, deployment)
-            .unwrap();
-        assert!(!plan.precompute_plan.materializations.is_empty());
-        assert!(plan
-            .precompute_plan
-            .materializations
-            .iter()
-            .all(|m| !matches!(
-                m.aggregation_type,
-                asap_types::AggregationType::Increase | asap_types::AggregationType::Rate
-            )));
-        let entry = plan.query_plan.entries.values().next().unwrap();
-        assert!(!entry.materialization_bindings().is_empty());
+        let Err(error) = DeploymentPlanCompiler.compile_metricsql(workload, deployment) else {
+            panic!("mixed local/external snapshots must fail deployment binding");
+        };
         assert!(
-            entry.nodes.values().any(|node| matches!(
-                node,
-                crate::query_plan::QueryPlanNode::ExternalExact { .. }
-                    | crate::query_plan::QueryPlanNode::Logical {
-                        operator:
-                            crate::query_plan::residual::ResidualQueryOperator::ExactSubquery { .. },
-                        ..
-                    }
-            ))
+            error.to_string().contains("common snapshot proof"),
+            "{error}"
         );
     }
 

control_plane/src/physical/maintained_population.rs

@@ -84,6 +84,7 @@ pub(super) fn operator(
             without: input.without,
         },
         lookback_ms: input.lookback_ms,
+        history_retention_ms: request.query_retention_margin_ms,
         max_k: spec.max_k as u64,
         quantiles: spec.quantiles,
         max_bytes,

control_plane/src/physical/workload_cost.rs

@@ -1069,7 +1069,7 @@ mod tests {
     }
 
     #[test]
-    fn mixed_materialization_masks_price_state_and_prometheus_subquery_sources() {
+    fn materialization_masks_reject_mixed_snapshots_before_costing() {
         let mut snapshot = fixture();
         let q = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0];
         q.query =
@@ -1082,18 +1082,30 @@ mod tests {
         assert_eq!(
             candidates.len(),
             5,
-            "four legal candidate key sets plus native"
+            "four proposed candidate key sets plus native"
         );
         let mut identities = BTreeSet::new();
+        let mut rejected = 0;
         for candidate in &candidates[..4] {
             let enabled = candidate
                 .enabled_materialization_keys
                 .as_ref()
                 .unwrap()
                 .len();
-            let plan = DeploymentPlanCompiler
-                .compile_promql(candidate.clone(), environment.clone())
-                .unwrap();
+            let result =
+                DeploymentPlanCompiler.compile_promql(candidate.clone(), environment.clone());
+            if enabled == 1 {
+                let Err(error) = result else {
+                    panic!("mixed snapshots must fail binding");
+                };
+                assert!(
+                    error.to_string().contains("common snapshot proof"),
+                    "{error}"
+                );
+                rejected += 1;
+                continue;
+            }
+            let plan = result.unwrap();
             assert!(identities.insert(plan.envelope.plan_id));
             let cost = manifest(&plan, &candidate.queries).unwrap();
             assert_eq!(
@@ -1128,6 +1140,8 @@ mod tests {
                     crate::query_plan::QueryPlanNode::ExactFallback { .. }
                 ))));
         }
+        assert_eq!(rejected, 2);
+        assert_eq!(identities.len(), 2);
         assert!(
             !candidates
                 .last()

crates/asap_types/src/query_plan.rs

@@ -62,6 +62,13 @@ impl QueryPlan {
     }
 
     pub fn lookup(&self, promql: &str) -> Result<&QueryPlanEntry, QueryPlanError> {
+        // Installed entries already carry validated canonical identities. The
+        // common exact spelling requires no serving-time parser invocation.
+        if let Some(entry) = self.entries.get(promql).filter(|entry| {
+            entry.language == QueryLanguage::PromQl && entry.canonical_query == promql
+        }) {
+            return Ok(entry);
+        }
         let identity = canonical_promql(promql)?;
         self.lookup_canonical(QueryLanguage::PromQl, &identity)
     }
@@ -375,15 +382,48 @@ impl QueryPlanEntry {
         topological_order(root, &self.nodes)
     }
 
+    /// The current external adapters provide evaluation time, not a snapshot
+    /// token compatible with local SDS or current-series revisions.
+    pub fn validate_snapshot_sources(&self) -> Result<(), QueryPlanError> {
+        use residual::ResidualQueryOperator;
+        let mut local = false;
+        let mut external = false;
+        for id in self.topological_order()? {
+            match &self.nodes[&id] {
+                QueryPlanNode::ReadMaterialization { .. }
+                | QueryPlanNode::Logical {
+                    operator: ResidualQueryOperator::CurrentSeries { .. },
+                    ..
+                } => local = true,
+                QueryPlanNode::ExternalExact { .. }
+                | QueryPlanNode::Logical {
+                    operator:
+                        ResidualQueryOperator::ExactSubquery { .. }
+                        | ResidualQueryOperator::CandidateExactSubquery { .. },
+                    ..
+                } => external = true,
+                _ => {}
+            }
+        }
+        if local && external {
+            return Err(QueryPlanError::UnsupportedNode(
+                "local state and external exact input have no common snapshot proof".into(),
+            ));
+        }
+        Ok(())
+    }
+
     /// Validate references, bindings, reachability, and cycles before activation.
     pub fn validate(&self, available: &BTreeSet<PolicyFingerprint>) -> Result<(), QueryPlanError> {
+        self.validate_snapshot_sources()?;
         if !self.nodes.contains_key(&self.root) {
             return Err(QueryPlanError::Invalid(format!(
                 "query `{}` has missing root {}",
                 self.query_id, self.root.0
             )));
         }
         for (id, node) in &self.nodes {
+            validate_native_relation(*id, node)?;
             if let QueryPlanNode::PhysicalFragment {
                 inputs,
                 dag,
@@ -853,3 +893,99 @@ mod contract_tests {
         assert_send_sync::<super::QueryPlanEntry>();
     }
 }
+
+/// Bind portable relation semantics before an installed plan can access its sources.
+fn validate_native_relation(id: QueryNodeId, node: &QueryPlanNode) -> Result<(), QueryPlanError> {
+    use planner_types::post_asap::{
+        ExecutableDagNode, ExecutableOperatorPayload as Payload, ExecutionDataState, PostAsapNodeId,
+    };
+    use std::sync::Arc;
+    let invalid = |error: String| QueryPlanError::Invalid(format!("query node {}: {error}", id.0));
+    let (payload, inputs, output) = match node {
+        QueryPlanNode::Relational {
+            operation,
+            input_schema,
+            output_schema,
+            ..
+        } => (
+            Payload::Value {
+                operation: serde_json::from_value(operation.clone())
+                    .map_err(|e| invalid(e.to_string()))?,
+            },
+            vec![Arc::new(input_schema.clone())],
+            output_schema,
+        ),
+        QueryPlanNode::RelationalJoin {
+            join_kind,
+            pred,
+            left_schema,
+            right_schema,
+            output_schema,
+            pruning,
+            ..
+        } => (
+            Payload::RelationalJoin {
+                join_kind: join_kind.clone(),
+                pred: serde_json::from_value(pred.clone()).map_err(|e| invalid(e.to_string()))?,
+                pruning: serde_json::from_value(
+                    serde_json::to_value(pruning).map_err(|e| invalid(e.to_string()))?,
+                )
+                .map_err(|e| invalid(e.to_string()))?,
+            },
+            vec![
+                Arc::new(left_schema.clone()),
+                Arc::new(right_schema.clone()),
+            ],
+            output_schema,
+        ),
+        _ => return Ok(()),
+    };
+    let node = ExecutableDagNode {
+        id: PostAsapNodeId(0),
+        payload,
+        output_state: ExecutionDataState::QUERY_ROWS,
+        output_schema: output.clone(),
+        guarantee: None,
+    };
+    asap_physical_operators::physical_planner::compile_node(&node, &inputs)
+        .map_err(|e| invalid(e.to_string()))?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod native_binding_tests {
+    use super::*;
+    use planner_types::{
+        post_asap::{SummaryFamilyType, SummaryField, SummarySchema, ValueOperation},
+        pre_asap::{DataType, Predicate, QueryExpr},
+    };
+
+    // Unsupported expressions fail installation without evaluating any source.
+    #[test]
+    fn rejects_unimplemented_relation_predicate_before_execution() {
+        let schema = SummarySchema {
+            fields: vec![SummaryField {
+                name: "value".into(),
+                dtype: SummaryFamilyType::Plain(DataType::Float64),
+                nullable: false,
+            }],
+            time_index: None,
+        };
+        let node = QueryPlanNode::Relational {
+            input: QueryNodeId(0),
+            operation: serde_json::to_value(ValueOperation::Filter {
+                pred: Predicate(
+                    QueryExpr::FunctionCall {
+                        name: "unimplemented_predicate".into(),
+                        args: vec![],
+                    }
+                    .into(),
+                ),
+            })
+            .unwrap(),
+            input_schema: schema.clone(),
+            output_schema: schema,
+        };
+        assert!(validate_native_relation(QueryNodeId(1), &node).is_err());
+    }
+}

crates/asap_types/src/query_plan/current_series.rs

@@ -13,6 +13,9 @@ pub struct SeriesPopulation {
     pub grouping: Grouping,
     pub lookback_ms: u64,
     pub max_input_lag_ms: u64,
+    /// Admitted historical distance behind ingestion; zero retains current state only.
+    #[serde(default)]
+    pub history_retention_ms: u64,
     pub max_series: usize,
     pub max_bytes: u64,
     pub max_k: u64,
@@ -28,6 +31,7 @@ impl SeriesPopulation {
             || self.lookback_ms > i64::MAX as u64
             || self.max_input_lag_ms == 0
             || self.max_input_lag_ms > self.lookback_ms
+            || self.history_retention_ms > i64::MAX as u64
             || self.max_series == 0
             || self.max_series > 100_000
             || self.max_bytes == 0
@@ -67,6 +71,7 @@ mod tests {
             },
             lookback_ms: 1_000,
             max_input_lag_ms: 1_000,
+            history_retention_ms: 0,
             max_series: 100,
             max_bytes: 1_000_000,
             max_k: 3,

data_plane/examples/audit_clickhouse_fallback.rs

@@ -172,6 +172,14 @@ async fn main() {
                 "clickhouse_exception_code": headers.get("x-clickhouse-exception-code").and_then(|v| v.to_str().ok()),
                 "result": body,
             }),
+            ClickHouseAccelerationOutcome::Failed(error) => json!({
+                "id": row.id,
+                "fallback_requested": false,
+                "exact_executed": false,
+                "exact_success": false,
+                "execution_error": error.to_string(),
+                "result": body,
+            }),
             ClickHouseAccelerationOutcome::Accelerated(_) => json!({
                 "id": row.id,
                 "fallback_requested": false,

data_plane/src/drivers/ingest/prometheus_remote_write.rs

@@ -1003,6 +1003,7 @@ mod tests {
             snapshot_sha256: reference.snapshot_sha256,
         };
         let active = RuntimePhysicalPlan {
+            readout_programs: Default::default(),
             envelope: envelope.clone(),
             summary_catalog: Some(catalog),
             precompute_plan: PrecomputePlan {