1use std::collections::{BTreeMap, HashMap, HashSet};
16
17use common_time::util::current_time_millis;
18use derive_builder::Builder;
19use serde::ser::SerializeSeq;
20use serde::{Deserialize, Deserializer, Serialize, Serializer};
21use store_api::region_engine::RegionRole;
22use store_api::storage::{RegionId, RegionNumber};
23use strum::AsRefStr;
24
25use crate::DatanodeId;
26use crate::key::RegionDistribution;
27use crate::peer::Peer;
28
29pub fn region_distribution(region_routes: &[RegionRoute]) -> RegionDistribution {
34 let mut regions_id_map = RegionDistribution::new();
35 for route in region_routes.iter() {
36 if let Some(peer) = route.leader_peer.as_ref() {
37 let region_number = route.region.id.region_number();
38 regions_id_map
39 .entry(peer.id)
40 .or_default()
41 .add_leader_region(region_number);
42 }
43 for peer in route.follower_peers.iter() {
44 let region_number = route.region.id.region_number();
45 regions_id_map
46 .entry(peer.id)
47 .or_default()
48 .add_follower_region(region_number);
49 }
50 }
51 for region_role_set in regions_id_map.values_mut() {
52 region_role_set.sort()
54 }
55 regions_id_map
56}
57
58pub fn find_leaders(region_routes: &[RegionRoute]) -> HashSet<Peer> {
60 region_routes
61 .iter()
62 .flat_map(|x| &x.leader_peer)
63 .cloned()
64 .collect()
65}
66
67pub fn find_followers(region_routes: &[RegionRoute]) -> HashSet<Peer> {
69 region_routes
70 .iter()
71 .flat_map(|x| &x.follower_peers)
72 .cloned()
73 .collect()
74}
75
76pub fn operating_leader_regions(region_routes: &[RegionRoute]) -> Vec<(RegionId, DatanodeId)> {
78 region_routes
79 .iter()
80 .filter_map(|route| {
81 route
82 .leader_peer
83 .as_ref()
84 .map(|leader| (route.region.id, leader.id))
85 })
86 .collect::<Vec<_>>()
87}
88
89pub fn operating_leader_region_roles(
91 region_routes: &[RegionRoute],
92) -> Vec<(RegionId, DatanodeId, RegionRole)> {
93 region_routes
94 .iter()
95 .filter_map(|route| {
96 let role = route.leader_region_role()?;
97 let leader = route.leader_peer.as_ref()?;
98 Some((route.region.id, leader.id, role))
99 })
100 .collect()
101}
102
103pub fn convert_to_region_leader_map(region_routes: &[RegionRoute]) -> HashMap<RegionNumber, &Peer> {
107 region_routes
108 .iter()
109 .filter_map(|x| {
110 x.leader_peer
111 .as_ref()
112 .map(|leader| (x.region.id.region_number(), leader))
113 })
114 .collect::<HashMap<_, _>>()
115}
116
117pub fn find_region_leader(
119 region_routes: &[RegionRoute],
120 region_number: RegionNumber,
121) -> Option<Peer> {
122 find_region_leader_ref(region_routes, region_number).cloned()
123}
124
125pub fn find_region_leader_ref(
129 region_routes: &[RegionRoute],
130 region_number: RegionNumber,
131) -> Option<&Peer> {
132 region_routes
133 .iter()
134 .find(|x| x.region.id.region_number() == region_number)
135 .and_then(|r| r.leader_peer.as_ref())
136}
137
138pub fn find_leader_regions(region_routes: &[RegionRoute], datanode: &Peer) -> Vec<RegionNumber> {
140 region_routes
141 .iter()
142 .filter_map(|x| {
143 if let Some(peer) = &x.leader_peer
144 && peer == datanode
145 {
146 return Some(x.region.id.region_number());
147 }
148 None
149 })
150 .collect()
151}
152
153pub fn find_follower_regions(region_routes: &[RegionRoute], datanode: &Peer) -> Vec<RegionNumber> {
155 region_routes
156 .iter()
157 .filter_map(|x| {
158 if x.follower_peers.contains(datanode) {
159 return Some(x.region.id.region_number());
160 }
161 None
162 })
163 .collect()
164}
165
166#[derive(Debug, Clone, Default, Deserialize, Serialize, PartialEq, Builder)]
167pub struct RegionRoute {
168 pub region: Region,
169 #[builder(setter(into, strip_option))]
170 pub leader_peer: Option<Peer>,
171 #[builder(setter(into), default)]
172 pub follower_peers: Vec<Peer>,
173 #[builder(setter(into, strip_option), default)]
175 #[serde(
176 default,
177 alias = "leader_status",
178 skip_serializing_if = "Option::is_none"
179 )]
180 pub leader_state: Option<LeaderState>,
181 #[serde(default)]
183 #[builder(default = "self.default_leader_down_since()")]
184 pub leader_down_since: Option<i64>,
185 #[builder(setter(into, strip_option), default)]
187 #[serde(default, skip_serializing_if = "Option::is_none")]
188 pub write_route_policy: Option<WriteRoutePolicy>,
189}
190
191impl RegionRouteBuilder {
192 fn default_leader_down_since(&self) -> Option<i64> {
193 match self.leader_state {
194 Some(Some(LeaderState::Downgrading)) => Some(current_time_millis()),
195 _ => None,
196 }
197 }
198}
199
200#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, AsRefStr)]
203#[strum(serialize_all = "UPPERCASE")]
204pub enum LeaderState {
205 #[serde(alias = "Downgraded")]
210 Downgrading,
211 Staging,
216}
217
218#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq)]
220pub enum WriteRoutePolicy {
221 Normal,
223 IgnoreAllWrites,
229}
230
231impl RegionRoute {
232 pub fn is_ignore_all_writes(&self) -> bool {
234 matches!(
235 self.write_route_policy,
236 Some(WriteRoutePolicy::IgnoreAllWrites)
237 )
238 }
239
240 pub fn set_ignore_all_writes(&mut self) {
242 self.write_route_policy = Some(WriteRoutePolicy::IgnoreAllWrites);
243 }
244
245 pub fn clear_ignore_all_writes(&mut self) {
247 if self.write_route_policy == Some(WriteRoutePolicy::IgnoreAllWrites) {
248 self.write_route_policy = None;
249 }
250 }
251
252 pub fn is_leader_downgrading(&self) -> bool {
260 matches!(self.leader_state, Some(LeaderState::Downgrading))
261 }
262
263 pub fn is_leader_staging(&self) -> bool {
265 matches!(self.leader_state, Some(LeaderState::Staging))
266 }
267
268 pub fn leader_region_role(&self) -> Option<RegionRole> {
270 self.leader_peer.as_ref().map(|_| {
271 if self.is_leader_staging() {
272 RegionRole::StagingLeader
273 } else if self.is_leader_downgrading() {
274 RegionRole::DowngradingLeader
275 } else {
276 RegionRole::Leader
277 }
278 })
279 }
280
281 pub fn downgrade_leader(&mut self) {
293 self.leader_down_since = Some(current_time_millis());
294 self.leader_state = Some(LeaderState::Downgrading)
295 }
296
297 pub fn set_leader_staging(&mut self) {
299 self.leader_state = Some(LeaderState::Staging);
300 self.leader_down_since = None;
302 }
303
304 pub fn clear_leader_staging(&mut self) {
306 if self.leader_state == Some(LeaderState::Staging) {
307 self.leader_state = None;
308 self.leader_down_since = None;
309 }
310 }
311
312 pub fn leader_down_millis(&self) -> Option<i64> {
314 self.leader_down_since
315 .map(|start| current_time_millis() - start)
316 }
317
318 pub fn set_leader_state(&mut self, state: Option<LeaderState>) -> bool {
322 let updated = self.leader_state != state;
323
324 match (state, updated) {
325 (Some(LeaderState::Downgrading), true) => {
326 self.leader_down_since = Some(current_time_millis());
327 }
328 (Some(LeaderState::Downgrading), false) => {
329 }
331 _ => {
332 self.leader_down_since = None;
333 }
334 }
335
336 self.leader_state = state;
337 updated
338 }
339}
340
341#[derive(Debug, Clone, Default, PartialEq, Serialize)]
342pub struct Region {
343 pub id: RegionId,
344 pub name: String,
345 pub attrs: BTreeMap<String, String>,
346 pub partition_expr: String,
348}
349
350#[derive(Debug, Deserialize)]
351struct RegionDe {
352 id: RegionId,
353 name: String,
354 #[serde(default)]
355 attrs: BTreeMap<String, String>,
356 #[serde(default)]
357 partition: Option<LegacyPartition>,
358 #[serde(default)]
359 partition_expr: String,
360}
361
362impl<'de> Deserialize<'de> for Region {
363 fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
364 where
365 D: Deserializer<'de>,
366 {
367 let de = RegionDe::deserialize(deserializer)?;
368 let partition_expr = if de.partition_expr.is_empty() {
371 if let Some(LegacyPartition { value_list, .. }) = &de.partition {
372 value_list
373 .first()
374 .map(|expr| String::from_utf8_lossy(expr).to_string())
375 .unwrap_or_default()
376 } else {
377 String::new()
378 }
379 } else {
380 de.partition_expr
381 };
382
383 Ok(Self {
384 id: de.id,
385 name: de.name,
386 attrs: de.attrs,
387 partition_expr,
388 })
389 }
390}
391
392impl Region {
393 #[cfg(any(test, feature = "testing"))]
394 pub fn new_test(id: RegionId) -> Self {
395 Self {
396 id,
397 ..Default::default()
398 }
399 }
400
401 pub fn partition_expr(&self) -> String {
403 self.partition_expr.clone()
404 }
405}
406
407#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
408pub struct LegacyPartition {
409 #[serde(serialize_with = "as_utf8_vec", deserialize_with = "from_utf8_vec")]
410 pub column_list: Vec<Vec<u8>>,
411 #[serde(serialize_with = "as_utf8_vec", deserialize_with = "from_utf8_vec")]
412 pub value_list: Vec<Vec<u8>>,
413}
414
415fn as_utf8_vec<S: Serializer>(
416 val: &[Vec<u8>],
417 serializer: S,
418) -> std::result::Result<S::Ok, S::Error> {
419 let mut seq = serializer.serialize_seq(Some(val.len()))?;
420 for v in val {
421 seq.serialize_element(&String::from_utf8_lossy(v))?;
422 }
423 seq.end()
424}
425
426pub fn from_utf8_vec<'de, D>(deserializer: D) -> std::result::Result<Vec<Vec<u8>>, D::Error>
427where
428 D: Deserializer<'de>,
429{
430 let values = Vec::<String>::deserialize(deserializer)?;
431
432 let values = values
433 .into_iter()
434 .map(|value| value.into_bytes())
435 .collect::<Vec<_>>();
436 Ok(values)
437}
438
439#[cfg(test)]
440mod tests {
441 use super::*;
442 use crate::key::RegionRoleSet;
443
444 fn new_test_region_route(region_id: RegionId) -> RegionRoute {
445 RegionRoute {
446 region: Region::new_test(region_id),
447 leader_peer: Some(Peer::new(1, "a1")),
448 follower_peers: vec![Peer::new(2, "a2")],
449 leader_state: None,
450 leader_down_since: None,
451 write_route_policy: None,
452 }
453 }
454
455 #[test]
456 fn test_find_region_leader() {
457 let mut routes = vec![
458 new_test_region_route(RegionId::new(1, 1)),
459 new_test_region_route(RegionId::new(1, 1)),
460 ];
461 routes[1].leader_peer = Some(Peer::new(2, "a2"));
462
463 assert_eq!(
464 find_region_leader_ref(&routes, 1),
465 routes[0].leader_peer.as_ref()
466 );
467 assert_eq!(find_region_leader(&routes, 1), routes[0].leader_peer);
468 assert_eq!(find_region_leader_ref(&routes, 2), None);
469 assert_eq!(find_region_leader(&routes, 2), None);
470
471 routes[0].leader_peer = None;
472 assert_eq!(find_region_leader_ref(&routes, 1), None);
473 assert_eq!(find_region_leader(&routes, 1), None);
474 }
475
476 #[test]
477 fn test_leader_is_downgraded() {
478 let mut region_route = RegionRoute {
479 region: Region {
480 id: 2.into(),
481 name: "r2".to_string(),
482 attrs: BTreeMap::new(),
483 partition_expr: "".to_string(),
484 },
485 leader_peer: Some(Peer::new(1, "a1")),
486 follower_peers: vec![Peer::new(2, "a2"), Peer::new(3, "a3")],
487 leader_state: None,
488 leader_down_since: None,
489 write_route_policy: None,
490 };
491
492 assert!(!region_route.is_leader_downgrading());
493
494 region_route.downgrade_leader();
495
496 assert!(region_route.is_leader_downgrading());
497 }
498
499 #[test]
500 fn test_region_route_decode() {
501 let region_route = RegionRoute {
502 region: Region {
503 id: 2.into(),
504 name: "r2".to_string(),
505 attrs: BTreeMap::new(),
506 partition_expr: "".to_string(),
507 },
508 leader_peer: Some(Peer::new(1, "a1")),
509 follower_peers: vec![Peer::new(2, "a2"), Peer::new(3, "a3")],
510 leader_state: None,
511 leader_down_since: None,
512 write_route_policy: None,
513 };
514
515 let input = r#"{"region":{"id":2,"name":"r2","partition":null,"attrs":{}},"leader_peer":{"id":1,"addr":"a1"},"follower_peers":[{"id":2,"addr":"a2"},{"id":3,"addr":"a3"}]}"#;
516
517 let decoded: RegionRoute = serde_json::from_str(input).unwrap();
518
519 assert_eq!(decoded, region_route);
520 }
521
522 #[test]
523 fn test_region_route_compatibility() {
524 let region_route = RegionRoute {
525 region: Region {
526 id: 2.into(),
527 name: "r2".to_string(),
528 attrs: BTreeMap::new(),
529 partition_expr: "".to_string(),
530 },
531 leader_peer: Some(Peer::new(1, "a1")),
532 follower_peers: vec![Peer::new(2, "a2"), Peer::new(3, "a3")],
533 leader_state: Some(LeaderState::Downgrading),
534 leader_down_since: None,
535 write_route_policy: None,
536 };
537 let input = r#"{"region":{"id":2,"name":"r2","partition":null,"attrs":{}},"leader_peer":{"id":1,"addr":"a1"},"follower_peers":[{"id":2,"addr":"a2"},{"id":3,"addr":"a3"}],"leader_state":"Downgraded","leader_down_since":null}"#;
538 let decoded: RegionRoute = serde_json::from_str(input).unwrap();
539 assert_eq!(decoded, region_route);
540
541 let region_route = RegionRoute {
542 region: Region {
543 id: 2.into(),
544 name: "r2".to_string(),
545 attrs: BTreeMap::new(),
546 partition_expr: "".to_string(),
547 },
548 leader_peer: Some(Peer::new(1, "a1")),
549 follower_peers: vec![Peer::new(2, "a2"), Peer::new(3, "a3")],
550 leader_state: Some(LeaderState::Downgrading),
551 leader_down_since: None,
552 write_route_policy: None,
553 };
554 let input = r#"{"region":{"id":2,"name":"r2","partition":null,"attrs":{}},"leader_peer":{"id":1,"addr":"a1"},"follower_peers":[{"id":2,"addr":"a2"},{"id":3,"addr":"a3"}],"leader_status":"Downgraded","leader_down_since":null}"#;
555 let decoded: RegionRoute = serde_json::from_str(input).unwrap();
556 assert_eq!(decoded, region_route);
557
558 let region_route = RegionRoute {
559 region: Region {
560 id: 2.into(),
561 name: "r2".to_string(),
562 attrs: BTreeMap::new(),
563 partition_expr: "".to_string(),
564 },
565 leader_peer: Some(Peer::new(1, "a1")),
566 follower_peers: vec![Peer::new(2, "a2"), Peer::new(3, "a3")],
567 leader_state: Some(LeaderState::Downgrading),
568 leader_down_since: None,
569 write_route_policy: None,
570 };
571 let input = r#"{"region":{"id":2,"name":"r2","partition":null,"attrs":{}},"leader_peer":{"id":1,"addr":"a1"},"follower_peers":[{"id":2,"addr":"a2"},{"id":3,"addr":"a3"}],"leader_state":"Downgrading","leader_down_since":null}"#;
572 let decoded: RegionRoute = serde_json::from_str(input).unwrap();
573 assert_eq!(decoded, region_route);
574
575 let region_route = RegionRoute {
576 region: Region {
577 id: 2.into(),
578 name: "r2".to_string(),
579 attrs: BTreeMap::new(),
580 partition_expr: "".to_string(),
581 },
582 leader_peer: Some(Peer::new(1, "a1")),
583 follower_peers: vec![Peer::new(2, "a2"), Peer::new(3, "a3")],
584 leader_state: Some(LeaderState::Downgrading),
585 leader_down_since: None,
586 write_route_policy: None,
587 };
588 let input = r#"{"region":{"id":2,"name":"r2","partition":null,"attrs":{}},"leader_peer":{"id":1,"addr":"a1"},"follower_peers":[{"id":2,"addr":"a2"},{"id":3,"addr":"a3"}],"leader_status":"Downgrading","leader_down_since":null}"#;
589 let decoded: RegionRoute = serde_json::from_str(input).unwrap();
590 assert_eq!(decoded, region_route);
591 }
592
593 #[test]
594 fn test_region_route_write_route_policy_decode_compatibility() {
595 let input = r#"{"region":{"id":2,"name":"r2","partition":null,"attrs":{}},"leader_peer":{"id":1,"addr":"a1"},"follower_peers":[{"id":2,"addr":"a2"}],"write_route_policy":"IgnoreAllWrites"}"#;
596 let decoded: RegionRoute = serde_json::from_str(input).unwrap();
597
598 assert!(decoded.is_ignore_all_writes());
599 }
600
601 #[test]
602 fn test_region_route_write_route_policy_default_not_serialized() {
603 let region_route = RegionRoute {
604 region: Region {
605 id: 2.into(),
606 name: "r2".to_string(),
607 attrs: BTreeMap::new(),
608 partition_expr: "".to_string(),
609 },
610 leader_peer: Some(Peer::new(1, "a1")),
611 follower_peers: vec![Peer::new(2, "a2"), Peer::new(3, "a3")],
612 leader_state: None,
613 leader_down_since: None,
614 write_route_policy: None,
615 };
616
617 let encoded = serde_json::to_string(®ion_route).unwrap();
618 assert!(!encoded.contains("write_route_policy"));
619 }
620
621 #[test]
622 fn test_region_route_write_route_policy_helpers() {
623 let mut region_route = RegionRoute {
624 region: Region::new_test(2.into()),
625 leader_peer: Some(Peer::new(1, "a1")),
626 follower_peers: vec![],
627 leader_state: None,
628 leader_down_since: None,
629 write_route_policy: None,
630 };
631
632 assert!(!region_route.is_ignore_all_writes());
633 region_route.set_ignore_all_writes();
634 assert!(region_route.is_ignore_all_writes());
635 region_route.clear_ignore_all_writes();
636 assert!(!region_route.is_ignore_all_writes());
637 }
638
639 #[test]
640 fn test_leader_region_role_without_leader_peer_returns_none() {
641 let region_route = RegionRoute {
642 leader_peer: None,
643 ..new_test_region_route(RegionId::new(1, 1))
644 };
645
646 assert_eq!(region_route.leader_region_role(), None);
647 }
648
649 #[test]
650 fn test_leader_region_role_variants() {
651 let normal = new_test_region_route(RegionId::new(1, 1));
652 let mut downgrading = new_test_region_route(RegionId::new(1, 2));
653 downgrading.leader_state = Some(LeaderState::Downgrading);
654 let mut staging = new_test_region_route(RegionId::new(1, 3));
655 staging.leader_state = Some(LeaderState::Staging);
656
657 assert_eq!(normal.leader_region_role(), Some(RegionRole::Leader));
658 assert_eq!(
659 downgrading.leader_region_role(),
660 Some(RegionRole::DowngradingLeader)
661 );
662 assert_eq!(
663 staging.leader_region_role(),
664 Some(RegionRole::StagingLeader)
665 );
666 }
667
668 #[test]
669 fn test_operating_leader_region_roles_returns_expected_roles() {
670 let no_leader_region = RegionRoute {
671 leader_peer: None,
672 ..new_test_region_route(RegionId::new(1, 4))
673 };
674 let mut downgrading = new_test_region_route(RegionId::new(1, 2));
675 downgrading.leader_peer = Some(Peer::new(2, "a2"));
676 downgrading.leader_state = Some(LeaderState::Downgrading);
677 let mut staging = new_test_region_route(RegionId::new(1, 3));
678 staging.leader_peer = Some(Peer::new(3, "a3"));
679 staging.leader_state = Some(LeaderState::Staging);
680
681 let roles = operating_leader_region_roles(&[
682 new_test_region_route(RegionId::new(1, 1)),
683 downgrading,
684 staging,
685 no_leader_region,
686 ]);
687
688 assert_eq!(
689 roles,
690 vec![
691 (RegionId::new(1, 1), 1, RegionRole::Leader),
692 (RegionId::new(1, 2), 2, RegionRole::DowngradingLeader),
693 (RegionId::new(1, 3), 3, RegionRole::StagingLeader),
694 ]
695 );
696 }
697
698 #[test]
699 fn test_region_distribution() {
700 let region_routes = vec![
701 RegionRoute {
702 region: Region {
703 id: RegionId::new(1, 1),
704 name: "r1".to_string(),
705 attrs: BTreeMap::new(),
706 partition_expr: "".to_string(),
707 },
708 leader_peer: Some(Peer::new(1, "a1")),
709 follower_peers: vec![Peer::new(2, "a2"), Peer::new(3, "a3")],
710 leader_state: None,
711 leader_down_since: None,
712 write_route_policy: None,
713 },
714 RegionRoute {
715 region: Region {
716 id: RegionId::new(1, 2),
717 name: "r2".to_string(),
718 attrs: BTreeMap::new(),
719 partition_expr: "".to_string(),
720 },
721 leader_peer: Some(Peer::new(2, "a2")),
722 follower_peers: vec![Peer::new(1, "a1"), Peer::new(3, "a3")],
723 leader_state: None,
724 leader_down_since: None,
725 write_route_policy: None,
726 },
727 ];
728
729 let distribution = region_distribution(®ion_routes);
730 assert_eq!(distribution.len(), 3);
731 assert_eq!(distribution[&1], RegionRoleSet::new(vec![1], vec![2]));
732 assert_eq!(distribution[&2], RegionRoleSet::new(vec![2], vec![1]));
733 assert_eq!(distribution[&3], RegionRoleSet::new(vec![], vec![1, 2]));
734 }
735
736 #[test]
737 fn test_de_serialize_partition() {
738 let p = LegacyPartition {
739 column_list: vec![b"a".to_vec(), b"b".to_vec()],
740 value_list: vec![b"hi".to_vec(), b",".to_vec()],
741 };
742
743 let output = serde_json::to_string(&p).unwrap();
744 let got: LegacyPartition = serde_json::from_str(&output).unwrap();
745
746 assert_eq!(got, p);
747 }
748}