fix

Author: zhyass

src/query/expression/src/block.rs

@@ -77,6 +77,10 @@ impl BlockEntry {
     pub fn to_column(&self, num_rows: usize) -> Column {
         self.value.convert_to_full_column(&self.data_type, num_rows)
     }
+
+    pub fn into_column(self, num_rows: usize) -> Column {
+        self.value.into_full_column(&self.data_type, num_rows)
+    }
 }
 
 #[typetag::serde(tag = "type")]

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

@@ -67,12 +67,7 @@ impl BlockThresholds {
     }
 
     #[inline]
-    pub fn calc_rows_per_block(
-        &self,
-        total_bytes: usize,
-        total_rows: usize,
-        block_per_seg: Option<usize>,
-    ) -> usize {
+    pub fn calc_rows_per_block(&self, total_bytes: usize, total_rows: usize) -> usize {
         if self.check_for_compact(total_rows, total_bytes) {
             return total_rows;
         }
@@ -83,35 +78,6 @@ impl BlockThresholds {
             return self.max_rows_per_block;
         }
 
-        let mut rows_per_block = total_rows.div_ceil(block_num_by_size);
-        if let Some(block_per_seg) = block_per_seg {
-            if block_num_by_size >= block_per_seg {
-                return block_num_by_size;
-            }
-        }
-
-        let max_bytes_per_block = match rows_per_block {
-            v if v < self.max_rows_per_block / 10 => {
-                // If block rows < 100_000, max_bytes_per_block set to 200M
-                2 * self.max_bytes_per_block
-            }
-            v if v < self.max_rows_per_block / 2 => {
-                // If block rows < 500_000, max_bytes_per_block set to 150M
-                3 * self.max_bytes_per_block / 2
-            }
-            v if v < self.min_rows_per_block => {
-                // If block rows < 800_000, max_bytes_per_block set to 125M
-                5 * self.max_bytes_per_block / 4
-            }
-            _ => self.max_bytes_per_block,
-        };
-
-        if max_bytes_per_block > self.max_bytes_per_block {
-            rows_per_block = std::cmp::max(
-                total_rows / (std::cmp::max(total_bytes / max_bytes_per_block, 1)),
-                1,
-            );
-        }
-        rows_per_block
+        total_rows.div_ceil(block_num_by_size)
     }
 }

src/query/expression/src/values.rs

@@ -329,6 +329,16 @@ impl Value<AnyType> {
         }
     }
 
