chore: codefmt

Author: KKould

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

@@ -101,7 +101,7 @@ pub use task::TaskDependentKey;
 pub use task::TaskDependentValue;
 pub use task::TaskMessage;
 pub use task::TaskRun;
-pub use task::TaskStateValue;
+pub use task::TaskSucceededStateValue;
 pub use task::WarehouseOptions;
 pub use task_ident::TaskIdent;
 pub use task_ident::TaskIdentRaw;

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

@@ -175,21 +175,22 @@ impl TaskMessage {
 }
 
 #[derive(Debug, Clone, PartialEq)]
-pub struct TaskStateKey {
-    pub current: String,
-    pub next: String,
+pub struct TaskSucceededStateKey {
+    pub before_task: String,
+    pub after_task: String,
 }
 
-impl TaskStateKey {
-    pub fn new(current: String, next: String) -> Self {
-        Self { current, next }
+impl TaskSucceededStateKey {
+    pub fn new(before_task: String, after_task: String) -> Self {
+        Self {
+            before_task,
+            after_task,
+        }
     }
 }
 
 #[derive(Debug, Clone, PartialEq)]
-pub struct TaskStateValue {
-    pub is_succeeded: bool,
-}
+pub struct TaskSucceededStateValue;
 
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub enum DependentType {
@@ -218,22 +219,25 @@ mod kvapi_key_impl {
     use databend_common_meta_kvapi::kvapi::KeyError;
     use databend_common_meta_kvapi::kvapi::KeyParser;
 
-    use crate::principal::task::TaskStateKey;
+    use crate::principal::task::TaskSucceededStateKey;
     use crate::principal::DependentType;
     use crate::principal::TaskDependentKey;
 
-    impl kvapi::KeyCodec for TaskStateKey {
+    impl kvapi::KeyCodec for TaskSucceededStateKey {
         fn encode_key(&self, b: KeyBuilder) -> KeyBuilder {
-            b.push_str(self.current.as_str())
-                .push_str(self.next.as_str())
+            b.push_str(self.before_task.as_str())
+                .push_str(self.after_task.as_str())
         }
 
         fn decode_key(parser: &mut KeyParser) -> Result<Self, KeyError>
         where Self: Sized {
-            let current = parser.next_str()?;
-            let next = parser.next_str()?;
+            let before_task = parser.next_str()?;
+            let after_task = parser.next_str()?;
 
-            Ok(Self { current, next })
+            Ok(Self {
+                before_task,
+                after_task,
+            })
         }
     }
 

src/meta/app/src/principal/task_state_ident.rs

@@ -14,39 +14,43 @@
 
 use crate::tenant_key::ident::TIdent;
 
-pub type TaskStateIdent = TIdent<Resource, TaskStateKey>;
-
-impl TaskStateIdent {
-    pub fn new(tenant: impl ToTenant, current: impl ToString, next: impl ToString) -> Self {
-        TaskStateIdent::new_generic(
+pub type TaskSucceededStateIdent = TIdent<Resource, TaskSucceededStateKey>;
+
+impl TaskSucceededStateIdent {
+    pub fn new(
+        tenant: impl ToTenant,
+        before_task: impl ToString,
+        after_task: impl ToString,
+    ) -> Self {
+        TaskSucceededStateIdent::new_generic(
             tenant,
-            TaskStateKey::new(current.to_string(), next.to_string()),
+            TaskSucceededStateKey::new(before_task.to_string(), after_task.to_string()),
         )
     }
 }
 
 pub use kvapi_impl::Resource;
 
-use crate::principal::task::TaskStateKey;
+use crate::principal::task::TaskSucceededStateKey;
 use crate::tenant::ToTenant;
 
 mod kvapi_impl {
     use databend_common_meta_kvapi::kvapi;
 
-    use crate::principal::task::TaskStateValue;
-    use crate::principal::task_state_ident::TaskStateIdent;
+    use crate::principal::task_state_ident::TaskSucceededStateIdent;
+    use crate::principal::TaskSucceededStateValue;
     use crate::tenant_key::resource::TenantResource;
 
     pub struct Resource;
     impl TenantResource for Resource {
         const PREFIX: &'static str = "__fd_task_states";
-        const TYPE: &'static str = "TaskStateIdent";
+        const TYPE: &'static str = "TaskSucceededStateIdent";
         const HAS_TENANT: bool = true;
-        type ValueType = TaskStateValue;
+        type ValueType = TaskSucceededStateValue;
     }
 
-    impl kvapi::Value for TaskStateValue {
-        type KeyType = TaskStateIdent;
+    impl kvapi::Value for TaskSucceededStateValue {
+        type KeyType = TaskSucceededStateIdent;
 
         fn dependency_keys(&self, _key: &Self::KeyType) -> impl IntoIterator<Item = String> {
             []

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

@@ -219,25 +219,22 @@ impl FromToProto for mt::TaskDependentValue {
     }
 }
 
-impl FromToProto for mt::TaskStateValue {
+impl FromToProto for mt::TaskSucceededStateValue {
     type PB = pb::TaskStateValue;
 
     fn get_pb_ver(p: &Self::PB) -> u64 {
         p.ver
     }
 
-    fn from_pb(p: Self::PB) -> Result<Self, Incompatible>
+    fn from_pb(_p: Self::PB) -> Result<Self, Incompatible>
     where Self: Sized {
-        Ok(Self {
-            is_succeeded: p.is_succeeded,
-        })
+        Ok(Self)
     }
 
     fn to_pb(&self) -> Result<Self::PB, Incompatible> {
         Ok(pb::TaskStateValue {
             ver: VER,
             min_reader_ver: MIN_READER_VER,
-            is_succeeded: self.is_succeeded,
         })
     }
 }

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

@@ -21,9 +21,9 @@ use crate::common;
 
 #[test]
 fn test_decode_v141_task_state() -> anyhow::Result<()> {
-    let task_state_v141 = vec![8, 1, 160, 6, 141, 1, 168, 6, 24];
+    let task_state_v141 = vec![160, 6, 141, 1, 168, 6, 24];
 
-    let want = || mt::TaskStateValue { is_succeeded: true };
+    let want = || mt::TaskSucceededStateValue;
     common::test_pb_from_to(func_name!(), want())?;
     common::test_load_old(func_name!(), task_state_v141.as_slice(), 141, want())?;
 

src/meta/protos/proto/task.proto

@@ -82,8 +82,6 @@ message Task {
 message TaskStateValue {
   uint64 ver = 100;
   uint64 min_reader_ver = 101;
-
-  bool is_succeeded = 1;
 }
 
 message TaskDependentValue {

src/query/management/src/task/task_mgr.rs

@@ -23,14 +23,15 @@ use databend_common_ast::ast::ScheduleOptions;
 use databend_common_meta_api::kv_pb_api::KVPbApi;
 use databend_common_meta_api::kv_pb_api::UpsertPB;
 use databend_common_meta_api::txn_cond_eq_seq;
+use databend_common_meta_api::txn_op_del;
 use databend_common_meta_api::util::txn_put_pb;
 use databend_common_meta_api::SchemaApi;
 use databend_common_meta_app::principal::task;
 use databend_common_meta_app::principal::task::TaskMessage;
-use databend_common_meta_app::principal::task::TaskStateValue;
+use databend_common_meta_app::principal::task::TaskSucceededStateValue;
 use databend_common_meta_app::principal::task_dependent_ident::TaskDependentIdent;
 use databend_common_meta_app::principal::task_message_ident::TaskMessageIdent;
-use databend_common_meta_app::principal::task_state_ident::TaskStateIdent;
+use databend_common_meta_app::principal::task_state_ident::TaskSucceededStateIdent;
 use databend_common_meta_app::principal::ScheduleType;
 use databend_common_meta_app::principal::Status;
 use databend_common_meta_app::principal::Task;
@@ -40,16 +41,16 @@ use databend_common_meta_app::tenant::Tenant;
 use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_kvapi::kvapi::DirName;
 use databend_common_meta_kvapi::kvapi::Key;
+use databend_common_meta_types::txn_condition::Target;
+use databend_common_meta_types::ConditionResult;
 use databend_common_meta_types::MatchSeq;
 use databend_common_meta_types::MetaError;
 use databend_common_meta_types::TxnCondition;
 use databend_common_meta_types::TxnOp;
 use databend_common_meta_types::TxnRequest;
 use databend_common_meta_types::With;
-use databend_common_proto_conv::FromToProto;
 use futures::StreamExt;
 use futures::TryStreamExt;
-use prost::Message;
 use seq_marked::SeqValue;
 
 use crate::task::errors::TaskApiError;
@@ -407,7 +408,7 @@ impl TaskMgr {
         ));
         let mut stream = self
             .kv_api
-            .list_pb_keys(&DirName::new(TaskStateIdent::new(
+            .list_pb_keys(&DirName::new(TaskSucceededStateIdent::new(
                 &self.tenant,
                 task_name,
                 "",
@@ -456,10 +457,6 @@ impl TaskMgr {
         task_name: &str,
     ) -> Result<Result<Vec<String>, TaskError>, TaskApiError> {
         let task_before_ident = TaskDependentIdent::new_before(&self.tenant, task_name);
-        let succeeded_value = TaskStateValue { is_succeeded: true };
-        let not_succeeded_value = TaskStateValue {
-            is_succeeded: false,
-        };
 
         let Some(task_before_dependent) = self.kv_api.get_pb(&task_before_ident).await? else {
             return Ok(Ok(Vec::new()));
@@ -478,25 +475,24 @@ impl TaskMgr {
             };
             let target_after = &target_after_ident.name().source;
             let this_task_to_target_state =
-                TaskStateIdent::new(&self.tenant, task_name, target_after);
+                TaskSucceededStateIdent::new(&self.tenant, task_name, target_after);
             let mut request = TxnRequest::new(vec![], vec![]).with_else(vec![txn_put_pb(
                 &this_task_to_target_state,
-                &succeeded_value,
+                &TaskSucceededStateValue,
             )?]);
 
             for before_target_after in target_after_dependent.0.iter() {
                 let task_ident =
-                    TaskStateIdent::new(&self.tenant, before_target_after, target_after);
+                    TaskSucceededStateIdent::new(&self.tenant, before_target_after, target_after);
                 // Only care about the predecessors of this task's successor tasks, excluding this task itself.
                 if before_target_after != task_name {
-                    request.condition.push(TxnCondition::eq_value(
-                        task_ident.to_string_key(),
-                        succeeded_value.to_pb()?.encode_to_vec(),
-                    ));
+                    request.condition.push(TxnCondition {
+                        key: task_ident.to_string_key(),
+                        expected: ConditionResult::Gt as i32,
+                        target: Some(Target::Seq(0)),
+                    });
                 }
-                request
-                    .if_then
-                    .push(txn_put_pb(&task_ident, &not_succeeded_value)?);
+                request.if_then.push(txn_op_del(&task_ident));
             }
             let reply = self.kv_api.transaction(request).await?;