refactor: share installed producer and publication contracts

control_plane/examples/audit_clickhouse_corpus.rs

@@ -4,7 +4,7 @@ use asap_frontend_sql::SqlCatalog;
 use asap_types::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind};
 use control_plane::clickhouse::{ClickHouseSqlWorkload, ClickHouseSqlWorkloadEntry};
 use control_plane::physical::compiler::{
-    PlanEnvelope, PrecomputePlan, TransmissionPlan, BACKEND_COMPAT, PLANNER_REVISION,
+    PlanEnvelope, PrecomputePlan, BACKEND_COMPAT, PLANNER_REVISION,
 };
 use planner_types::pre_asap::{Column, DataType, Schema};
 use planner_types::types::AccuracyTarget;
@@ -56,8 +56,12 @@ fn publication_inputs(schema: &Schema, sql: String) -> ClickHouseSqlWorkload {
     let mut precompute_plan =
         PrecomputePlan::build_backend_local(envelope.clone(), vec![materialization.clone()])
             .unwrap();
-    let mut transmission_plan =
-        TransmissionPlan::build(envelope, &precompute_plan, &Default::default()).unwrap();
+    let mut transmission_plan = control_plane::physical::compiler::compile_transmission_plan(
+        envelope,
+        &precompute_plan,
+        &Default::default(),
+    )
+    .unwrap();
     let sds = asap_types::summary_catalog::SummaryCatalog::from_materializations(
         27,
         1,

control_plane/src/clickhouse.rs

@@ -183,7 +183,7 @@ pub async fn compile_automatic_clickhouse_workload(
             .map_err(|error| ClickHousePlanningError::Lower(error.to_string()))?,
     );
     precompute.executable_dags = installed_dags;
-    let mut transmission = TransmissionPlan::build(
+    let mut transmission = crate::physical::compiler::compile_transmission_plan(
         request.envelope.clone(),
         &precompute,
         &std::collections::BTreeMap::new(),
@@ -1010,9 +1010,12 @@ mod tests {
         let mut precompute =
             PrecomputePlan::build_backend_local(envelope.clone(), vec![config]).unwrap();
         precompute.summary_catalog = Some(sds.reference().unwrap());
-        let mut transmission =
-            TransmissionPlan::build(envelope, &precompute, &std::collections::BTreeMap::new())
-                .unwrap();
+        let mut transmission = crate::physical::compiler::compile_transmission_plan(
+            envelope,
+            &precompute,
+            &std::collections::BTreeMap::new(),
+        )
+        .unwrap();
         transmission.summary_catalog = Some(sds.reference().unwrap());
         let timestamped = |time_name: &str, value_name: &str| {
             Schema::with_time_index(
@@ -1204,7 +1207,7 @@ mod tests {
         request.precompute_plan =
             PrecomputePlan::build_backend_local(envelope.clone(), vec![config]).unwrap();
         request.precompute_plan.summary_catalog = Some(request.sds.reference().unwrap());
-        request.transmission_plan = TransmissionPlan::build(
+        request.transmission_plan = crate::physical::compiler::compile_transmission_plan(
             envelope,
             &request.precompute_plan,
             &std::collections::BTreeMap::new(),

control_plane/src/emit/backend_push.rs

@@ -260,7 +260,7 @@ async fn push_documents_coupled(
         warn!(%error, "failed to bind compatibility PrecomputePlan to SummaryCatalog");
         return (false, false, 0);
     }
-    let transmission_plan = match crate::physical::compiler::TransmissionPlan::build(
+    let transmission_plan = match crate::physical::compiler::compile_transmission_plan(
         precompute_plan.envelope.clone(),
         &precompute_plan,
         &Default::default(),