feat: define proto for InsertExpr (grpc) (#79)

* feat: implement InsertExpr

* 1.InsertExpr reverted to previous version 2.add InsertBatch message

* add two SemanticTypes: TAG, TIMESTAMP

* chore: format proto files

* chore: add some comments about "Column"

* fix: rename "semanticType" -> "semantic_type"

* fix: unique number in InsertBatch

* fix: type of f64_values

* chore: move insertbatch and column to insert.proto

* chore: rename "ExprHeader" to "Header"

* fix: ExprHeader not found in this scope

src/api/greptime/v1/database.proto

@@ -3,29 +3,31 @@ syntax = "proto3";
 package greptime.v1;
 
 message DatabaseRequest {
-    string name = 1;
-    repeated ObjectExpr exprs = 2;
+  string name = 1;
+  repeated ObjectExpr exprs = 2;
 }
 
 message DatabaseResponse {
-    repeated ObjectResult results = 1;
+  repeated ObjectResult results = 1;
 }
 
 message ObjectExpr {
-    ExprHeader header = 1;
+  Header header = 1;
+  oneof expr {
     InsertExpr insert = 2;
     SelectExpr select = 3;
     UpdateExpr update = 4;
     DeleteExpr delete = 5;
+  }
 }
 
 message SelectExpr {
-    string select = 1; // sql, promql, etc.
+  string select = 1; // sql, promql, etc.
 }
 
 message InsertExpr {
-    string table_name = 1;
-    repeated bytes values = 2;
+  string table_name = 1;
+  repeated bytes values = 2;
 }
 
 // TODO(jiachun)
@@ -34,19 +36,19 @@ message UpdateExpr {}
 message DeleteExpr {}
 
 message ObjectResult {
-    ResultHeader header = 1;
-    string schema = 2;
-    repeated bytes results = 3;
+  ResultHeader header = 1;
+  string schema = 2;
+  repeated bytes results = 3;
 }
 
-message ExprHeader {
-    uint32 version = 1;
+message Header {
+  uint32 version = 1;
 }
 
 message ResultHeader {
-    uint32 version = 1;
-    uint32 success = 2;
-    uint32 failure = 3;
-    uint32 code = 4;
-    string err_msg = 5;
+  uint32 version = 1;
+  uint32 success = 2;
+  uint32 failure = 3;
+  uint32 code = 4;
+  string err_msg = 5;
 }

src/api/greptime/v1/greptime.proto

@@ -6,15 +6,15 @@ import "greptime/v1/admin.proto";
 import "greptime/v1/database.proto";
 
 service Greptime {
-    rpc Batch(BatchRequest) returns (BatchResponse) {}
+  rpc Batch(BatchRequest) returns (BatchResponse) {}
 }
 
 message BatchRequest {
-    repeated AdminRequest admins = 1;
-    repeated DatabaseRequest databases = 2;
+  repeated AdminRequest admins = 1;
+  repeated DatabaseRequest databases = 2;
 }
 
 message BatchResponse {
-    repeated AdminResponse admins = 1;
-    repeated DatabaseResponse databases = 2;
+  repeated AdminResponse admins = 1;
+  repeated DatabaseResponse databases = 2;
 }

src/api/greptime/v1/insert.proto

@@ -0,0 +1,47 @@
+message InsertBatch {
+  repeated Column columns = 1;
+  uint32 row_count = 2;
+}
+
+message Column {
+  string column_name = 1;
+
+  enum SemanticType {
+    TAG = 0;
+    FIELD = 1;
+    TIMESTAMP = 2;
+  }
+  SemanticType semantic_type = 2;
+
+  message Values {
+    repeated int32 i8_values = 1;
+    repeated int32 i16_values = 2;
+    repeated int32 i32_values = 3;
+    repeated int64 i64_values = 4;
+
+    repeated uint32 u8_values = 5;
+    repeated uint32 u16_values = 6;
+    repeated uint32 u32_values = 7;
+    repeated uint64 u64_values = 8;
+
+    repeated float f32_values = 9;
+    repeated double f64_values = 10;
+
+    repeated bool bool_values = 11;
+    repeated bytes bytes_values = 12;
+    repeated string string_values = 13;
+  }
+  // The array of non-null values in this column.
+  //
+  // For example: suppose there is a column "foo" that contains some int32 values (1, 2, 3, 4, 5, null, 7, 8, 9, null);
+  //   column:
+  //     column_name: foo
+  //     semantic_type: Tag
+  //     values: 1, 2, 3, 4, 5, 7, 8, 9
+  //     null_masks: 00100000 00000010
+  Values values = 3;
+
+  // Mask maps the positions of null values.
+  // If a bit in null_mask is 1, it indicates that the column value at that position is null.
+  bytes null_mask = 4;
+}

src/client/src/database.rs

@@ -1,3 +1,4 @@
+use api::v1::object_expr;
 use api::v1::*;
 use snafu::ensure;
 
@@ -27,7 +28,7 @@ impl Database {
     }
 
     pub async fn insert(&self, table: impl Into<String>, values: Vec<Bytes>) -> Result<()> {
-        let header = ExprHeader {
+        let header = Header {
             version: PROTOCOL_VERSION,
         };
         let insert = InsertExpr {
@@ -36,8 +37,7 @@ impl Database {
         };
         let expr = ObjectExpr {
             header: Some(header),
-            insert: Some(insert),
-            ..Default::default()
+            expr: Some(object_expr::Expr::Insert(insert)),
         };
 
         self.object(expr).await?;