adapter: apply missing optimizer feature overrides (#35181)

For indexes, MVs, and CTs, optimizer features can be overridden on a
per-cluster basis. This PR fixes two places where we forgot to apply
these overrides. This fixes a bug where dataflows would be incorrectly
re-planned after an envd restart because we'd drop their cluster
overrides.

Author: teskje

src/adapter/src/catalog/state.rs

@@ -55,7 +55,7 @@ use mz_repr::namespaces::{
     UNSTABLE_SCHEMAS,
 };
 use mz_repr::network_policy_id::NetworkPolicyId;
-use mz_repr::optimize::OptimizerFeatures;
+use mz_repr::optimize::{OptimizerFeatures, OverrideFrom};
 use mz_repr::role_id::RoleId;
 use mz_repr::{
     CatalogItemId, GlobalId, RelationDesc, RelationVersion, RelationVersionSelector,
@@ -1328,8 +1328,13 @@ impl CatalogState {
                     .collect();
 
                 // Collect optimizer parameters.
+                let system_vars = session_catalog.system_vars();
+                let overrides = self
+                    .get_cluster(materialized_view.cluster_id)
+                    .config
+                    .features();
                 let optimizer_config =
-                    optimize::OptimizerConfig::from(session_catalog.system_vars());
+                    optimize::OptimizerConfig::from(system_vars).override_from(&overrides);
                 let previous_exprs = previous_item.map(|item| match item {
                     CatalogItem::MaterializedView(materialized_view) => {
                         (materialized_view.raw_expr, materialized_view.optimized_expr)

src/adapter/src/coord.rs

@@ -134,7 +134,7 @@ use mz_persist_client::usage::{ShardsUsageReferenced, StorageUsageClient};
 use mz_repr::adt::numeric::Numeric;
 use mz_repr::explain::{ExplainConfig, ExplainFormat};
 use mz_repr::global_id::TransientIdGen;
-use mz_repr::optimize::OptimizerFeatures;
+use mz_repr::optimize::{OptimizerFeatures, OverrideFrom};
 use mz_repr::role_id::RoleId;
 use mz_repr::{CatalogItemId, Diff, GlobalId, RelationDesc, SqlRelationType, Timestamp};
 use mz_secrets::cache::CachingSecretsReader;
@@ -3050,7 +3050,11 @@ impl Coordinator {
         let mut instance_snapshots = BTreeMap::new();
         let mut uncached_expressions = BTreeMap::new();
 
-        let optimizer_config = OptimizerConfig::from(self.catalog().system_config());
+        let optimizer_config = |catalog: &Catalog, cluster_id| {
+            let system_config = catalog.system_config();
+            let overrides = catalog.get_cluster(cluster_id).config.features();
+            OptimizerConfig::from(system_config).override_from(&overrides)
+        };
 
         for entry in ordered_catalog_entries {
             match entry.item() {
@@ -3069,6 +3073,8 @@ impl Coordinator {
                         continue;
                     }
 
+                    let optimizer_config = optimizer_config(&self.catalog, idx.cluster_id);
+
                     let (optimized_plan, physical_plan, metainfo) =
                         match cached_global_exprs.remove(&global_id) {
                             Some(global_expressions)
@@ -3089,7 +3095,7 @@ impl Coordinator {
                                         self.owned_catalog(),
                                         compute_instance.clone(),
                                         global_id,
-                                        optimizer_config.clone(),
+                                        optimizer_config,
                                         self.optimizer_metrics(),
                                     );
 
@@ -3154,6 +3160,8 @@ impl Coordinator {
                         });
                     let global_id = mv.global_id_writes();
 
+                    let optimizer_config = optimizer_config(&self.catalog, mv.cluster_id);
+
                     let (optimized_plan, physical_plan, metainfo) =
                         match cached_global_exprs.remove(&global_id) {
                             Some(global_expressions)
@@ -3186,7 +3194,7 @@ impl Coordinator {
                                         mv.non_null_assertions.clone(),
                                         mv.refresh_schedule.clone(),
                                         debug_name,
-                                        optimizer_config.clone(),
+                                        optimizer_config,
                                         self.optimizer_metrics(),
                                         force_non_monotonic,
                                     );
@@ -3252,6 +3260,8 @@ impl Coordinator {
                         });
                     let global_id = ct.global_id();
 
+                    let optimizer_config = optimizer_config(&self.catalog, ct.cluster_id);
+
                     let (optimized_plan, physical_plan, metainfo) =
                         match cached_global_exprs.remove(&global_id) {
                             Some(global_expressions)

test/cluster/mzcompose.py

@@ -6467,3 +6467,51 @@ def workflow_github_10086(c: Composition) -> None:
         """
     )[0][0]
     assert not upper_empty
+
+
+def workflow_test_optimizer_feature_override_after_restart(c: Composition) -> None:
+    """
+    Test that verifies that optimizer feature overrides survive envd restarts.
+    """
+
+    with c.override(
+        Materialized(
+            additional_system_parameter_defaults={
+                "enable_eager_delta_joins": "false",
+            },
+        ),
+    ):
+        c.up("materialized")
+
+        c.sql(
+            """
+            CREATE CLUSTER test (SIZE 'scale=1,workers=1')
+                FEATURES (ENABLE EAGER DELTA JOINS = true);
+            GRANT ALL ON CLUSTER test TO materialize;
+            """,
+            port=6877,
+            user="mz_system",
+        )
+
+        c.sql(
+            """
+            SET cluster = test;
+
+            CREATE TABLE t1 (x int, y int);
+            CREATE TABLE t2 (x int, y int);
+            CREATE TABLE t3 (x int, y int);
+            CREATE MATERIALIZED VIEW mv2 IN CLUSTER test AS
+                SELECT t1.y as f1, t2.y as f2, t3.y as f3
+                FROM t1, t2, t3
+                WHERE t1.x = t2.x AND t2.y = t3.y;
+            """
+        )
+
+        plan1 = c.sql_query("EXPLAIN MATERIALIZED VIEW mv2")[0][0]
+
+        c.kill("materialized")
+        c.up("materialized")
+
+        plan2 = c.sql_query("EXPLAIN MATERIALIZED VIEW mv2")[0][0]
+
+        assert plan1 == plan2, f"before={plan1}\nafter={plan2}"