fix

Author: zhyass

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

@@ -27,16 +27,20 @@ use databend_common_expression::LimitType;
 use databend_common_expression::SortColumnDescription;
 use databend_common_functions::BUILTIN_FUNCTIONS;
 use databend_common_pipeline_core::Pipeline;
+use databend_common_pipeline_transforms::build_compact_block_pipeline;
 use databend_common_pipeline_transforms::processors::create_dummy_item;
 use databend_common_pipeline_transforms::processors::TransformPipelineHelper;
 use databend_common_pipeline_transforms::processors::TransformSortPartial;
 use databend_common_sql::evaluator::BlockOperator;
 use databend_common_sql::evaluator::CompoundBlockOperator;
+use databend_common_sql::executor::physical_plans::MutationKind;
 use databend_storages_common_table_meta::meta::TableMetaTimestamps;
 use databend_storages_common_table_meta::table::ClusterType;
 
 use crate::operations::TransformBlockWriter;
+use crate::operations::TransformSerializeBlock;
 use crate::statistics::ClusterStatsGenerator;
+use crate::FuseStorageFormat;
 use crate::FuseTable;
 
 impl FuseTable {
@@ -46,16 +50,44 @@ impl FuseTable {
         pipeline: &mut Pipeline,
         table_meta_timestamps: TableMetaTimestamps,
     ) -> Result<()> {
-        pipeline.add_transform(|input, output| {
-            TransformBlockWriter::try_create(
-                ctx.clone(),
-                input,
-                output,
-                self,
-                table_meta_timestamps,
-                false,
-            )
-        })?;
+        match self.storage_format {
+            FuseStorageFormat::Parquet => {
+                pipeline.add_transform(|input, output| {
+                    TransformBlockWriter::try_create(
+                        ctx.clone(),
+                        input,
+                        output,
+                        self,
+                        table_meta_timestamps,
+                        false,
+                    )
+                })?;
+            }
+            FuseStorageFormat::Native => {
+                let block_thresholds = self.get_block_thresholds();
+                build_compact_block_pipeline(pipeline, block_thresholds)?;
+
+                let schema = DataSchema::from(self.schema()).into();
+                let cluster_stats_gen = self.cluster_gen_for_append(
+                    ctx.clone(),
+                    pipeline,
+                    block_thresholds,
+                    Some(schema),
+                )?;
+                pipeline.add_transform(|input, output| {
+                    let proc = TransformSerializeBlock::try_create(
+                        ctx.clone(),
+                        input,
+                        output,
+                        self,
+                        cluster_stats_gen.clone(),
+                        MutationKind::Insert,
+                        table_meta_timestamps,
+                    )?;
+                    proc.into_processor()
+                })?;
+            }
+        }
 
         Ok(())
     }

tests/suites/1_stateful/05_formats/05_01_compact/05_01_02_load_compact_copy_row_per_block.sh

@@ -15,7 +15,7 @@ echo "drop table if exists t1 all" | $BENDSQL_CLIENT_CONNECT
 echo "CREATE TABLE t1
 (
     c0 string
-) engine=fuse row_per_block=800;
+) engine=fuse row_per_block=500;
 " | $BENDSQL_CLIENT_CONNECT