spill block build and block write

Author: zhyass

src/query/expression/src/utils/block_thresholds.rs

@@ -153,7 +153,7 @@ impl BlockThresholds {
         let bytes_per_block = total_bytes.div_ceil(block_num_by_compressed);
         // Adjust the number of blocks based on block size thresholds.
         let max_bytes_per_block = self.max_bytes_per_block.min(400 * 1024 * 1024);
-        let min_bytes_per_block = max_bytes_per_block / 2;
+        let min_bytes_per_block = (self.min_bytes_per_block / 2).min(50 * 1024 * 1024);
         let block_nums = if bytes_per_block > max_bytes_per_block {
             // Case 1: If the block size is too bigger.
             total_bytes.div_ceil(max_bytes_per_block)
@@ -201,7 +201,7 @@ impl BlockThresholds {
         // Adjust block count based on byte size thresholds.
         let bytes_per_block = total_bytes.div_ceil(by_compressed);
         let max_bytes = self.max_bytes_per_block.min(400 * 1024 * 1024);
-        let min_bytes = max_bytes / 2;
+        let min_bytes = (self.min_bytes_per_block / 2).min(50 * 1024 * 1024);
         let total_partitions = if bytes_per_block > max_bytes {
             // Block size is too large.
             total_bytes / max_bytes

src/query/service/src/pipelines/builders/builder_hilbert_partition.rs

@@ -20,19 +20,21 @@ use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::Result;
 use databend_common_pipeline_core::processors::ProcessorPtr;
 use databend_common_pipeline_transforms::MemorySettings;
+use databend_common_pipeline_transforms::TransformPipelineHelper;
 use databend_common_sql::executor::physical_plans::HilbertPartition;
 use databend_common_sql::executor::physical_plans::MutationKind;
+use databend_common_storages_fuse::io::StreamBlockProperties;
 use databend_common_storages_fuse::operations::TransformBlockWriter;
 use databend_common_storages_fuse::operations::TransformSerializeBlock;
 use databend_common_storages_fuse::statistics::ClusterStatsGenerator;
 use databend_common_storages_fuse::FuseTable;
 use databend_storages_common_cache::TempDirManager;
 
 use crate::pipelines::memory_settings::MemorySettingsExt;
-use crate::pipelines::processors::transforms::CompactStrategy;
-use crate::pipelines::processors::transforms::HilbertPartitionExchange;
-use crate::pipelines::processors::transforms::TransformHilbertCollect;
-use crate::pipelines::processors::transforms::TransformWindowPartitionCollect;
+use crate::pipelines::processors::transforms::CompactPartitionStrategy;
+use crate::pipelines::processors::transforms::ReclusterPartitionExchange;
+use crate::pipelines::processors::transforms::ReclusterPartitionStrategy;
+use crate::pipelines::processors::transforms::TransformPartitionCollect;
 use crate::pipelines::PipelineBuilder;
 use crate::spillers::SpillerDiskConfig;
 
@@ -49,7 +51,7 @@ impl PipelineBuilder {
 
         self.main_pipeline.exchange(
             num_processors,
-            HilbertPartitionExchange::create(partition.range_start, partition.range_width),
+            ReclusterPartitionExchange::create(partition.range_start, partition.range_width),
         );
 
         let settings = self.ctx.get_settings();
@@ -66,9 +68,15 @@ impl PipelineBuilder {
         let processor_id = AtomicUsize::new(0);
 
         if enable_stream_writer {
+            let properties = StreamBlockProperties::try_create(
+                self.ctx.clone(),
+                table,
+                partition.table_meta_timestamps,
+            )?;
+
             self.main_pipeline.add_transform(|input, output| {
                 Ok(ProcessorPtr::create(Box::new(
-                    TransformHilbertCollect::new(
+                    TransformPartitionCollect::new(
                         self.ctx.clone(),
                         input,
                         output,
@@ -78,28 +86,24 @@ impl PipelineBuilder {
                         partition.range_width,
                         window_spill_settings.clone(),
                         disk_spill.clone(),
-                        partition.rows_per_block,
-                        partition.bytes_per_block,
+                        ReclusterPartitionStrategy::new(properties.clone()),
                     )?,
                 )))
             })?;
 
-            self.main_pipeline.add_transform(|input, output| {
-                TransformBlockWriter::try_create(
+            self.main_pipeline.add_async_accumulating_transformer(|| {
+                TransformBlockWriter::create(
                     self.ctx.clone(),
-                    input,
-                    output,
                     MutationKind::Recluster,
                     table,
-                    partition.table_meta_timestamps,
                     false,
-                    Some(partition.bytes_per_block),
                 )
-            })
+            });
+            Ok(())
         } else {
             self.main_pipeline.add_transform(|input, output| {
                 Ok(ProcessorPtr::create(Box::new(
-                    TransformWindowPartitionCollect::new(
+                    TransformPartitionCollect::new(
                         self.ctx.clone(),
                         input,
                         output,
@@ -109,24 +113,26 @@ impl PipelineBuilder {
                         partition.range_width,
                         window_spill_settings.clone(),
                         disk_spill.clone(),
-                        CompactStrategy::new(partition.rows_per_block, partition.bytes_per_block),
+                        CompactPartitionStrategy::new(
+                            partition.rows_per_block,
+                            partition.bytes_per_block,
+                        ),
                     )?,
                 )))
             })?;
 
-            self.main_pipeline
-                .add_transform(|transform_input_port, transform_output_port| {
-                    let proc = TransformSerializeBlock::try_create(
-                        self.ctx.clone(),
-                        transform_input_port,
-                        transform_output_port,
-                        table,
-                        ClusterStatsGenerator::default(),
-                        MutationKind::Recluster,
-                        partition.table_meta_timestamps,
-                    )?;
-                    proc.into_processor()
-                })
+            self.main_pipeline.add_transform(|input, output| {
+                let proc = TransformSerializeBlock::try_create(
+                    self.ctx.clone(),
+                    input,
+                    output,
+                    table,
+                    ClusterStatsGenerator::default(),
+                    MutationKind::Recluster,
+                    partition.table_meta_timestamps,
+                )?;
+                proc.into_processor()
+            })
         }
     }
 }

src/query/service/src/pipelines/builders/builder_window.rs

@@ -30,11 +30,11 @@ use databend_storages_common_cache::TempDirManager;
 
 use crate::pipelines::memory_settings::MemorySettingsExt;
 use crate::pipelines::processors::transforms::FrameBound;
-use crate::pipelines::processors::transforms::SortStrategy;
+use crate::pipelines::processors::transforms::TransformPartitionCollect;
 use crate::pipelines::processors::transforms::TransformWindow;
-use crate::pipelines::processors::transforms::TransformWindowPartitionCollect;
 use crate::pipelines::processors::transforms::WindowFunctionInfo;
 use crate::pipelines::processors::transforms::WindowPartitionExchange;
+use crate::pipelines::processors::transforms::WindowPartitionStrategy;
 use crate::pipelines::processors::transforms::WindowPartitionTopNExchange;
 use crate::pipelines::processors::transforms::WindowSortDesc;
 use crate::pipelines::PipelineBuilder;
@@ -203,14 +203,14 @@ impl PipelineBuilder {
 
         let processor_id = AtomicUsize::new(0);
         self.main_pipeline.add_transform(|input, output| {
-            let strategy = SortStrategy::try_create(
+            let strategy = WindowPartitionStrategy::try_create(
                 &settings,
                 sort_desc.clone(),
                 plan_schema.clone(),
                 have_order_col,
             )?;
             Ok(ProcessorPtr::create(Box::new(
-                TransformWindowPartitionCollect::new(
+                TransformPartitionCollect::new(
                     self.ctx.clone(),
                     input,
                     output,

src/query/service/src/pipelines/processors/transforms/recluster/compact_strategy.rs

@@ -12,10 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod compact_strategy;
-mod hilbert_partition_exchange;
-mod transform_hilbert_collect;
+mod recluster_partition_exchange;
+mod recluster_partition_strategy;
 
-pub use compact_strategy::CompactStrategy;
-pub use hilbert_partition_exchange::HilbertPartitionExchange;
-pub use transform_hilbert_collect::TransformHilbertCollect;
+pub use recluster_partition_exchange::ReclusterPartitionExchange;
+pub use recluster_partition_strategy::CompactPartitionStrategy;
+pub use recluster_partition_strategy::ReclusterPartitionStrategy;

src/query/service/src/pipelines/processors/transforms/recluster/mod.rs

@@ -20,19 +20,19 @@ use databend_common_pipeline_core::processors::Exchange;
 
 use crate::pipelines::processors::transforms::WindowPartitionMeta;
 
-pub struct HilbertPartitionExchange {
+pub struct ReclusterPartitionExchange {
     start: u64,
     width: usize,
 }
 
-impl HilbertPartitionExchange {
-    pub fn create(start: u64, width: usize) -> Arc<HilbertPartitionExchange> {
-        Arc::new(HilbertPartitionExchange { start, width })
+impl ReclusterPartitionExchange {
+    pub fn create(start: u64, width: usize) -> Arc<ReclusterPartitionExchange> {
+        Arc::new(ReclusterPartitionExchange { start, width })
     }
 }
 
-impl Exchange for HilbertPartitionExchange {
-    const NAME: &'static str = "Hilbert";
+impl Exchange for ReclusterPartitionExchange {
+    const NAME: &'static str = "Recluster";
     fn partition(&self, data_block: DataBlock, n: usize) -> Result<Vec<DataBlock>> {
         let mut data_block = data_block;
         let range_ids = data_block
@@ -51,16 +51,10 @@ impl Exchange for HilbertPartitionExchange {
 
         let scatter_indices = DataBlock::divide_indices_by_scatter_size(&indices, self.width);
         // Partition the data blocks to different processors.
-        let base = self.width / n;
-        let remainder = self.width % n;
         let mut output_data_blocks = vec![vec![]; n];
         for (partition_id, indices) in scatter_indices.into_iter().take(self.width).enumerate() {
             if !indices.is_empty() {
-                let target = if partition_id < remainder * (base + 1) {
-                    partition_id / (base + 1)
-                } else {
-                    (partition_id - remainder) / base
-                };
+                let target = (partition_id * n) / self.width;
                 let block = data_block.take_with_optimize_size(&indices)?;
                 output_data_blocks[target].push((partition_id, block));
             }