dbx_core/sharding/
router.rs

1//! Shard 라우터 — FNV1a 해시 기반 키→샤드 매핑
2
3/// FNV-1a 32-bit 해시 (결정론적, 빠른 비암호학적 해시)
4pub fn fnv1a_hash(data: &[u8]) -> u64 {
5    const FNV_OFFSET: u64 = 14_695_981_039_346_656_037;
6    const FNV_PRIME: u64 = 1_099_511_628_211;
7    let mut hash = FNV_OFFSET;
8    for &byte in data {
9        hash ^= byte as u64;
10        hash = hash.wrapping_mul(FNV_PRIME);
11    }
12    hash
13}
14
15/// 샤드 노드 정보
16#[derive(Debug, Clone, PartialEq)]
17pub struct ShardNode {
18    /// 고유 샤드 ID (0-based)
19    pub id: usize,
20    /// 연결 주소 (예: "10.0.0.1:5432")
21    pub address: String,
22    /// 노드 가중치 — 1.0이 기본
23    /// 가중치가 높을수록 vnode가 더 많이 배정되어 더 많은 데이터를 담당
24    pub weight: f64,
25}
26
27use crate::sharding::node_ring::NodeRing;
28
29/// 샤드 라우터
30///
31/// 키를 받아 어느 ShardNode로 라우팅할지 결정합니다.
32/// 일관된 해싱(Consistent Hashing) 링을 통해 노드 간 부하 분산을 수행합니다.
33#[derive(Debug)]
34pub struct ShardRouter {
35    shards: Vec<ShardNode>,
36    ring: NodeRing,
37}
38
39impl ShardRouter {
40    /// n개의 로컬 샤드 라우터 생성 (테스트/개발용)
41    pub fn new_local(n: usize) -> Self {
42        let shards: Vec<ShardNode> = (0..n)
43            .map(|i| ShardNode {
44                id: i,
45                address: format!("127.0.0.1:{}", 5000 + i),
46                weight: 1.0, // 기본 가중치
47            })
48            .collect();
49
50        let mut ring = NodeRing::new(100);
51        for s in &shards {
52            ring.add_node(s);
53        }
54
55        Self { shards, ring }
56    }
57
58    /// 커스텀 노드 목록으로 라우터 생성
59    pub fn new(shards: Vec<ShardNode>) -> Self {
60        use crate::sharding::node_ring::NodeRing;
61        let mut ring = NodeRing::new(100);
62        for s in &shards {
63            ring.add_node(s);
64        }
65        Self { shards, ring }
66    }
67
68    /// 노드 주소 목록으로 라우터 생성
69    pub fn new_with_addresses(addresses: Vec<String>) -> Self {
70        let shards: Vec<ShardNode> = addresses
71            .into_iter()
72            .enumerate()
73            .map(|(i, addr)| ShardNode {
74                id: i,
75                address: addr,
76                weight: 1.0,
77            })
78            .collect();
79        Self::new(shards)
80    }
81
82    /// 샤드 수
83    pub fn num_shards(&self) -> usize {
84        self.shards.len()
85    }
86
87    /// key 바이트로부터 물리 샤드 노드 ID 인덱스 결정 (Consistent Hashing)
88    pub fn shard_index(&self, key: &[u8]) -> usize {
89        if self.shards.is_empty() {
90            return 0;
91        }
92        let hash = fnv1a_hash(key);
93        self.ring.get_node(hash).unwrap_or(0)
94    }
95
96    /// key로부터 담당 ShardNode 반환
97    pub fn route(&self, key: &[u8]) -> Option<&ShardNode> {
98        if self.shards.is_empty() {
99            return None;
100        }
101        let idx = self.shard_index(key);
102        Some(&self.shards[idx])
103    }
104
105    /// 모든 샤드 목록 반환 (scatter-gather broadcast 용)
106    pub fn all_shards(&self) -> &[ShardNode] {
107        &self.shards
108    }
109
110    /// sub-table 이름 생성: `{base_table}__shard_{idx}`
111    pub fn sub_table_name(&self, base_table: &str, key: &[u8]) -> String {
112        let idx = self.shard_index(key);
113        format!("{}__shard_{}", base_table, idx)
114    }
115}
116
117#[cfg(test)]
118mod tests {
119    use super::*;
120
121    #[test]
122    fn test_shard_routing_deterministic() {
123        let router = ShardRouter::new_local(4);
124        let idx1 = router.shard_index(b"user:42");
125        let idx2 = router.shard_index(b"user:42");
126        assert_eq!(idx1, idx2, "동일 키는 항상 같은 샤드");
127    }
128
129    #[test]
130    fn test_shard_index_in_range() {
131        let router = ShardRouter::new_local(8);
132        for i in 0u64..200 {
133            let key = format!("row:{}", i);
134            let idx = router.shard_index(key.as_bytes());
135            assert!(idx < 8, "인덱스 {}가 샤드 수 이상", idx);
136        }
137    }
138
139    #[test]
140    fn test_route_returns_node() {
141        let router = ShardRouter::new_local(4);
142        let node = router.route(b"my_key").unwrap();
143        assert!(node.address.starts_with("127.0.0.1:"), "로컬 주소");
144    }
145
146    #[test]
147    fn test_sub_table_name() {
148        let router = ShardRouter::new_local(4);
149        let name = router.sub_table_name("orders", b"order:1");
150        assert!(name.starts_with("orders__shard_"));
151        let idx: usize = name.split('_').last().unwrap().parse().unwrap();
152        assert!(idx < 4);
153    }
154
155    #[test]
156    fn test_num_shards() {
157        let router = ShardRouter::new_local(6);
158        assert_eq!(router.num_shards(), 6);
159    }
160
161    #[test]
162    fn test_all_shards() {
163        let router = ShardRouter::new_local(3);
164        assert_eq!(router.all_shards().len(), 3);
165    }
166}
dbx_core/sharding/router.rs

dbx_core/sharding/
router.rs
