chore: optimize building bloom index for null (#17625)

* optimize building bloom index for null

calc scalar memory size

* fix

* fix

Author: zhyass

src/query/expression/src/block.rs

@@ -606,6 +606,26 @@ impl DataBlock {
         }
         self.columns[col].value.index(row)
     }
+
+    /// Calculates the memory size of a `DataBlock` for writing purposes.
+    /// This function is used to estimate the memory footprint of a `DataBlock` when writing it to storage.
+    pub fn estimate_block_size(&self) -> usize {
+        let num_rows = self.num_rows();
+        self.columns()
+            .iter()
+            .map(|entry| match &entry.value {
+                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()
+                }
+                Value::Scalar(v) => v
+                    .as_ref()
+                    .estimated_scalar_repeat_size(num_rows, &entry.data_type),
+                _ => entry.memory_size(),
+            })
+            .sum()
+    }
 }
 
 impl BlockEntry {

src/query/expression/src/values.rs

@@ -756,6 +756,56 @@ impl ScalarRef<'_> {
             },
         }
     }
+
+    /// Estimates the memory size of a scalar value if it were repeated `n` times,
+    /// without actually converting it into a column. This avoids unnecessary allocations
+    /// and provides a direct calculation based on the scalar type and its associated DataType.
+    ///
+    /// # Parameters:
+    /// - `scalar`: The scalar value to estimate memory size for.
+    /// - `n`: The number of times the scalar is hypothetically repeated.
+    /// - `data_type`: The data type of the scalar, used for correct size calculations.
+    ///
+    /// # Returns:
+    /// The estimated memory size (in bytes) that `n` repetitions of `scalar` would occupy.
+    pub fn estimated_scalar_repeat_size(&self, n: usize, data_type: &DataType) -> usize {
+        if let DataType::Nullable(ty) = data_type {
+            let mut memory_size = (n + 7) / 8;
+            if !self.is_null() {
+                memory_size += self.estimated_scalar_repeat_size(n, ty);
+            }
+            return memory_size;
+        }
+
+        match self {
+            ScalarRef::Null => std::mem::size_of::<usize>(),
+            ScalarRef::EmptyArray | ScalarRef::EmptyMap => std::mem::size_of::<usize>(),
+            ScalarRef::Number(_) => n * self.memory_size(),
+            ScalarRef::Decimal(_) => n * self.memory_size(),
+            ScalarRef::Boolean(_) => (n + 7) / 8,
+            ScalarRef::Binary(s) => s.len() * n + (n + 1) * 8,
+            ScalarRef::String(s) => s.len() * n + n * 12,
+            ScalarRef::Timestamp(_) => n * 8,
+            ScalarRef::Date(_) => n * 4,
+            ScalarRef::Interval(_) => n * 16,
+            ScalarRef::Array(col) => col.memory_size() * n + (n + 1) * 8,
+            ScalarRef::Map(col) => col.memory_size() * n + (n + 1) * 8,
+            ScalarRef::Bitmap(b) => b.len() * n + (n + 1) * 8,
+            ScalarRef::Tuple(fields) => {
+                let DataType::Tuple(fields_ty) = data_type else {
+                    unreachable!()
+                };
+                fields
+                    .iter()
+                    .zip(fields_ty.iter())
+                    .map(|(v, ty)| v.estimated_scalar_repeat_size(n, ty))
+                    .sum()
+            }
+            ScalarRef::Variant(s) => s.len() * n + (n + 1) * 8,
+            ScalarRef::Geometry(s) => s.len() * n + (n + 1) * 8,
+            ScalarRef::Geography(s) => s.0.len() * n + (n + 1) * 8,
+        }
+    }
 }
 
 impl PartialOrd for Scalar {
@@ -1708,9 +1758,8 @@ impl ColumnBuilder {
             ScalarRef::Map(col) => ColumnBuilder::Map(Box::new(ArrayColumnBuilder::repeat(col, n))),
             ScalarRef::Bitmap(b) => ColumnBuilder::Bitmap(BinaryColumnBuilder::repeat(b, n)),
             ScalarRef::Tuple(fields) => {
-                let fields_ty = match data_type {
-                    DataType::Tuple(fields_ty) => fields_ty,
-                    _ => unreachable!(),
+                let DataType::Tuple(fields_ty) = data_type else {
+                    unreachable!()
                 };
                 ColumnBuilder::Tuple(
                     fields

src/query/expression/tests/it/kernel.rs

@@ -14,11 +14,17 @@
 
 use core::ops::Range;
 
+use databend_common_base::base::OrderedFloat;
 use databend_common_column::bitmap::Bitmap;
 use databend_common_expression::block_debug::assert_block_value_eq;
 use databend_common_expression::types::number::*;
 use databend_common_expression::types::AnyType;
 use databend_common_expression::types::DataType;
+use databend_common_expression::types::DecimalDataType;
+use databend_common_expression::types::DecimalScalar;
+use databend_common_expression::types::DecimalSize;
+use databend_common_expression::types::IntervalType;
+use databend_common_expression::types::NullableType;
 use databend_common_expression::types::NumberDataType;
 use databend_common_expression::types::StringType;
 use databend_common_expression::types::ValueType;
@@ -31,6 +37,7 @@ use databend_common_expression::FilterVisitor;
 use databend_common_expression::FromData;
 use databend_common_expression::IterationStrategy;
 use databend_common_expression::Scalar;
+use databend_common_expression::ScalarRef;
 use databend_common_expression::Value;
 use goldenfile::Mint;
 
@@ -636,3 +643,112 @@ fn test_builder() {
 
     assert_eq!(r1, r2)
 }
+
+fn assert_estimated_scalar_repeat_size(scalar: ScalarRef, num_rows: usize, ty: DataType) {
+    let builder = ColumnBuilder::repeat(&scalar, num_rows, &ty);
+    let col = builder.build();
+    assert_eq!(
+        scalar.estimated_scalar_repeat_size(num_rows, &ty),
+        col.memory_size()
+    );
+}
+
+#[test]
+fn test_estimated_scalar_repeat_size() {
+    let num_rows = 108;
+
+    // null
+    {
+        let scalar = ScalarRef::Null;
+        let ty = DataType::Null;
+        assert_estimated_scalar_repeat_size(scalar, num_rows, ty);
+    }
+
+    // nullable
+    {
+        let scalar = ScalarRef::Null;
+        let ty = DataType::Nullable(Box::new(DataType::Interval));
+        let builder = ColumnBuilder::repeat(&scalar, num_rows, &ty);
+        let col = builder.build();
+        let col = NullableType::<IntervalType>::try_downcast_column(&col).unwrap();
+        assert_eq!(
+            scalar.estimated_scalar_repeat_size(num_rows, &ty),
+            col.validity.as_slice().0.len()
+        );
+    }
+
+    // nullable(float32)
+    {
+        let scalar = ScalarRef::Number(NumberScalar::Float32(OrderedFloat(2.33)));
+        let ty = DataType::Nullable(Box::new(DataType::Number(NumberDataType::Float32)));
+        assert_estimated_scalar_repeat_size(scalar, num_rows, ty);
+    }
+
+    // decimal
+    {
+        let scalar = ScalarRef::Decimal(DecimalScalar::Decimal128(233, DecimalSize {
+            precision: 46,
+            scale: 6,
+        }));
+        let ty = DataType::Decimal(DecimalDataType::Decimal256(DecimalSize {
+            precision: 46,
+            scale: 6,
+        }));
+        assert_estimated_scalar_repeat_size(scalar, num_rows, ty);
+    }
+
+    // string
+    {
+        let scalar = ScalarRef::String("abc");
+        let ty = DataType::String;
+        assert_estimated_scalar_repeat_size(scalar, num_rows, ty);
+    }
+
+    // string
+    {
+        let scalar = ScalarRef::String("abcdefghijklmn123");
+        let ty = DataType::String;
+        assert_estimated_scalar_repeat_size(scalar, num_rows, ty);
+    }
+
+    // binary
+    {
+        let scalar = ScalarRef::Binary(&[1, 133, 244, 123]);
+        let ty = DataType::Binary;
+        assert_estimated_scalar_repeat_size(scalar, num_rows, ty);
+    }
+
+    // boolean
+    {
+        let scalar = ScalarRef::Boolean(true);
+        let ty = DataType::Boolean;
+        assert_estimated_scalar_repeat_size(scalar, num_rows, ty);
+    }
+
+    // bitmap
+    {
+        let scalar = ScalarRef::Bitmap(&[1, 133, 244, 123]);
+        let ty = DataType::Bitmap;
+        assert_estimated_scalar_repeat_size(scalar, num_rows, ty);
+    }
+
+    // array
+    {
+        let scalar = ScalarRef::Array(StringType::from_data(vec!["abc", "abcdefghijklmn123"]));
+        let ty = DataType::Array(Box::new(DataType::String));
+        assert_estimated_scalar_repeat_size(scalar, num_rows, ty);
+    }
+
+    // map
+    {
+        let scalar = ScalarRef::Map(Column::Tuple(vec![
+            UInt8Type::from_data(vec![1, 2]),
+            StringType::from_data(vec!["a", "b"]),
+        ]));
+        let ty = DataType::Map(Box::new(DataType::Tuple(vec![
+            DataType::Number(NumberDataType::UInt8),
+            DataType::String,
+        ])));
+        assert_estimated_scalar_repeat_size(scalar, num_rows, ty);
+    }
+}

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

@@ -12,58 +12,28 @@
 // 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.
@@ -93,7 +63,7 @@ impl AccumulatingTransform for BlockCompactBuilder {
 
     fn transform(&mut self, data: DataBlock) -> Result<Vec<DataBlock>> {
         let num_rows = data.num_rows();
-        let num_bytes = memory_size(&data);
+        let num_bytes = data.estimate_block_size();
 
         if !self.thresholds.check_for_compact(num_rows, num_bytes) {
             // holding slices of blocks to merge later may lead to oom, so
@@ -142,18 +112,3 @@ impl AccumulatingTransform for BlockCompactBuilder {
         }
     }
 }
-
-pub fn memory_size(data_block: &DataBlock) -> usize {
-    data_block
-        .columns()
-        .iter()
-        .map(|entry| match &entry.value {
-            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/pipeline/transforms/src/processors/transforms/transform_compact_no_split_builder.rs

@@ -79,7 +79,7 @@ impl AccumulatingTransform for BlockCompactNoSplitBuilder {
 
     fn transform(&mut self, data: DataBlock) -> Result<Vec<DataBlock>> {
         self.accumulated_rows += data.num_rows();
-        self.accumulated_bytes += crate::processors::memory_size(&data);
+        self.accumulated_bytes += data.estimate_block_size();
         if !self
             .thresholds
             .check_large_enough(self.accumulated_rows, self.accumulated_bytes)

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

@@ -17,7 +17,6 @@ use databend_common_expression::BlockThresholds;
 use databend_common_expression::DataBlock;
 use databend_common_expression::DataSchemaRef;
 use databend_common_expression::SortColumnDescription;
-use databend_common_pipeline_transforms::memory_size;
 use databend_common_pipeline_transforms::sort_merge;
 use databend_common_settings::Settings;
 
@@ -56,7 +55,7 @@ impl DataProcessorStrategy for CompactStrategy {
         let mut result = Vec::with_capacity(blocks_num);
         for block in data_blocks {
             accumulated_rows += block.num_rows();
-            accumulated_bytes += memory_size(&block);
+            accumulated_bytes += block.estimate_block_size();
             if !self
                 .thresholds
                 .check_large_enough(accumulated_rows, accumulated_bytes)

src/query/service/src/test_kits/fixture.rs

@@ -563,7 +563,7 @@ impl TestFixture {
         num_of_blocks: usize,
         start: i32,
     ) -> (TableSchemaRef, Vec<Result<DataBlock>>) {
-        Self::gen_sample_blocks_ex(num_of_blocks, 3, start)
+        Self::gen_sample_blocks_ex(num_of_blocks, 2, start)
     }
 
     pub fn gen_sample_blocks_ex(

src/query/service/src/test_kits/fuse.rs

@@ -177,7 +177,7 @@ async fn generate_blocks(
     let mut block_metas = vec![];
 
     // does not matter in this suite
-    let rows_per_block = 1;
+    let rows_per_block = 2;
     let value_start_from = 1;
 
     let stream =