Skip to content

Commit 7ef3131

Browse files
KKouldKKould
committed
chore: codefmt
1 parent 5a9de35 commit 7ef3131

File tree

4 files changed

+55
-68
lines changed

4 files changed

+55
-68
lines changed

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313
// limitations under the License.
1414

1515
use std::collections::BTreeMap;
16-
use std::collections::HashSet;
16+
use std::collections::BTreeSet;
1717

1818
use chrono::DateTime;
1919
use chrono::Utc;
@@ -198,7 +198,7 @@ impl TaskDependentKey {
198198
}
199199

200200
#[derive(Debug, Clone, PartialEq)]
201-
pub struct TaskDependentValue(pub HashSet<String>);
201+
pub struct TaskDependentValue(pub BTreeSet<String>);
202202

203203
mod kvapi_key_impl {
204204
use databend_common_meta_kvapi::kvapi;

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

Lines changed: 2 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -12,13 +12,12 @@
1212
// See the License for the specific language governing permissions and
1313
// limitations under the License.
1414

15-
use std::collections::HashSet;
15+
use std::collections::BTreeSet;
1616

1717
use chrono::DateTime;
1818
use chrono::Utc;
1919
use databend_common_meta_app::principal as mt;
2020
use databend_common_meta_app::principal::task::Status;
21-
use databend_common_meta_app::principal::DependentType;
2221
use databend_common_protos::pb;
2322
use databend_common_protos::pb::task_message::DeleteTask;
2423
use databend_common_protos::pb::task_message::Message;
@@ -199,36 +198,6 @@ impl FromToProto for mt::TaskMessage {
199198
}
200199
}
201200

