Merge branch 'stack/requested-766' into stack/requested-761

Author: zzylol

control_plane/tests/issue754_level1.rs

@@ -123,7 +123,7 @@ fn assert_selected_plan(name: &str, plan: &CompiledPhysicalPlan) -> Option<Strin
     for materialization in &plan.precompute_plan.materializations {
         let definition = plan
             .summary_catalog
-            .materializations
+            .definitions
             .get(&materialization.policy_fingerprint().into())
             .expect("precompute producer has no catalog definition");
         let descriptor =

crates/asap_types/src/derived_input.rs

@@ -37,11 +37,12 @@ impl DerivedInputIdentity {
         root: PostAsapNodeId,
         frontiers: &BTreeMap<PostAsapNodeId, SummaryDefinitionId>,
     ) -> Result<Self, String> {
-        if !matches!(
-            document.schema_version,
-            crate::executable_plan::OWNED_POST_ASAP_DAG_SCHEMA_VERSION
-                | crate::executable_plan::MAINTENANCE_DAG_SCHEMA_VERSION
-        ) {
+        if ![
+            crate::executable_plan::OWNED_POST_ASAP_DAG_SCHEMA_VERSION,
+            crate::executable_plan::MAINTENANCE_DAG_SCHEMA_VERSION,
+        ]
+        .contains(&document.schema_version)
+        {
             return Err("unsupported derived program document version".into());
         }
         let decoded = document.decode()?;

data_plane/examples/overhead/fixture.rs

@@ -78,13 +78,16 @@ impl Fixture {
         let plan = physical_fixture::artifact_from_materializations(vec![config.clone()]);
         let store = Arc::new(SketchStore::new());
         store
-            .install_summary_catalog(Arc::new(plan.summary_catalog.clone()))
+            .install_precompute_plan(
+                Arc::new(plan.summary_catalog.clone()),
+                &plan.precompute_plan,
+            )
             .map_err(anyhow::Error::msg)?;
         let skconfig = SketchConfig::DDSketch {
             relative_accuracy: 0.01,
         };
         store.register(SummarySeriesMetadata {
-            sid: 1,
+            storage_handle: 1,
             metric_name: "overhead_values".into(),
             group_by_keys: BTreeSet::new(),
             capability: Some(Capability::QuantileApprox(Some(SketchAlgorithm::DDSketch))),

data_plane/src/precompute_engine/subdag_scheduler.rs

@@ -114,7 +114,8 @@ where
     S: IdempotentCommitSink<V>,
 {
     let _span = tracing::debug_span!(target: "asap_runtime_debug", "precompute_dag",
-        output_count = outputs.len(), node_count = dag.nodes.len(), edge_count = dag.edges.len()).entered();
+        output_count = outputs.len(), node_count = dag.nodes.len(), edge_count = dag.edges.len())
+    .entered();
     tracing::debug!(target: "asap_runtime_debug", "precompute DAG execution started");
     let mut unique = BTreeSet::new();
     for (node, key) in outputs {
@@ -211,7 +212,19 @@ where
         let node = nodes
             .get(&id)
             .ok_or_else(|| ScheduleError::Invalid(format!("missing node {id}")))?;
-        let op = node.operator;
+        let op = match &node.payload {
+            ExecutableOperatorPayload::Fallback { .. } => "Fallback",
+            ExecutableOperatorPayload::Binary { .. } => "Binary",
+            ExecutableOperatorPayload::CandidateTopK { .. } => "CandidateTopK",
+            ExecutableOperatorPayload::Value { .. } => "Value",
+            ExecutableOperatorPayload::RelationalJoin { .. } => "RelationalJoin",
+            ExecutableOperatorPayload::SummaryAgg { .. } => "SummaryAgg",
+            ExecutableOperatorPayload::SummaryJoin { .. } => "SummaryJoin",
+            ExecutableOperatorPayload::SummarySubtract => "SummarySubtract",
+            ExecutableOperatorPayload::SummaryDelete { .. } => "SummaryDelete",
+            ExecutableOperatorPayload::SummaryEstimate { .. } => "SummaryEstimate",
+            ExecutableOperatorPayload::SummaryMerge => "SummaryMerge",
+        };
         if node.output_state == ExecutionDataState::READ_ROWS {
             return Err(ScheduleError::Invalid(format!(
                 "query-time node {id} in precompute dependency path"

data_plane/src/precompute_engine/worker.rs

@@ -570,7 +570,7 @@ impl Worker {
             return Ok(());
         }
         let state = self.group_states.get_mut(&sid).unwrap();
-        tracing::debug!(target: "asap_runtime_debug", worker_id = self.id, policy_fp = %agg_id,
+        tracing::debug!(target: "asap_runtime_debug", worker_id = self.id, policy_fp = %policy_fp,
             "precompute update route resolved");
         #[cfg(not(test))]
         if state.program.is_none() {

data_plane/src/storage_engines/sketch_db/sds.rs

@@ -158,8 +158,11 @@ impl SummaryDescriptorRegistry {
             snapshot_sha256: reference.snapshot_sha256,
         });
         *self.authoritative_catalog.write().unwrap() = Some((catalog, generation, outputs));
-        tracing::info!(plan_id = reference.plan_id, plan_version = reference.plan_version,
-            "storage descriptor catalog installed");
+        tracing::info!(
+            plan_id = reference.plan_id,
+            plan_version = reference.plan_version,
+            "storage descriptor catalog installed"
+        );
         Ok(())
     }
 
@@ -265,8 +268,10 @@ impl SummaryDescriptorRegistry {
             }
         };
 
-        let catalog_generation = authoritative.as_ref().map(|(_, generation, _)| generation.clone());
-        tracing::debug!(target: "asap_runtime_debug", sid = metadata.sid,
+        let catalog_generation = authoritative
+            .as_ref()
+            .map(|(_, generation, _)| generation.clone());
+        tracing::debug!(target: "asap_runtime_debug", sid = metadata.storage_handle,
             policy_fp = %metadata.policy_fp,
             summary_descriptor_hash = format_args!("{:016x}", xxhash_rust::xxh64::xxh64(summary_descriptor.id().canonical().as_bytes(), 0)),
             data_descriptor_hash = format_args!("{:016x}", xxhash_rust::xxh64::xxh64(data_id.canonical().as_bytes(), 0)),