reset

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 = max_bytes_per_block / 3;
         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)
@@ -190,29 +190,11 @@ impl BlockThresholds {
             return 1;
         }
 
-        // Estimate the number of blocks based on row count and compressed size.
-        let by_rows = std::cmp::max(total_rows / self.max_rows_per_block, 1);
-        let by_compressed = total_compressed / self.max_compressed_per_block;
-        // If row-based block count is greater, use max rows per block as limit.
-        if by_rows >= by_compressed {
-            return by_rows;
-        }
-
-        // 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 total_partitions = if bytes_per_block > max_bytes {
-            // Block size is too large.
-            total_bytes / max_bytes
-        } else if bytes_per_block < min_bytes {
-            // Block size is too small.
-            total_bytes / min_bytes
-        } else {
-            // Block size is acceptable.
-            by_compressed
-        };
-
-        std::cmp::max(total_partitions, 1)
+        let by_rows = std::cmp::max(total_rows.div_ceil(65536), 1);
+        let by_bytes = std::cmp::max(
+            total_compressed.div_ceil(self.min_compressed_per_block),
+            total_bytes.div_ceil(self.min_bytes_per_block),
+        );
+        std::cmp::max(by_rows, by_bytes)
     }
 }

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

@@ -79,7 +79,6 @@ impl PipelineBuilder {
                         window_spill_settings.clone(),
                         disk_spill.clone(),
                         partition.rows_per_block,
-                        partition.bytes_per_block,
                     )?,
                 )))
             })?;
