fix

Signed-off-by: coldWater <forsaken628@gmail.com>

Author: forsaken628

src/query/pipeline/core/src/processors/mod.rs

@@ -39,6 +39,5 @@ pub use resize_processor::create_resize_item;
 pub use resize_processor::ResizeProcessor;
 pub use shuffle_processor::Exchange;
 pub use shuffle_processor::MergePartitionProcessor;
-pub use shuffle_processor::MultiwayStrategy;
 pub use shuffle_processor::PartitionProcessor;
 pub use shuffle_processor::ShuffleProcessor;

src/query/pipeline/core/src/processors/shuffle_processor.rs

@@ -345,10 +345,7 @@ impl<T: Exchange> Processor for MergePartitionProcessor<T> {
             input.set_need_data();
         }
 
-        if all_inputs_finished
-            && (!matches!(T::STRATEGY, MultiwayStrategy::Custom)
-                || self.inputs_data.iter().all(Option::is_none))
-        {
+        if all_inputs_finished {
             self.output.finish();
             return Ok(Event::Finished);
         }
@@ -360,11 +357,6 @@ impl<T: Exchange> Processor for MergePartitionProcessor<T> {
                 self.output.push_data(Ok(block));
                 return Ok(Event::NeedConsume);
             }
-
-            if all_inputs_finished && self.inputs_data.iter().all(Option::is_none) {
-                self.output.finish();
-                return Ok(Event::Finished);
-            }
         }
 
         Ok(Event::NeedData)