add validation for cluster & keeper configs

Author: DatGuyJonathan

apps/framework-cli/src/framework/core/plan_validator.rs

@@ -70,12 +70,59 @@ fn validate_cluster_references(project: &Project, plan: &InfraPlan) -> Result<()
     Ok(())
 }
 
+/// Validates that replicated engines either have keeper path/replica name OR a cluster defined
+fn validate_replicated_engine_args(plan: &InfraPlan) -> Result<(), ValidationError> {
+    use crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine;
+
+    for table in plan.target_infra_map.tables.values() {
+        let needs_args = match &table.engine {
+            Some(ClickhouseEngine::ReplicatedMergeTree {
+                keeper_path,
+                replica_name,
+            }) => keeper_path.is_none() && replica_name.is_none(),
+            Some(ClickhouseEngine::ReplicatedReplacingMergeTree {
+                keeper_path,
+                replica_name,
+                ..
+            }) => keeper_path.is_none() && replica_name.is_none(),
+            Some(ClickhouseEngine::ReplicatedAggregatingMergeTree {
+                keeper_path,
+                replica_name,
+            }) => keeper_path.is_none() && replica_name.is_none(),
+            Some(ClickhouseEngine::ReplicatedSummingMergeTree {
+                keeper_path,
+                replica_name,
+                ..
+            }) => keeper_path.is_none() && replica_name.is_none(),
+            _ => false,
+        };
+
+        // If engine args are missing AND no cluster is defined, that's an error
+        if needs_args && table.cluster_name.is_none() {
+            return Err(ValidationError::TableValidation(format!(
+                "Table '{}' uses a replicated engine but neither cluster nor keeper path/replica name are specified.\n\
+                \n\
+                You must either:\n\
+                1. Specify a cluster in the table config: cluster = \"prod_cluster\"\n\
+                   (and define it in moose.config.toml)\n\
+                2. Or provide explicit keeper path and replica name in the engine config\n",
+                table.name
+            )));
+        }
+    }
+
+    Ok(())
+}
+
 pub fn validate(project: &Project, plan: &InfraPlan) -> Result<(), ValidationError> {
     stream::validate_changes(project, &plan.changes.streaming_engine_changes)?;
 
     // Validate cluster references
     validate_cluster_references(project, plan)?;
 
+    // Validate replicated engine args
+    validate_replicated_engine_args(plan)?;
+
     // Check for validation errors in OLAP changes
     for change in &plan.changes.olap_changes {
         if let OlapChange::Table(TableChange::ValidationError { message, .. }) = change {
@@ -306,4 +353,177 @@ mod tests {
             _ => panic!("Expected ClusterValidation error"),
         }
     }
+
+    // Helper to create a table with a specific engine
+    fn create_table_with_engine(
+        name: &str,
+        cluster_name: Option<String>,
+        engine: Option<crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine>,
+    ) -> Table {
+        Table {
+            name: name.to_string(),
+            columns: vec![Column {
+                name: "id".to_string(),
+                data_type: ColumnType::String,
+                required: true,
+                unique: false,
+                primary_key: true,
+                default: None,
+                annotations: vec![],
+                comment: None,
+                ttl: None,
+            }],
+            order_by: OrderBy::Fields(vec!["id".to_string()]),
+            partition_by: None,
+            sample_by: None,
+            engine,
+            version: Some(Version::from_string("1.0.0".to_string())),
+            source_primitive: PrimitiveSignature {
+                name: name.to_string(),
+                primitive_type: PrimitiveTypes::DataModel,
+            },
+            metadata: None,
+            life_cycle: LifeCycle::FullyManaged,
+            engine_params_hash: None,
+            table_settings: None,
+            indexes: vec![],
+            database: None,
+            table_ttl_setting: None,
+            cluster_name,
+        }
+    }
+
+    #[test]
+    fn test_replicated_engine_without_args_or_cluster_fails() {
+        use crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine;
+
+        let project = create_test_project(None);
+        let table = create_table_with_engine(
+            "test_table",
+            None,
+            Some(ClickhouseEngine::ReplicatedMergeTree {
+                keeper_path: None,
+                replica_name: None,
+            }),
+        );
+        let plan = create_test_plan(vec![table]);
+
+        let result = validate(&project, &plan);
+
+        assert!(result.is_err());
+        match result {
+            Err(ValidationError::TableValidation(msg)) => {
+                assert!(msg.contains("test_table"));
+                assert!(msg.contains("replicated engine"));
+                assert!(msg.contains("cluster"));
+            }
+            _ => panic!("Expected TableValidation error"),
+        }
+    }
+
+    #[test]
+    fn test_replicated_engine_with_cluster_but_no_args_succeeds() {
+        use crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine;
+
+        let project = create_test_project(Some(vec![ClusterConfig {
+            name: "prod_cluster".to_string(),
+        }]));
+        let table = create_table_with_engine(
+            "test_table",
+            Some("prod_cluster".to_string()),
+            Some(ClickhouseEngine::ReplicatedMergeTree {
+                keeper_path: None,
+                replica_name: None,
+            }),
+        );
+        let plan = create_test_plan(vec![table]);
+
+        let result = validate(&project, &plan);
+
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn test_replicated_engine_with_args_but_no_cluster_succeeds() {
+        use crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine;
+
+        let project = create_test_project(None);
+        let table = create_table_with_engine(
+            "test_table",
+            None,
+            Some(ClickhouseEngine::ReplicatedMergeTree {
+                keeper_path: Some("/clickhouse/tables/{database}/{table}".to_string()),
+                replica_name: Some("{replica}".to_string()),
+            }),
+        );
+        let plan = create_test_plan(vec![table]);
+
+        let result = validate(&project, &plan);
+
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn test_replicated_engine_with_both_args_and_cluster_succeeds() {
+        use crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine;
+
+        let project = create_test_project(Some(vec![ClusterConfig {
+            name: "prod_cluster".to_string(),
+        }]));
+        let table = create_table_with_engine(
+            "test_table",
+            Some("prod_cluster".to_string()),
+            Some(ClickhouseEngine::ReplicatedMergeTree {
+                keeper_path: Some("/clickhouse/tables/{database}/{table}".to_string()),
+                replica_name: Some("{replica}".to_string()),
+            }),
+        );
+        let plan = create_test_plan(vec![table]);
+
+        let result = validate(&project, &plan);
+
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn test_replicated_replacing_merge_tree_without_args_or_cluster_fails() {
+        use crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine;
+
+        let project = create_test_project(None);
+        let table = create_table_with_engine(
+            "test_table",
+            None,
+            Some(ClickhouseEngine::ReplicatedReplacingMergeTree {
+                keeper_path: None,
+                replica_name: None,
+                ver: Some("version".to_string()),
+                is_deleted: None,
+            }),
+        );
+        let plan = create_test_plan(vec![table]);
+
+        let result = validate(&project, &plan);
+
+        assert!(result.is_err());
+        match result {
+            Err(ValidationError::TableValidation(msg)) => {
+                assert!(msg.contains("test_table"));
+                assert!(msg.contains("replicated engine"));
+            }
+            _ => panic!("Expected TableValidation error"),
+        }
+    }
+
+    #[test]
+    fn test_non_replicated_engine_without_cluster_succeeds() {
+        use crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine;
+
+        let project = create_test_project(None);
+        let table = create_table_with_engine("test_table", None, Some(ClickhouseEngine::MergeTree));
+        let plan = create_test_plan(vec![table]);
+
+        let result = validate(&project, &plan);
+
+        assert!(result.is_ok());
+    }
 }

packages/py-moose-lib/moose_lib/dmv2/olap_table.py

@@ -121,6 +121,10 @@ class OlapConfig(BaseModel):
         life_cycle: Determines how changes in code will propagate to the resources.
         settings: Optional table-level settings that can be modified with ALTER TABLE MODIFY SETTING.
                   These are alterable settings that can be changed without recreating the table.
+        cluster: Optional cluster name for ON CLUSTER support in ClickHouse.
+                 Use this to enable replicated tables across ClickHouse clusters.
+                 The cluster must be defined in moose.config.toml (dev environment only).
+                 Example: cluster="prod_cluster"
     """
     order_by_fields: list[str] = []
     order_by_expression: Optional[str] = None
@@ -133,6 +137,8 @@ class OlapConfig(BaseModel):
     settings: Optional[dict[str, str]] = None
     # Optional table-level TTL expression (without leading 'TTL')
     ttl: Optional[str] = None
+    # Optional cluster name for ON CLUSTER support in ClickHouse
+    cluster: Optional[str] = None
 
     # Optional secondary/data-skipping indexes
     class TableIndex(BaseModel):

packages/py-moose-lib/moose_lib/internal.py

@@ -198,6 +198,7 @@ class TableConfig(BaseModel):
         metadata: Optional metadata for the table.
         life_cycle: Lifecycle management setting for the table.
         table_settings: Optional table-level settings that can be modified with ALTER TABLE MODIFY SETTING.
+        cluster: Optional cluster name for ON CLUSTER support in ClickHouse.
     """
     model_config = model_config
 
@@ -214,6 +215,7 @@ class TableConfig(BaseModel):
     indexes: list[OlapConfig.TableIndex] = []
     ttl: Optional[str] = None
     database: Optional[str] = None
+    cluster: Optional[str] = None
 
 
 class TopicConfig(BaseModel):
@@ -696,6 +698,7 @@ def to_infra_map() -> dict:
             indexes=table.config.indexes,
             ttl=table.config.ttl,
             database=table.config.database,
+            cluster=table.config.cluster,
         )
 
     for name, stream in get_streams().items():