feat: modify proto for distribute insert (#327)

src/api/greptime/v1/common.proto

@@ -2,6 +2,10 @@ syntax = "proto3";
 
 package greptime.v1;
 
+message RequestHeader {
+  string tenant = 1;
+}
+
 message ExprHeader {
   uint32 version = 1;
 }

src/api/greptime/v1/database.proto

@@ -55,6 +55,8 @@ message InsertExpr {
     // The "sql" field is meant to be removed in the future.
     string sql = 3;
   }
+  
+  map<string, bytes> options = 4;
 }
 
 // TODO(jiachun)

src/api/greptime/v1/greptime.proto

@@ -3,6 +3,7 @@ syntax = "proto3";
 package greptime.v1;
 
 import "greptime/v1/admin.proto";
+import "greptime/v1/common.proto";
 import "greptime/v1/database.proto";
 
 service Greptime {
@@ -10,8 +11,9 @@ service Greptime {
 }
 
 message BatchRequest {
-  repeated AdminRequest admins = 1;
-  repeated DatabaseRequest databases = 2;
+  RequestHeader header = 1;
+  repeated AdminRequest admins = 2;
+  repeated DatabaseRequest databases = 3;
 }
 
 message BatchResponse {

src/client/examples/insert.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use api::v1::{codec::InsertBatch, *};
 use client::{Client, Database};
 
@@ -18,6 +20,7 @@ async fn run() {
         expr: Some(insert_expr::Expr::Values(insert_expr::Values {
             values: insert_batches(),
         })),
+        options: HashMap::default(),
     };
     db.insert(expr).await.unwrap();
 }

src/datanode/src/tests/grpc_test.rs

@@ -162,6 +162,7 @@ async fn insert_and_assert(db: &Database) {
     let expr = InsertExpr {
         table_name: "demo".to_string(),
         expr: Some(insert_expr::Expr::Values(insert_expr::Values { values })),
+        options: HashMap::default(),
     };
     let result = db.insert(expr).await;
     assert!(result.is_ok());

src/frontend/src/instance.rs

@@ -93,6 +93,7 @@ impl SqlQueryHandler for Instance {
                 let expr = InsertExpr {
                     table_name,
                     expr: Some(insert_expr::Expr::Sql(query.to_string())),
+                    options: HashMap::default(),
                 };
                 self.db
                     .insert(expr)
@@ -444,6 +445,7 @@ mod tests {
         let insert_expr = InsertExpr {
             table_name: "demo".to_string(),
             expr: Some(insert_expr::Expr::Values(insert_expr::Values { values })),
+            options: HashMap::default(),
         };
         let object_expr = ObjectExpr {
             header: Some(ExprHeader::default()),

src/servers/src/influxdb.rs

@@ -98,6 +98,7 @@ impl TryFrom<&InfluxdbRequest> for Vec<InsertExpr> {
                 expr: Some(Expr::Values(insert_expr::Values {
                     values: vec![writer.finish().into()],
                 })),
+                options: HashMap::default(),
             })
             .collect())
     }

src/servers/src/opentsdb/codec.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use api::v1::codec::InsertBatch;
 use api::v1::{column, column::SemanticType, insert_expr, Column, ColumnDataType, InsertExpr};
 
@@ -159,6 +161,7 @@ impl DataPoint {
             expr: Some(insert_expr::Expr::Values(insert_expr::Values {
                 values: vec![batch.into()],
             })),
+            options: HashMap::default(),
         }
     }