feat: Replace SelectResult with FlightData (#776)

* feat: replace SelectResult with FlightData

* Update tests/runner/src/env.rs

Co-authored-by: Ruihang Xia <waynestxia@gmail.com>

Co-authored-by: Ruihang Xia <waynestxia@gmail.com>

Author: MichaelScofield

Cargo.lock

@@ -20,7 +20,6 @@ fn main() {
         .file_descriptor_set_path(default_out_dir.join("greptime_fd.bin"))
         .compile(
             &[
-                "greptime/v1/select.proto",
                 "greptime/v1/greptime.proto",
                 "greptime/v1/meta/common.proto",
                 "greptime/v1/meta/heartbeat.proto",

src/api/build.rs

@@ -48,17 +48,11 @@ message InsertExpr {
 message ObjectResult {
   ResultHeader header = 1;
   oneof result {
-    SelectResult select = 2;
-    MutateResult mutate = 3;
-    FlightDataRaw flight_data = 4;
+    MutateResult mutate = 2;
+    FlightDataRaw flight_data = 3;
   }
 }
 
-// TODO(LFC): replace with flight data
-message SelectResult {
-  bytes raw_data = 1;
-}
-
 message FlightDataRaw {
   repeated bytes raw_data = 1;
 }

src/api/greptime/v1/database.proto

@@ -14,10 +14,8 @@
 
 use common_error::prelude::ErrorExt;
 
-use crate::v1::codec::SelectResult;
 use crate::v1::{
     admin_result, object_result, AdminResult, MutateResult, ObjectResult, ResultHeader,
-    SelectResult as SelectResultRaw,
 };
 
 pub const PROTOCOL_VERSION: u32 = 1;
@@ -36,7 +34,6 @@ pub struct ObjectResultBuilder {
 
 pub enum Body {
     Mutate((Success, Failure)),
-    Select(SelectResult),
     FlightDataRaw(FlightDataRaw),
 }
 
@@ -69,11 +66,6 @@ impl ObjectResultBuilder {
         self
     }
 
-    pub fn select_result(mut self, select_result: SelectResult) -> Self {
-        self.result = Some(Body::Select(select_result));
-        self
-    }
-
     pub fn flight_data(mut self, flight_data: FlightDataRaw) -> Self {
         self.result = Some(Body::FlightDataRaw(flight_data));
         self
@@ -93,9 +85,6 @@ impl ObjectResultBuilder {
                     failure,
                 }))
             }
-            Some(Body::Select(select)) => Some(object_result::Result::Select(SelectResultRaw {
-                raw_data: select.into(),
-            })),
             Some(Body::FlightDataRaw(raw_data)) => Some(object_result::Result::FlightData(
                 crate::v1::FlightDataRaw { raw_data },
             )),

src/api/greptime/v1/select.proto

@@ -15,7 +15,6 @@
 pub use prost::DecodeError;
 use prost::Message;
 
-use crate::v1::codec::SelectResult;
 use crate::v1::meta::TableRouteValue;
 
 macro_rules! impl_convert_with_bytes {
@@ -36,80 +35,4 @@ macro_rules! impl_convert_with_bytes {
     };
 }
 
-impl_convert_with_bytes!(SelectResult);
 impl_convert_with_bytes!(TableRouteValue);
-
-#[cfg(test)]
-mod tests {
-    use std::ops::Deref;
-
-    use crate::v1::codec::*;
-    use crate::v1::{column, Column};
-
-    const SEMANTIC_TAG: i32 = 0;
-
-    #[test]
-    fn test_convert_select_result() {
-        let select_result = mock_select_result();
-
-        let bytes: Vec<u8> = select_result.into();
-        let result: SelectResult = bytes.deref().try_into().unwrap();
-
-        assert_eq!(8, result.row_count);
-        assert_eq!(1, result.columns.len());
-
-        let column = &result.columns[0];
-        assert_eq!("foo", column.column_name);
-        assert_eq!(SEMANTIC_TAG, column.semantic_type);
-        assert_eq!(vec![1], column.null_mask);
-        assert_eq!(
-            vec![2, 3, 4, 5, 6, 7, 8],
-            column.values.as_ref().unwrap().i32_values
-        );
-    }
-
-    #[should_panic]
-    #[test]
-    fn test_convert_select_result_wrong() {
-        let select_result = mock_select_result();
-
-        let mut bytes: Vec<u8> = select_result.into();
-
-        // modify some bytes
-        bytes[0] = 0b1;
-        bytes[1] = 0b1;
-
-        let result: SelectResult = bytes.deref().try_into().unwrap();
-
-        assert_eq!(8, result.row_count);
-        assert_eq!(1, result.columns.len());
-
-        let column = &result.columns[0];
-        assert_eq!("foo", column.column_name);
-        assert_eq!(SEMANTIC_TAG, column.semantic_type);
-        assert_eq!(vec![1], column.null_mask);
-        assert_eq!(
-            vec![2, 3, 4, 5, 6, 7, 8],
-            column.values.as_ref().unwrap().i32_values
-        );
-    }
-
-    fn mock_select_result() -> SelectResult {
-        let values = column::Values {
-            i32_values: vec![2, 3, 4, 5, 6, 7, 8],
-            ..Default::default()
-        };
-        let null_mask = vec![1];
-        let column = Column {
-            column_name: "foo".to_string(),
-            semantic_type: SEMANTIC_TAG,
-            values: Some(values),
-            null_mask,
-            ..Default::default()
-        };
-        SelectResult {
-            columns: vec![column],
-            row_count: 8,
-        }
-    }
-}

src/api/src/result.rs

@@ -17,9 +17,5 @@ tonic::include_proto!("greptime.v1");
 
 pub const GREPTIME_FD_SET: &[u8] = tonic::include_file_descriptor_set!("greptime_fd");
 
-pub mod codec {
-    tonic::include_proto!("greptime.v1.codec");
-}
-
 mod column_def;
 pub mod meta;

src/api/src/serde.rs

@@ -12,24 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
-
-use api::v1::codec::SelectResult as GrpcSelectResult;
-use api::v1::column::SemanticType;
 use api::v1::{
     object_expr, object_result, query_request, DatabaseRequest, ExprHeader, InsertExpr,
     MutateResult as GrpcMutateResult, ObjectExpr, ObjectResult as GrpcObjectResult, QueryRequest,
 };
 use common_error::status_code::StatusCode;
 use common_grpc::flight::{raw_flight_data_to_message, FlightMessage};
-use common_grpc_expr::column_to_vector;
 use common_query::Output;
-use common_recordbatch::{RecordBatch, RecordBatches};
-use datatypes::prelude::*;
-use datatypes::schema::{ColumnSchema, Schema};
 use snafu::{ensure, OptionExt, ResultExt};
 
-use crate::error::{ColumnToVectorSnafu, ConvertSchemaSnafu, DatanodeSnafu, DecodeSelectSnafu};
+use crate::error::DatanodeSnafu;
 use crate::{error, Client, Result};
 
 pub const PROTOCOL_VERSION: u32 = 1;
@@ -141,7 +133,6 @@ impl Database {
 
 #[derive(Debug)]
 pub enum ObjectResult {
-    Select(GrpcSelectResult),
     FlightData(Vec<FlightMessage>),
     Mutate(GrpcMutateResult),
 }
@@ -165,10 +156,6 @@ impl TryFrom<api::v1::ObjectResult> for ObjectResult {
             actual: 0_usize,
         })?;
         Ok(match obj_result {
-            object_result::Result::Select(select) => {
-                let result = (*select.raw_data).try_into().context(DecodeSelectSnafu)?;
-                ObjectResult::Select(result)
-            }
             object_result::Result::Mutate(mutate) => ObjectResult::Mutate(mutate),
             object_result::Result::FlightData(flight_data) => {
                 let flight_messages = raw_flight_data_to_message(flight_data.raw_data)
@@ -188,41 +175,6 @@ impl TryFrom<ObjectResult> for Output {
 
     fn try_from(value: ObjectResult) -> Result<Self> {
         let output = match value {
-            ObjectResult::Select(select) => {
-                let vectors = select
-                    .columns
-                    .iter()
-                    .map(|column| {
-                        column_to_vector(column, select.row_count).context(ColumnToVectorSnafu)
-                    })
-                    .collect::<Result<Vec<VectorRef>>>()?;
-
-                let column_schemas = select
-                    .columns
-                    .iter()
-                    .zip(vectors.iter())
-                    .map(|(column, vector)| {
-                        let datatype = vector.data_type();
-                        // nullable or not, does not affect the output
-                        let mut column_schema =
-                            ColumnSchema::new(&column.column_name, datatype, true);
-                        if column.semantic_type == SemanticType::Timestamp as i32 {
-                            column_schema = column_schema.with_time_index(true);
-                        }
-                        column_schema
-                    })
-                    .collect::<Vec<ColumnSchema>>();
-
-                let schema = Arc::new(Schema::try_new(column_schemas).context(ConvertSchemaSnafu)?);
-                let recordbatches = if vectors.is_empty() {
-                    RecordBatches::try_new(schema, vec![])
-                } else {
-                    RecordBatch::new(schema, vectors)
-                        .and_then(|batch| RecordBatches::try_new(batch.schema.clone(), vec![batch]))
-                }
-                .context(error::CreateRecordBatchesSnafu)?;
-                Output::RecordBatches(recordbatches)
-            }
             ObjectResult::Mutate(mutate) => {
                 if mutate.failure != 0 {
                     return error::MutateFailureSnafu {
@@ -240,17 +192,19 @@ impl TryFrom<ObjectResult> for Output {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
+
     use api::helper::ColumnDataTypeWrapper;
     use api::v1::Column;
     use common_grpc::select::{null_mask, values};
+    use common_grpc_expr::column_to_vector;
+    use datatypes::prelude::{Vector, VectorRef};
     use datatypes::vectors::{
         BinaryVector, BooleanVector, DateTimeVector, DateVector, Float32Vector, Float64Vector,
         Int16Vector, Int32Vector, Int64Vector, Int8Vector, StringVector, UInt16Vector,
         UInt32Vector, UInt64Vector, UInt8Vector,
     };
 
-    use super::*;
-
     #[test]
     fn test_column_to_vector() {
         let mut column = create_test_column(Arc::new(BooleanVector::from(vec![true])));