1use std::collections::{BTreeMap, HashMap};
52use std::fmt;
53use std::fs;
54use std::ops;
55use std::path::{Path, PathBuf};
56use std::sync::{Mutex, OnceLock};
57
58use mongreldb_types::errors::ErrorCategory;
59use mongreldb_types::ids::{NodeId, RaftGroupId, TableId, TabletId};
60use serde::{Deserialize, Serialize};
61use sha2::{Digest, Sha256};
62
63use crate::node::ClusterError;
64
65#[derive(
75 Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize,
76)]
77#[repr(transparent)]
78pub struct ColumnId(pub u16);
79
80impl ColumnId {
81 pub const fn new(id: u16) -> Self {
83 Self(id)
84 }
85
86 pub const fn get(self) -> u16 {
88 self.0
89 }
90}
91
92impl fmt::Display for ColumnId {
93 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
94 self.0.fmt(f)
95 }
96}
97
98#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
109pub struct Key(Vec<u8>);
110
111impl Key {
112 pub const fn from_bytes(bytes: Vec<u8>) -> Self {
114 Self(bytes)
115 }
116
117 pub fn as_bytes(&self) -> &[u8] {
119 &self.0
120 }
121
122 pub fn into_bytes(self) -> Vec<u8> {
124 self.0
125 }
126}
127
128impl AsRef<[u8]> for Key {
129 fn as_ref(&self) -> &[u8] {
130 self.as_bytes()
131 }
132}
133
134impl fmt::Display for Key {
135 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
137 f.write_str(&hex_encode(&self.0))
138 }
139}
140
141impl Serialize for Key {
142 fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
145 if serializer.is_human_readable() {
146 serializer.serialize_str(&hex_encode(&self.0))
147 } else {
148 serializer.serialize_bytes(&self.0)
149 }
150 }
151}
152
153impl<'de> Deserialize<'de> for Key {
154 fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
155 struct KeyVisitor;
156
157 impl<'v> serde::de::Visitor<'v> for KeyVisitor {
158 type Value = Key;
159
160 fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
161 f.write_str("a hex string or byte sequence")
162 }
163
164 fn visit_str<E: serde::de::Error>(self, text: &str) -> Result<Key, E> {
165 hex_decode(text)
166 .map(Key)
167 .map_err(|detail| E::custom(format!("invalid key hex: {detail}")))
168 }
169
170 fn visit_bytes<E: serde::de::Error>(self, bytes: &[u8]) -> Result<Key, E> {
171 Ok(Key(bytes.to_vec()))
172 }
173
174 fn visit_byte_buf<E: serde::de::Error>(self, bytes: Vec<u8>) -> Result<Key, E> {
175 Ok(Key(bytes))
176 }
177
178 fn visit_seq<A: serde::de::SeqAccess<'v>>(self, mut seq: A) -> Result<Key, A::Error> {
179 let mut bytes = Vec::with_capacity(seq.size_hint().unwrap_or(0));
180 while let Some(byte) = seq.next_element::<u8>()? {
181 bytes.push(byte);
182 }
183 Ok(Key(bytes))
184 }
185 }
186
187 if deserializer.is_human_readable() {
188 deserializer.deserialize_str(KeyVisitor)
189 } else {
190 deserializer.deserialize_bytes(KeyVisitor)
191 }
192 }
193}
194
195fn hex_encode(bytes: &[u8]) -> String {
196 const HEX: &[u8; 16] = b"0123456789abcdef";
197 let mut out = String::with_capacity(bytes.len() * 2);
198 for byte in bytes {
199 out.push(HEX[(byte >> 4) as usize] as char);
200 out.push(HEX[(byte & 0x0f) as usize] as char);
201 }
202 out
203}
204
205fn hex_decode(text: &str) -> Result<Vec<u8>, String> {
206 if !text.len().is_multiple_of(2) {
207 return Err("odd number of hex digits".to_owned());
208 }
209 let bytes = text.as_bytes();
210 let mut out = Vec::with_capacity(bytes.len() / 2);
211 for pair in bytes.chunks_exact(2) {
212 let digit = |byte: u8| -> Result<u8, String> {
213 (byte as char)
214 .to_digit(16)
215 .map(|value| value as u8)
216 .ok_or_else(|| format!("invalid hex character `{}`", byte as char))
217 };
218 out.push((digit(pair[0])? << 4) | digit(pair[1])?);
219 }
220 Ok(out)
221}
222
223#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
234pub enum Bound<T> {
235 Unbounded,
237 Included(T),
239 Excluded(T),
241}
242
243impl<T: Clone> Bound<T> {
244 pub fn from_std(bound: ops::Bound<T>) -> Self {
246 match bound {
247 ops::Bound::Unbounded => Self::Unbounded,
248 ops::Bound::Included(value) => Self::Included(value),
249 ops::Bound::Excluded(value) => Self::Excluded(value),
250 }
251 }
252}
253
254#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
262#[serde(deny_unknown_fields)]
263pub struct PartitionBounds {
264 pub low: Bound<Key>,
266 pub high: Bound<Key>,
268}
269
270impl PartitionBounds {
271 pub fn new(low: Bound<Key>, high: Bound<Key>) -> Result<Self, TabletError> {
273 let bounds = Self { low, high };
274 bounds.validate()?;
275 Ok(bounds)
276 }
277
278 pub fn unbounded() -> Self {
280 Self {
281 low: Bound::Unbounded,
282 high: Bound::Unbounded,
283 }
284 }
285
286 pub fn validate(&self) -> Result<(), TabletError> {
289 if let (Some(low), Some(high)) = (bound_key(&self.low), bound_key(&self.high)) {
290 if low > high {
291 return Err(TabletError::InvalidBounds(format!(
292 "low endpoint {low} is above high endpoint {high}"
293 )));
294 }
295 let single_point =
296 matches!(self.low, Bound::Included(_)) && matches!(self.high, Bound::Included(_));
297 if low == high && !single_point {
298 return Err(TabletError::InvalidBounds(format!(
299 "empty range at endpoint {low}"
300 )));
301 }
302 }
303 Ok(())
304 }
305
306 pub fn contains(&self, key: &Key) -> bool {
308 let above_low = match &self.low {
309 Bound::Unbounded => true,
310 Bound::Included(low) => key >= low,
311 Bound::Excluded(low) => key > low,
312 };
313 let below_high = match &self.high {
314 Bound::Unbounded => true,
315 Bound::Included(high) => key <= high,
316 Bound::Excluded(high) => key < high,
317 };
318 above_low && below_high
319 }
320
321 pub fn overlaps(&self, other: &Self) -> bool {
323 !ends_before(&self.high, &other.low) && !ends_before(&other.high, &self.low)
324 }
325
326 pub fn is_adjacent_to(&self, other: &Self) -> bool {
331 meets(&self.high, &other.low) || meets(&other.high, &self.low)
332 }
333
334 pub fn meets_start_of(&self, other: &Self) -> bool {
337 meets(&self.high, &other.low)
338 }
339
340 pub fn split_at(&self, key: &Key) -> Option<(Self, Self)> {
349 if !self.contains(key) {
350 return None;
351 }
352 let lower = Self::new(self.low.clone(), Bound::Excluded(key.clone())).ok()?;
353 let upper = Self::new(Bound::Included(key.clone()), self.high.clone()).ok()?;
354 Some((lower, upper))
355 }
356
357 pub fn union_adjacent(&self, other: &Self) -> Option<Self> {
362 let (lower, upper) = if self.meets_start_of(other) {
363 (self, other)
364 } else if other.meets_start_of(self) {
365 (other, self)
366 } else {
367 return None;
368 };
369 Self::new(lower.low.clone(), upper.high.clone()).ok()
370 }
371}
372
373fn bound_key(bound: &Bound<Key>) -> Option<&Key> {
375 match bound {
376 Bound::Unbounded => None,
377 Bound::Included(key) | Bound::Excluded(key) => Some(key),
378 }
379}
380
381fn ends_before(high: &Bound<Key>, low: &Bound<Key>) -> bool {
385 let (Some(high_key), Some(low_key)) = (bound_key(high), bound_key(low)) else {
386 return false;
387 };
388 if high_key != low_key {
389 return high_key < low_key;
390 }
391 !(matches!(high, Bound::Included(_)) && matches!(low, Bound::Included(_)))
393}
394
395fn meets(high: &Bound<Key>, low: &Bound<Key>) -> bool {
397 match (high, low) {
398 (Bound::Included(high), Bound::Excluded(low))
399 | (Bound::Excluded(high), Bound::Included(low)) => high == low,
400 _ => false,
401 }
402}
403
404#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
411pub enum TabletState {
412 Creating,
415 Active,
417 Splitting,
420 Merging,
423 Retiring,
426 Retired,
429}
430
431impl TabletState {
432 pub fn can_transition_to(self, next: Self) -> bool {
445 use TabletState::{Active, Creating, Merging, Retired, Retiring, Splitting};
446 matches!(
447 (self, next),
448 (Creating, Active)
449 | (Creating, Retired)
450 | (Active, Splitting)
451 | (Active, Merging)
452 | (Active, Retiring)
453 | (Splitting, Active)
454 | (Splitting, Retiring)
455 | (Merging, Active)
456 | (Merging, Retiring)
457 | (Retiring, Retired)
458 )
459 }
460
461 pub fn is_routable(self) -> bool {
467 matches!(self, Self::Active | Self::Splitting | Self::Merging)
468 }
469}
470
471impl fmt::Display for TabletState {
472 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
473 let name = match self {
474 Self::Creating => "Creating",
475 Self::Active => "Active",
476 Self::Splitting => "Splitting",
477 Self::Merging => "Merging",
478 Self::Retiring => "Retiring",
479 Self::Retired => "Retired",
480 };
481 f.write_str(name)
482 }
483}
484
485#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
489pub enum ReplicaRole {
490 Voter,
492 Learner,
495 ReadReplica,
497 AiReadReplica,
499 AnalyticsReplica,
501 BackupReplica,
503}
504
505impl ReplicaRole {
506 pub const fn counts_toward_quorum(self) -> bool {
508 matches!(self, Self::Voter)
509 }
510
511 pub const fn applies_log(self) -> bool {
513 true
514 }
515
516 pub const fn serves_reads(self) -> bool {
518 !matches!(self, Self::BackupReplica)
519 }
520}
521
522impl fmt::Display for ReplicaRole {
523 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
524 let name = match self {
525 Self::Voter => "Voter",
526 Self::Learner => "Learner",
527 Self::ReadReplica => "ReadReplica",
528 Self::AiReadReplica => "AiReadReplica",
529 Self::AnalyticsReplica => "AnalyticsReplica",
530 Self::BackupReplica => "BackupReplica",
531 };
532 f.write_str(name)
533 }
534}
535
536#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
545#[serde(deny_unknown_fields)]
546pub struct ReplicaDescriptor {
547 pub node_id: NodeId,
549 pub role: ReplicaRole,
551 pub raft_node_id: u64,
553}
554
555#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
581#[serde(deny_unknown_fields)]
582pub struct TabletDescriptor {
583 pub tablet_id: TabletId,
585 pub table_id: TableId,
587 #[serde(default = "crate::tablet::zero_database_id")]
592 pub database_id: mongreldb_types::ids::DatabaseId,
593 pub raft_group_id: RaftGroupId,
595 pub partition: PartitionBounds,
597 pub replicas: Vec<ReplicaDescriptor>,
599 pub leader_hint: Option<NodeId>,
601 pub generation: u64,
603 pub state: TabletState,
605}
606
607fn zero_database_id() -> mongreldb_types::ids::DatabaseId {
608 mongreldb_types::ids::DatabaseId::ZERO
609}
610
611impl TabletDescriptor {
612 pub fn validate(&self) -> Result<(), TabletError> {
616 if self.tablet_id == TabletId::ZERO {
617 return Err(TabletError::InvalidDescriptor(
618 "reserved all-zero tablet id".to_owned(),
619 ));
620 }
621 if self.raft_group_id == RaftGroupId::ZERO {
622 return Err(TabletError::InvalidDescriptor(
623 "reserved all-zero raft group id".to_owned(),
624 ));
625 }
626 self.partition.validate()?;
627 if self.replicas.is_empty() {
628 return Err(TabletError::InvalidDescriptor(
629 "tablet has no replicas".to_owned(),
630 ));
631 }
632 for (index, replica) in self.replicas.iter().enumerate() {
633 if self.replicas[..index]
634 .iter()
635 .any(|prior| prior.node_id == replica.node_id)
636 {
637 return Err(TabletError::InvalidDescriptor(format!(
638 "node {} holds more than one replica",
639 replica.node_id
640 )));
641 }
642 if self.replicas[..index]
643 .iter()
644 .any(|prior| prior.raft_node_id == replica.raft_node_id)
645 {
646 return Err(TabletError::InvalidDescriptor(format!(
647 "raft node id {} is used by more than one replica",
648 replica.raft_node_id
649 )));
650 }
651 }
652 if let Some(leader) = self.leader_hint {
653 if !self
654 .replicas
655 .iter()
656 .any(|replica| replica.node_id == leader)
657 {
658 return Err(TabletError::InvalidDescriptor(format!(
659 "leader hint {leader} is not a replica of the tablet"
660 )));
661 }
662 }
663 if self.state != TabletState::Creating && self.voter_count() == 0 {
664 return Err(TabletError::InvalidDescriptor(format!(
665 "tablet in state {} has no voters",
666 self.state
667 )));
668 }
669 Ok(())
670 }
671
672 pub fn voters(&self) -> impl Iterator<Item = &ReplicaDescriptor> {
674 self.replicas
675 .iter()
676 .filter(|replica| replica.role == ReplicaRole::Voter)
677 }
678
679 pub fn learners(&self) -> impl Iterator<Item = &ReplicaDescriptor> {
681 self.replicas
682 .iter()
683 .filter(|replica| replica.role == ReplicaRole::Learner)
684 }
685
686 pub fn voter_count(&self) -> usize {
688 self.voters().count()
689 }
690
691 pub fn replica_on(&self, node: NodeId) -> Option<&ReplicaDescriptor> {
693 self.replicas.iter().find(|replica| replica.node_id == node)
694 }
695
696 pub fn try_transition(&mut self, next: TabletState) -> Result<(), TabletError> {
699 if !self.state.can_transition_to(next) {
700 return Err(TabletError::InvalidStateTransition {
701 tablet: self.tablet_id,
702 from: self.state,
703 to: next,
704 });
705 }
706 self.state = next;
707 Ok(())
708 }
709
710 pub fn published_transition(&self, next: TabletState) -> Result<Self, TabletError> {
716 let mut published = self.clone();
717 published.try_transition(next)?;
718 published.generation =
719 published
720 .generation
721 .checked_add(1)
722 .ok_or(TabletError::InvalidDescriptor(
723 "descriptor generation overflows u64".to_owned(),
724 ))?;
725 Ok(published)
726 }
727}
728
729#[derive(Debug, thiserror::Error)]
736pub enum TabletError {
737 #[error("tablet metadata operation failed: {0}")]
740 Metadata(#[from] ClusterError),
741 #[error("tablet metadata is missing at {0}")]
743 MissingMetadata(PathBuf),
744 #[error(
747 "tablet directory {path} holds metadata for tablet {found} / group {found_group}, \
748 expected tablet {expected} / group {expected_group}"
749 )]
750 TabletMismatch {
751 path: PathBuf,
753 expected: TabletId,
755 found: TabletId,
757 expected_group: RaftGroupId,
759 found_group: RaftGroupId,
761 },
762 #[error("tablet directory {0} already holds different tablet metadata")]
764 MetadataConflict(PathBuf),
765 #[error("invalid tablet state transition for tablet {tablet}: {from} -> {to}")]
767 InvalidStateTransition {
768 tablet: TabletId,
770 from: TabletState,
772 to: TabletState,
774 },
775 #[error("invalid partition bounds: {0}")]
777 InvalidBounds(String),
778 #[error("invalid tablet descriptor: {0}")]
780 InvalidDescriptor(String),
781 #[error("partitioning failed: {0}")]
783 Partition(#[from] PartitionError),
784 #[error("tablet storage core at {path} is already owned in this process by tablet {tablet}")]
787 AlreadyOwned {
788 tablet: TabletId,
790 path: PathBuf,
792 },
793}
794
795#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
797pub enum PartitionError {
798 #[error("invalid partitioning: {0}")]
800 InvalidPartitioning(String),
801 #[error("partition key column {column} is missing from the supplied values")]
803 MissingPartitionColumn {
804 column: ColumnId,
806 },
807 #[error("partition key column {column} is {found}, expected {expected}")]
809 PartitionColumnType {
810 column: ColumnId,
812 expected: &'static str,
814 found: &'static str,
816 },
817 #[error("malformed encoded key: {0}")]
819 MalformedKey(String),
820 #[error("time-range partition slot is negative for pre-epoch timestamp {micros} micros")]
822 NegativeSlot {
823 micros: i64,
825 },
826}
827
828#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
831pub enum RoutingError {
832 #[error(
834 "tablet {tablet_id} moved: request used generation {used_generation}, \
835 current generation is {current_generation}"
836 )]
837 TabletMoved {
838 tablet_id: TabletId,
840 used_generation: u64,
842 current_generation: u64,
844 },
845 #[error(
847 "tablet {tablet_id} is splitting: request used generation {used_generation}, \
848 current generation is {current_generation}"
849 )]
850 TabletSplit {
851 tablet_id: TabletId,
853 used_generation: u64,
855 current_generation: u64,
857 },
858 #[error(
860 "stale tablet metadata for tablet {tablet_id}: request used generation \
861 {used_generation}, current generation is {current_generation}"
862 )]
863 StaleMetadata {
864 tablet_id: TabletId,
866 used_generation: u64,
868 current_generation: u64,
870 },
871}
872
873impl RoutingError {
874 pub fn category(&self) -> ErrorCategory {
878 match self {
879 Self::TabletMoved { .. } => ErrorCategory::TabletMoved,
880 Self::TabletSplit { .. } => ErrorCategory::TabletSplitting,
881 Self::StaleMetadata { .. } => ErrorCategory::StaleMetadata,
882 }
883 }
884}
885
886#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
898pub enum KeyValue {
899 Null,
901 Bool(bool),
903 Int(i64),
905 TimestampMicros(i64),
907 Text(String),
909 Bytes(Vec<u8>),
911}
912
913impl KeyValue {
914 pub fn type_name(&self) -> &'static str {
916 match self {
917 Self::Null => "null",
918 Self::Bool(_) => "bool",
919 Self::Int(_) => "int",
920 Self::TimestampMicros(_) => "timestamp-micros",
921 Self::Text(_) => "text",
922 Self::Bytes(_) => "bytes",
923 }
924 }
925}
926
927const TAG_NULL: u8 = 0x01;
930const TAG_BOOL: u8 = 0x02;
931const TAG_INT: u8 = 0x03;
932const TAG_TIMESTAMP_MICROS: u8 = 0x04;
933const TAG_TEXT: u8 = 0x05;
934const TAG_BYTES: u8 = 0x06;
935
936pub struct RowKeyEncoder;
951
952impl RowKeyEncoder {
953 pub fn encode_key(values: &[KeyValue]) -> Key {
955 let mut out = Vec::new();
956 for value in values {
957 Self::encode_component(&mut out, value);
958 }
959 Key::from_bytes(out)
960 }
961
962 pub fn encode_component(out: &mut Vec<u8>, value: &KeyValue) {
964 match value {
965 KeyValue::Null => out.push(TAG_NULL),
966 KeyValue::Bool(bit) => {
967 out.push(TAG_BOOL);
968 out.push(u8::from(*bit));
969 }
970 KeyValue::Int(int) => {
971 out.push(TAG_INT);
972 out.extend_from_slice(&orderable_i64(*int).to_be_bytes());
973 }
974 KeyValue::TimestampMicros(micros) => {
975 out.push(TAG_TIMESTAMP_MICROS);
976 out.extend_from_slice(&orderable_i64(*micros).to_be_bytes());
977 }
978 KeyValue::Text(text) => {
979 out.push(TAG_TEXT);
980 encode_escaped(out, text.as_bytes());
981 }
982 KeyValue::Bytes(bytes) => {
983 out.push(TAG_BYTES);
984 encode_escaped(out, bytes);
985 }
986 }
987 }
988
989 pub fn decode_components(key: &Key) -> Result<Vec<KeyValue>, PartitionError> {
992 let bytes = key.as_bytes();
993 let mut values = Vec::new();
994 let mut cursor = 0;
995 while cursor < bytes.len() {
996 let tag = bytes[cursor];
997 cursor += 1;
998 let value = match tag {
999 TAG_NULL => KeyValue::Null,
1000 TAG_BOOL => {
1001 let byte = take(bytes, &mut cursor, 1)?[0];
1002 match byte {
1003 0 => KeyValue::Bool(false),
1004 1 => KeyValue::Bool(true),
1005 other => {
1006 return Err(PartitionError::MalformedKey(format!(
1007 "invalid bool payload {other}"
1008 )))
1009 }
1010 }
1011 }
1012 TAG_INT => {
1013 let raw = take(bytes, &mut cursor, 8)?;
1014 KeyValue::Int(unorderable_i64(raw))
1015 }
1016 TAG_TIMESTAMP_MICROS => {
1017 let raw = take(bytes, &mut cursor, 8)?;
1018 KeyValue::TimestampMicros(unorderable_i64(raw))
1019 }
1020 TAG_TEXT => {
1021 let payload = decode_escaped(bytes, &mut cursor)?;
1022 let text = String::from_utf8(payload)
1023 .map_err(|error| PartitionError::MalformedKey(error.to_string()))?;
1024 KeyValue::Text(text)
1025 }
1026 TAG_BYTES => KeyValue::Bytes(decode_escaped(bytes, &mut cursor)?),
1027 other => {
1028 return Err(PartitionError::MalformedKey(format!(
1029 "unknown type tag 0x{other:02x}"
1030 )))
1031 }
1032 };
1033 values.push(value);
1034 }
1035 Ok(values)
1036 }
1037
1038 pub fn fnv1a64(bytes: &[u8]) -> u64 {
1042 let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
1043 for byte in bytes {
1044 hash ^= u64::from(*byte);
1045 hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
1046 }
1047 hash
1048 }
1049}
1050
1051fn orderable_i64(value: i64) -> u64 {
1053 (value as u64) ^ (1 << 63)
1054}
1055
1056fn unorderable_i64(raw: &[u8]) -> i64 {
1058 let mut bytes = [0u8; 8];
1059 bytes.copy_from_slice(raw);
1060 (u64::from_be_bytes(bytes) ^ (1 << 63)) as i64
1061}
1062
1063fn take<'a>(bytes: &'a [u8], cursor: &mut usize, count: usize) -> Result<&'a [u8], PartitionError> {
1065 let end = cursor
1066 .checked_add(count)
1067 .filter(|end| *end <= bytes.len())
1068 .ok_or_else(|| PartitionError::MalformedKey("truncated payload".to_owned()))?;
1069 let slice = &bytes[*cursor..end];
1070 *cursor = end;
1071 Ok(slice)
1072}
1073
1074fn encode_escaped(out: &mut Vec<u8>, payload: &[u8]) {
1077 for byte in payload {
1078 out.push(*byte);
1079 if *byte == 0x00 {
1080 out.push(0xFF);
1081 }
1082 }
1083 out.extend_from_slice(&[0x00, 0x00]);
1084}
1085
1086fn decode_escaped(bytes: &[u8], cursor: &mut usize) -> Result<Vec<u8>, PartitionError> {
1088 let mut payload = Vec::new();
1089 loop {
1090 let byte = *bytes
1091 .get(*cursor)
1092 .ok_or_else(|| PartitionError::MalformedKey("unterminated payload".to_owned()))?;
1093 *cursor += 1;
1094 if byte != 0x00 {
1095 payload.push(byte);
1096 continue;
1097 }
1098 let escape = *bytes
1099 .get(*cursor)
1100 .ok_or_else(|| PartitionError::MalformedKey("unterminated payload".to_owned()))?;
1101 *cursor += 1;
1102 match escape {
1103 0x00 => return Ok(payload),
1104 0xFF => payload.push(0x00),
1105 other => {
1106 return Err(PartitionError::MalformedKey(format!(
1107 "invalid escape byte 0x{other:02x}"
1108 )))
1109 }
1110 }
1111 }
1112}
1113
1114#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
1124#[serde(deny_unknown_fields)]
1125pub struct TimeInterval {
1126 micros: u64,
1127}
1128
1129impl TimeInterval {
1130 pub fn micros(micros: u64) -> Result<Self, TabletError> {
1132 if micros == 0 {
1133 return Err(TabletError::InvalidDescriptor(
1134 "time interval must be positive".to_owned(),
1135 ));
1136 }
1137 Ok(Self { micros })
1138 }
1139
1140 pub fn hours(hours: u64) -> Result<Self, TabletError> {
1142 Self::micros(
1143 hours
1144 .checked_mul(3_600_000_000)
1145 .ok_or(TabletError::InvalidDescriptor(
1146 "time interval overflows u64 micros".to_owned(),
1147 ))?,
1148 )
1149 }
1150
1151 pub fn days(days: u64) -> Result<Self, TabletError> {
1153 Self::micros(
1154 days.checked_mul(86_400_000_000)
1155 .ok_or(TabletError::InvalidDescriptor(
1156 "time interval overflows u64 micros".to_owned(),
1157 ))?,
1158 )
1159 }
1160
1161 pub const fn as_micros(self) -> u64 {
1163 self.micros
1164 }
1165}
1166
1167#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1170pub enum Partitioning {
1171 Hash {
1174 columns: Vec<ColumnId>,
1176 buckets: u32,
1178 },
1179 Range {
1183 columns: Vec<ColumnId>,
1185 splits: Vec<Key>,
1187 },
1188 Tenant {
1192 tenant_column: ColumnId,
1194 buckets_per_tenant: u32,
1196 },
1197 TimeRange {
1200 timestamp_column: ColumnId,
1202 interval: TimeInterval,
1204 },
1205}
1206
1207impl Partitioning {
1208 pub fn validate(&self) -> Result<(), PartitionError> {
1211 match self {
1212 Self::Hash { columns, buckets } => {
1213 if columns.is_empty() {
1214 return Err(PartitionError::InvalidPartitioning(
1215 "hash partitioning needs at least one column".to_owned(),
1216 ));
1217 }
1218 if *buckets == 0 {
1219 return Err(PartitionError::InvalidPartitioning(
1220 "hash partitioning needs at least one bucket".to_owned(),
1221 ));
1222 }
1223 }
1224 Self::Range { columns, splits } => {
1225 if columns.is_empty() {
1226 return Err(PartitionError::InvalidPartitioning(
1227 "range partitioning needs at least one column".to_owned(),
1228 ));
1229 }
1230 if splits.windows(2).any(|window| window[0] >= window[1]) {
1231 return Err(PartitionError::InvalidPartitioning(
1232 "range splits must be strictly increasing".to_owned(),
1233 ));
1234 }
1235 }
1236 Self::Tenant {
1237 buckets_per_tenant, ..
1238 } => {
1239 if *buckets_per_tenant == 0 {
1240 return Err(PartitionError::InvalidPartitioning(
1241 "tenant partitioning needs at least one bucket per tenant".to_owned(),
1242 ));
1243 }
1244 }
1245 Self::TimeRange { interval, .. } => {
1246 if interval.as_micros() == 0 {
1247 return Err(PartitionError::InvalidPartitioning(
1248 "time-range interval must be positive".to_owned(),
1249 ));
1250 }
1251 }
1252 }
1253 Ok(())
1254 }
1255
1256 pub fn partition_columns(&self) -> Vec<ColumnId> {
1259 match self {
1260 Self::Hash { columns, .. } | Self::Range { columns, .. } => columns.clone(),
1261 Self::Tenant { tenant_column, .. } => vec![*tenant_column],
1262 Self::TimeRange {
1263 timestamp_column, ..
1264 } => vec![*timestamp_column],
1265 }
1266 }
1267
1268 pub fn partition_key(
1277 &self,
1278 values: &BTreeMap<ColumnId, KeyValue>,
1279 ) -> Result<Key, PartitionError> {
1280 self.validate()?;
1281 let columns = self.partition_columns();
1282 let mut components = Vec::with_capacity(columns.len());
1283 for column in columns {
1284 let value = values
1285 .get(&column)
1286 .ok_or(PartitionError::MissingPartitionColumn { column })?;
1287 if let Self::TimeRange { .. } = self {
1288 if !matches!(value, KeyValue::TimestampMicros(_)) {
1289 return Err(PartitionError::PartitionColumnType {
1290 column,
1291 expected: "timestamp-micros",
1292 found: value.type_name(),
1293 });
1294 }
1295 }
1296 components.push(value.clone());
1297 }
1298 Ok(RowKeyEncoder::encode_key(&components))
1299 }
1300
1301 pub fn route(&self, partition_key: &Key) -> Result<u64, PartitionError> {
1312 self.validate()?;
1313 match self {
1314 Self::Hash { buckets, .. } => {
1315 Ok(RowKeyEncoder::fnv1a64(partition_key.as_bytes()) % u64::from(*buckets))
1316 }
1317 Self::Range { splits, .. } => {
1318 Ok(splits.partition_point(|split| split <= partition_key) as u64)
1319 }
1320 Self::Tenant {
1321 buckets_per_tenant, ..
1322 } => {
1323 Ok(RowKeyEncoder::fnv1a64(partition_key.as_bytes())
1324 % u64::from(*buckets_per_tenant))
1325 }
1326 Self::TimeRange { interval, .. } => {
1327 let mut components = RowKeyEncoder::decode_components(partition_key)?;
1328 if components.len() != 1 {
1329 return Err(PartitionError::MalformedKey(format!(
1330 "time-range partition key has {} components, expected 1",
1331 components.len()
1332 )));
1333 }
1334 let Some(KeyValue::TimestampMicros(micros)) = components.pop() else {
1335 return Err(PartitionError::MalformedKey(
1336 "time-range partition key does not start with a timestamp".to_owned(),
1337 ));
1338 };
1339 let slot =
1340 micros.div_euclid(i64::try_from(interval.as_micros()).map_err(|_| {
1341 PartitionError::InvalidPartitioning(
1342 "time-range interval exceeds i64 micros".to_owned(),
1343 )
1344 })?);
1345 u64::try_from(slot).map_err(|_| PartitionError::NegativeSlot { micros })
1346 }
1347 }
1348 }
1349
1350 pub fn routed_key(&self, partition_key: &Key) -> Result<Key, PartitionError> {
1362 match self {
1363 Self::Range { .. } | Self::TimeRange { .. } => Ok(partition_key.clone()),
1364 Self::Hash { .. } => Ok(Key::from_bytes(
1365 self.route(partition_key)?.to_be_bytes().to_vec(),
1366 )),
1367 Self::Tenant { .. } => {
1368 let mut bytes = partition_key.as_bytes().to_vec();
1369 bytes.extend_from_slice(&self.route(partition_key)?.to_be_bytes());
1370 Ok(Key::from_bytes(bytes))
1371 }
1372 }
1373 }
1374
1375 pub fn range_bounds(&self, index: u64) -> Option<PartitionBounds> {
1378 let Self::Range { splits, .. } = self else {
1379 return None;
1380 };
1381 let index = usize::try_from(index).ok()?;
1382 if index > splits.len() {
1383 return None;
1384 }
1385 let low = index.checked_sub(1).map_or(Bound::Unbounded, |previous| {
1386 Bound::Included(splits[previous].clone())
1387 });
1388 let high = splits
1389 .get(index)
1390 .map_or(Bound::Unbounded, |split| Bound::Excluded(split.clone()));
1391 Some(PartitionBounds { low, high })
1392 }
1393}
1394
1395pub fn hash_slot_bounds(low_slot: u64, high_slot_exclusive: u64) -> PartitionBounds {
1399 PartitionBounds {
1400 low: Bound::Included(Key::from_bytes(low_slot.to_be_bytes().to_vec())),
1401 high: Bound::Excluded(Key::from_bytes(high_slot_exclusive.to_be_bytes().to_vec())),
1402 }
1403}
1404
1405#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
1409pub enum PartitioningOrigin {
1410 Declared,
1412 AutomaticDefault,
1415}
1416
1417#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
1423#[repr(transparent)]
1424pub struct ColocatedWith(pub TableId);
1425
1426#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1430#[serde(deny_unknown_fields)]
1431pub struct TablePartitioningRecord {
1432 pub table_id: TableId,
1434 pub partitioning: Partitioning,
1436 pub origin: PartitioningOrigin,
1438 pub colocated_with: Option<ColocatedWith>,
1440}
1441
1442impl TablePartitioningRecord {
1443 pub fn automatic_default(
1447 table_id: TableId,
1448 primary_key_columns: Vec<ColumnId>,
1449 buckets: u32,
1450 ) -> Self {
1451 Self {
1452 table_id,
1453 partitioning: Partitioning::Hash {
1454 columns: primary_key_columns,
1455 buckets,
1456 },
1457 origin: PartitioningOrigin::AutomaticDefault,
1458 colocated_with: None,
1459 }
1460 }
1461
1462 pub fn validate(&self) -> Result<(), TabletError> {
1465 if self.table_id == TableId::ZERO {
1466 return Err(TabletError::InvalidDescriptor(
1467 "reserved zero table id".to_owned(),
1468 ));
1469 }
1470 self.partitioning.validate()?;
1471 if self
1472 .colocated_with
1473 .is_some_and(|target| target.0 == self.table_id)
1474 {
1475 return Err(TabletError::InvalidDescriptor(
1476 "a table cannot be colocated with itself".to_owned(),
1477 ));
1478 }
1479 Ok(())
1480 }
1481}
1482
1483pub fn find_tablet_for_key<'a>(
1494 tablets: &'a [TabletDescriptor],
1495 table_id: TableId,
1496 routed_key: &Key,
1497) -> Option<&'a TabletDescriptor> {
1498 tablets.iter().find(|tablet| {
1499 tablet.table_id == table_id
1500 && tablet.state.is_routable()
1501 && tablet.partition.contains(routed_key)
1502 })
1503}
1504
1505pub fn tablets_overlapping<'a>(
1508 tablets: &'a [TabletDescriptor],
1509 table_id: TableId,
1510 range: &PartitionBounds,
1511) -> Vec<&'a TabletDescriptor> {
1512 let mut matched: Vec<&TabletDescriptor> = tablets
1513 .iter()
1514 .filter(|tablet| {
1515 tablet.table_id == table_id
1516 && tablet.state.is_routable()
1517 && tablet.partition.overlaps(range)
1518 })
1519 .collect();
1520 matched.sort_by(|left, right| {
1521 cmp_low_bounds(&left.partition.low, &right.partition.low)
1522 .then_with(|| left.tablet_id.cmp(&right.tablet_id))
1523 });
1524 matched
1525}
1526
1527fn cmp_low_bounds(left: &Bound<Key>, right: &Bound<Key>) -> std::cmp::Ordering {
1530 use std::cmp::Ordering;
1531 match (left, right) {
1532 (Bound::Unbounded, Bound::Unbounded) => Ordering::Equal,
1533 (Bound::Unbounded, _) => Ordering::Less,
1534 (_, Bound::Unbounded) => Ordering::Greater,
1535 (Bound::Included(left), Bound::Included(right))
1536 | (Bound::Excluded(left), Bound::Excluded(right)) => left.cmp(right),
1537 (Bound::Included(left), Bound::Excluded(right)) => left.cmp(right).then(Ordering::Less),
1538 (Bound::Excluded(left), Bound::Included(right)) => left.cmp(right).then(Ordering::Greater),
1539 }
1540}
1541
1542pub fn check_generation(
1554 descriptor: &TabletDescriptor,
1555 used_generation: u64,
1556) -> Result<(), RoutingError> {
1557 if used_generation == descriptor.generation {
1558 return Ok(());
1559 }
1560 let tablet_id = descriptor.tablet_id;
1561 let current_generation = descriptor.generation;
1562 Err(match descriptor.state {
1566 TabletState::Splitting if used_generation < current_generation => {
1567 RoutingError::TabletSplit {
1568 tablet_id,
1569 used_generation,
1570 current_generation,
1571 }
1572 }
1573 TabletState::Retiring | TabletState::Retired => RoutingError::TabletMoved {
1574 tablet_id,
1575 used_generation,
1576 current_generation,
1577 },
1578 _ => RoutingError::StaleMetadata {
1579 tablet_id,
1580 used_generation,
1581 current_generation,
1582 },
1583 })
1584}
1585
1586pub const TABLETS_DIR: &str = "tablets";
1592pub const GROUPS_DIR: &str = "groups";
1594pub const TABLET_STATE_DIR: &str = "state";
1596pub const TABLET_RUNS_DIR: &str = "runs";
1598pub const TABLET_INDEXES_DIR: &str = "indexes";
1600pub const TABLET_TEMP_DIR: &str = "temp";
1602pub const GROUP_RAFT_DIR: &str = "raft";
1604pub const GROUP_SNAPSHOTS_DIR: &str = "snapshots";
1610pub const TABLET_META_FILENAME: &str = "tablet.json";
1612pub const TABLET_META_FORMAT_VERSION: u32 = 1;
1614pub const MIN_SUPPORTED_TABLET_META_FORMAT_VERSION: u32 = 1;
1616
1617#[derive(Clone, Debug, PartialEq, Eq)]
1632pub struct TabletLayout {
1633 node_data: PathBuf,
1634 tablet_id: TabletId,
1635 raft_group_id: RaftGroupId,
1636}
1637
1638impl TabletLayout {
1639 pub fn new(
1641 node_data: impl Into<PathBuf>,
1642 tablet_id: TabletId,
1643 raft_group_id: RaftGroupId,
1644 ) -> Self {
1645 Self {
1646 node_data: node_data.into(),
1647 tablet_id,
1648 raft_group_id,
1649 }
1650 }
1651
1652 pub fn tablet_id(&self) -> TabletId {
1654 self.tablet_id
1655 }
1656
1657 pub fn raft_group_id(&self) -> RaftGroupId {
1659 self.raft_group_id
1660 }
1661
1662 pub fn node_data(&self) -> &Path {
1664 &self.node_data
1665 }
1666
1667 pub fn tablet_dir(&self) -> PathBuf {
1669 self.node_data
1670 .join(TABLETS_DIR)
1671 .join(self.tablet_id.to_hex())
1672 }
1673
1674 pub fn state_dir(&self) -> PathBuf {
1676 self.tablet_dir().join(TABLET_STATE_DIR)
1677 }
1678
1679 pub fn runs_dir(&self) -> PathBuf {
1681 self.tablet_dir().join(TABLET_RUNS_DIR)
1682 }
1683
1684 pub fn indexes_dir(&self) -> PathBuf {
1686 self.tablet_dir().join(TABLET_INDEXES_DIR)
1687 }
1688
1689 pub fn temp_dir(&self) -> PathBuf {
1691 self.tablet_dir().join(TABLET_TEMP_DIR)
1692 }
1693
1694 pub fn group_dir(&self) -> PathBuf {
1696 self.node_data
1697 .join(GROUPS_DIR)
1698 .join(self.raft_group_id.to_hex())
1699 }
1700
1701 pub fn raft_dir(&self) -> PathBuf {
1703 self.group_dir().join(GROUP_RAFT_DIR)
1704 }
1705
1706 pub fn snapshots_dir(&self) -> PathBuf {
1709 self.group_dir().join(GROUP_SNAPSHOTS_DIR)
1710 }
1711
1712 pub fn raft_snapshot_dir(&self) -> PathBuf {
1715 self.raft_dir().join("snapshot")
1716 }
1717
1718 pub fn metadata_path(&self) -> PathBuf {
1720 self.tablet_dir().join(TABLET_META_FILENAME)
1721 }
1722
1723 fn required_dirs(&self) -> [PathBuf; 6] {
1725 [
1726 self.state_dir(),
1727 self.runs_dir(),
1728 self.indexes_dir(),
1729 self.temp_dir(),
1730 self.raft_dir(),
1731 self.snapshots_dir(),
1732 ]
1733 }
1734
1735 pub fn create(&self, descriptor: &TabletDescriptor) -> Result<(), TabletError> {
1744 self.check_descriptor_identity(descriptor)?;
1745 descriptor.validate()?;
1746 for dir in self.required_dirs() {
1747 fs::create_dir_all(&dir).map_err(ClusterError::Io)?;
1748 }
1749 let file = TabletMetaFile::envelope(descriptor)?;
1750 let bytes = crate::node::encode_json(TABLET_META_FILENAME, &file)?;
1751 match crate::node::create_meta_file(&self.tablet_dir(), TABLET_META_FILENAME, &bytes)
1752 .map_err(ClusterError::Io)?
1753 {
1754 true => Ok(()),
1755 false => match self.load_metadata()? {
1757 existing if existing == *descriptor => Ok(()),
1758 _ => Err(TabletError::MetadataConflict(self.tablet_dir())),
1759 },
1760 }
1761 }
1762
1763 pub fn store_metadata(&self, descriptor: &TabletDescriptor) -> Result<(), TabletError> {
1767 self.check_descriptor_identity(descriptor)?;
1768 descriptor.validate()?;
1769 let file = TabletMetaFile::envelope(descriptor)?;
1770 let bytes = crate::node::encode_json(TABLET_META_FILENAME, &file)?;
1771 crate::node::write_meta_atomic(&self.tablet_dir(), TABLET_META_FILENAME, &bytes)
1772 .map_err(ClusterError::Io)?;
1773 Ok(())
1774 }
1775
1776 pub fn load_metadata(&self) -> Result<TabletDescriptor, TabletError> {
1780 let Some(bytes) = crate::node::read_meta_file(&self.metadata_path())? else {
1781 return Err(TabletError::MissingMetadata(self.metadata_path()));
1782 };
1783 let file: TabletMetaFile = crate::node::decode_json(TABLET_META_FILENAME, &bytes)?;
1784 if file.format_version < MIN_SUPPORTED_TABLET_META_FORMAT_VERSION
1785 || file.format_version > TABLET_META_FORMAT_VERSION
1786 {
1787 return Err(ClusterError::UnsupportedFormatVersion {
1788 file: TABLET_META_FILENAME,
1789 found: file.format_version,
1790 min: MIN_SUPPORTED_TABLET_META_FORMAT_VERSION,
1791 max: TABLET_META_FORMAT_VERSION,
1792 }
1793 .into());
1794 }
1795 if file.checksum != tablet_checksum(&file.tablet)? {
1796 return Err(ClusterError::CorruptMetadata {
1797 file: TABLET_META_FILENAME,
1798 detail: "checksum mismatch".to_owned(),
1799 }
1800 .into());
1801 }
1802 self.check_descriptor_identity(&file.tablet)?;
1803 file.tablet.validate()?;
1804 Ok(file.tablet)
1805 }
1806
1807 pub fn validate(&self) -> Result<TabletDescriptor, TabletError> {
1811 let descriptor = self.load_metadata()?;
1812 for dir in self.required_dirs() {
1813 if !dir.is_dir() {
1814 return Err(ClusterError::CorruptMetadata {
1815 file: TABLET_META_FILENAME,
1816 detail: format!("required directory {} is missing", dir.display()),
1817 }
1818 .into());
1819 }
1820 }
1821 Ok(descriptor)
1822 }
1823
1824 fn check_descriptor_identity(&self, descriptor: &TabletDescriptor) -> Result<(), TabletError> {
1826 if descriptor.tablet_id != self.tablet_id || descriptor.raft_group_id != self.raft_group_id
1827 {
1828 return Err(TabletError::TabletMismatch {
1829 path: self.tablet_dir(),
1830 expected: self.tablet_id,
1831 found: descriptor.tablet_id,
1832 expected_group: self.raft_group_id,
1833 found_group: descriptor.raft_group_id,
1834 });
1835 }
1836 Ok(())
1837 }
1838
1839 pub fn teardown(&self) -> Result<(), TabletError> {
1846 if self.tablet_dir().is_dir() {
1847 match self.load_metadata() {
1848 Ok(_) | Err(TabletError::MissingMetadata(_)) => {}
1849 Err(error) => return Err(error),
1851 }
1852 fs::remove_dir_all(self.tablet_dir()).map_err(ClusterError::Io)?;
1853 }
1854 if self.group_dir().is_dir() {
1855 fs::remove_dir_all(self.group_dir()).map_err(ClusterError::Io)?;
1856 }
1857 Ok(())
1858 }
1859}
1860
1861#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1864#[serde(deny_unknown_fields)]
1865struct TabletMetaFile {
1866 format_version: u32,
1868 checksum: String,
1870 tablet: TabletDescriptor,
1872}
1873
1874impl TabletMetaFile {
1875 fn envelope(tablet: &TabletDescriptor) -> Result<Self, TabletError> {
1876 Ok(Self {
1877 format_version: TABLET_META_FORMAT_VERSION,
1878 checksum: tablet_checksum(tablet)?,
1879 tablet: tablet.clone(),
1880 })
1881 }
1882}
1883
1884fn tablet_checksum(tablet: &TabletDescriptor) -> Result<String, ClusterError> {
1886 let bytes = serde_json::to_vec(tablet).map_err(|error| ClusterError::CorruptMetadata {
1887 file: TABLET_META_FILENAME,
1888 detail: format!("encode: {error}"),
1889 })?;
1890 Ok(hex_encode(&Sha256::digest(&bytes)))
1891}
1892
1893pub fn list_tablets_on_disk(
1900 node_data: impl AsRef<Path>,
1901) -> Result<(Vec<TabletDescriptor>, Vec<String>), ClusterError> {
1902 let root = node_data.as_ref().join(TABLETS_DIR);
1903 if !root.is_dir() {
1904 return Ok((Vec::new(), Vec::new()));
1905 }
1906 let mut tablets = Vec::new();
1907 let mut issues = Vec::new();
1908 let entries = fs::read_dir(&root).map_err(ClusterError::Io)?;
1909 for entry in entries {
1910 let entry = entry.map_err(ClusterError::Io)?;
1911 let path = entry.path();
1912 if !path.is_dir() {
1913 continue;
1914 }
1915 let meta = path.join(TABLET_META_FILENAME);
1916 if !meta.is_file() {
1917 continue;
1918 }
1919 match load_tablet_meta_file(&meta) {
1922 Ok(descriptor) => tablets.push(descriptor),
1923 Err(error) => issues.push(format!("{}: {error}", meta.display())),
1924 }
1925 }
1926 tablets.sort_by_key(|t| t.tablet_id);
1927 Ok((tablets, issues))
1928}
1929
1930fn load_tablet_meta_file(path: &Path) -> Result<TabletDescriptor, ClusterError> {
1931 let Some(bytes) = crate::node::read_meta_file(path)? else {
1932 return Err(ClusterError::CorruptMetadata {
1933 file: TABLET_META_FILENAME,
1934 detail: format!("missing {}", path.display()),
1935 });
1936 };
1937 let file: TabletMetaFile = crate::node::decode_json(TABLET_META_FILENAME, &bytes)?;
1938 if file.format_version < MIN_SUPPORTED_TABLET_META_FORMAT_VERSION
1939 || file.format_version > TABLET_META_FORMAT_VERSION
1940 {
1941 return Err(ClusterError::UnsupportedFormatVersion {
1942 file: TABLET_META_FILENAME,
1943 found: file.format_version,
1944 min: MIN_SUPPORTED_TABLET_META_FORMAT_VERSION,
1945 max: TABLET_META_FORMAT_VERSION,
1946 });
1947 }
1948 if file.checksum != tablet_checksum(&file.tablet)? {
1949 return Err(ClusterError::CorruptMetadata {
1950 file: TABLET_META_FILENAME,
1951 detail: "checksum mismatch".to_owned(),
1952 });
1953 }
1954 file.tablet
1955 .validate()
1956 .map_err(|e| ClusterError::CorruptMetadata {
1957 file: TABLET_META_FILENAME,
1958 detail: e.to_string(),
1959 })?;
1960 Ok(file.tablet)
1961}
1962
1963#[derive(Debug, Default)]
1975pub struct TabletOwnershipRegistry {
1976 reservations: Mutex<HashMap<PathBuf, TabletId>>,
1977}
1978
1979impl TabletOwnershipRegistry {
1980 pub fn new() -> Self {
1982 Self::default()
1983 }
1984
1985 pub fn global() -> &'static Self {
1987 static REGISTRY: OnceLock<TabletOwnershipRegistry> = OnceLock::new();
1988 REGISTRY.get_or_init(Self::new)
1989 }
1990
1991 pub fn try_reserve(
1995 &self,
1996 layout: &TabletLayout,
1997 ) -> Result<TabletOwnershipGuard<'_>, TabletError> {
1998 let path = layout
1999 .tablet_dir()
2000 .canonicalize()
2001 .map_err(ClusterError::Io)?;
2002 let mut reservations = self
2003 .reservations
2004 .lock()
2005 .expect("tablet ownership registry lock poisoned");
2006 if let Some(holder) = reservations.get(&path) {
2007 return Err(TabletError::AlreadyOwned {
2008 tablet: *holder,
2009 path,
2010 });
2011 }
2012 reservations.insert(path.clone(), layout.tablet_id());
2013 Ok(TabletOwnershipGuard {
2014 registry: self,
2015 path,
2016 tablet_id: layout.tablet_id(),
2017 })
2018 }
2019
2020 pub fn len(&self) -> usize {
2022 self.reservations
2023 .lock()
2024 .expect("tablet ownership registry lock poisoned")
2025 .len()
2026 }
2027
2028 pub fn is_empty(&self) -> bool {
2030 self.len() == 0
2031 }
2032}
2033
2034#[derive(Debug)]
2037pub struct TabletOwnershipGuard<'a> {
2038 registry: &'a TabletOwnershipRegistry,
2039 path: PathBuf,
2040 tablet_id: TabletId,
2041}
2042
2043impl TabletOwnershipGuard<'_> {
2044 pub fn tablet_id(&self) -> TabletId {
2046 self.tablet_id
2047 }
2048
2049 pub fn path(&self) -> &Path {
2051 &self.path
2052 }
2053}
2054
2055impl Drop for TabletOwnershipGuard<'_> {
2056 fn drop(&mut self) {
2057 let mut reservations = self
2058 .registry
2059 .reservations
2060 .lock()
2061 .expect("tablet ownership registry lock poisoned");
2062 if reservations.get(&self.path) == Some(&self.tablet_id) {
2063 reservations.remove(&self.path);
2064 }
2065 }
2066}
2067
2068#[cfg(test)]
2073mod tests {
2074 use super::*;
2075
2076 fn node(byte: u8) -> NodeId {
2077 NodeId::from_bytes([byte; 16])
2078 }
2079
2080 fn tablet_id(byte: u8) -> TabletId {
2081 TabletId::from_bytes([byte; 16])
2082 }
2083
2084 fn group_id(byte: u8) -> RaftGroupId {
2085 RaftGroupId::from_bytes([byte; 16])
2086 }
2087
2088 fn key(bytes: &[u8]) -> Key {
2089 Key::from_bytes(bytes.to_vec())
2090 }
2091
2092 fn text_key(text: &str) -> Key {
2093 RowKeyEncoder::encode_key(&[KeyValue::Text(text.to_owned())])
2094 }
2095
2096 fn int_key(value: i64) -> Key {
2097 RowKeyEncoder::encode_key(&[KeyValue::Int(value)])
2098 }
2099
2100 fn bounds(low: Bound<Key>, high: Bound<Key>) -> PartitionBounds {
2101 PartitionBounds::new(low, high).unwrap()
2102 }
2103
2104 fn descriptor(state: TabletState) -> TabletDescriptor {
2105 TabletDescriptor {
2106 tablet_id: tablet_id(9),
2107 table_id: TableId::new(3),
2108 database_id: mongreldb_types::ids::DatabaseId::ZERO,
2109 raft_group_id: group_id(7),
2110 partition: bounds(
2111 Bound::Included(text_key("a")),
2112 Bound::Excluded(text_key("m")),
2113 ),
2114 replicas: vec![
2115 ReplicaDescriptor {
2116 node_id: node(1),
2117 role: ReplicaRole::Voter,
2118 raft_node_id: 11,
2119 },
2120 ReplicaDescriptor {
2121 node_id: node(2),
2122 role: ReplicaRole::Voter,
2123 raft_node_id: 12,
2124 },
2125 ReplicaDescriptor {
2126 node_id: node(3),
2127 role: ReplicaRole::Learner,
2128 raft_node_id: 13,
2129 },
2130 ],
2131 leader_hint: Some(node(1)),
2132 generation: 7,
2133 state,
2134 }
2135 }
2136
2137 #[test]
2140 fn descriptor_round_trips_serde_in_every_state() {
2141 for state in [
2142 TabletState::Creating,
2143 TabletState::Active,
2144 TabletState::Splitting,
2145 TabletState::Merging,
2146 TabletState::Retiring,
2147 TabletState::Retired,
2148 ] {
2149 let descriptor = descriptor(state);
2150 let json = serde_json::to_vec(&descriptor).unwrap();
2151 let back: TabletDescriptor = serde_json::from_slice(&json).unwrap();
2152 assert_eq!(back, descriptor);
2153 }
2154 }
2155
2156 #[test]
2157 fn descriptor_rejects_unknown_fields() {
2158 let descriptor = descriptor(TabletState::Active);
2159 let mut value: serde_json::Value =
2160 serde_json::from_slice(&serde_json::to_vec(&descriptor).unwrap()).unwrap();
2161 value["unexpected"] = serde_json::json!(1);
2162 assert!(serde_json::from_value::<TabletDescriptor>(value).is_err());
2163 }
2164
2165 #[test]
2168 fn descriptor_validation_catches_structural_violations() {
2169 let mut zero_tablet = descriptor(TabletState::Active);
2170 zero_tablet.tablet_id = TabletId::ZERO;
2171 assert!(matches!(
2172 zero_tablet.validate(),
2173 Err(TabletError::InvalidDescriptor(_))
2174 ));
2175
2176 let mut duplicate_nodes = descriptor(TabletState::Active);
2177 duplicate_nodes.replicas[1].node_id = node(1);
2178 assert!(matches!(
2179 duplicate_nodes.validate(),
2180 Err(TabletError::InvalidDescriptor(_))
2181 ));
2182
2183 let mut duplicate_raft_ids = descriptor(TabletState::Active);
2184 duplicate_raft_ids.replicas[1].raft_node_id = 11;
2185 assert!(matches!(
2186 duplicate_raft_ids.validate(),
2187 Err(TabletError::InvalidDescriptor(_))
2188 ));
2189
2190 let mut foreign_leader = descriptor(TabletState::Active);
2191 foreign_leader.leader_hint = Some(node(8));
2192 assert!(matches!(
2193 foreign_leader.validate(),
2194 Err(TabletError::InvalidDescriptor(_))
2195 ));
2196
2197 let mut learner_only = descriptor(TabletState::Active);
2199 learner_only.replicas.iter_mut().for_each(|replica| {
2200 replica.role = ReplicaRole::Learner;
2201 });
2202 assert!(matches!(
2203 learner_only.validate(),
2204 Err(TabletError::InvalidDescriptor(_))
2205 ));
2206 let mut creating = learner_only.clone();
2208 creating.state = TabletState::Creating;
2209 creating.validate().unwrap();
2210
2211 descriptor(TabletState::Active).validate().unwrap();
2212 }
2213
2214 #[test]
2217 fn transition_graph_allows_exactly_the_documented_edges() {
2218 use TabletState::{Active, Creating, Merging, Retired, Retiring, Splitting};
2219 let allowed = [
2220 (Creating, Active),
2221 (Creating, Retired),
2222 (Active, Splitting),
2223 (Active, Merging),
2224 (Active, Retiring),
2225 (Splitting, Active),
2226 (Splitting, Retiring),
2227 (Merging, Active),
2228 (Merging, Retiring),
2229 (Retiring, Retired),
2230 ];
2231 let states = [Creating, Active, Splitting, Merging, Retiring, Retired];
2232 for from in states {
2233 for to in states {
2234 assert_eq!(
2235 from.can_transition_to(to),
2236 allowed.contains(&(from, to)),
2237 "unexpected graph edge {from} -> {to}"
2238 );
2239 }
2240 }
2241 }
2242
2243 #[test]
2244 fn try_transition_enforces_the_graph() {
2245 let mut tablet = descriptor(TabletState::Active);
2246 let error = tablet.try_transition(TabletState::Creating).unwrap_err();
2247 assert!(matches!(
2248 error,
2249 TabletError::InvalidStateTransition { tablet, from, to }
2250 if tablet == tablet_id(9)
2251 && from == TabletState::Active
2252 && to == TabletState::Creating
2253 ));
2254 tablet.try_transition(TabletState::Splitting).unwrap();
2255 tablet.try_transition(TabletState::Retiring).unwrap();
2256 tablet.try_transition(TabletState::Retired).unwrap();
2257 assert!(tablet.try_transition(TabletState::Active).is_err());
2259 }
2260
2261 #[test]
2262 fn published_transition_stages_an_atomic_publication() {
2263 let tablet = descriptor(TabletState::Active);
2264 let splitting = tablet.published_transition(TabletState::Splitting).unwrap();
2265 assert_eq!(splitting.state, TabletState::Splitting);
2267 assert_eq!(splitting.generation, 8);
2268 assert_eq!(tablet.state, TabletState::Active);
2269 assert_eq!(tablet.generation, 7);
2270 splitting.validate().unwrap();
2271
2272 assert!(matches!(
2274 tablet.published_transition(TabletState::Creating),
2275 Err(TabletError::InvalidStateTransition { .. })
2276 ));
2277 }
2278
2279 #[test]
2280 fn routability_matches_the_split_merge_protocol() {
2281 assert!(TabletState::Active.is_routable());
2282 assert!(TabletState::Splitting.is_routable());
2283 assert!(TabletState::Merging.is_routable());
2284 assert!(!TabletState::Creating.is_routable());
2285 assert!(!TabletState::Retiring.is_routable());
2286 assert!(!TabletState::Retired.is_routable());
2287 }
2288
2289 #[test]
2292 fn bounds_containment_respects_inclusion() {
2293 let range = bounds(Bound::Included(key(b"b")), Bound::Excluded(key(b"f")));
2294 assert!(!range.contains(&key(b"a")));
2295 assert!(range.contains(&key(b"b")));
2296 assert!(range.contains(&key(b"e")));
2297 assert!(!range.contains(&key(b"f")));
2298 assert!(!range.contains(&key(b"z")));
2299
2300 let open = bounds(Bound::Excluded(key(b"b")), Bound::Included(key(b"f")));
2301 assert!(!open.contains(&key(b"b")));
2302 assert!(open.contains(&key(b"f")));
2303
2304 let everything = PartitionBounds::unbounded();
2305 assert!(everything.contains(&key(b"")));
2306 assert!(everything.contains(&key(b"zzzz")));
2307 }
2308
2309 #[test]
2310 fn bounds_overlap_and_adjacency_matrix() {
2311 let lower = bounds(Bound::Included(key(b"a")), Bound::Excluded(key(b"c")));
2312 let upper = bounds(Bound::Included(key(b"c")), Bound::Excluded(key(b"e")));
2313 assert!(!lower.overlaps(&upper));
2315 assert!(lower.is_adjacent_to(&upper));
2316 assert!(upper.is_adjacent_to(&lower));
2317
2318 let lower_closed = bounds(Bound::Included(key(b"a")), Bound::Included(key(b"c")));
2320 assert!(lower_closed.overlaps(&upper));
2321 assert!(!lower_closed.is_adjacent_to(&upper));
2322
2323 let upper_open = bounds(Bound::Excluded(key(b"c")), Bound::Included(key(b"e")));
2325 assert!(!lower.overlaps(&upper_open));
2326 assert!(!lower.is_adjacent_to(&upper_open));
2327
2328 let nested = bounds(Bound::Included(key(b"a0")), Bound::Excluded(key(b"b")));
2330 assert!(lower.overlaps(&nested));
2331 assert!(PartitionBounds::unbounded().overlaps(&lower));
2332 assert!(!PartitionBounds::unbounded().is_adjacent_to(&lower));
2333
2334 let far = bounds(Bound::Included(key(b"x")), Bound::Unbounded);
2336 assert!(!lower.overlaps(&far));
2337 assert!(!lower.is_adjacent_to(&far));
2338 }
2339
2340 #[test]
2341 fn bounds_validation_rejects_empty_and_inverted_ranges() {
2342 assert!(matches!(
2343 PartitionBounds::new(Bound::Included(key(b"m")), Bound::Excluded(key(b"a"))),
2344 Err(TabletError::InvalidBounds(_))
2345 ));
2346 assert!(matches!(
2348 PartitionBounds::new(Bound::Included(key(b"k")), Bound::Excluded(key(b"k"))),
2349 Err(TabletError::InvalidBounds(_))
2350 ));
2351 assert!(
2353 PartitionBounds::new(Bound::Excluded(key(b"k")), Bound::Included(key(b"k"))).is_err()
2354 );
2355 let point = bounds(Bound::Included(key(b"k")), Bound::Included(key(b"k")));
2357 assert!(point.contains(&key(b"k")));
2358 assert!(!point.contains(&key(b"j")));
2359 }
2360
2361 #[test]
2362 fn split_at_partitions_the_range_with_no_gap_or_overlap() {
2363 let range = bounds(Bound::Included(key(b"b")), Bound::Excluded(key(b"f")));
2364 let (lower, upper) = range.split_at(&key(b"d")).unwrap();
2365 assert_eq!(
2366 lower,
2367 bounds(Bound::Included(key(b"b")), Bound::Excluded(key(b"d")))
2368 );
2369 assert_eq!(
2370 upper,
2371 bounds(Bound::Included(key(b"d")), Bound::Excluded(key(b"f")))
2372 );
2373 assert!(lower.meets_start_of(&upper));
2375 assert!(lower.is_adjacent_to(&upper));
2376 assert!(!lower.overlaps(&upper));
2377 for candidate in [b"a", b"b", b"c", b"d", b"e", b"f", b"g"] {
2378 let candidate = key(candidate);
2379 assert_eq!(
2380 range.contains(&candidate),
2381 lower.contains(&candidate) || upper.contains(&candidate),
2382 "coverage mismatch at {candidate}"
2383 );
2384 assert!(
2385 !(lower.contains(&candidate) && upper.contains(&candidate)),
2386 "double coverage at {candidate}"
2387 );
2388 }
2389
2390 let whole = PartitionBounds::unbounded();
2392 let (low_half, high_half) = whole.split_at(&key(b"k")).unwrap();
2393 assert!(matches!(low_half.low, Bound::Unbounded));
2394 assert!(matches!(high_half.high, Bound::Unbounded));
2395 assert!(!low_half.contains(&key(b"k")));
2396 assert!(high_half.contains(&key(b"k")));
2397
2398 assert!(range.split_at(&key(b"a")).is_none());
2400 assert!(range.split_at(&key(b"f")).is_none());
2401 assert!(range.split_at(&key(b"b")).is_none()); let point = bounds(Bound::Included(key(b"b")), Bound::Included(key(b"b")));
2404 assert!(point.split_at(&key(b"b")).is_none());
2405 let closed = bounds(Bound::Included(key(b"b")), Bound::Included(key(b"f")));
2407 let (_, upper) = closed.split_at(&key(b"f")).unwrap();
2408 assert_eq!(
2409 upper,
2410 bounds(Bound::Included(key(b"f")), Bound::Included(key(b"f")))
2411 );
2412 }
2413
2414 #[test]
2415 fn union_adjacent_combines_only_adjacent_ranges() {
2416 let lower = bounds(Bound::Included(key(b"a")), Bound::Excluded(key(b"c")));
2417 let upper = bounds(Bound::Included(key(b"c")), Bound::Excluded(key(b"e")));
2418 let combined = bounds(Bound::Included(key(b"a")), Bound::Excluded(key(b"e")));
2419 assert_eq!(lower.union_adjacent(&upper).unwrap(), combined);
2421 assert_eq!(upper.union_adjacent(&lower).unwrap(), combined);
2422
2423 let overlapping = bounds(Bound::Included(key(b"a")), Bound::Included(key(b"c")));
2425 assert!(lower.union_adjacent(&overlapping).is_none());
2426 let gapped = bounds(Bound::Excluded(key(b"c")), Bound::Excluded(key(b"e")));
2427 assert!(lower.union_adjacent(&gapped).is_none());
2428 assert!(PartitionBounds::unbounded()
2429 .union_adjacent(&lower)
2430 .is_none());
2431
2432 let left = bounds(Bound::Unbounded, Bound::Excluded(key(b"c")));
2434 assert_eq!(
2435 left.union_adjacent(&upper).unwrap(),
2436 bounds(Bound::Unbounded, Bound::Excluded(key(b"e")))
2437 );
2438 }
2439
2440 #[test]
2443 fn encoded_keys_preserve_typed_order() {
2444 let ints: Vec<Key> = [i64::MIN, -7, -1, 0, 1, 42, i64::MAX]
2445 .into_iter()
2446 .map(int_key)
2447 .collect();
2448 let mut shuffled = ints.clone();
2449 shuffled.reverse();
2450 shuffled.sort();
2451 assert_eq!(shuffled, ints);
2452
2453 let mut texts = [
2455 text_key("aa"),
2456 text_key("a"),
2457 RowKeyEncoder::encode_key(&[KeyValue::Text("a\0".to_owned())]),
2458 text_key("b"),
2459 ];
2460 texts.sort();
2461 assert_eq!(texts[0], text_key("a"));
2462 assert_eq!(texts[2], text_key("aa"));
2463
2464 let values = [
2466 KeyValue::Null,
2467 KeyValue::Bool(true),
2468 KeyValue::Int(1),
2469 KeyValue::TimestampMicros(1),
2470 KeyValue::Text("x".to_owned()),
2471 KeyValue::Bytes(vec![0xFF]),
2472 ];
2473 let mut cross: Vec<Key> = values
2474 .iter()
2475 .map(|value| RowKeyEncoder::encode_key(std::slice::from_ref(value)))
2476 .collect();
2477 let sorted = cross.clone();
2478 cross.reverse();
2479 cross.sort();
2480 assert_eq!(cross, sorted);
2481 }
2482
2483 #[test]
2484 fn encoded_keys_decode_back_to_their_components() {
2485 let values = vec![
2486 KeyValue::Null,
2487 KeyValue::Bool(true),
2488 KeyValue::Int(-42),
2489 KeyValue::TimestampMicros(1_700_000_000_000_000),
2490 KeyValue::Text("tenant\0-42".to_owned()),
2491 KeyValue::Bytes(vec![0x00, 0x01, 0xFF]),
2492 ];
2493 let encoded = RowKeyEncoder::encode_key(&values);
2494 assert_eq!(RowKeyEncoder::decode_components(&encoded).unwrap(), values);
2495
2496 assert!(matches!(
2498 RowKeyEncoder::decode_components(&key(&[0x7F])),
2499 Err(PartitionError::MalformedKey(_))
2500 ));
2501 assert!(matches!(
2502 RowKeyEncoder::decode_components(&key(&[TAG_TEXT, b'a'])),
2503 Err(PartitionError::MalformedKey(_))
2504 ));
2505 }
2506
2507 #[test]
2510 fn partitioning_validation_fails_closed_on_bad_declarations() {
2511 let no_columns = Partitioning::Hash {
2512 columns: vec![],
2513 buckets: 16,
2514 };
2515 assert!(matches!(
2516 no_columns.validate(),
2517 Err(PartitionError::InvalidPartitioning(_))
2518 ));
2519 let no_buckets = Partitioning::Hash {
2520 columns: vec![ColumnId(1)],
2521 buckets: 0,
2522 };
2523 assert!(no_buckets.validate().is_err());
2524 let unsorted = Partitioning::Range {
2525 columns: vec![ColumnId(1)],
2526 splits: vec![int_key(20), int_key(10)],
2527 };
2528 assert!(unsorted.validate().is_err());
2529 let no_tenant_buckets = Partitioning::Tenant {
2530 tenant_column: ColumnId(1),
2531 buckets_per_tenant: 0,
2532 };
2533 assert!(no_tenant_buckets.validate().is_err());
2534 assert!(TimeInterval::micros(0).is_err());
2535 assert!(TimeInterval::days(1).is_ok());
2536 }
2537
2538 #[test]
2539 fn partition_key_extracts_declared_columns_in_declared_order() {
2540 let partitioning = Partitioning::Hash {
2541 columns: vec![ColumnId(5), ColumnId(2)],
2542 buckets: 16,
2543 };
2544 let mut values = BTreeMap::new();
2545 values.insert(ColumnId(2), KeyValue::Int(2));
2546 values.insert(ColumnId(5), KeyValue::Text("tenant".to_owned()));
2547 values.insert(ColumnId(9), KeyValue::Bool(true)); let key = partitioning.partition_key(&values).unwrap();
2549 assert_eq!(
2551 RowKeyEncoder::decode_components(&key).unwrap(),
2552 vec![KeyValue::Text("tenant".to_owned()), KeyValue::Int(2)]
2553 );
2554
2555 let mut incomplete = BTreeMap::new();
2557 incomplete.insert(ColumnId(2), KeyValue::Int(2));
2558 assert_eq!(
2559 partitioning.partition_key(&incomplete).unwrap_err(),
2560 PartitionError::MissingPartitionColumn {
2561 column: ColumnId(5)
2562 }
2563 );
2564 }
2565
2566 #[test]
2569 fn hash_route_is_deterministic_and_bounded() {
2570 let partitioning = Partitioning::Hash {
2571 columns: vec![ColumnId(1)],
2572 buckets: 16,
2573 };
2574 let mut values = BTreeMap::new();
2575 values.insert(ColumnId(1), KeyValue::Int(42));
2576 let key = partitioning.partition_key(&values).unwrap();
2577 let slot = partitioning.route(&key).unwrap();
2578 assert_eq!(slot, partitioning.route(&key).unwrap());
2579 assert!(slot < 16);
2580
2581 let buckets: std::collections::BTreeSet<u64> = (0..100)
2583 .map(|i| {
2584 let key = RowKeyEncoder::encode_key(&[KeyValue::Int(i)]);
2585 partitioning.route(&key).unwrap()
2586 })
2587 .collect();
2588 assert!(buckets.len() > 8, "poor spread: {buckets:?}");
2589 }
2590
2591 #[test]
2592 fn range_route_indexes_the_split_points() {
2593 let partitioning = Partitioning::Range {
2594 columns: vec![ColumnId(1)],
2595 splits: vec![int_key(10), int_key(20)],
2596 };
2597 let route = |value: i64| partitioning.route(&int_key(value)).unwrap();
2598 assert_eq!(route(i64::MIN), 0);
2599 assert_eq!(route(9), 0);
2600 assert_eq!(route(10), 1); assert_eq!(route(19), 1);
2602 assert_eq!(route(20), 2);
2603 assert_eq!(route(i64::MAX), 2);
2604
2605 let zero = partitioning.range_bounds(0).unwrap();
2607 let one = partitioning.range_bounds(1).unwrap();
2608 let two = partitioning.range_bounds(2).unwrap();
2609 assert!(partitioning.range_bounds(3).is_none());
2610 assert!(zero.is_adjacent_to(&one));
2611 assert!(one.is_adjacent_to(&two));
2612 assert!(!zero.is_adjacent_to(&two));
2613 assert!(zero.contains(&int_key(9)));
2614 assert!(!zero.contains(&int_key(10)));
2615 assert!(two.contains(&int_key(20)));
2616 assert!(matches!(zero.low, Bound::Unbounded));
2617 assert!(matches!(two.high, Bound::Unbounded));
2618 }
2619
2620 #[test]
2621 fn tenant_route_is_per_tenant_deterministic_and_bounded() {
2622 let partitioning = Partitioning::Tenant {
2623 tenant_column: ColumnId(1),
2624 buckets_per_tenant: 8,
2625 };
2626 let key_for = |tenant: &str| {
2627 let mut values = BTreeMap::new();
2628 values.insert(ColumnId(1), KeyValue::Text(tenant.to_owned()));
2629 partitioning.partition_key(&values).unwrap()
2630 };
2631 let acme = key_for("acme");
2632 let slot = partitioning.route(&acme).unwrap();
2633 assert_eq!(slot, partitioning.route(&acme).unwrap());
2634 assert!(slot < 8);
2635 let routed = partitioning.routed_key(&acme).unwrap();
2637 let mut expected = acme.as_bytes().to_vec();
2638 expected.extend_from_slice(&slot.to_be_bytes());
2639 assert_eq!(routed, key(&expected));
2640
2641 let initech = key_for("initech");
2642 assert_eq!(
2643 partitioning.route(&initech).unwrap(),
2644 partitioning.route(&key_for("initech")).unwrap()
2645 );
2646 }
2647
2648 #[test]
2649 fn time_range_route_buckets_by_interval() {
2650 let interval = TimeInterval::hours(1).unwrap();
2651 let partitioning = Partitioning::TimeRange {
2652 timestamp_column: ColumnId(3),
2653 interval,
2654 };
2655 let key_for = |micros: i64| {
2656 let mut values = BTreeMap::new();
2657 values.insert(ColumnId(3), KeyValue::TimestampMicros(micros));
2658 partitioning.partition_key(&values).unwrap()
2659 };
2660 let width = i64::try_from(interval.as_micros()).unwrap();
2661 assert_eq!(partitioning.route(&key_for(0)).unwrap(), 0);
2662 assert_eq!(partitioning.route(&key_for(width - 1)).unwrap(), 0);
2663 assert_eq!(partitioning.route(&key_for(width)).unwrap(), 1);
2664 assert_eq!(partitioning.route(&key_for(5 * width / 2)).unwrap(), 2);
2665
2666 assert!(matches!(
2668 partitioning.route(&key_for(-1)),
2669 Err(PartitionError::NegativeSlot { micros: -1 })
2670 ));
2671
2672 let mut wrong_type = BTreeMap::new();
2674 wrong_type.insert(ColumnId(3), KeyValue::Int(0));
2675 assert_eq!(
2676 partitioning.partition_key(&wrong_type).unwrap_err(),
2677 PartitionError::PartitionColumnType {
2678 column: ColumnId(3),
2679 expected: "timestamp-micros",
2680 found: "int",
2681 }
2682 );
2683 }
2684
2685 #[test]
2688 fn table_partitioning_record_round_trips_with_colocation() {
2689 let record = TablePartitioningRecord {
2690 table_id: TableId::new(4),
2691 partitioning: Partitioning::Range {
2692 columns: vec![ColumnId(1)],
2693 splits: vec![int_key(100)],
2694 },
2695 origin: PartitioningOrigin::Declared,
2696 colocated_with: Some(ColocatedWith(TableId::new(2))),
2697 };
2698 record.validate().unwrap();
2699 let json = serde_json::to_vec(&record).unwrap();
2700 let back: TablePartitioningRecord = serde_json::from_slice(&json).unwrap();
2701 assert_eq!(back, record);
2702 }
2703
2704 #[test]
2705 fn automatic_defaults_are_visible_in_schema_metadata() {
2706 let record =
2707 TablePartitioningRecord::automatic_default(TableId::new(4), vec![ColumnId(1)], 64);
2708 assert_eq!(record.origin, PartitioningOrigin::AutomaticDefault);
2709 assert_eq!(record.partitioning.partition_columns(), vec![ColumnId(1)]);
2710 assert_eq!(record.colocated_with, None);
2711 record.validate().unwrap();
2712
2713 let json = serde_json::to_value(&record).unwrap();
2715 assert_eq!(json["origin"], serde_json::json!("AutomaticDefault"));
2716 }
2717
2718 #[test]
2719 fn record_validation_rejects_zero_table_and_self_colocation() {
2720 let mut zero =
2721 TablePartitioningRecord::automatic_default(TableId::ZERO, vec![ColumnId(1)], 4);
2722 assert!(matches!(
2723 zero.validate(),
2724 Err(TabletError::InvalidDescriptor(_))
2725 ));
2726 zero.table_id = TableId::new(4);
2727 zero.colocated_with = Some(ColocatedWith(TableId::new(4)));
2728 assert!(matches!(
2729 zero.validate(),
2730 Err(TabletError::InvalidDescriptor(_))
2731 ));
2732 }
2733
2734 fn routed_hash_tablet(byte: u8, state: TabletState, slots: (u64, u64)) -> TabletDescriptor {
2737 let mut tablet = descriptor(state);
2738 tablet.tablet_id = tablet_id(byte);
2739 tablet.table_id = TableId::new(5);
2740 tablet.partition = hash_slot_bounds(slots.0, slots.1);
2741 tablet
2742 }
2743
2744 #[test]
2745 fn point_reads_and_writes_route_directly() {
2746 let partitioning = Partitioning::Hash {
2747 columns: vec![ColumnId(1)],
2748 buckets: 4,
2749 };
2750 let tablets = vec![
2751 routed_hash_tablet(1, TabletState::Active, (0, 2)),
2752 routed_hash_tablet(2, TabletState::Active, (2, 4)),
2753 ];
2754 for value in 0..50 {
2755 let key = RowKeyEncoder::encode_key(&[KeyValue::Int(value)]);
2756 let routed = partitioning.routed_key(&key).unwrap();
2757 let slot = partitioning.route(&key).unwrap();
2758 let expected = tablet_id(if slot < 2 { 1 } else { 2 });
2759 assert_eq!(
2760 find_tablet_for_key(&tablets, TableId::new(5), &routed).map(|t| t.tablet_id),
2761 Some(expected),
2762 "value {value} routed wrong"
2763 );
2764 assert!(find_tablet_for_key(&tablets, TableId::new(6), &routed).is_none());
2766 }
2767 let children = vec![
2769 routed_hash_tablet(1, TabletState::Active, (0, 2)),
2770 routed_hash_tablet(3, TabletState::Creating, (0, 1)),
2771 ];
2772 let routed = partitioning
2773 .routed_key(&RowKeyEncoder::encode_key(&[KeyValue::Int(1)]))
2774 .unwrap();
2775 let slot = partitioning
2776 .route(&RowKeyEncoder::encode_key(&[KeyValue::Int(1)]))
2777 .unwrap();
2778 if slot < 2 {
2779 assert_eq!(
2780 find_tablet_for_key(&children, TableId::new(5), &routed).map(|t| t.tablet_id),
2781 Some(tablet_id(1))
2782 );
2783 }
2784 }
2785
2786 #[test]
2787 fn range_queries_route_to_all_overlapping_tablets_in_order() {
2788 let t = |byte: u8, low: Bound<Key>, high: Bound<Key>| {
2789 let mut tablet = descriptor(TabletState::Active);
2790 tablet.tablet_id = tablet_id(byte);
2791 tablet.table_id = TableId::new(5);
2792 tablet.partition = bounds(low, high);
2793 tablet
2794 };
2795 let tablets = vec![
2796 t(3, Bound::Included(text_key("m")), Bound::Unbounded),
2797 t(1, Bound::Unbounded, Bound::Excluded(text_key("e"))),
2798 t(
2799 2,
2800 Bound::Included(text_key("e")),
2801 Bound::Excluded(text_key("m")),
2802 ),
2803 {
2804 let mut retired = t(4, Bound::Unbounded, Bound::Excluded(text_key("e")));
2805 retired.state = TabletState::Retired;
2806 retired
2807 },
2808 ];
2809 let narrow = bounds(
2811 Bound::Included(text_key("f")),
2812 Bound::Excluded(text_key("g")),
2813 );
2814 assert_eq!(
2815 tablets_overlapping(&tablets, TableId::new(5), &narrow)
2816 .iter()
2817 .map(|tablet| tablet.tablet_id)
2818 .collect::<Vec<_>>(),
2819 vec![tablet_id(2)]
2820 );
2821 assert_eq!(
2823 tablets_overlapping(&tablets, TableId::new(5), &PartitionBounds::unbounded())
2824 .iter()
2825 .map(|tablet| tablet.tablet_id)
2826 .collect::<Vec<_>>(),
2827 vec![tablet_id(1), tablet_id(2), tablet_id(3)]
2828 );
2829 }
2830
2831 #[test]
2832 fn stale_generations_classify_per_tablet_state() {
2833 let tablet = descriptor(TabletState::Active);
2834 assert!(check_generation(&tablet, 7).is_ok());
2835
2836 let mut splitting = tablet.clone();
2837 splitting.state = TabletState::Splitting;
2838 let error = check_generation(&splitting, 6).unwrap_err();
2839 assert_eq!(
2840 error,
2841 RoutingError::TabletSplit {
2842 tablet_id: tablet_id(9),
2843 used_generation: 6,
2844 current_generation: 7,
2845 }
2846 );
2847 assert_eq!(error.category(), ErrorCategory::TabletSplitting);
2848
2849 let mut retiring = tablet.clone();
2850 retiring.state = TabletState::Retiring;
2851 let error = check_generation(&retiring, 6).unwrap_err();
2852 assert!(matches!(error, RoutingError::TabletMoved { .. }));
2853 assert_eq!(error.category(), ErrorCategory::TabletMoved);
2854
2855 let error = check_generation(&tablet, 6).unwrap_err();
2858 assert!(matches!(error, RoutingError::StaleMetadata { .. }));
2859 assert_eq!(error.category(), ErrorCategory::StaleMetadata);
2860 let error = check_generation(&tablet, 8).unwrap_err();
2861 assert!(matches!(error, RoutingError::StaleMetadata { .. }));
2862
2863 let error = check_generation(&splitting, 9).unwrap_err();
2866 assert!(
2867 matches!(error, RoutingError::StaleMetadata { .. }),
2868 "got {error:?}"
2869 );
2870 }
2871
2872 #[test]
2873 fn list_tablets_on_disk_reads_real_metadata() {
2874 let dir = tempfile::tempdir().unwrap();
2875 let desc = descriptor(TabletState::Active);
2876 let layout = TabletLayout::new(dir.path(), desc.tablet_id, desc.raft_group_id);
2877 layout.create(&desc).unwrap();
2878 let (listed, issues) = list_tablets_on_disk(dir.path()).unwrap();
2879 assert!(issues.is_empty());
2880 assert_eq!(listed.len(), 1);
2881 assert_eq!(listed[0].tablet_id, desc.tablet_id);
2882 assert_eq!(listed[0].replicas.len(), desc.replicas.len());
2883 }
2884
2885 fn layout_fixture(root: &Path) -> (TabletLayout, TabletDescriptor) {
2888 let descriptor = descriptor(TabletState::Active);
2889 let layout = TabletLayout::new(root, descriptor.tablet_id, descriptor.raft_group_id);
2890 (layout, descriptor)
2891 }
2892
2893 #[test]
2894 fn layout_create_makes_the_spec_directory_tree_and_metadata() {
2895 let dir = tempfile::tempdir().unwrap();
2896 let (layout, descriptor) = layout_fixture(dir.path());
2897 layout.create(&descriptor).unwrap();
2898
2899 for path in [
2900 layout.state_dir(),
2901 layout.runs_dir(),
2902 layout.indexes_dir(),
2903 layout.temp_dir(),
2904 layout.raft_dir(),
2905 layout.snapshots_dir(),
2906 ] {
2907 assert!(path.is_dir(), "missing {}", path.display());
2908 }
2909 assert!(layout.metadata_path().is_file());
2910 assert_eq!(
2912 layout.tablet_dir(),
2913 dir.path()
2914 .join(TABLETS_DIR)
2915 .join(descriptor.tablet_id.to_hex())
2916 );
2917 assert_eq!(
2918 layout.group_dir(),
2919 dir.path()
2920 .join(GROUPS_DIR)
2921 .join(descriptor.raft_group_id.to_hex())
2922 );
2923
2924 assert_eq!(layout.validate().unwrap(), descriptor);
2926 assert_eq!(layout.load_metadata().unwrap(), descriptor);
2927 }
2928
2929 #[test]
2930 fn layout_create_is_idempotent_but_never_repurposes_a_directory() {
2931 let dir = tempfile::tempdir().unwrap();
2932 let (layout, descriptor) = layout_fixture(dir.path());
2933 layout.create(&descriptor).unwrap();
2934 layout.create(&descriptor).unwrap();
2936 let mut other = descriptor.clone();
2938 other.generation = 8;
2939 assert!(matches!(
2940 layout.create(&other),
2941 Err(TabletError::MetadataConflict(_))
2942 ));
2943 }
2944
2945 #[test]
2946 fn layout_store_metadata_advances_the_descriptor_atomically() {
2947 let dir = tempfile::tempdir().unwrap();
2948 let (layout, descriptor) = layout_fixture(dir.path());
2949 layout.create(&descriptor).unwrap();
2950 let mut advanced = descriptor.clone();
2951 advanced.generation = 8;
2952 advanced.state = TabletState::Splitting;
2953 layout.store_metadata(&advanced).unwrap();
2954 assert_eq!(layout.load_metadata().unwrap(), advanced);
2955
2956 let mut foreign = advanced.clone();
2958 foreign.tablet_id = tablet_id(8);
2959 assert!(matches!(
2960 layout.store_metadata(&foreign),
2961 Err(TabletError::TabletMismatch { .. })
2962 ));
2963 }
2964
2965 #[test]
2966 fn layout_open_fails_closed_on_missing_or_corrupt_metadata() {
2967 let dir = tempfile::tempdir().unwrap();
2968 let (layout, descriptor) = layout_fixture(dir.path());
2969
2970 assert!(matches!(
2972 layout.load_metadata(),
2973 Err(TabletError::MissingMetadata(_))
2974 ));
2975 layout.create(&descriptor).unwrap();
2976
2977 std::fs::write(layout.metadata_path(), b"{ not json").unwrap();
2979 assert!(matches!(
2980 layout.load_metadata(),
2981 Err(TabletError::Metadata(ClusterError::CorruptMetadata { .. }))
2982 ));
2983
2984 layout.store_metadata(&descriptor).unwrap();
2986 let mut value: serde_json::Value =
2987 serde_json::from_slice(&std::fs::read(layout.metadata_path()).unwrap()).unwrap();
2988 value["format_version"] = serde_json::json!(99);
2989 std::fs::write(layout.metadata_path(), serde_json::to_vec(&value).unwrap()).unwrap();
2990 assert!(matches!(
2991 layout.load_metadata(),
2992 Err(TabletError::Metadata(
2993 ClusterError::UnsupportedFormatVersion { found: 99, .. }
2994 ))
2995 ));
2996 }
2997
2998 #[test]
2999 fn layout_open_fails_closed_on_checksum_or_identity_drift() {
3000 let dir = tempfile::tempdir().unwrap();
3001 let (layout, descriptor) = layout_fixture(dir.path());
3002 layout.create(&descriptor).unwrap();
3003
3004 let path = layout.metadata_path();
3006 let mut value: serde_json::Value =
3007 serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
3008 value["tablet"]["generation"] = serde_json::json!(8);
3009 std::fs::write(&path, serde_json::to_vec(&value).unwrap()).unwrap();
3010 assert!(matches!(
3011 layout.load_metadata(),
3012 Err(TabletError::Metadata(ClusterError::CorruptMetadata { .. }))
3013 ));
3014
3015 let other_dir = tempfile::tempdir().unwrap();
3018 let mut foreign = descriptor.clone();
3019 foreign.tablet_id = tablet_id(8);
3020 let foreign_layout =
3021 TabletLayout::new(other_dir.path(), foreign.tablet_id, foreign.raft_group_id);
3022 foreign_layout.create(&foreign).unwrap();
3023 std::fs::write(
3024 &path,
3025 std::fs::read(foreign_layout.metadata_path()).unwrap(),
3026 )
3027 .unwrap();
3028 assert!(matches!(
3029 layout.load_metadata(),
3030 Err(TabletError::TabletMismatch { .. })
3031 ));
3032 }
3033
3034 #[test]
3035 fn layout_validate_requires_the_full_directory_tree() {
3036 let dir = tempfile::tempdir().unwrap();
3037 let (layout, descriptor) = layout_fixture(dir.path());
3038 layout.create(&descriptor).unwrap();
3039 layout.validate().unwrap();
3040
3041 std::fs::remove_dir(layout.runs_dir()).unwrap();
3042 assert!(matches!(
3043 layout.validate(),
3044 Err(TabletError::Metadata(ClusterError::CorruptMetadata { .. }))
3045 ));
3046 }
3047
3048 #[test]
3049 fn teardown_removes_the_replica_idempotently_but_never_foreign_state() {
3050 let dir = tempfile::tempdir().unwrap();
3051 let (layout, descriptor) = layout_fixture(dir.path());
3052 layout.create(&descriptor).unwrap();
3053 assert!(layout.tablet_dir().is_dir() && layout.group_dir().is_dir());
3054
3055 layout.teardown().unwrap();
3056 assert!(!layout.tablet_dir().exists());
3057 assert!(!layout.group_dir().exists());
3058 layout.teardown().unwrap();
3060
3061 let other_dir = tempfile::tempdir().unwrap();
3063 let mut foreign = descriptor.clone();
3064 foreign.tablet_id = tablet_id(8);
3065 let foreign_layout =
3066 TabletLayout::new(other_dir.path(), foreign.tablet_id, foreign.raft_group_id);
3067 foreign_layout.create(&foreign).unwrap();
3068 layout.create(&descriptor).unwrap();
3069 std::fs::write(
3070 layout.metadata_path(),
3071 std::fs::read(foreign_layout.metadata_path()).unwrap(),
3072 )
3073 .unwrap();
3074 assert!(matches!(
3075 layout.teardown(),
3076 Err(TabletError::TabletMismatch { .. })
3077 ));
3078 assert!(layout.tablet_dir().is_dir(), "foreign state was deleted");
3079 }
3080
3081 #[test]
3084 fn one_tablet_storage_core_is_owned_by_one_process() {
3085 let dir = tempfile::tempdir().unwrap();
3086 let (layout, descriptor) = layout_fixture(dir.path());
3087 layout.create(&descriptor).unwrap();
3088
3089 let registry = TabletOwnershipRegistry::new();
3090 let guard = registry.try_reserve(&layout).unwrap();
3091 assert_eq!(guard.tablet_id(), descriptor.tablet_id);
3092 assert_eq!(registry.len(), 1);
3093
3094 let error = registry.try_reserve(&layout).unwrap_err();
3096 assert!(matches!(
3097 error,
3098 TabletError::AlreadyOwned { tablet, .. } if tablet == descriptor.tablet_id
3099 ));
3100
3101 let mut distinct = descriptor.clone();
3103 distinct.tablet_id = tablet_id(10);
3104 let distinct_layout =
3105 TabletLayout::new(dir.path(), distinct.tablet_id, distinct.raft_group_id);
3106 distinct_layout.create(&distinct).unwrap();
3107 let _distinct_guard = registry.try_reserve(&distinct_layout).unwrap();
3108 assert_eq!(registry.len(), 2);
3109
3110 drop(guard);
3112 assert_eq!(registry.len(), 1);
3113 let _reacquired = registry.try_reserve(&layout).unwrap();
3114 assert_eq!(registry.len(), 2);
3115 }
3116
3117 #[test]
3118 fn ownership_reservation_keys_on_the_canonical_path() {
3119 let dir = tempfile::tempdir().unwrap();
3120 let (layout, descriptor) = layout_fixture(dir.path());
3121 layout.create(&descriptor).unwrap();
3122
3123 let registry = TabletOwnershipRegistry::new();
3124 let _guard = registry.try_reserve(&layout).unwrap();
3125 let aliased_root = dir.path().join(".");
3128 let aliased =
3129 TabletLayout::new(aliased_root, descriptor.tablet_id, descriptor.raft_group_id);
3130 assert!(matches!(
3131 registry.try_reserve(&aliased),
3132 Err(TabletError::AlreadyOwned { .. })
3133 ));
3134 }
3135}