fix test

Author: zhyass

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

@@ -102,11 +102,12 @@ impl ColumnStatisticsState {
             } else if let Some(estimator) = self.distinct_columns.get(&id) {
                 col_stats.distinct_of_values = Some(estimator.finalize());
             } else {
-                assert_eq!(col_stats.min, col_stats.max);
-                if col_stats.min.is_null() {
-                    col_stats.distinct_of_values = Some(0);
-                } else {
-                    col_stats.distinct_of_values = Some(1);
+                if col_stats.min == col_stats.max {
+                    if col_stats.min.is_null() {
+                        col_stats.distinct_of_values = Some(0);
+                    } else {
+                        col_stats.distinct_of_values = Some(1);
+                    }
                 }
             }
             statistics.insert(id, col_stats);

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

@@ -63,7 +63,7 @@ pub struct TransformBlockBuilder {
     input_data_size: usize,
     input_num_rows: usize,
 
-    input_data: VecDeque<DataBlock>,
+    input_data: VecDeque<(usize, DataBlock)>,
     output_data: Option<DataBlock>,
 }
 
@@ -97,14 +97,15 @@ impl TransformBlockBuilder {
     }
 
     fn split_input(&self, input: DataBlock) -> Vec<DataBlock> {
-        let min_bytes_per_block = self.properties.block_thresholds.min_bytes_per_block;
         let block_size = input.estimate_block_size();
-        if block_size <= min_bytes_per_block {
-            return vec![input];
-        }
         let num_rows = input.num_rows();
         let average_row_size = block_size.div_ceil(num_rows);
-        let max_rows = min_bytes_per_block.div_ceil(average_row_size);
+        let max_rows = self
+            .properties
+            .block_thresholds
+            .min_bytes_per_block
+            .div_ceil(average_row_size)
+            .min(self.properties.block_thresholds.max_rows_per_block);
         input.split_by_rows_no_tail(max_rows)
     }
 }
@@ -181,14 +182,16 @@ impl Processor for TransformBlockBuilder {
             State::Collect(block) => {
                 // Check if the datablock is valid, this is needed to ensure data is correct
                 block.check_valid()?;
-                self.input_data_size += block.estimate_block_size();
                 self.input_num_rows += block.num_rows();
-                let blocks = self.split_input(block);
-                self.input_data.extend(blocks);
+                for block in self.split_input(block) {
+                    let block_size = block.estimate_block_size();
+                    self.input_data_size += block_size;
+                    self.input_data.push_back((block_size, block));
+                }
             }
             State::Serialize => {
-                while let Some(b) = self.input_data.pop_front() {
-                    self.input_data_size -= b.estimate_block_size();
+                while let Some((block_size, b)) = self.input_data.pop_front() {
+                    self.input_data_size -= block_size;
                     self.input_num_rows -= b.num_rows();
 
                     let builder = self.get_or_create_builder()?;
@@ -201,7 +204,7 @@ impl Processor for TransformBlockBuilder {
                 }
             }
             State::Finalize => {
-                while let Some(b) = self.input_data.pop_front() {
+                while let Some((_, b)) = self.input_data.pop_front() {
                     let builder = self.get_or_create_builder()?;
                     builder.write(b)?;
                 }

tests/sqllogictests/suites/base/09_fuse_engine/09_0004_remote_insert_into_select.test

@@ -7,9 +7,6 @@ CREATE DATABASE db_09_004
 statement ok
 USE db_09_004
 
-statement ok
-set enable_block_stream_write = 1
-
 statement ok
 CREATE TABLE IF NOT EXISTS t1(a UInt8 not null, b UInt64 not null, c Int8 not null, d Int64 not null, e Date not null, f Date not null, g DateTime not null, h String not null) Engine = Fuse
 

tests/sqllogictests/suites/base/09_fuse_engine/09_0008_fuse_optimize_table.test

@@ -7,9 +7,6 @@ CREATE DATABASE db_09_0008
 statement ok
 USE db_09_0008
 
-statement ok
-set enable_block_stream_write = 1
-
 statement ok
 create table t(a uint64 not null)
 

tests/sqllogictests/suites/base/issues/issue_18275.test

@@ -14,9 +14,6 @@ CREATE OR REPLACE TABLE product_test (
     stock INT
 );
 
-statement ok
-set enable_block_stream_write = 1;
-
 statement ok
 INSERT INTO product_test (id, name, category, price, stock)
    VALUES(6, 'Keyboard', 'Electronics', 79.99, 25),