chore(query): add enforce_local setting for disable distribute plan (#18402)

Author: zhang2014

src/query/settings/src/settings_default.rs

@@ -492,6 +492,13 @@ impl DefaultSettings {
                     scope: SettingScope::Both,
                     range: Some(SettingRange::Numeric(0..=1)),
                 }),
+                ("enforce_local", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(0),
+                    desc: "Enforce local plan.",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Both,
+                    range: Some(SettingRange::Numeric(0..=1)),
+                }),
                 ("enforce_broadcast_join", DefaultSettingValue {
                     value: UserSettingValue::UInt64(0),
                     desc: "Enforce broadcast join.",

src/query/settings/src/settings_getter_setter.rs

@@ -1013,4 +1013,8 @@ impl Settings {
     pub fn get_enable_parallel_union_all(&self) -> Result<bool> {
         Ok(self.try_get_u64("enable_parallel_union_all")? == 1)
     }
+
+    pub fn get_enforce_local(&self) -> Result<bool> {
+        Ok(self.try_get_u64("enforce_local")? == 1)
+    }
 }

src/query/sql/src/planner/planner.rs

@@ -272,7 +272,9 @@ impl Planner {
 
         // Step 4: Optimize the SExpr with optimizers, and generate optimized physical SExpr
         let opt_ctx = OptimizerContext::new(self.ctx.clone(), metadata.clone())
-            .set_enable_distributed_optimization(!self.ctx.get_cluster().is_empty())
+            .set_enable_distributed_optimization(
+                !self.ctx.get_cluster().is_empty() && !settings.get_enforce_local()?,
+            )
             .set_enable_join_reorder(unsafe { !settings.get_disable_join_reorder()? })
             .set_enable_dphyp(settings.get_enable_dphyp()?)
             .set_max_push_down_limit(settings.get_max_push_down_limit()?)

src/query/sql/src/planner/plans/recluster.rs

@@ -86,7 +86,9 @@ pub async fn plan_hilbert_sql(
     let plan = binder.bind(&stmt).await?;
 
     let opt_ctx = OptimizerContext::new(ctx.clone(), metadata)
-        .set_enable_distributed_optimization(!ctx.get_cluster().is_empty())
+        .set_enable_distributed_optimization(
+            !ctx.get_cluster().is_empty() && !settings.get_enforce_local()?,
+        )
         .set_enable_join_reorder(unsafe { !settings.get_disable_join_reorder()? })
         .set_enable_dphyp(settings.get_enable_dphyp()?)
         .set_max_push_down_limit(settings.get_max_push_down_limit()?)

tests/sqllogictests/suites/mode/cluster/enfore_local.test

@@ -0,0 +1,28 @@
+statement ok
+set enforce_local = 1;
+
+query T
+EXPLAIN SELECT number FROM numbers_mt(100000) GROUP BY number;
+----
+AggregateFinal
+├── output columns: [numbers_mt.number (#0)]
+├── group by: [number]
+├── aggregate functions: []
+├── estimated rows: 100000.00
+└── AggregatePartial
+    ├── group by: [number]
+    ├── aggregate functions: []
+    ├── estimated rows: 100000.00
+    └── TableScan
+        ├── table: default.system.numbers_mt
+        ├── output columns: [number (#0)]
+        ├── read rows: 100000
+        ├── read size: 781.25 KiB
+        ├── partitions total: 2
+        ├── partitions scanned: 2
+        ├── push downs: [filters: [], limit: NONE]
+        └── estimated rows: 100000.00
+
+
+statement ok
+set enforce_local = 0;