adapter: gate materialized view key forwarding behind a dyncfg

misc/python/materialize/mzcompose/__init__.py

@@ -318,6 +318,9 @@ def get_variable_system_parameters(
             ["100", "1000", "10000"],
         ),
         VariableSystemParameter("persist_fast_path_order", "true", ["true", "false"]),
+        VariableSystemParameter(
+            "enable_materialized_view_keys", "false", ["true", "false"]
+        ),
         VariableSystemParameter(
             "persist_gc_use_active_gc",
             ("true" if version > MzVersion.parse_mz("v0.143.0-dev") else "false"),

misc/python/materialize/parallel_workload/action.py

@@ -1553,6 +1553,7 @@ def __init__(
             BOOLEAN_FLAG_VALUES
         )
         self.flags_with_values["enable_eager_delta_joins"] = BOOLEAN_FLAG_VALUES
+        self.flags_with_values["enable_materialized_view_keys"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_public_metrics_endpoint"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_scoped_system_parameters"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["persist_batch_structured_key_lower_len"] = [

src/adapter-types/src/dyncfgs.rs

@@ -373,10 +373,24 @@ pub const DEFAULT_CLUSTER_RECONFIGURATION_TIMEOUT: Config<Duration> = Config::ne
     "The reconfiguration deadline written when a config-shape ALTER CLUSTER omits WITH (WAIT ...).",
 );
 
+/// Whether a materialized view's persisted schema records the unique keys the
+/// optimizer inferred from its plan.
+///
+/// These keys are not stable. An optimizer change or a replacement materialized
+/// view can produce different real keys while downstream consumers were planned
+/// against the old ones, which is unsound. Defaults to off so no consumer relies
+/// on them.
+pub const ENABLE_MATERIALIZED_VIEW_KEYS: Config<bool> = Config::new(
+    "enable_materialized_view_keys",
+    false,
+    "Whether to record the optimizer-inferred unique keys in a materialized view's persisted schema.",
+);
+
 /// Adds the full set of all adapter `Config`s.
 pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
     configs
         .add(&ALLOW_USER_SESSIONS)
+        .add(&ENABLE_MATERIALIZED_VIEW_KEYS)
         .add(&ENABLE_CLUSTER_CONTROLLER)
         .add(&CLUSTER_CONTROLLER_TICK_INTERVAL)
         .add(&ENABLE_BACKGROUND_ALTER_CLUSTER)

src/adapter/src/catalog/state.rs

@@ -1411,6 +1411,9 @@ impl CatalogState {
                     .features();
                 let optimizer_config =
                     optimize::OptimizerConfig::from(system_vars).override_from(&overrides);
+                // Read before `optimizer_config` is moved into the optimizer below.
+                let enable_materialized_view_keys =
+                    optimizer_config.enable_materialized_view_keys();
                 let previous_exprs = previous_item.map(|item| match item {
                     CatalogItem::MaterializedView(materialized_view) => (
                         materialized_view.raw_expr,
@@ -1456,6 +1459,12 @@ impl CatalogState {
                 for &i in &materialized_view.non_null_assertions {
                     typ.column_types[i].nullable = false;
                 }
+                if !enable_materialized_view_keys {
+                    // Keep the persisted schema consistent with the creation
+                    // path, which drops the unstable optimizer-inferred keys
+                    // when the feature is disabled.
+                    typ.keys.clear();
+                }
                 let desc = RelationDesc::new(typ, materialized_view.column_names);
                 let desc = VersionedRelationDesc::new(desc);
 

src/adapter/src/optimize.rs

@@ -63,7 +63,7 @@ pub mod view;
 use std::fmt::Debug;
 
 use mz_adapter_types::connection::ConnectionId;
-use mz_adapter_types::dyncfgs::PERSIST_FAST_PATH_ORDER;
+use mz_adapter_types::dyncfgs::{ENABLE_MATERIALIZED_VIEW_KEYS, PERSIST_FAST_PATH_ORDER};
 use mz_catalog::memory::objects::{CatalogCollectionEntry, CatalogEntry, Index};
 use mz_compute_types::ComputeInstanceId;
 use mz_compute_types::dataflows::DataflowDescription;
@@ -285,10 +285,21 @@ pub struct OptimizerConfig {
     persist_fast_path_order: bool,
     // Enable calculating with_snapshot metadata for subscribes.
     subscribe_snapshot_optimization: bool,
+    // Whether to record the optimizer-inferred unique keys in a materialized
+    // view's persisted schema.
+    enable_materialized_view_keys: bool,
     /// Optimizer feature flags.
     pub features: OptimizerFeatures,
 }
 
+impl OptimizerConfig {
+    /// Whether to record the optimizer-inferred unique keys in a materialized
+    /// view's persisted schema.
+    pub fn enable_materialized_view_keys(&self) -> bool {
+        self.enable_materialized_view_keys
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum OptimizeMode {
     /// A mode where the optimized statement is executed.
@@ -305,6 +316,7 @@ impl From<&SystemVars> for OptimizerConfig {
             no_fast_path: false,
             persist_fast_path_order: PERSIST_FAST_PATH_ORDER.get(vars.dyncfgs()),
             subscribe_snapshot_optimization: SUBSCRIBE_SNAPSHOT_OPTIMIZATION.get(vars.dyncfgs()),
+            enable_materialized_view_keys: ENABLE_MATERIALIZED_VIEW_KEYS.get(vars.dyncfgs()),
             features: OptimizerFeatures::from(vars),
         }
     }

src/adapter/src/optimize/materialized_view.rs

@@ -230,6 +230,13 @@ impl Optimize<LocalMirPlan> for Optimizer {
         for &i in self.non_null_assertions.iter() {
             rel_typ.column_types[i].nullable = false;
         }
+        if !self.config.enable_materialized_view_keys() {
+            // The optimizer-inferred keys are not stable across optimizer
+            // changes or replacement materialized views, so downstream
+            // consumers must not rely on them. Drop them before they reach the
+            // persisted schema.
+            rel_typ.keys.clear();
+        }
         let rel_desc = RelationDesc::new(rel_typ, self.column_names.clone());
 
         let mut df_builder = {