fix(batch): fix sequential exchange (#14924)

Author: chenzl25

proto/batch_plan.proto

@@ -241,6 +241,7 @@ message ExchangeSource {
 
 message ExchangeNode {
   repeated ExchangeSource sources = 1;
+  // sequential means each tasks of the exchange node will be executed sequentially.
   bool sequential = 2;
   repeated plan_common.Field input_schema = 3;
 }

src/frontend/src/optimizer/plan_node/batch_exchange.rs

@@ -30,6 +30,8 @@ use crate::optimizer::property::{Distribution, DistributionDisplay, Order, Order
 pub struct BatchExchange {
     pub base: PlanBase<Batch>,
     input: PlanRef,
+    // sequential means each tasks of the exchange node will be executed sequentially.
+    // Currently, it is used to avoid spawn too many tasks for limit operator.
     sequential: bool,
 }
 

src/frontend/src/scheduler/distributed/stage.rs

@@ -935,12 +935,12 @@ impl StageRunner {
                 let exchange_sources = child_stage.all_exchange_sources_for(task_id);
 
                 match &execution_plan_node.node {
-                    NodeBody::Exchange(_exchange_node) => PlanNodePb {
+                    NodeBody::Exchange(exchange_node) => PlanNodePb {
                         children: vec![],
                         identity,
                         node_body: Some(NodeBody::Exchange(ExchangeNode {
                             sources: exchange_sources,
-                            sequential: true,
+                            sequential: exchange_node.sequential,
                             input_schema: execution_plan_node.schema.clone(),
                         })),
                     },
@@ -950,7 +950,7 @@ impl StageRunner {
                         node_body: Some(NodeBody::MergeSortExchange(MergeSortExchangeNode {
                             exchange: Some(ExchangeNode {
                                 sources: exchange_sources,
-                                sequential: true,
+                                sequential: false,
                                 input_schema: execution_plan_node.schema.clone(),
                             }),
                             column_orders: sort_merge_exchange_node.column_orders.clone(),