enable sort spill in recluster

databendlabs · Jan 15, 2025 · d177cc4 · d177cc4
1 parent 13d1dcd
commit d177cc4
Show file tree

Hide file tree

Showing 4 changed files with 10 additions and 13 deletions.
diff --git a/src/query/service/src/interpreters/hook/compact_hook.rs b/src/query/service/src/interpreters/hook/compact_hook.rs
@@ -16,6 +16,7 @@ use std::sync::Arc;
 use std::time::Instant;
 
 use databend_common_base::runtime::GlobalIORuntime;
+use databend_common_base::runtime::GLOBAL_MEM_STAT;
 use databend_common_catalog::lock::LockTableOption;
 use databend_common_catalog::table::CompactionLimits;
 use databend_common_catalog::table_context::TableContext;
@@ -203,6 +204,13 @@ async fn compact_table(
                     &compact_target.database,
                     &compact_target.table,
                 )?;
+                let avail_memory_usage =
+                    settings.get_max_memory_usage()? - GLOBAL_MEM_STAT.get_memory_usage().max(0) as u64;
+                let recluster_block_size = settings
+                    .get_recluster_block_size()?
+                    .min(avail_memory_usage * 30 / 100);
+                settings.set_recluster_block_size(recluster_block_size)?;
+                ctx.set_enable_sort_spill(false);
                 let recluster = RelOperator::Recluster(Recluster {
                     catalog: compact_target.catalog,
                     database: compact_target.database,

diff --git a/src/query/service/src/pipelines/builders/builder_recluster.rs b/src/query/service/src/pipelines/builders/builder_recluster.rs
@@ -154,7 +154,6 @@ impl PipelineBuilder {
                 let sort_block_size =
                     block_thresholds.calc_rows_per_block(task.total_bytes, task.total_rows);
 
-                self.ctx.set_enable_sort_spill(false);
                 let sort_pipeline_builder =
                     SortPipelineBuilder::create(self.ctx.clone(), schema, Arc::new(sort_descs))?
                         .with_block_size_hit(sort_block_size)

diff --git a/src/query/settings/src/settings_default.rs b/src/query/settings/src/settings_default.rs
@@ -128,7 +128,6 @@ impl DefaultSettings {
         Ok(Arc::clone(DEFAULT_SETTINGS.get_or_try_init(|| -> Result<Arc<DefaultSettings>> {
             let num_cpus = Self::num_cpus();
             let max_memory_usage = Self::max_memory_usage()?;
-            let recluster_block_size = Self::recluster_block_size(max_memory_usage);
             let default_max_spill_io_requests = Self::spill_io_requests(num_cpus);
             let default_max_storage_io_requests = Self::storage_io_requests(num_cpus);
             let data_retention_time_in_days_max = Self::data_retention_time_in_days_max();
@@ -815,7 +814,7 @@ impl DefaultSettings {
                     range: Some(SettingRange::Numeric(0..=1)),
                 }),
                 ("recluster_block_size", DefaultSettingValue {
-                    value: UserSettingValue::UInt64(recluster_block_size),
+                    value: UserSettingValue::UInt64(100 * 1024 * 1024 * 1024),
                     desc: "Sets the maximum byte size of blocks for recluster",
                     mode: SettingMode::Both,
                     scope: SettingScope::Both,
@@ -1254,12 +1253,6 @@ impl DefaultSettings {
         })
     }
 
-    fn recluster_block_size(max_memory_usage: u64) -> u64 {
-        // The sort merge consumes more than twice as much memory,
-        // so the block size is set relatively conservatively here.
-        std::cmp::min(max_memory_usage * 30 / 100, 80 * 1024 * 1024 * 1024)
-    }
-
     /// Converts and validates a setting value based on its key.
     pub fn convert_value(k: String, v: String) -> Result<(String, UserSettingValue)> {
         // Retrieve the default settings instance

diff --git a/src/query/storages/fuse/src/operations/mutation/mutator/recluster_mutator.rs b/src/query/storages/fuse/src/operations/mutation/mutator/recluster_mutator.rs
@@ -195,10 +195,7 @@ impl ReclusterMutator {
             }
         }
 
-        // Compute memory threshold and maximum number of blocks allowed for reclustering
-        let mem_info = sys_info::mem_info().map_err(ErrorCode::from_std_error)?;
-        let recluster_block_size = self.ctx.get_settings().get_recluster_block_size()? as usize;
-        let memory_threshold = recluster_block_size.min(mem_info.avail as usize * 1024 * 30 / 100);
+        let memory_threshold = self.ctx.get_settings().get_recluster_block_size()? as usize;
         // specify a rather small value, so that `recluster_block_size` might be tuned to lower value.
         let max_blocks_num =
             (memory_threshold / self.block_thresholds.max_bytes_per_block).max(2) * self.max_tasks;