Fix event.

databendlabs · Dec 21, 2023 · 51df83c · 51df83c
1 parent 59b47b0
commit 51df83c
Showing 1 changed file with 9 additions and 2 deletions.
diff --git a/src/query/pipeline/transforms/src/processors/transforms/transform_multi_sort_merge.rs b/src/query/pipeline/transforms/src/processors/transforms/transform_multi_sort_merge.rs
@@ -233,6 +233,7 @@ where R: Rows
     output: Arc<OutputPort>,
 
     output_data: VecDeque<DataBlock>,
+    need_data: bool,
 }
 
 impl<R> MultiSortMergeProcessor<R>
@@ -257,11 +258,11 @@ where R: Rows
             inputs,
             output,
             output_data: VecDeque::new(),
+            need_data: false,
         })
     }
 }
 
-// #[async_trait::async_trait]
 impl<R> Processor for MultiSortMergeProcessor<R>
 where R: Rows + Send + 'static
 {
@@ -298,13 +299,19 @@ where R: Rows + Send + 'static
             return Ok(Event::Finished);
         }
 
-        Ok(Event::Sync)
+        if self.need_data {
+            self.need_data = false;
+            Ok(Event::NeedData)
+        } else {
+            Ok(Event::Sync)
+        }
     }
 
     fn process(&mut self) -> Result<()> {
         while let Some(block) = self.merger.next_block()? {
             self.output_data.push_back(block);
         }
+        self.need_data = true;
         Ok(())
     }
 }