Skip to content

Commit a500532

Browse files
authored
chore(query): add enforce_local setting for disable distribute plan (#18402)
1 parent 3d6f946 commit a500532

File tree

5 files changed

+45
-2
lines changed

5 files changed

+45
-2
lines changed

โ€Žsrc/query/settings/src/settings_default.rs

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -492,6 +492,13 @@ impl DefaultSettings {
492492
scope: SettingScope::Both,
493493
range: Some(SettingRange::Numeric(0..=1)),
494494
}),
495+
("enforce_local", DefaultSettingValue {
496+
value: UserSettingValue::UInt64(0),
497+
desc: "Enforce local plan.",
498+
mode: SettingMode::Both,
499+
scope: SettingScope::Both,
500+
range: Some(SettingRange::Numeric(0..=1)),
501+
}),
495502
("enforce_broadcast_join", DefaultSettingValue {
496503
value: UserSettingValue::UInt64(0),
497504
desc: "Enforce broadcast join.",

โ€Žsrc/query/settings/src/settings_getter_setter.rs

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1013,4 +1013,8 @@ impl Settings {
10131013
pub fn get_enable_parallel_union_all(&self) -> Result<bool> {
10141014
Ok(self.try_get_u64("enable_parallel_union_all")? == 1)
10151015
}
1016+
1017+
pub fn get_enforce_local(&self) -> Result<bool> {
1018+
Ok(self.try_get_u64("enforce_local")? == 1)
1019+
}
10161020
}

โ€Žsrc/query/sql/src/planner/planner.rs

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -272,7 +272,9 @@ impl Planner {
272272

273273
// Step 4: Optimize the SExpr with optimizers, and generate optimized physical SExpr
274274
let opt_ctx = OptimizerContext::new(self.ctx.clone(), metadata.clone())
275-
.set_enable_distributed_optimization(!self.ctx.get_cluster().is_empty())
275+
.set_enable_distributed_optimization(
276+
!self.ctx.get_cluster().is_empty() && !settings.get_enforce_local()?,
277+
)
276278
.set_enable_join_reorder(unsafe { !settings.get_disable_join_reorder()? })
277279
.set_enable_dphyp(settings.get_enable_dphyp()?)
278280
.set_max_push_down_limit(settings.get_max_push_down_limit()?)

โ€Žsrc/query/sql/src/planner/plans/recluster.rs

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -86,7 +86,9 @@ pub async fn plan_hilbert_sql(
8686
let plan = binder.bind(&stmt).await?;
8787

8888
let opt_ctx = OptimizerContext::new(ctx.clone(), metadata)
89-
.set_enable_distributed_optimization(!ctx.get_cluster().is_empty())
89+
.set_enable_distributed_optimization(
90+
!ctx.get_cluster().is_empty() && !settings.get_enforce_local()?,
91+
)
9092
.set_enable_join_reorder(unsafe { !settings.get_disable_join_reorder()? })
9193
.set_enable_dphyp(settings.get_enable_dphyp()?)
9294
.set_max_push_down_limit(settings.get_max_push_down_limit()?)
Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
statement ok
2+
set enforce_local = 1;
3+
4+
query T
5+
EXPLAIN SELECT number FROM numbers_mt(100000) GROUP BY number;
6+
----
7+
AggregateFinal
8+
โ”œโ”€โ”€ output columns: [numbers_mt.number (#0)]
9+
โ”œโ”€โ”€ group by: [number]
10+
โ”œโ”€โ”€ aggregate functions: []
11+
โ”œโ”€โ”€ estimated rows: 100000.00
12+
โ””โ”€โ”€ AggregatePartial
13+
โ”œโ”€โ”€ group by: [number]
14+
โ”œโ”€โ”€ aggregate functions: []
15+
โ”œโ”€โ”€ estimated rows: 100000.00
16+
โ””โ”€โ”€ TableScan
17+
โ”œโ”€โ”€ table: default.system.numbers_mt
18+
โ”œโ”€โ”€ output columns: [number (#0)]
19+
โ”œโ”€โ”€ read rows: 100000
20+
โ”œโ”€โ”€ read size: 781.25 KiB
21+
โ”œโ”€โ”€ partitions total: 2
22+
โ”œโ”€โ”€ partitions scanned: 2
23+
โ”œโ”€โ”€ push downs: [filters: [], limit: NONE]
24+
โ””โ”€โ”€ estimated rows: 100000.00
25+
26+
27+
statement ok
28+
set enforce_local = 0;

0 commit comments

Comments
ย (0)