202-
impl FromToProto for mt::TaskDependentKey {
203-
type PB = pb::TaskDependentKey;
204-
205-
fn get_pb_ver(p: &Self::PB) -> u64 {
206-
p.ver
207-
}
208-
209-
fn from_pb(p: Self::PB) -> Result<Self, Incompatible>
210-
where Self: Sized {
211-
Ok(Self {
212-
ty: match p.ty {
213-
0 => DependentType::After,
214-
1 => DependentType::Before,
215-
_ => return Err(Incompatible::new(format!("invalid task type {}", p.ty))),
216-
},
217-
source: p.source,
218-
})
219-
}
220-
221-
fn to_pb(&self) -> Result<Self::PB, Incompatible> {
222-
Ok(pb::TaskDependentKey {
223-
ver: VER,
224-
min_reader_ver: MIN_READER_VER,
225-
226-
source: self.source.clone(),
227-
ty: self.ty as i32,
228-
})
229-
}
230-
}
231-
232201
impl FromToProto for mt::TaskDependentValue {
233202
type PB = pb::TaskDependentValue;
234203

@@ -238,7 +207,7 @@ impl FromToProto for mt::TaskDependentValue {
238207

239208
fn from_pb(p: Self::PB) -> Result<Self, Incompatible>
240209
where Self: Sized {
241-
Ok(Self(HashSet::from_iter(p.names)))
210+
Ok(Self(BTreeSet::from_iter(p.names)))
242211
}
243212

244213
fn to_pb(&self) -> Result<Self::PB, Incompatible> {

‎src/meta/protos/proto/task.proto

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -94,8 +94,8 @@ message TaskDependentKey {
9494
After = 0;
9595
Before = 1;
9696
}
97-
string source = 1;
98-
DependentType ty = 3;
97+
DependentType ty = 1;
98+
string source = 2;
9999
}
100100

101101
message TaskDependentValue {

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

Lines changed: 49 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212
// See the License for the specific language governing permissions and
1313
// limitations under the License.
1414

15-
use std::collections::HashSet;
15+
use std::collections::BTreeSet;
1616
use std::str::FromStr;
1717
use std::sync::Arc;
1818

@@ -28,7 +28,6 @@ use databend_common_meta_app::principal::task;
2828
use databend_common_meta_app::principal::task::TaskMessage;
2929
use databend_common_meta_app::principal::task::TaskState;
3030
use databend_common_meta_app::principal::task_dependent_ident::TaskDependentIdent;
31-
use databend_common_meta_app::principal::task_dependent_ident::TaskDependentResource;
3231
use databend_common_meta_app::principal::task_message_ident::TaskMessageIdent;
3332
use databend_common_meta_app::principal::task_state_ident::TaskStateIdent;
3433
use databend_common_meta_app::principal::DependentType;
@@ -40,7 +39,6 @@ use databend_common_meta_app::principal::TaskDependentValue;
4039
use databend_common_meta_app::principal::TaskIdent;
4140
use databend_common_meta_app::schema::CreateOption;
4241
use databend_common_meta_app::tenant::Tenant;
43-
use databend_common_meta_app::tenant_key::ident::TIdent;
4442
use databend_common_meta_kvapi::kvapi;
4543
use databend_common_meta_kvapi::kvapi::DirName;
4644
use databend_common_meta_kvapi::kvapi::Key;
@@ -278,12 +276,11 @@ impl TaskMgr {
278276
let after_dependent_ident =
279277
TaskDependentIdent::new_generic(&self.tenant, after_dependent.clone());
280278

281-
self.check_and_set(
279+
self.check_and_add(
282280
&mut check_ops,
283281
&mut update_ops,
284282
new_afters,
285283
&after_dependent_ident,
286-
true,
287284
)
288285
.await?;
289286

@@ -292,12 +289,11 @@ impl TaskMgr {
292289
let before_dependent_ident =
293290
TaskDependentIdent::new_generic(&self.tenant, before_dependent);
294291

295-
self.check_and_set(
292+
self.check_and_add(
296293
&mut check_ops,
297294
&mut update_ops,
298295
&[task_name.to_string()],
299296
&before_dependent_ident,
300-
true,
301297
)
302298
.await?;
303299
}
@@ -328,12 +324,11 @@ impl TaskMgr {
328324
let after_dependent_ident =
329325
TaskDependentIdent::new_generic(&self.tenant, after_dependent.clone());
330326

331-
self.check_and_set(
327+
self.check_and_remove(
332328
&mut check_ops,
333329
&mut update_ops,
334330
remove_afters,
335331
&after_dependent_ident,
336-
false,
337332
)
338333
.await?;
339334

@@ -342,12 +337,11 @@ impl TaskMgr {
342337
let before_dependent_ident =
343338
TaskDependentIdent::new_generic(&self.tenant, before_dependent.clone());
344339

345-
self.check_and_set(
340+
self.check_and_remove(
346341
&mut check_ops,
347342
&mut update_ops,
348343
&[task_name.to_string()],
349344
&before_dependent_ident,
350-
false,
351345
)
352346
.await?;
353347
}
@@ -556,55 +550,79 @@ impl TaskMgr {
556550
let target_ident =
557551
TaskDependentIdent::new_generic(&self.tenant, target_key.clone());
558552

559-
self.check_and_set(
553+
self.check_and_remove(
560554
check_ops,
561555
update_ops,
562556
&[task_name.to_string()],
563557
&target_ident,
564-
false,
565558
)
566559
.await?;
567560
}
568561
}
569562
Ok(())
570563
}
571564

572-
async fn check_and_set(
565+
async fn check_and_add(
573566
&self,
574567
check_ops: &mut Vec<TxnCondition>,
575568
update_ops: &mut Vec<TxnOp>,
576569
task_names: &[String],
577-
dependent_ident: &TIdent<TaskDependentResource, TaskDependentKey>,
578-
is_add: bool,
570+
dependent_ident: &TaskDependentIdent,
579571
) -> Result<(), TaskApiError> {
580-
match self.kv_api.get(dependent_ident).await? {
572+
match self.kv_api.get_pb(dependent_ident).await? {
581573
None => {
582574
check_ops.push(TxnCondition {
583575
key: dependent_ident.to_string_key(),
584576
expected: ConditionResult::Eq as i32,
585577
target: Some(Target::Seq(0)),
586578
});
587579

588-
if is_add {
589-
update_ops.push(TxnOp::put(
590-
dependent_ident.to_string_key(),
591-
TaskDependentValue(HashSet::from_iter(task_names.iter().cloned()))
592-
.to_pb()?
593-
.encode_to_vec(),
594-
));
595-
}
580+
update_ops.push(TxnOp::put(
581+
dependent_ident.to_string_key(),
582+
TaskDependentValue(BTreeSet::from_iter(task_names.iter().cloned()))
583+
.to_pb()?
584+
.encode_to_vec(),
585+
));
596586
}
597587
Some(mut old_dependent) => {
598-
check_ops.push(TxnCondition::eq_value(
588+
check_ops.push(TxnCondition::eq_seq(
589+
dependent_ident.to_string_key(),
590+
old_dependent.seq,
591+
));
592+
593+
old_dependent.0.extend(task_names.iter().cloned());
594+
595+
update_ops.push(TxnOp::put(
599596
dependent_ident.to_string_key(),
600597
old_dependent.to_pb()?.encode_to_vec(),
601598
));
599+
}
600+
}
601+
Ok(())
602+
}
602603

603-
if is_add {
604-
old_dependent.0.extend(task_names.iter().cloned());
605-
} else {
606-
old_dependent.0.retain(|name| !task_names.contains(name));
607-
}
604+
async fn check_and_remove(
605+
&self,
606+
check_ops: &mut Vec<TxnCondition>,
607+
update_ops: &mut Vec<TxnOp>,
608+
task_names: &[String],
609+
dependent_ident: &TaskDependentIdent,
610+
) -> Result<(), TaskApiError> {
611+
match self.kv_api.get_pb(dependent_ident).await? {
612+
None => {
613+
check_ops.push(TxnCondition {
614+
key: dependent_ident.to_string_key(),
615+
expected: ConditionResult::Eq as i32,
616+
target: Some(Target::Seq(0)),
617+
});
618+
}
619+
Some(mut old_dependent) => {
620+
check_ops.push(TxnCondition::eq_seq(
621+
dependent_ident.to_string_key(),
622+
old_dependent.seq,
623+
));
624+
625+
old_dependent.0.retain(|name| !task_names.contains(name));
608626

609627
update_ops.push(TxnOp::put(
610628
dependent_ident.to_string_key(),

0 commit comments

Comments
 (0)