refactor(spill): unify window/cte spill to async_buffer path

Migrate WindowPartitionBufferV2 and MaterializedCTE from the
BackpressureSpiller -> WriterCreator -> SpillWriter -> AnyFileWriter
chain to the SpillsBufferPool -> SpillsDataWriter/SpillsDataReader
path used by aggregate and join operators.

Key changes:
- Rewrite window_partition_buffer_v2 to use SpillsDataWriter directly
- Rewrite materialized_cte to use SpillsBufferPool.writer()
- Remove BackpressureSpiller, WriterCreator, SpillWriter, SpillReader
- Remove row_group_encoder.rs (FileWriter, LocalWriter, AnyFileWriter,
  RowGroupEncoder, Properties, RangeFetchPlan, FileReader)
- Remove V1 WindowPartitionBuffer (async Spiller path)
- Drop local DMA support for window/cte spill
- Add SpillsDataWriter::flush_row_group() for explicit row group control

Reduce from 5-state machine (Collect/Spill/Process/Restore/Finish) to
3-state (Collect/Output/Finish). Since spill and restore are now both
synchronous, the indirection through separate Spill/Restore steps with
async fallback is unnecessary.

- Collect: pull data, buffer, spill inline when memory pressure is high
- Output: restore next partition, process (sort), push downstream
- Remove next_step(), collect(), output() helper methods
- Remove is_collect_finished, restored_data_blocks fields

Author: dqhl76

src/query/service/src/physical_plans/physical_materialized_cte.rs

@@ -32,7 +32,6 @@ use crate::physical_plans::format::PhysicalFormat;
 use crate::pipelines::PipelineBuilder;
 use crate::pipelines::memory_settings::MemorySettingsExt;
 use crate::pipelines::processors::transforms::MaterializedCteSink;
-use crate::pipelines::processors::transforms::create_materialized_cte_spiller;
 
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct MaterializedCTE {
@@ -121,15 +120,19 @@ impl IPhysicalPlan for MaterializedCTE {
             self.ref_count,
             self.channel_size,
         );
-        let spiller =
-            create_materialized_cte_spiller(builder.ctx.clone(), builder.settings.clone())?;
         let memory_settings =
             databend_common_pipeline_transforms::MemorySettings::from_materialized_cte_settings(
                 &builder.ctx,
                 &builder.settings,
             )?;
         builder.main_pipeline.add_sink(|input| {
-            MaterializedCteSink::create(input, tx.clone(), spiller.clone(), memory_settings.clone())
+            MaterializedCteSink::create(
+                builder.ctx.clone(),
+                input,
+                tx.clone(),
+                &builder.settings,
+                memory_settings.clone(),
+            )
         })
     }
 }

src/query/service/src/physical_plans/physical_recluster.rs

@@ -21,7 +21,6 @@ use databend_common_catalog::plan::DataSourceInfo;
 use databend_common_catalog::plan::DataSourcePlan;
 use databend_common_catalog::plan::ReclusterTask;
 use databend_common_catalog::table::Table;
-use databend_common_config::GlobalConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::DataSchemaRef;
@@ -45,7 +44,6 @@ use databend_common_storages_fuse::FUSE_OPT_KEY_BLOCK_IN_MEM_SIZE_THRESHOLD;
 use databend_common_storages_fuse::FuseTable;
 use databend_common_storages_fuse::operations::TransformSerializeBlock;
 use databend_common_storages_fuse::statistics::ClusterStatsGenerator;
-use databend_storages_common_cache::TempDirManager;
 use databend_storages_common_table_meta::meta::TableMetaTimestamps;
 
 use crate::physical_plans::physical_plan::IPhysicalPlan;
@@ -58,9 +56,7 @@ use crate::pipelines::processors::transforms::CompactStrategy;
 use crate::pipelines::processors::transforms::HilbertPartitionExchange;
 use crate::pipelines::processors::transforms::TransformWindowPartitionCollect;
 use crate::sessions::TableContextPartitionStats;
-use crate::sessions::TableContextQueryIdentity;
 use crate::sessions::TableContextSettings;
-use crate::spillers::SpillerDiskConfig;
 
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct Recluster {
@@ -334,17 +330,6 @@ impl IPhysicalPlan for HilbertPartition {
         )?;
 
         let settings = builder.settings.clone();
-        let temp_dir_manager = TempDirManager::instance();
-
-        let disk_bytes_limit = GlobalConfig::instance()
-            .spill
-            .window_partition_spill_bytes_limit();
-
-        let enable_dio = settings.get_enable_dio()?;
-        let disk_spill = temp_dir_manager
-            .get_disk_spill_dir(disk_bytes_limit, &builder.ctx.get_id())
-            .map(|temp_dir| SpillerDiskConfig::new(temp_dir, enable_dio))
-            .transpose()?;
 
         let window_spill_settings = MemorySettings::from_window_settings(&builder.ctx)?;
         let processor_id = AtomicUsize::new(0);
@@ -366,7 +351,6 @@ impl IPhysicalPlan for HilbertPartition {
                     num_processors,
                     self.num_partitions,
                     window_spill_settings.clone(),
-                    disk_spill.clone(),
                     CompactStrategy::new(self.rows_per_block, max_bytes_per_block),
                 )?,
             )))

