fix ProcessorExecutor lost data.
@@ -122,7 +122,8 @@ impl PipelinePullingExecutor {
pub fn try_pull_data<F>(&mut self, f: F) -> Result<Option<DataBlock>>
where F: Fn() -> bool {
- while !f() && !self.executor.is_finished() {
+ if !self.executor.is_finished() {
+ while !f() {
return match self.receiver.recv_timeout(Duration::from_millis(100)) {
Ok(data_block) => data_block,
Err(RecvTimeoutError::Timeout) => {
@@ -133,8 +134,17 @@ impl PipelinePullingExecutor {
}
};
}
-
Ok(None)
+ } else {
+ return match self.receiver.try_recv() {
+ Ok(data_block) => data_block,
+ // puller will not pull again once it received a None
+ Err(err) => Err(ErrorCode::LogicalError(format!(
+ "Logical error, try receiver error. after executor finish {}",
+ err
+ ))),
+ };
+ }
}
}