+    pub fn into_full_column(self, ty: &DataType, num_rows: usize) -> Column {
+        match self {
+            Value::Scalar(s) => {
+                let builder = ColumnBuilder::repeat(&s.as_ref(), num_rows, ty);
+                builder.build()
+            }
+            Value::Column(c) => c,
+        }
+    }
+
     pub fn try_downcast<T: ValueType>(&self) -> Option<Value<T>> {
         Some(match self {
             Value::Scalar(scalar) => Value::Scalar(T::to_owned_scalar(T::try_downcast_scalar(

src/query/pipeline/transforms/src/processors/transforms/transform_compact_builder.rs

@@ -12,30 +12,58 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
 use databend_common_exception::Result;
 use databend_common_expression::BlockThresholds;
 use databend_common_expression::Column;
 use databend_common_expression::DataBlock;
 use databend_common_expression::Value;
+use databend_common_pipeline_core::processors::InputPort;
+use databend_common_pipeline_core::processors::OutputPort;
+use databend_common_pipeline_core::processors::ProcessorPtr;
 use databend_common_pipeline_core::Pipeline;
 
 use crate::processors::AccumulatingTransform;
 use crate::processors::BlockCompactMeta;
 use crate::processors::TransformCompactBlock;
 use crate::processors::TransformPipelineHelper;
+use crate::Transform;
+use crate::Transformer;
 
 pub fn build_compact_block_pipeline(
     pipeline: &mut Pipeline,
     thresholds: BlockThresholds,
 ) -> Result<()> {
     let output_len = pipeline.output_len();
+    pipeline.add_transform(ConvertToFullTransform::create)?;
     pipeline.try_resize(1)?;
     pipeline.add_accumulating_transformer(|| BlockCompactBuilder::new(thresholds));
     pipeline.try_resize(output_len)?;
     pipeline.add_block_meta_transformer(TransformCompactBlock::default);
     Ok(())
 }
 
+pub(crate) struct ConvertToFullTransform;
+
+impl ConvertToFullTransform {
+    pub(crate) fn create(input: Arc<InputPort>, output: Arc<OutputPort>) -> Result<ProcessorPtr> {
+        Ok(ProcessorPtr::create(Transformer::create(
+            input,
+            output,
+            ConvertToFullTransform {},
+        )))
+    }
+}
+
+impl Transform for ConvertToFullTransform {
+    const NAME: &'static str = "ConvertToFullTransform";
+
+    fn transform(&mut self, data: DataBlock) -> Result<DataBlock> {
+        Ok(data.consume_convert_to_full())
+    }
+}
+
 pub struct BlockCompactBuilder {
     thresholds: BlockThresholds,
     // Holds blocks that are partially accumulated but haven't reached the threshold.
@@ -71,9 +99,7 @@ impl AccumulatingTransform for BlockCompactBuilder {
             // holding slices of blocks to merge later may lead to oom, so
             // 1. we expect blocks from file formats are not slice.
             // 2. if block is split here, cut evenly and emit them at once.
-            let rows_per_block = self
-                .thresholds
-                .calc_rows_per_block(num_bytes, num_rows, None);
+            let rows_per_block = self.thresholds.calc_rows_per_block(num_bytes, num_rows);
             Ok(vec![DataBlock::empty_with_meta(Box::new(
                 BlockCompactMeta::Split {
                     block: data,
@@ -122,7 +148,11 @@ pub fn memory_size(data_block: &DataBlock) -> usize {
         .columns()
         .iter()
         .map(|entry| match &entry.value {
-            Value::Column(Column::Nullable(col)) if col.validity.true_count() == 0 => 0,
+            Value::Column(Column::Nullable(col)) if col.validity.true_count() == 0 => {
+                // For `Nullable` columns with no valid values,
+                // only the size of the validity bitmap is counted.
+                col.validity.as_slice().0.len()
+            }
             _ => entry.memory_size(),
         })
         .sum()

src/query/service/src/interpreters/interpreter_table_recluster.rs

@@ -60,8 +60,6 @@ use databend_common_sql::MetadataRef;
 use databend_common_sql::NameResolutionContext;
 use databend_common_sql::ScalarExpr;
 use databend_common_sql::TypeChecker;
-use databend_common_storages_fuse::DEFAULT_BLOCK_PER_SEGMENT;
-use databend_common_storages_fuse::FUSE_OPT_KEY_BLOCK_PER_SEGMENT;
 use databend_enterprise_hilbert_clustering::get_hilbert_clustering_handler;
 use databend_storages_common_table_meta::meta::TableSnapshot;
 use databend_storages_common_table_meta::table::ClusterType;
@@ -307,17 +305,10 @@ impl ReclusterTableInterpreter {
         let table_info = tbl.get_table_info().clone();
 
         let block_thresholds = tbl.get_block_thresholds();
-        let block_per_seg = table_info
-            .options()
-            .get(FUSE_OPT_KEY_BLOCK_PER_SEGMENT)
-            .and_then(|s| s.parse().ok())
-            .unwrap_or(DEFAULT_BLOCK_PER_SEGMENT);
-
         let total_bytes = recluster_info.removed_statistics.uncompressed_byte_size as usize;
         let total_rows = recluster_info.removed_statistics.row_count as usize;
 
-        let rows_per_block =
-            block_thresholds.calc_rows_per_block(total_bytes, total_rows, Some(block_per_seg));
+        let rows_per_block = block_thresholds.calc_rows_per_block(total_bytes, total_rows);
         let total_partitions = std::cmp::max(total_rows / rows_per_block, 1);
         warn!("Do hilbert recluster, total_bytes: {}, total_rows: {}, total_partitions: {}, rows_per_block: {}",
             total_bytes, total_rows, total_partitions, rows_per_block);

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

@@ -152,7 +152,7 @@ impl PipelineBuilder {
 
                 // merge sort
                 let sort_block_size =
-                    block_thresholds.calc_rows_per_block(task.total_bytes, task.total_rows, None);
+                    block_thresholds.calc_rows_per_block(task.total_bytes, task.total_rows);
 
                 let sort_pipeline_builder =
                     SortPipelineBuilder::create(self.ctx.clone(), schema, Arc::new(sort_descs))?

src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_aggregate_spill_writer.rs

@@ -206,7 +206,7 @@ pub fn agg_spilling_aggregate_payload(
         let mut columns_data = Vec::with_capacity(columns.len());
         let mut columns_layout = Vec::with_capacity(columns.len());
         for column in columns.into_iter() {
-            let column = column.to_column(data_block.num_rows());
+            let column = column.into_column(data_block.num_rows());
             let column_data = serialize_column(&column);
             write_size += column_data.len() as u64;
             columns_layout.push(column_data.len() as u64);

src/query/service/src/pipelines/processors/transforms/aggregator/serde/transform_exchange_aggregate_serializer.rs

@@ -210,7 +210,7 @@ fn exchange_agg_spilling_aggregate_payload(
         let mut columns_layout = Vec::with_capacity(columns.len());
 
         for column in columns.into_iter() {
-            let column = column.to_column(data_block.num_rows());
+            let column = column.into_column(data_block.num_rows());
             let column_data = serialize_column(&column);
             write_size += column_data.len() as u64;
             columns_layout.push(column_data.len() as u64);

src/query/service/src/spillers/serialize.rs

@@ -80,13 +80,12 @@ impl BlocksEncoder {
             let block = if blocks.len() == 1 {
                 blocks.remove(0)
             } else {
-                DataBlock::concat(&blocks).unwrap()
+                DataBlock::concat(&std::mem::take(&mut blocks)).unwrap()
             };
+            let num_rows = block.num_rows();
             let columns_layout = std::iter::once(self.size())
-                .chain(block.columns().iter().map(|entry| {
-                    let column = entry
-                        .value
-                        .convert_to_full_column(&entry.data_type, block.num_rows());
+                .chain(block.take_columns().into_iter().map(|entry| {
+                    let column = entry.value.into_full_column(&entry.data_type, num_rows);
                     write_column(&column, &mut self.buf).unwrap();
                     self.size()
                 }))