chore(binder): add enable_materialized_cte settings (#16950)

src/query/settings/src/settings_default.rs

@@ -335,6 +335,12 @@ impl DefaultSettings {
                     mode: SettingMode::Both,
                     range: Some(SettingRange::Numeric(0..=u64::MAX)),
                 }),
+                ("enable_materialized_cte", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(1),
+                    desc: "Enable materialized common table expression.",
+                    mode: SettingMode::Both,
+                    range: Some(SettingRange::Numeric(0..=1)),
+                }),
                 ("inlist_to_join_threshold", DefaultSettingValue {
                     value: UserSettingValue::UInt64(1024),
                     desc: "Set the threshold for converting IN list to JOIN.",

src/query/settings/src/settings_getter_setter.rs

@@ -363,6 +363,10 @@ impl Settings {
         Ok(self.try_get_u64("max_cte_recursive_depth")? as usize)
     }
 
+    pub fn get_enable_materialized_cte(&self) -> Result<bool> {
+        Ok(self.try_get_u64("enable_materialized_cte")? != 0)
+    }
+
     pub fn get_sql_dialect(&self) -> Result<Dialect> {
         match self.try_get_string("sql_dialect")?.to_lowercase().as_str() {
             "hive" => Ok(Dialect::Hive),

src/query/sql/src/planner/binder/bind_table_reference/bind_table.rs

@@ -84,7 +84,9 @@ impl Binder {
                 ))
                 .set_span(*span));
             }
-            return if cte_info.materialized {
+            return if cte_info.materialized
+                && self.ctx.get_settings().get_enable_materialized_cte()?
+            {
                 self.bind_m_cte(bind_context, cte_info, &table_name, alias, span)
             } else if cte_info.recursive {
                 if self.bind_recursive_cte {
@@ -119,10 +121,12 @@ impl Binder {
                         let bind_context = parent.unwrap().as_mut();
                         let cte_map = bind_context.cte_context.cte_map.clone();
                         if let Some(cte_info) = cte_map.get(&table_name) {
-                            return if !cte_info.materialized {
-                                self.bind_cte(*span, bind_context, &table_name, alias, cte_info)
-                            } else {
+                            return if cte_info.materialized
+                                && self.ctx.get_settings().get_enable_materialized_cte()?
+                            {
                                 self.bind_m_cte(bind_context, cte_info, &table_name, alias, span)
+                            } else {
+                                self.bind_cte(*span, bind_context, &table_name, alias, cte_info)
                             };
                         }
                         parent = bind_context.parent.as_mut();

tests/sqllogictests/suites/mode/standalone/explain/explain_cte.test

@@ -0,0 +1,36 @@
+query T
+EXPLAIN WITH t AS MATERIALIZED (SELECT * FROM numbers(5)) SELECT * FROM t;
+----
+MaterializedCTE
+├── output columns: [number (#1)]
+├── TableScan
+│   ├── table: default.system.numbers
+│   ├── output columns: [number (#0)]
+│   ├── read rows: 5
+│   ├── read size: < 1 KiB
+│   ├── partitions total: 1
+│   ├── partitions scanned: 1
+│   ├── push downs: [filters: [], limit: NONE]
+│   └── estimated rows: 5.00
+└── CTEScan
+    ├── CTE index: 0, sub index: 1
+    └── estimated rows: 5.00
+
+statement ok
+set enable_materialized_cte = 0;
+
+query T
+EXPLAIN WITH t AS MATERIALIZED (SELECT * FROM numbers(5)) SELECT * FROM t;
+----
+TableScan
+├── table: default.system.numbers
+├── output columns: [number (#0)]
+├── read rows: 5
+├── read size: < 1 KiB
+├── partitions total: 1
+├── partitions scanned: 1
+├── push downs: [filters: [], limit: NONE]
+└── estimated rows: 5.00
+
+statement ok
+set enable_materialized_cte = 1;