feat(query): add distributed pruning settings (#16864)

chore(query): add distributed pruning settings

Author: zhang2014

src/query/settings/src/settings_default.rs

@@ -275,7 +275,7 @@ impl DefaultSettings {
                     mode: SettingMode::Both,
                     range: Some(SettingRange::Numeric(0..=1)),
                 }),
-                ("enable_dio", DefaultSettingValue{
+                ("enable_dio", DefaultSettingValue {
                     value: UserSettingValue::UInt64(1),
                     desc: "Enables Direct IO.",
                     mode: SettingMode::Both,
@@ -880,7 +880,6 @@ impl DefaultSettings {
                     mode: SettingMode::Both,
                     range: Some(SettingRange::Numeric(0..=u64::MAX)),
                 }),
-
                 ("enable_auto_fix_missing_bloom_index", DefaultSettingValue {
                     value: UserSettingValue::UInt64(0),
                     desc: "Enables auto fix missing bloom index",
@@ -939,7 +938,13 @@ impl DefaultSettings {
                     value: UserSettingValue::UInt64(128),
                     desc: "Sets the maximum length for truncating SQL queries in short_sql function.",
                     mode: SettingMode::Both,
-                    range: Some(SettingRange::Numeric(1..=1024*1024)),
+                    range: Some(SettingRange::Numeric(1..=1024 * 1024)),
+                }),
+                ("enable_distributed_pruning", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(1),
+                    desc: "Enable distributed index pruning, as it is very necessary and should remain enabled in the vast majority of cases.",
+                    mode: SettingMode::Both,
+                    range: Some(SettingRange::Numeric(0..=1)),
                 }),
             ]);
 

src/query/settings/src/settings_getter_setter.rs

@@ -809,4 +809,8 @@ impl Settings {
     pub fn set_short_sql_max_length(&self, val: u64) -> Result<()> {
         self.try_set_u64("short_sql_max_length", val)
     }
+
+    pub fn get_enable_distributed_pruning(&self) -> Result<bool> {
+        Ok(self.try_get_u64("enable_distributed_pruning")? == 1)
+    }
 }

src/query/storages/fuse/src/operations/read_partitions.rs

@@ -61,6 +61,7 @@ impl FuseTable {
         push_downs: Option<PushDownInfo>,
         dry_run: bool,
     ) -> Result<(PartStatistics, Partitions)> {
+        let distributed_pruning = ctx.get_settings().get_enable_distributed_pruning()?;
         debug!("fuse table do read partitions, push downs:{:?}", push_downs);
         if let Some(changes_desc) = &self.changes_desc {
             // For "ANALYZE TABLE" statement, we need set the default change type to "Insert".
@@ -86,7 +87,7 @@ impl FuseTable {
                     nodes_num = cluster.nodes.len();
                 }
 
-                if !dry_run && snapshot.segments.len() > nodes_num {
+                if !dry_run && snapshot.segments.len() > nodes_num && distributed_pruning {
                     let mut segments = Vec::with_capacity(snapshot.segments.len());
                     for (idx, segment_location) in snapshot.segments.iter().enumerate() {
                         segments.push(FuseLazyPartInfo::create(idx, segment_location.clone()))

tests/sqllogictests/suites/base/09_fuse_engine/09_0102_distributed_pruning.sql

@@ -0,0 +1,41 @@
+
+statement ok
+create or replace database distributed_pruning;
+
+statement ok
+use distributed_pruning;
+
+statement ok
+CREATE or replace TABLE test_table(TEXT String);
+
+
+statement ok
+INSERT INTO test_table VALUES('test_text1');
+
+statement ok
+INSERT INTO test_table VALUES('test_text2');
+
+statement ok
+INSERT INTO test_table VALUES('test_text3');
+
+statement ok
+INSERT INTO test_table VALUES('test_text4');
+
+statement ok
+SET enable_distributed_pruning = 1;
+
+query TTTTT
+SELECT COUNT() FROM test_table;
+----
+4
+
+statement ok
+SET enable_distributed_pruning = 0;
+
+query TTTTT
+SELECT COUNT() FROM test_table;
+----
+4
+
+statement ok
+SET enable_distributed_pruning = 1;