fix test

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.
@@ -64,7 +92,6 @@ impl AccumulatingTransform for BlockCompactBuilder {
     const NAME: &'static str = "BlockCompactBuilder";
 
     fn transform(&mut self, data: DataBlock) -> Result<Vec<DataBlock>> {
-        let data = data.consume_convert_to_full();
         let num_rows = data.num_rows();
         let num_bytes = memory_size(&data);
 

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

@@ -78,7 +78,6 @@ impl AccumulatingTransform for BlockCompactNoSplitBuilder {
     const NAME: &'static str = "BlockCompactNoSplitBuilder";
 
     fn transform(&mut self, data: DataBlock) -> Result<Vec<DataBlock>> {
-        let data = data.consume_convert_to_full();
         self.accumulated_rows += data.num_rows();
         self.accumulated_bytes += crate::processors::memory_size(&data);
         if !self

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

@@ -27,7 +27,6 @@ use crate::processors::transforms::Transformer;
 pub struct TransformDummy;
 
 impl TransformDummy {
-    #[allow(dead_code)]
     pub fn create(input: Arc<InputPort>, output: Arc<OutputPort>) -> ProcessorPtr {
         ProcessorPtr::create(Transformer::create(input, output, TransformDummy {}))
     }