@@ -93,7 +92,6 @@ impl PipelineBuilder {
                     table,
                     partition.table_meta_timestamps,
                     false,
-                    Some(partition.bytes_per_block),
                 )
             })
         } else {

src/query/service/src/pipelines/processors/transforms/window/partition/transform_hilbert_collect.rs

@@ -37,26 +37,21 @@ use crate::spillers::SpillerType;
 
 enum State {
     Collect,
-    Flush,
     Spill,
     Restore,
-    Concat(Vec<DataBlock>),
 }
 
 pub struct TransformHilbertCollect {
     input: Arc<InputPort>,
     output: Arc<OutputPort>,
 
-    immediate_output_blocks: Vec<(usize, DataBlock)>,
     output_data_blocks: VecDeque<DataBlock>,
 
     // The partition id is used to map the partition id to the new partition id.
     partition_id: Vec<usize>,
-    partition_sizes: Vec<usize>,
     // The buffer is used to control the memory usage of the window operator.
     buffer: WindowPartitionBuffer,
 
-    max_block_size: usize,
     // Event variables.
     state: State,
 }
@@ -74,7 +69,6 @@ impl TransformHilbertCollect {
         memory_settings: MemorySettings,
         disk_spill: Option<SpillerDiskConfig>,
         max_block_rows: usize,
-        max_block_size: usize,
     ) -> Result<Self> {
         // Calculate the partition ids collected by the processor.
         let partitions: Vec<usize> = (0..num_partitions)
@@ -110,9 +104,6 @@ impl TransformHilbertCollect {
             output,
             partition_id,
             buffer,
-            immediate_output_blocks: vec![],
-            partition_sizes: vec![0; num_partitions],
-            max_block_size,
             output_data_blocks: VecDeque::new(),
             state: State::Collect,
         })
@@ -130,11 +121,7 @@ impl Processor for TransformHilbertCollect {
     }
 
     fn event(&mut self) -> Result<Event> {
-        if matches!(self.state, State::Concat(_)) {
-            return Ok(Event::Sync);
-        }
-
-        if matches!(self.state, State::Flush | State::Spill | State::Restore) {
+        if matches!(self.state, State::Spill | State::Restore) {
             return Ok(Event::Async);
         }
 
@@ -157,11 +144,6 @@ impl Processor for TransformHilbertCollect {
             return Ok(Event::Async);
         }
 
-        if !self.immediate_output_blocks.is_empty() {
-            self.state = State::Flush;
-            return Ok(Event::Async);
-        }
-
         if self.input.is_finished() {
             if !self.buffer.is_empty() {
                 self.state = State::Restore;
@@ -179,42 +161,18 @@ impl Processor for TransformHilbertCollect {
                 self.state = State::Spill;
                 return Ok(Event::Async);
             }
-
-            if !self.immediate_output_blocks.is_empty() {
-                self.state = State::Flush;
-                return Ok(Event::Async);
-            }
         }
 
         self.input.set_need_data();
         Ok(Event::NeedData)
     }
 
-    fn process(&mut self) -> Result<()> {
-        match std::mem::replace(&mut self.state, State::Collect) {
-            State::Concat(blocks) => {
-                let output = DataBlock::concat(&blocks)?;
-                self.output_data_blocks.push_back(output);
-            }
-            _ => unreachable!(),
-        }
-        Ok(())
-    }
-
     #[async_backtrace::framed]
     async fn async_process(&mut self) -> Result<()> {
         match std::mem::replace(&mut self.state, State::Collect) {
             State::Spill => {
                 self.buffer.spill().await?;
             }
-            State::Flush => {
-                if let Some((partition_id, data_block)) = self.immediate_output_blocks.pop() {
-                    let mut restored_data_blocks =
-                        self.buffer.restore_by_id(partition_id, true).await?;
-                    restored_data_blocks.push(data_block);
-                    self.state = State::Concat(restored_data_blocks);
-                }
-            }
             State::Restore => {
                 let restored_data_blocks = self.buffer.restore().await?;
                 self.output_data_blocks.extend(restored_data_blocks);
@@ -234,12 +192,6 @@ impl TransformHilbertCollect {
         {
             for (partition_id, data_block) in meta.partitioned_data.into_iter() {
                 let new_id = self.partition_id[partition_id];
-                self.partition_sizes[new_id] += data_block.estimate_block_size();
-                if self.partition_sizes[new_id] >= self.max_block_size {
-                    self.immediate_output_blocks.push((new_id, data_block));
-                    self.partition_sizes[new_id] = 0;
-                    continue;
-                }
                 self.buffer.add_data_block(new_id, data_block);
             }
         }

src/query/storages/fuse/src/io/write/stream/block_builder.rs

@@ -233,16 +233,11 @@ impl StreamBlockBuilder {
     }
 
     pub fn need_flush(&self) -> bool {
-        if let Some(max_block_bytes) = self.properties.max_block_bytes {
-            if self.block_size >= max_block_bytes {
-                return true;
-            }
-        };
         let file_size = self.block_writer.compressed_size();
         self.row_count >= self.properties.block_thresholds.min_rows_per_block
             || self.block_size >= self.properties.block_thresholds.max_bytes_per_block
             || (file_size >= self.properties.block_thresholds.min_compressed_per_block
-                && self.block_size >= self.properties.block_thresholds.min_bytes_per_block)
+                && self.block_size >= self.properties.block_thresholds.max_bytes_per_block / 4)
     }
 
     pub fn write(&mut self, block: DataBlock) -> Result<()> {
@@ -355,7 +350,6 @@ pub struct StreamBlockProperties {
     pub(crate) ctx: Arc<dyn TableContext>,
     pub(crate) write_settings: WriteSettings,
     pub(crate) block_thresholds: BlockThresholds,
-    pub(crate) max_block_bytes: Option<usize>,
 
     meta_locations: TableMetaLocationGenerator,
     source_schema: TableSchemaRef,
@@ -374,7 +368,6 @@ impl StreamBlockProperties {
         ctx: Arc<dyn TableContext>,
         table: &FuseTable,
         table_meta_timestamps: TableMetaTimestamps,
-        max_block_bytes: Option<usize>,
     ) -> Result<Arc<Self>> {
         // remove virtual computed fields.
         let fields = table
@@ -434,7 +427,6 @@ impl StreamBlockProperties {
             ngram_args,
             inverted_index_builders,
             table_meta_timestamps,
-            max_block_bytes,
         }))
     }
 }

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

@@ -61,7 +61,6 @@ impl FuseTable {
                     self,
                     table_meta_timestamps,
                     false,
-                    None,
                 )
             })?;
         } else {

src/query/storages/fuse/src/operations/common/processors/transform_block_writer.rs

@@ -85,14 +85,12 @@ impl TransformBlockWriter {
         table: &FuseTable,
         table_meta_timestamps: TableMetaTimestamps,
         with_tid: bool,
-        max_block_bytes: Option<usize>,
     ) -> Result<ProcessorPtr> {
         let max_block_rows = std::cmp::min(
             ctx.get_settings().get_max_block_size()? as usize,
             table.get_option(FUSE_OPT_KEY_ROW_PER_BLOCK, DEFAULT_BLOCK_ROW_COUNT),
         );
-        let properties =
-            StreamBlockProperties::try_create(ctx, table, table_meta_timestamps, max_block_bytes)?;
+        let properties = StreamBlockProperties::try_create(ctx, table, table_meta_timestamps)?;
         Ok(ProcessorPtr::create(Box::new(TransformBlockWriter {
             state: State::Consume,
             input,
@@ -207,13 +205,9 @@ impl Processor for TransformBlockWriter {
                 block.check_valid()?;
                 self.input_data_size += block.estimate_block_size();
                 self.input_num_rows += block.num_rows();
-                if self.properties.max_block_bytes.is_some() {
-                    self.input_data.push_back(block);
-                } else {
-                    let max_rows_per_block = self.calc_max_block_rows(&block);
-                    let blocks = block.split_by_rows_no_tail(max_rows_per_block);
-                    self.input_data.extend(blocks);
-                }
+                let max_rows_per_block = self.calc_max_block_rows(&block);
+                let blocks = block.split_by_rows_no_tail(max_rows_per_block);
+                self.input_data.extend(blocks);
             }
             State::Serialize => {
                 while let Some(b) = self.input_data.pop_front() {