Skip to content

Commit c6c7373

Browse files
KKouldKKould
committed
chore: codefmt
1 parent 8b34211 commit c6c7373

File tree

1 file changed

+35
-65
lines changed

1 file changed

+35
-65
lines changed

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

Lines changed: 35 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -42,8 +42,6 @@ use databend_common_meta_app::tenant::Tenant;
4242
use databend_common_meta_kvapi::kvapi;
4343
use databend_common_meta_kvapi::kvapi::DirName;
4444
use databend_common_meta_kvapi::kvapi::Key;
45-
use databend_common_meta_types::txn_condition::Target;
46-
use databend_common_meta_types::ConditionResult;
4745
use databend_common_meta_types::MatchSeq;
4846
use databend_common_meta_types::MetaError;
4947
use databend_common_meta_types::TxnCondition;
@@ -277,65 +275,55 @@ impl TaskMgr {
277275
TaskDependentIdent::new_generic(&self.tenant, after_dependent.clone());
278276
let after_key = after_dependent_ident.to_string_key();
279277

280-
match self.kv_api.get_pb(&after_dependent_ident).await? {
278+
let seq = match self.kv_api.get_pb(&after_dependent_ident).await? {
281279
None => {
282-
check_ops.push(TxnCondition {
283-
key: after_key.clone(),
284-
expected: ConditionResult::Eq as i32,
285-
target: Some(Target::Seq(0)),
286-
});
287-
288280
update_ops.push(TxnOp::put(
289-
after_key,
281+
after_key.clone(),
290282
TaskDependentValue(BTreeSet::from_iter(new_afters.iter().cloned()))
291283
.to_pb()?
292284
.encode_to_vec(),
293285
));
286+
0
294287
}
295288
Some(mut old_dependent) => {
296-
check_ops.push(TxnCondition::eq_seq(after_key.clone(), old_dependent.seq));
297-
298289
old_dependent.0.extend(new_afters.iter().cloned());
299290

300291
update_ops.push(TxnOp::put(
301-
after_key,
292+
after_key.clone(),
302293
old_dependent.to_pb()?.encode_to_vec(),
303294
));
295+
old_dependent.seq
304296
}
305-
}
297+
};
298+
check_ops.push(TxnCondition::eq_seq(after_key, seq));
306299

307300
for after in new_afters {
308301
let before_dependent = TaskDependentKey::new(DependentType::Before, after.clone());
309302
let before_dependent_ident =
310303
TaskDependentIdent::new_generic(&self.tenant, before_dependent);
311304
let before_key = before_dependent_ident.to_string_key();
312305

313-
match self.kv_api.get_pb(&before_dependent_ident).await? {
306+
let seq = match self.kv_api.get_pb(&before_dependent_ident).await? {
314307
None => {
315-
check_ops.push(TxnCondition {
316-
key: before_key.clone(),
317-
expected: ConditionResult::Eq as i32,
318-
target: Some(Target::Seq(0)),
319-
});
320-
321308
update_ops.push(TxnOp::put(
322-
before_key,
309+
before_key.clone(),
323310
TaskDependentValue(BTreeSet::from_iter([task_name.to_string()]))
324311
.to_pb()?
325312
.encode_to_vec(),
326313
));
314+
0
327315
}
328316
Some(mut old_dependent) => {
329-
check_ops.push(TxnCondition::eq_seq(before_key.clone(), old_dependent.seq));
330-
331317
old_dependent.0.insert(task_name.to_string());
332318

333319
update_ops.push(TxnOp::put(
334-
before_key,
320+
before_key.clone(),
335321
old_dependent.to_pb()?.encode_to_vec(),
336322
));
323+
old_dependent.seq
337324
}
338-
}
325+
};
326+
check_ops.push(TxnCondition::eq_seq(before_key, seq));
339327
}
340328
let request = TxnRequest::new(check_ops, update_ops);
341329
let reply = self.kv_api.transaction(request).await?;
@@ -365,52 +353,41 @@ impl TaskMgr {
365353
TaskDependentIdent::new_generic(&self.tenant, after_dependent.clone());
366354
let after_key = after_dependent_ident.to_string_key();
367355

368-
match self.kv_api.get_pb(&after_dependent_ident).await? {
369-
None => {
370-
check_ops.push(TxnCondition {
371-
key: after_key.clone(),
372-
expected: ConditionResult::Eq as i32,
373-
target: Some(Target::Seq(0)),
374-
});
375-
}
356+
let seq = match self.kv_api.get_pb(&after_dependent_ident).await? {
357+
None => 0,
376358
Some(mut old_dependent) => {
377-
check_ops.push(TxnCondition::eq_seq(after_key.clone(), old_dependent.seq));
378-
379359
for remove_after in remove_afters {
380360
old_dependent.0.remove(remove_after);
381361
}
382362
update_ops.push(TxnOp::put(
383-
after_key,
363+
after_key.clone(),
384364
old_dependent.to_pb()?.encode_to_vec(),
385365
));
366+
367+
old_dependent.seq
386368
}
387-
}
369+
};
370+
check_ops.push(TxnCondition::eq_seq(after_key, seq));
388371

389372
for after in remove_afters {
390373
let before_dependent = TaskDependentKey::new(DependentType::Before, after.clone());
391374
let before_dependent_ident =
392375
TaskDependentIdent::new_generic(&self.tenant, before_dependent);
393376
let before_key = before_dependent_ident.to_string_key();
394377

395-
match self.kv_api.get_pb(&before_dependent_ident).await? {
396-
None => {
397-
check_ops.push(TxnCondition {
398-
key: before_key.clone(),
399-
expected: ConditionResult::Eq as i32,
400-
target: Some(Target::Seq(0)),
401-
});
402-
}
378+
let seq = match self.kv_api.get_pb(&before_dependent_ident).await? {
379+
None => 0,
403380
Some(mut old_dependent) => {
404-
check_ops.push(TxnCondition::eq_seq(before_key.clone(), old_dependent.seq));
405-
406381
old_dependent.0.remove(task_name);
407382

408383
update_ops.push(TxnOp::put(
409-
before_key,
384+
before_key.clone(),
410385
old_dependent.to_pb()?.encode_to_vec(),
411386
));
387+
old_dependent.seq
412388
}
413-
}
389+
};
390+
check_ops.push(TxnCondition::eq_seq(before_key, seq));
414391
}
415392
let request = TxnRequest::new(check_ops, update_ops);
416393
let reply = self.kv_api.transaction(request).await?;
@@ -616,36 +593,29 @@ impl TaskMgr {
616593
let target_key = TaskDependentKey::new(other_dependent, dependent_target.clone());
617594
let target_ident =
618595
TaskDependentIdent::new_generic(&self.tenant, target_key.clone());
596+
let target_key = target_ident.to_string_key();
619597

620-
match self.kv_api.get_pb(&target_ident).await? {
598+
let seq = match self.kv_api.get_pb(&target_ident).await? {
621599
None => {
622-
check_ops.push(TxnCondition {
623-
key: target_ident.to_string_key(),
624-
expected: ConditionResult::Eq as i32,
625-
target: Some(Target::Seq(0)),
626-
});
627-
628600
update_ops.push(TxnOp::put(
629-
target_ident.to_string_key(),
601+
target_key.clone(),
630602
TaskDependentValue(BTreeSet::from_iter([task_name.to_string()]))
631603
.to_pb()?
632604
.encode_to_vec(),
633605
));
606+
0
634607
}
635608
Some(mut old_dependent) => {
636-
check_ops.push(TxnCondition::eq_seq(
637-
target_ident.to_string_key(),
638-
old_dependent.seq,
639-
));
640-
641609
old_dependent.0.remove(task_name);
642610

643611
update_ops.push(TxnOp::put(
644-
target_ident.to_string_key(),
612+
target_key.clone(),
645613
old_dependent.to_pb()?.encode_to_vec(),
646614
));
615+
old_dependent.seq
647616
}
648-
}
617+
};
618+
check_ops.push(TxnCondition::eq_seq(target_key, seq));
649619
}
650620
}
651621
Ok(())

0 commit comments

Comments
 (0)