refactor: execute query DAGs with shared native physical operators

control_plane/src/clickhouse.rs

@@ -1453,14 +1453,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,

control_plane/src/physical/executable_binding.rs

@@ -4,44 +4,17 @@ pub use asap_types::executable_plan::*;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum OperatorExecution {
-    Maintenance,
+    Ingestion,
     Query,
 }
 
-/// Keep the backend's ownership decision exhaustive over Planner's physical IR.
-/// Adding a payload variant upstream must therefore choose an executor here.
-fn operator_execution(
-    node: &planner_types::post_asap::ExecutableDagNode,
-) -> Result<OperatorExecution, String> {
-    use planner_types::post_asap::{ExecutableOperatorPayload as Payload, ExecutionTiming};
-
-    let declared = match &node.payload {
-        Payload::Binary { timing, .. }
-        | Payload::Value { timing, .. }
-        | Payload::SummaryMerge { timing } => *timing,
-        Payload::MembershipFilter { .. } | Payload::SummaryEstimate { .. } => {
-            ExecutionTiming::QueryTime
-        }
-        Payload::SummaryAgg { .. }
-        | Payload::SummaryJoin { .. }
-        | Payload::SummarySubtract
-        | Payload::SummaryDelete { .. } => ExecutionTiming::IngestionTime,
-        // These operators can be placed on either side of the stored-state
-        // boundary. Planner's validated output state is authoritative.
-        Payload::Fallback { .. } | Payload::RelationalJoin { .. } => node.output_state.timing,
-    };
-    if declared != node.output_state.timing {
-        return Err(format!(
-            "post-ASAP node {:?} has operator timing {} but output state {}",
-            node.id,
-            declared.as_str(),
-            node.output_state
-        ));
+/// Every physical operator uses its node's placement; payload kind does not
+/// restrict execution phase. Runtime capability is checked separately.
+fn operator_execution(node: &planner_types::post_asap::ExecutableDagNode) -> OperatorExecution {
+    match node.output_state.timing {
+        planner_types::post_asap::ExecutionTiming::IngestionTime => OperatorExecution::Ingestion,
+        planner_types::post_asap::ExecutionTiming::QueryTime => OperatorExecution::Query,
     }
-    Ok(match declared {
-        ExecutionTiming::IngestionTime => OperatorExecution::Maintenance,
-        ExecutionTiming::QueryTime => OperatorExecution::Query,
-    })
 }
 
 /// Assign backend phases to a selected semantic DAG without changing its nodes.
@@ -67,11 +40,11 @@ pub fn install_selected_dag(
             asap_physical_operators::capability::validate_summary_kernel(family, input, grouping)
                 .map_err(|reason| format!("post-ASAP node {:?}: {reason}", node.id))?;
         }
-        let execution = operator_execution(node)?;
+        let execution = operator_execution(node);
         let binding = if let Some(summary_definition) = materialization(node.id) {
             precompute_sinks.push(node.id);
             BackendNodeBinding::Materialization { summary_definition }
-        } else if execution == OperatorExecution::Maintenance {
+        } else if execution == OperatorExecution::Ingestion {
             BackendNodeBinding::MaintenanceInput
         } else {
             query_node(node.id).map_or(BackendNodeBinding::QueryInput, |query_node| {

control_plane/src/query_plan.rs

@@ -386,6 +386,7 @@ where
                     child.schema.fields.get(column).map(|field| &field.dtype),
                     Some(SummaryFamilyType::Plain(
                         planner_types::pre_asap::DataType::Float64
+                            | planner_types::pre_asap::DataType::Int64
                     )) | Some(SummaryFamilyType::ExactAggregate(..))
                 ) {
                     return Err(QueryPlanError::Invalid(

crates/asap_types/src/query_plan.rs

@@ -366,6 +366,7 @@ impl QueryPlanEntry {
             )));
         }
         for (id, node) in &self.nodes {
+            validate_native_relation(*id, node)?;
             if let QueryPlanNode::Logical { operator, inputs } = node {
                 operator.validate(inputs.len())?;
             }
@@ -746,3 +747,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::dag::planner::bind_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());
+    }
+}

data_plane/src/precompute_engine/worker.rs

@@ -4624,9 +4624,9 @@ mod dag_execution_tests {
         .unwrap();
         installed.document.schema_version =
             asap_types::executable_plan::MAINTENANCE_DAG_SCHEMA_VERSION;
-        let error = InstalledPrecomputePlan::from_precompute_plan(plan)
+        assert!(InstalledPrecomputePlan::from_precompute_plan(plan)
             .unwrap_err()
-            .to_string();
-        assert!(error.contains("update"), "{error}");
+            .to_string()
+            .contains("update"));
     }
 }

data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs

@@ -1,5 +1,6 @@
 //! Catalog-backed ClickHouse acceleration boundary.
 
+use asap_physical_operators::accumulators::SumAccumulator;
 use async_trait::async_trait;
 use axum::{
     body::Bytes,
@@ -856,7 +857,12 @@ mod tests {
             "SELECT sum(value) FROM requests WHERE timestamp >= 2000 AND timestamp < 3000".into();
         let uncovered = accelerator.execute(&request).await;
         assert!(
-            matches!(&uncovered, ClickHouseAccelerationOutcome::Fallback(ClickHouseAccelerationFallback::Execution(detail)) if detail.contains("NoCandidates")),
+            matches!(
+                &uncovered,
+                ClickHouseAccelerationOutcome::Fallback(
+                    ClickHouseAccelerationFallback::IncompleteCoverage
+                )
+            ),
             "{uncovered:?}"
         );
         request.sql =