refactor: store Task's dependents in the single key

Author: KKould

src/meta/app/src/principal/mod.rs

@@ -97,7 +97,8 @@ pub use task::ScheduleType;
 pub use task::State;
 pub use task::Status;
 pub use task::Task;
-pub use task::TaskDependent;
+pub use task::TaskDependentKey;
+pub use task::TaskDependentValue;
 pub use task::TaskMessage;
 pub use task::TaskRun;
 pub use task::TaskState;

src/meta/app/src/principal/task.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::collections::BTreeMap;
+use std::collections::HashSet;
 
 use chrono::DateTime;
 use chrono::Utc;
@@ -185,33 +186,34 @@ pub enum DependentType {
 }
 
 #[derive(Debug, Clone, PartialEq)]
-pub struct TaskDependent {
+pub struct TaskDependentKey {
     pub ty: DependentType,
     pub source: String,
-    pub target: String,
 }
 
-impl TaskDependent {
-    pub fn new(ty: DependentType, source: String, target: String) -> Self {
-        Self { ty, source, target }
+impl TaskDependentKey {
+    pub fn new(ty: DependentType, source: String) -> Self {
+        Self { ty, source }
     }
 }
 
+#[derive(Debug, Clone, PartialEq)]
+pub struct TaskDependentValue(pub HashSet<String>);
+
 mod kvapi_key_impl {
     use databend_common_meta_kvapi::kvapi;
     use databend_common_meta_kvapi::kvapi::KeyError;
 
     use crate::principal::DependentType;
-    use crate::principal::TaskDependent;
+    use crate::principal::TaskDependentKey;
 
-    impl kvapi::KeyCodec for TaskDependent {
+    impl kvapi::KeyCodec for TaskDependentKey {
         fn encode_key(&self, b: kvapi::KeyBuilder) -> kvapi::KeyBuilder {
             match self.ty {
                 DependentType::After => b.push_str("After"),
                 DependentType::Before => b.push_str("Before"),
             }
             .push_str(self.source.as_str())
-            .push_str(self.target.as_str())
         }
 
         fn decode_key(parser: &mut kvapi::KeyParser) -> Result<Self, kvapi::KeyError> {
@@ -226,9 +228,8 @@ mod kvapi_key_impl {
                 }
             };
             let source = parser.next_str()?;
-            let target = parser.next_str()?;
 
-            Ok(Self { ty, source, target })
+            Ok(Self { ty, source })
         }
     }
 }

src/meta/app/src/principal/task_dependent_ident.rs

@@ -14,16 +14,16 @@
 
 use crate::tenant_key::ident::TIdent;
 
-pub type TaskDependentIdent = TIdent<TaskDependentResource, TaskDependent>;
+pub type TaskDependentIdent = TIdent<TaskDependentResource, TaskDependentKey>;
 
 pub use kvapi_impl::TaskDependentResource;
 
-use crate::principal::TaskDependent;
+use crate::principal::TaskDependentKey;
 
 mod kvapi_impl {
     use databend_common_meta_kvapi::kvapi;
 
-    use crate::principal::task::TaskDependent;
+    use crate::principal::task::TaskDependentValue;
     use crate::principal::task_dependent_ident::TaskDependentIdent;
     use crate::tenant_key::resource::TenantResource;
 
@@ -32,10 +32,10 @@ mod kvapi_impl {
         const PREFIX: &'static str = "__fd_task_dependents";
         const TYPE: &'static str = "TaskDependentIdent";
         const HAS_TENANT: bool = true;
-        type ValueType = TaskDependent;
+        type ValueType = TaskDependentValue;
     }
 
-    impl kvapi::Value for TaskDependent {
+    impl kvapi::Value for TaskDependentValue {
         type KeyType = TaskDependentIdent;
 
         fn dependency_keys(&self, _key: &Self::KeyType) -> impl IntoIterator<Item = String> {

src/meta/proto-conv/src/task_from_to_protobuf_impl.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashSet;
+
 use chrono::DateTime;
 use chrono::Utc;
 use databend_common_meta_app::principal as mt;
@@ -197,8 +199,8 @@ impl FromToProto for mt::TaskMessage {
     }
 }
 
-impl FromToProto for mt::TaskDependent {
-    type PB = pb::TaskDependent;
+impl FromToProto for mt::TaskDependentKey {
+    type PB = pb::TaskDependentKey;
 
     fn get_pb_ver(p: &Self::PB) -> u64 {
         p.ver
@@ -213,22 +215,41 @@ impl FromToProto for mt::TaskDependent {
                 _ => return Err(Incompatible::new(format!("invalid task type {}", p.ty))),
             },
             source: p.source,
-            target: p.target,
         })
     }
 
     fn to_pb(&self) -> Result<Self::PB, Incompatible> {
-        Ok(pb::TaskDependent {
+        Ok(pb::TaskDependentKey {
             ver: VER,
             min_reader_ver: MIN_READER_VER,
 
             source: self.source.clone(),
-            target: self.target.clone(),
             ty: self.ty as i32,
         })
     }
 }
 
+impl FromToProto for mt::TaskDependentValue {
+    type PB = pb::TaskDependentValue;
+
+    fn get_pb_ver(p: &Self::PB) -> u64 {
+        p.ver
+    }
+
+    fn from_pb(p: Self::PB) -> Result<Self, Incompatible>
+    where Self: Sized {
+        Ok(Self(HashSet::from_iter(p.names)))
+    }
+
+    fn to_pb(&self) -> Result<Self::PB, Incompatible> {
+        Ok(pb::TaskDependentValue {
+            ver: VER,
+            min_reader_ver: MIN_READER_VER,
+            names: Vec::from_iter(self.0.iter().cloned()),
+        })
+    }
+}
+
 impl FromToProto for mt::TaskState {
     type PB = pb::TaskState;
 

src/meta/proto-conv/tests/it/v141_task_state.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashSet;
+
 use databend_common_meta_app::principal as mt;
 use fastrace::func_name;
 
@@ -30,15 +32,31 @@ fn test_decode_v141_task_state() -> anyhow::Result<()> {
 
 #[test]
 fn test_decode_v141_task_dependent() -> anyhow::Result<()> {
-    let task_dependent_v141 = vec![10, 1, 97, 18, 1, 99, 160, 6, 141, 1, 168, 6, 24];
-
-    let want = || mt::TaskDependent {
-        ty: mt::DependentType::After,
-        source: s("a"),
-        target: s("c"),
-    };
-    common::test_pb_from_to(func_name!(), want())?;
-    common::test_load_old(func_name!(), task_dependent_v141.as_slice(), 141, want())?;
+    {
+        let task_dependent_key_v141 = vec![10, 1, 97, 160, 6, 141, 1, 168, 6, 24];
+        let want = || mt::TaskDependentKey {
+            ty: mt::DependentType::After,
+            source: s("a"),
+        };
+        common::test_pb_from_to(func_name!(), want())?;
+        common::test_load_old(
+            func_name!(),
+            task_dependent_key_v141.as_slice(),
+            141,
+            want(),
+        )?;
+    }
+    {
+        let task_dependent_value_v141 = vec![10, 1, 97, 10, 1, 98, 160, 6, 141, 1, 168, 6, 24];
+        let want = || mt::TaskDependentValue(HashSet::from([s("a"), s("b")]));
+        common::test_pb_from_to(func_name!(), want())?;
+        common::test_load_old(
+            func_name!(),
+            task_dependent_value_v141.as_slice(),
+            141,
+            want(),
+        )?;
+    }
 
     Ok(())
 }

src/meta/protos/proto/task.proto

@@ -86,7 +86,7 @@ message TaskState {
   bool is_succeeded = 1;
 }
 
-message TaskDependent {
+message TaskDependentKey {
   uint64 ver = 100;
   uint64 min_reader_ver = 101;
 
@@ -95,6 +95,12 @@ message TaskDependent {
     Before = 1;
   }
   string source = 1;
-  string target = 2;
   DependentType ty = 3;
 }
+
+message TaskDependentValue {
+  uint64 ver = 100;
+  uint64 min_reader_ver = 101;
+
+  repeated string names = 1;
+}