fix stats sender

Author: SkyFan2002

src/query/service/src/servers/flight/v1/exchange/exchange_manager.rs

@@ -876,20 +876,20 @@ impl QueryCoordinator {
         let query_ctx = info_mut.query_ctx.clone();
         let request_server_exchanges = std::mem::take(&mut self.statistics_exchanges);
 
-        if request_server_exchanges.len() != 1 {
-            return Err(ErrorCode::Internal(
-                "Request server must less than 1 if is not request server.",
-            ));
+        if request_server_exchanges.len() > 1 {
+            return Err(ErrorCode::Internal(format!(
+                "Request server exchanges must less than or equal to 1, but it's {}",
+                request_server_exchanges.len()
+            )));
         }
 
         let ctx = query_ctx.clone();
-        let (_, request_server_exchange) = request_server_exchanges.into_iter().next().unwrap();
-        let mut statistics_sender = StatisticsSender::spawn(
-            &query_id,
-            ctx,
-            request_server_exchange,
-            executor.get_inner(),
-        );
+        let statistics_sender = request_server_exchanges
+            .into_iter()
+            .next()
+            .map(|(_, exchange)| {
+                StatisticsSender::spawn(&query_id, ctx, exchange, executor.get_inner())
+            });
 
         let span = if let Some(parent) = SpanContext::current_local_parent() {
             Span::root("Distributed-Executor", parent)
@@ -900,7 +900,9 @@ impl QueryCoordinator {
         Thread::named_spawn(Some(String::from("Distributed-Executor")), move || {
             let _g = span.set_local_parent();
             let error = executor.execute().err();
-            statistics_sender.shutdown(error.clone());
+            if let Some(mut statistics_sender) = statistics_sender {
+                statistics_sender.shutdown(error.clone());
+            }
             query_ctx
                 .get_exchange_manager()
                 .on_finished_query(&query_id, error);