fix(query): double panic if broken state in block builder (#17091)

* fix(query): fix call double finish in transform accumulating

* fix(query): fix DataBlock corrupted, column length mismatch panic

* fix(query): remove useless file

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

@@ -66,6 +66,7 @@ impl<T: AccumulatingTransform + 'static> Drop for AccumulatingTransformer<T> {
     fn drop(&mut self) {
         drop_guard(move || {
             if !self.called_on_finish {
+                self.called_on_finish = true;
                 self.inner.on_finish(false).unwrap();
             }
         })

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

@@ -223,5 +223,6 @@ pub fn sort_merge(
     for block in data_blocks {
         processor.transform(block)?;
     }
+
     processor.on_finish(true)
 }

src/query/storages/stage/src/read/row_based/processors/block_builder.rs

@@ -148,7 +148,11 @@ impl AccumulatingTransform for BlockBuilder {
         Ok(blocks)
     }
 
-    fn on_finish(&mut self, _output: bool) -> Result<Vec<DataBlock>> {
-        self.flush_block(true)
+    fn on_finish(&mut self, output: bool) -> Result<Vec<DataBlock>> {
+        if output {
+            return self.flush_block(true);
+        }
+
+        Ok(vec![])
     }
 }