feat: add source_tables and source_table_ids (#151)

* feat: add `source_tables`

* feat: use `SchemaScopedTableName`

* chore: include flow proto

* chore: rename

* feat: add `CreateFlowTask` and `DropFlowTask`

proto/greptime/v1/ddl.proto

@@ -33,17 +33,17 @@ message DdlRequest {
     AlterExpr alter = 3;
     DropTableExpr drop_table = 4;
     TruncateTableExpr truncate_table = 7;
-    CreateTaskExpr create_task = 8;
-    RemoveTaskExpr remove_task = 9;
+    CreateFlowTaskExpr create_flow_task = 8;
+    DropFlowTaskExpr drop_flow_task = 9;
   }
 }
 
 // Create a flow task to run the SQL when new data arrives.
-message CreateTaskExpr {
+message CreateFlowTaskExpr {
   string catalog_name = 1;
   string task_name = 2;
-  string output_schema_name = 3;
-  string output_table_name = 4;
+  repeated SchemaScopedTableName source_table_names = 3;
+  SchemaScopedTableName sink_table_name = 4;
   bool create_if_not_exists = 5;
   string expire_when = 6;
   string comment = 7;
@@ -51,8 +51,14 @@ message CreateTaskExpr {
   map<string, string> task_options = 9;
 }
 
+// Table name
+message SchemaScopedTableName {
+  string schema_name = 1;
+  string table_name = 2;
+}
+
 // Remove a flow task.
-message RemoveTaskExpr {
+message DropFlowTaskExpr {
   string catalog_name = 1;
   string task_name = 2;
 }

proto/greptime/v1/flow/server.proto

@@ -42,8 +42,8 @@ message InsertRequest {
 
 message FlowRequest {
   oneof body {
-    FlowCreateRequest create = 1;
-    FlowRemoveRequest remove = 2;
+    CreateRequest create = 1;
+    RemoveRequest remove = 2;
   }
 }
 
@@ -56,16 +56,17 @@ message FlowResponse {
 }
 
 // very similar to `ddl.CreateTaskExpr` just replace `task_name` with `task_id`
-message FlowCreateRequest {
-  TaskId task_id = 1;
-  TableId output_table_id = 2;
-  bool create_if_not_exists = 3;
-  string expire_when = 4;
-  string comment = 5;
-  string sql = 6;
-  map<string, string> task_options = 7;
+message CreateRequest {
+  TaskId task_id = 1; 
+  repeated TableId source_table_ids = 2;
+  TableId sink_table_id = 3;
+  bool create_if_not_exists = 4;
+  string expire_when = 5;
+  string comment = 6;
+  string sql = 7;
+  map<string, string> task_options = 8;
 }
 
-message FlowRemoveRequest { TaskId task_id = 1; }
+message RemoveRequest { TaskId task_id = 1; }
 
 message TaskId { uint32 id = 1; }

proto/greptime/v1/meta/ddl.proto

@@ -59,6 +59,14 @@ message DropDatabaseTask {
     DropDatabaseExpr drop_database = 1;
 }
 
+message CreateFlowTask {
+    CreateFlowTaskExpr create_flow_task = 1;
+}
+
+message DropFlowTask {
+    DropFlowTaskExpr drop_flow_task = 1;
+}
+
 message DdlTaskRequest {
     RequestHeader header = 1;
 
@@ -72,6 +80,8 @@ message DdlTaskRequest {
         AlterTableTasks alter_table_tasks = 8;
         DropDatabaseTask drop_database_task = 9;
         CreateDatabaseTask create_database_task = 10;
+        CreateFlowTask create_flow_task = 11;
+        DropFlowTask drop_flow_task = 12;
      }
 }