fix: avoid send empty record batch to client (#796)

Author: ShiKaiWi

common_types/src/record_batch.rs

@@ -1,4 +1,4 @@
-// Copyright 2022 CeresDB Project Authors. Licensed under Apache-2.0.
+// Copyright 2022-2023 CeresDB Project Authors. Licensed under Apache-2.0.
 
 //! Record batch
 
@@ -234,23 +234,33 @@ impl RecordBatch {
         &self.schema
     }
 
+    #[inline]
+    pub fn is_empty(&self) -> bool {
+        self.num_rows() == 0
+    }
+
     // REQUIRE: index is valid
+    #[inline]
     pub fn column(&self, index: usize) -> &ColumnBlock {
         &self.data.column_blocks[index]
     }
 
+    #[inline]
     pub fn num_columns(&self) -> usize {
         self.schema.num_columns()
     }
 
+    #[inline]
     pub fn num_rows(&self) -> usize {
         self.data.num_rows()
     }
 
+    #[inline]
     pub fn as_arrow_record_batch(&self) -> &ArrowRecordBatch {
         &self.data.arrow_record_batch
     }
 
+    #[inline]
     pub fn into_arrow_record_batch(self) -> ArrowRecordBatch {
         self.data.arrow_record_batch
     }

components/arrow_ext/src/ipc.rs

@@ -1,4 +1,4 @@
-// Copyright 2022 CeresDB Project Authors. Licensed under Apache-2.0.
+// Copyright 2022-2023 CeresDB Project Authors. Licensed under Apache-2.0.
 
 //! Utilities for `RecordBatch` serialization using Arrow IPC
 
@@ -109,6 +109,7 @@ impl RecordBatchesEncoder {
         let stream_writer = if let Some(v) = &mut self.stream_writer {
             v
         } else {
+            // TODO: pre-allocate the buffer.
             let buffer: Vec<u8> = Vec::new();
             let stream_writer =
                 StreamWriter::try_new(buffer, &batch.schema()).context(ArrowError)?;

server/src/proxy/grpc/sql_query.rs

@@ -453,6 +453,10 @@ impl QueryResponseWriter {
     }
 
     pub fn write(&mut self, batch: &RecordBatch) -> Result<()> {
+        if batch.is_empty() {
+            return Ok(());
+        }
+
         self.encoder
             .write(batch.as_arrow_record_batch())
             .box_err()