optimize: add reoptimize_imported_views config flag

Forcing `DataflowBuilder::reoptimize_imported_views` to reoptimize all
catalog-backed views imported in the `DataflowDescription` in every
possible `EXPLAIN` run is a bit of an overkill.

Instead, we should enable work in this method call only when one of the
overrides of `OptimizerConfig` (`CREATE CLUSTER ... FEATURES(...)` or
`EXPLAIN PLAN WITH(...)`) explicitly sets the value of a feature flag
that influences an optimizer stage that runs before we produce the
locally optimized `MirRelationExpr`.

src/adapter/src/optimize/dataflows.rs

@@ -43,7 +43,7 @@ use tracing::warn;
 
 use crate::catalog::CatalogState;
 use crate::coord::id_bundle::CollectionIdBundle;
-use crate::optimize::{view, Optimize, OptimizeMode, OptimizerConfig, OptimizerError};
+use crate::optimize::{view, Optimize, OptimizerConfig, OptimizerError};
 use crate::session::{Session, SERVER_MAJOR_VERSION, SERVER_MINOR_VERSION};
 use crate::util::viewable_variables;
 
@@ -308,7 +308,7 @@ impl<'a> DataflowBuilder<'a> {
         df_desc: &mut DataflowDesc,
         config: &OptimizerConfig,
     ) -> Result<(), OptimizerError> {
-        if config.mode == OptimizeMode::Explain {
+        if config.reoptimize_imported_views {
             for desc in df_desc.objects_to_build.iter_mut().rev() {
                 if matches!(desc.id, GlobalId::Explain | GlobalId::Transient(_)) {
                     // Skip descriptions that do not reference proper views.

src/adapter/src/optimize/mod.rs

@@ -162,6 +162,9 @@ pub struct OptimizerConfig {
     /// This means that it will not consider catalog items (more specifically
     /// indexes) with [`GlobalId`] greater or equal than the one provided here.
     pub replan: Option<GlobalId>,
+    /// Reoptimize imported views when building and optimizing a
+    /// [`DataflowDescription`] in the global MIR optimization phase.
+    pub reoptimize_imported_views: bool,
     /// Enable fast path optimization.
     pub enable_fast_path: bool,
     /// Enable consolidation of unions that happen immediately after negate.
@@ -198,6 +201,7 @@ impl From<&SystemVars> for OptimizerConfig {
         Self {
             mode: OptimizeMode::Execute,
             replan: None,
+            reoptimize_imported_views: false,
             enable_fast_path: true, // Always enable fast path if available.
             enable_consolidate_after_union_negate: vars.enable_consolidate_after_union_negate(),
             enable_specialized_arrangements: vars.enable_specialized_arrangements(),
@@ -233,6 +237,7 @@ where
 impl OverrideFrom<ClusterFeatures> for OptimizerConfig {
     fn override_from(mut self, features: &ClusterFeatures) -> Self {
         if let Some(feature_value) = features.enable_new_outer_join_lowering {
+            self.reoptimize_imported_views = true;
             self.enable_new_outer_join_lowering = feature_value;
         }
         if let Some(feature_value) = features.enable_eager_delta_joins {
@@ -256,6 +261,7 @@ impl OverrideFrom<ExplainContext> for OptimizerConfig {
 
         // Override feature flags that can be enabled in the EXPLAIN config.
         if let Some(explain_flag) = ctx.config.enable_new_outer_join_lowering {
+            self.reoptimize_imported_views = true;
             self.enable_new_outer_join_lowering = explain_flag;
         }
         if let Some(explain_flag) = ctx.config.enable_eager_delta_joins {