testdrive: Disable expr cache in consistency check

This commit disables the expression cache in testdrive during catalog
consistency checks. We were noticing elevated CPU usage in testdrive
after merging the expression cache during catalog consistency checks.
The expression cache is still used by environmentd during testdrive
tests unless it's disabled.

Resolves #MaterializeInc/database-issues/8759

src/adapter/src/catalog.rs

@@ -550,6 +550,7 @@ impl Catalog {
             environment_id,
             system_parameter_defaults,
             bootstrap_args,
+            None,
         )
         .await
     }
@@ -580,6 +581,7 @@ impl Catalog {
             environment_id,
             system_parameter_defaults,
             bootstrap_args,
+            None,
         )
         .await
     }
@@ -595,6 +597,7 @@ impl Catalog {
         system_parameter_defaults: BTreeMap<String, String>,
         version: semver::Version,
         bootstrap_args: &BootstrapArgs,
+        enable_expression_cache_override: Option<bool>,
     ) -> Result<Catalog, anyhow::Error> {
         let openable_storage = TestCatalogStateBuilder::new(persist_client.clone())
             .with_organization_id(environment_id.organization_id())
@@ -609,6 +612,7 @@ impl Catalog {
             Some(environment_id),
             system_parameter_defaults,
             bootstrap_args,
+            enable_expression_cache_override,
         )
         .await
     }
@@ -620,6 +624,7 @@ impl Catalog {
         environment_id: Option<EnvironmentId>,
         system_parameter_defaults: BTreeMap<String, String>,
         bootstrap_args: &BootstrapArgs,
+        enable_expression_cache_override: Option<bool>,
     ) -> Result<Catalog, anyhow::Error> {
         let metrics_registry = &MetricsRegistry::new();
         let active_connection_count = Arc::new(std::sync::Mutex::new(ConnectionCounter::new(0, 0)));
@@ -665,6 +670,7 @@ impl Catalog {
                 active_connection_count,
                 builtin_item_migration_config: BuiltinItemMigrationConfig::Legacy,
                 persist_client,
+                enable_expression_cache_override,
                 helm_chart_version: None,
             },
         })

src/adapter/src/catalog/open.rs

@@ -435,7 +435,9 @@ impl Catalog {
         info!("startup: coordinator init: catalog open: expr cache open beginning");
         // We wait until after the `pre_item_updates` to open the cache so we can get accurate
         // dyncfgs because the `pre_item_updates` contains `SystemConfiguration` updates.
-        let expr_cache_enabled = ENABLE_EXPRESSION_CACHE.get(state.system_config().dyncfgs());
+        let expr_cache_enabled = config
+            .enable_expression_cache_override
+            .unwrap_or_else(|| ENABLE_EXPRESSION_CACHE.get(state.system_config().dyncfgs()));
         let (expr_cache_handle, cached_local_exprs, cached_global_exprs) = if expr_cache_enabled {
             info!("using expression cache for startup");
             let current_ids = txn

src/adapter/src/coord.rs

@@ -3858,6 +3858,7 @@ pub fn serve(
                 http_host_name,
                 builtin_item_migration_config,
                 persist_client: persist_client.clone(),
+                enable_expression_cache_override: None,
                 helm_chart_version,
             },
         })

src/catalog-debug/src/main.rs

@@ -611,6 +611,7 @@ async fn upgrade_check(
             active_connection_count: Arc::new(Mutex::new(ConnectionCounter::new(0, 0))),
             builtin_item_migration_config: BuiltinItemMigrationConfig::Legacy,
             persist_client,
+            enable_expression_cache_override: None,
             helm_chart_version: None,
         },
         &mut storage,

src/catalog/src/config.rs

@@ -88,6 +88,9 @@ pub struct StateConfig {
     pub active_connection_count: Arc<std::sync::Mutex<ConnectionCounter>>,
     pub builtin_item_migration_config: BuiltinItemMigrationConfig,
     pub persist_client: PersistClient,
+    /// Overrides the current value of the [`mz_adapter_types::dyncfgs::ENABLE_EXPRESSION_CACHE`]
+    /// feature flag.
+    pub enable_expression_cache_override: Option<bool>,
     /// Helm chart version
     pub helm_chart_version: Option<String>,
 }

src/testdrive/src/action.rs

@@ -359,6 +359,7 @@ impl State {
         system_parameter_defaults: BTreeMap<String, String>,
         version: semver::Version,
         bootstrap_args: &BootstrapArgs,
+        enable_expression_cache_override: Option<bool>,
         f: F,
     ) -> Result<Option<T>, anyhow::Error>
     where
@@ -394,6 +395,7 @@ impl State {
                 system_parameter_defaults,
                 version,
                 bootstrap_args,
+                enable_expression_cache_override,
             )
             .await?;
             let res = f(catalog.for_session(&Session::dummy()));

src/testdrive/src/action/consistency.rs

@@ -191,6 +191,8 @@ async fn check_catalog_state(state: &State) -> Result<(), anyhow::Error> {
             system_parameter_defaults,
             version,
             &state.materialize.bootstrap_args,
+            // The expression cache can be taxing on the CPU and is unnecessary for consistency checks.
+            Some(false),
             |catalog| catalog.state().clone(),
         )
         .await