fix: write cancel when flush pending write queue (#940)

## Related Issues
Closes #

## Detailed Changes
Currently, the writer for the first request in the pending write queue
is responsible for flushing all the writes in the queue, but it can't be
ensured to always trigger flush because it may be cancelled when await
for the serial exec lock. And this change set replaces the
`try_join_all` in the rpc services to avoid such cancellation.

## Test Plan 
Test it manually.

Author: ShiKaiWi

analytic_engine/src/instance/write.rs

@@ -353,7 +353,7 @@ impl<'a> MemTableWriter<'a> {
 
 impl<'a> Writer<'a> {
     pub(crate) async fn write(&mut self, request: WriteRequest) -> Result<usize> {
-        let _timer = self.table_data.metrics.start_table_write_timer();
+        let _timer = self.table_data.metrics.start_table_write_execute_timer();
         self.table_data.metrics.on_write_request_begin();
 
         self.validate_before_write(&request)?;

analytic_engine/src/table/metrics.rs

@@ -144,6 +144,7 @@ pub struct Metrics {
     table_write_space_flush_wait_duration: Histogram,
     table_write_instance_flush_wait_duration: Histogram,
     table_write_flush_wait_duration: Histogram,
+    table_write_execute_duration: Histogram,
     table_write_total_duration: Histogram,
 }
 
@@ -175,6 +176,8 @@ impl Default for Metrics {
                 .with_label_values(&["wait_instance_flush"]),
             table_write_flush_wait_duration: TABLE_WRITE_DURATION_HISTOGRAM
                 .with_label_values(&["wait_flush"]),
+            table_write_execute_duration: TABLE_WRITE_DURATION_HISTOGRAM
+                .with_label_values(&["execute"]),
             table_write_total_duration: TABLE_WRITE_DURATION_HISTOGRAM
                 .with_label_values(&["total"]),
         }
@@ -210,10 +213,15 @@ impl Metrics {
     }
 
     #[inline]
-    pub fn start_table_write_timer(&self) -> HistogramTimer {
+    pub fn start_table_total_timer(&self) -> HistogramTimer {
         self.table_write_total_duration.start_timer()
     }
 
+    #[inline]
+    pub fn start_table_write_execute_timer(&self) -> HistogramTimer {
+        self.table_write_execute_duration.start_timer()
+    }
+
     #[inline]
     pub fn start_table_write_encode_timer(&self) -> HistogramTimer {
         self.table_write_encode_duration.start_timer()

analytic_engine/src/table/mod.rs

@@ -379,6 +379,12 @@ impl Table for TableImpl {
     }
 
     async fn write(&self, request: WriteRequest) -> Result<usize> {
+        let _timer = self
+            .space_table
+            .table_data()
+            .metrics
+            .start_table_total_timer();
+
         if self.should_queue_write_request(&request) {
             return self.write_with_pending_queue(request).await;
         }

server/src/grpc/remote_engine_service/mod.rs

@@ -17,10 +17,7 @@ use ceresdbproto::{
 };
 use common_types::record_batch::RecordBatch;
 use common_util::{error::BoxError, time::InstantExt};
-use futures::{
-    future::try_join_all,
-    stream::{self, BoxStream, StreamExt},
-};
+use futures::stream::{self, BoxStream, StreamExt};
 use log::error;
 use proxy::instance::InstanceRef;
 use query_engine::executor::Executor as QueryExecutor;
@@ -162,31 +159,43 @@ impl<Q: QueryExecutor + 'static> RemoteEngineServiceImpl<Q> {
     ) -> std::result::Result<Response<WriteResponse>, Status> {
         let begin_instant = Instant::now();
         let request = request.into_inner();
-        let mut write_table_futures = Vec::with_capacity(request.batch.len());
+        let mut write_table_handles = Vec::with_capacity(request.batch.len());
         for one_request in request.batch {
             let ctx = self.handler_ctx();
-            write_table_futures.push(async move { handle_write(ctx, one_request).await });
+            let handle = self
+                .runtimes
+                .write_runtime
+                .spawn(handle_write(ctx, one_request));
+            write_table_handles.push(handle);
         }
 
-        let handle = self
-            .runtimes
-            .write_runtime
-            .spawn(try_join_all(write_table_futures));
-        let batch_result = handle.await.box_err().context(ErrWithCause {
-            code: StatusCode::Internal,
-            msg: "fail to run the join task",
-        });
-
-        let mut batch_resp = WriteResponse::default();
-        match batch_result {
-            Ok(Ok(v)) => {
-                batch_resp.header = Some(error::build_ok_header());
-                batch_resp.affected_rows = v.into_iter().map(|resp| resp.affected_rows).sum();
-            }
-            Ok(Err(e)) | Err(e) => {
-                batch_resp.header = Some(error::build_err_header(e));
-            }
+        let mut batch_resp = WriteResponse {
+            header: Some(error::build_ok_header()),
+            affected_rows: 0,
         };
+        for write_handle in write_table_handles {
+            let write_result = write_handle.await.box_err().context(ErrWithCause {
+                code: StatusCode::Internal,
+                msg: "fail to run the join task",
+            });
+            // The underlying write can't be cancelled, so just ignore the left write
+            // handles (don't abort them) if any error is encountered.
+            match write_result {
+                Ok(res) => match res {
+                    Ok(resp) => batch_resp.affected_rows += resp.affected_rows,
+                    Err(e) => {
+                        error!("Failed to write batches, err:{e}");
+                        batch_resp.header = Some(error::build_err_header(e));
+                        break;
+                    }
+                },
+                Err(e) => {
+                    error!("Failed to write batches, err:{e}");
+                    batch_resp.header = Some(error::build_err_header(e));
+                    break;
+                }
+            };
+        }
 
         REMOTE_ENGINE_GRPC_HANDLER_DURATION_HISTOGRAM_VEC
             .write_batch