src/query/service/src/physical_plans/physical_window.rs

@@ -17,7 +17,6 @@ use std::sync::atomic;
 use std::sync::atomic::AtomicUsize;
 
 use databend_common_catalog::plan::DataSourcePlan;
-use databend_common_config::GlobalConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::Constant;
@@ -48,7 +47,6 @@ use databend_common_sql::optimizer::ir::SExpr;
 use databend_common_sql::plans::WindowFuncFrame;
 use databend_common_sql::plans::WindowFuncFrameBound;
 use databend_common_sql::plans::WindowFuncType;
-use databend_storages_common_cache::TempDirManager;
 
 use super::LagLeadDefault;
 use super::LagLeadFunctionDesc;
@@ -79,8 +77,6 @@ use crate::pipelines::processors::transforms::WindowFunctionInfo;
 use crate::pipelines::processors::transforms::WindowPartitionExchange;
 use crate::pipelines::processors::transforms::WindowPartitionTopNExchange;
 use crate::pipelines::processors::transforms::WindowSortDesc;
-use crate::sessions::TableContextQueryIdentity;
-use crate::spillers::SpillerDiskConfig;
 
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct Window {
@@ -473,16 +469,6 @@ fn apply_window_partition(
         )?;
     }
 
-    let temp_dir_manager = TempDirManager::instance();
-    let disk_bytes_limit = GlobalConfig::instance()
-        .spill
-        .window_partition_spill_bytes_limit();
-    let enable_dio = settings.get_enable_dio()?;
-    let disk_spill = temp_dir_manager
-        .get_disk_spill_dir(disk_bytes_limit, &builder.ctx.get_id())
-        .map(|temp_dir| SpillerDiskConfig::new(temp_dir, enable_dio))
-        .transpose()?;
-
     let window_spill_settings = MemorySettings::from_window_settings(&builder.ctx)?;
     let plan_schema = DataSchemaRefExt::create(input_schema.fields().clone());
     let processor_id = AtomicUsize::new(0);
@@ -498,7 +484,6 @@ fn apply_window_partition(
                 num_processors,
                 num_partitions,
                 window_spill_settings.clone(),
-                disk_spill.clone(),
                 strategy,
             )?,
         )))

src/query/service/src/physical_plans/physical_window_partition.rs

@@ -17,14 +17,12 @@ use std::sync::atomic;
 use std::sync::atomic::AtomicUsize;
 
 use databend_common_catalog::plan::DataSourcePlan;
-use databend_common_config::GlobalConfig;
 use databend_common_exception::Result;
 use databend_common_expression::SortColumnDescription;
 use databend_common_pipeline::core::ProcessorPtr;
 use databend_common_pipeline_transforms::MemorySettings;
 use databend_common_sql::Symbol;
 use databend_common_sql::executor::physical_plans::SortDesc;
-use databend_storages_common_cache::TempDirManager;
 
 use crate::physical_plans::explain::PlanStatsInfo;
 use crate::physical_plans::format::PhysicalFormat;
@@ -38,8 +36,6 @@ use crate::pipelines::processors::transforms::SortStrategy;
 use crate::pipelines::processors::transforms::TransformWindowPartitionCollect;
 use crate::pipelines::processors::transforms::WindowPartitionExchange;
 use crate::pipelines::processors::transforms::WindowPartitionTopNExchange;
-use crate::sessions::TableContextQueryIdentity;
-use crate::spillers::SpillerDiskConfig;
 
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct WindowPartition {
@@ -145,16 +141,6 @@ impl IPhysicalPlan for WindowPartition {
             )?;
         }
 
-        let temp_dir_manager = TempDirManager::instance();
-        let disk_bytes_limit = GlobalConfig::instance()
-            .spill
-            .window_partition_spill_bytes_limit();
-        let enable_dio = settings.get_enable_dio()?;
-        let disk_spill = temp_dir_manager
-            .get_disk_spill_dir(disk_bytes_limit, &builder.ctx.get_id())
-            .map(|temp_dir| SpillerDiskConfig::new(temp_dir, enable_dio))
-            .transpose()?;
-
         let window_spill_settings = MemorySettings::from_window_settings(&builder.ctx)?;
 
         let processor_id = AtomicUsize::new(0);
@@ -171,7 +157,6 @@ impl IPhysicalPlan for WindowPartition {
                     num_processors,
                     num_partitions,
                     window_spill_settings.clone(),
-                    disk_spill.clone(),
                     strategy,
                 )?,
             )))