diff --git a/ROADMAP.md b/ROADMAP.md index 451cb19..073215c 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -78,12 +78,12 @@ - [x] Array литерали - [x] Типов анализатор (type checker) - [x] IR (Intermediate Representation) -- [ ] Оптимизатор на заявки (predicate pushdown, projection pushdown) +- [x] Оптимизатор на заявки (predicate pushdown, projection pushdown) +- [x] GROUP BY, HAVING +- [x] JOIN (inner, left, right, full, cross) +- [x] CTE (WITH) +- [x] Агрегатни функции (count, sum, avg, min, max) - [ ] Codegen → storage операции -- [ ] GROUP BY, HAVING -- [ ] JOIN (inner, left, right, full) -- [ ] CTE (WITH) -- [ ] Агрегатни функции (count, sum, avg, min, max) - [ ] Потребителски функции (UDF) ### Фаза 3: Мултимодален storage 🟡 @@ -170,11 +170,11 @@ - [ ] Interactive query editor с autocomplete - [ ] Import/Export (JSON, CSV, Parquet) -### Фаза 11: Кластеризация и разпределение ⬜ -- [ ] Raft консенсус протокол -- [ ] Sharding (hash-based, range-based) +### Фаза 11: Кластеризация и разпределение 🟡 +- [x] Raft консенсус протокол (leader election + log replication) +- [x] Sharding (hash-based, range-based, consistent hashing) - [ ] Replication (sync, async) -- [ ] Leader election +- [ ] Leader election за multi-node - [ ] Gossip protocol за membership - [ ] Distributed transactions - [ ] Auto-rebalancing @@ -196,16 +196,16 @@ | Фаза | Статус | Напредък | |------|--------|----------| | 1. Ядро | ✅ Завършена | 95% | -| 2. BaraQL | 🟡 В процес | 60% | +| 2. BaraQL | ✅ Основно завършена | 85% | | 3. Мултимодален storage | 🟡 В процес | 75% | | 4. Транзакции | ✅ Основно завършена | 85% | -| 5. Протокол | 🟡 В процес | 85% | +| 5. Протокол | ✅ Основно завършена | 85% | | 6. Schema | ✅ Основно завършена | 75% | | 7. Векторен engine | ✅ Завършена | 95% | | 8. Graph engine | ✅ Завършена | 90% | | 9. FTS | ✅ Завършена | 85% | | 10. Клиенти и CLI | 🟡 В процес | 50% | -| 11. Кластер | ⬜ Не стартирана | 0% | +| 11. Кластер | 🟡 В процес | 30% | | 12. Оптимизации | ⬜ Не стартирана | 0% | **Легенда:** ⬜ Не стартирана | 🟡 В процес | ✅ Завършена diff --git a/src/barabadb/core/raft.nim b/src/barabadb/core/raft.nim new file mode 100644 index 0000000..93ada6a --- /dev/null +++ b/src/barabadb/core/raft.nim @@ -0,0 +1,280 @@ +## Raft Consensus — leader election + log replication +import std/tables +import std/sets +import std/deques +import std/algorithm +import std/random + +type + RaftState* = enum + rsFollower + rsCandidate + rsLeader + + LogEntry* = object + term*: uint64 + index*: uint64 + command*: string + data*: seq[byte] + + RaftNode* = ref object + id*: string + state*: RaftState + currentTerm*: uint64 + votedFor*: string + log*: seq[LogEntry] + commitIndex*: uint64 + lastApplied*: uint64 + # Leader state + nextIndex*: Table[string, uint64] + matchIndex*: Table[string, uint64] + # Cluster + peers*: seq[string] + leaderId*: string + # Timing + electionTimeout*: int + heartbeatTimeout*: int + votesReceived*: HashSet[string] + + RaftMessageKind* = enum + rmkRequestVote + rmkRequestVoteReply + rmkAppendEntries + rmkAppendEntriesReply + + RaftMessage* = object + kind*: RaftMessageKind + term*: uint64 + senderId*: string + # RequestVote + lastLogIndex*: uint64 + lastLogTerm*: uint64 + # AppendEntries + prevLogIndex*: uint64 + prevLogTerm*: uint64 + entries*: seq[LogEntry] + leaderCommit*: uint64 + # Reply + success*: bool + matchIdx*: uint64 + + RaftCluster* = ref object + nodes*: Table[string, RaftNode] + messageQueue*: Deque[RaftMessage] + +proc newRaftNode*(id: string, peers: seq[string]): RaftNode = + RaftNode( + id: id, + state: rsFollower, + currentTerm: 0, + votedFor: "", + log: @[], + commitIndex: 0, + lastApplied: 0, + nextIndex: initTable[string, uint64](), + matchIndex: initTable[string, uint64](), + peers: peers, + leaderId: "", + electionTimeout: 150 + rand(150), + heartbeatTimeout: 50, + votesReceived: initHashSet[string](), + ) + +proc newRaftCluster*(): RaftCluster = + RaftCluster( + nodes: initTable[string, RaftNode](), + messageQueue: initDeque[RaftMessage](), + ) + +proc addNode*(cluster: RaftCluster, id: string) = + var peers: seq[string] = @[] + for existingId in cluster.nodes.keys: + peers.add(existingId) + cluster.nodes[existingId].peers.add(id) + cluster.nodes[id] = newRaftNode(id, peers) + +proc lastLogIndex*(node: RaftNode): uint64 = + if node.log.len == 0: + return 0 + return node.log[^1].index + +proc lastLogTerm*(node: RaftNode): uint64 = + if node.log.len == 0: + return 0 + return node.log[^1].term + +proc becomeFollower*(node: RaftNode, term: uint64) = + node.state = rsFollower + node.currentTerm = term + node.votedFor = "" + node.votesReceived.clear() + +proc becomeCandidate*(node: RaftNode) = + node.state = rsCandidate + inc node.currentTerm + node.votedFor = node.id + node.votesReceived.clear() + node.votesReceived.incl(node.id) + +proc becomeLeader*(node: RaftNode) = + node.state = rsLeader + node.leaderId = node.id + for peer in node.peers: + node.nextIndex[peer] = node.lastLogIndex + 1 + node.matchIndex[peer] = 0 + +proc handleRequestVote*(node: RaftNode, msg: RaftMessage): RaftMessage = + var reply = RaftMessage( + kind: rmkRequestVoteReply, + term: node.currentTerm, + senderId: node.id, + success: false, + ) + + if msg.term < node.currentTerm: + return reply + + if msg.term > node.currentTerm: + node.becomeFollower(msg.term) + + let canVote = node.votedFor == "" or node.votedFor == msg.senderId + let logOk = msg.lastLogTerm > node.lastLogTerm or + (msg.lastLogTerm == node.lastLogTerm and msg.lastLogIndex >= node.lastLogIndex) + + if canVote and logOk: + node.votedFor = msg.senderId + reply.success = true + reply.term = node.currentTerm + + return reply + +proc handleAppendEntries*(node: RaftNode, msg: RaftMessage): RaftMessage = + var reply = RaftMessage( + kind: rmkAppendEntriesReply, + term: node.currentTerm, + senderId: node.id, + success: false, + matchIdx: 0, + ) + + if msg.term < node.currentTerm: + return reply + + if msg.term >= node.currentTerm: + node.becomeFollower(msg.term) + node.leaderId = msg.senderId + + # Check if log contains entry at prevLogIndex with prevLogTerm + if msg.prevLogIndex > 0: + if msg.prevLogIndex > uint64(node.log.len): + return reply + if node.log[msg.prevLogIndex - 1].term != msg.prevLogTerm: + # Delete conflicting entries + node.log.setLen(int(msg.prevLogIndex - 1)) + return reply + + # Append new entries + for entry in msg.entries: + let idx = int(entry.index - 1) + if idx < node.log.len: + if node.log[idx].term != entry.term: + node.log.setLen(idx) + node.log.add(entry) + else: + node.log.add(entry) + + # Update commit index + if msg.leaderCommit > node.commitIndex: + node.commitIndex = min(msg.leaderCommit, node.lastLogIndex) + + reply.success = true + reply.matchIdx = node.lastLogIndex + return reply + +proc requestVote*(node: RaftNode): seq[RaftMessage] = + result = @[] + for peer in node.peers: + result.add(RaftMessage( + kind: rmkRequestVote, + term: node.currentTerm, + senderId: node.id, + lastLogIndex: node.lastLogIndex, + lastLogTerm: node.lastLogTerm, + )) + +proc appendEntries*(node: RaftNode, peerId: string): RaftMessage = + let nextIdx = node.nextIndex.getOrDefault(peerId, node.lastLogIndex + 1) + let prevIdx = nextIdx - 1 + var prevTerm: uint64 = 0 + if prevIdx > 0 and prevIdx <= uint64(node.log.len): + prevTerm = node.log[prevIdx - 1].term + + var entries: seq[LogEntry] = @[] + for i in int(nextIdx - 1).. node.currentTerm: + node.becomeFollower(reply.term) + return + + if node.state != rsCandidate: + return + + if reply.success: + node.votesReceived.incl(reply.senderId) + if node.votesReceived.len > (node.peers.len + 1) div 2: + node.becomeLeader() + +proc handleAppendReply*(node: RaftNode, peerId: string, reply: RaftMessage) = + if reply.term > node.currentTerm: + node.becomeFollower(reply.term) + return + + if node.state != rsLeader: + return + + if reply.success: + node.matchIndex[peerId] = reply.matchIdx + node.nextIndex[peerId] = reply.matchIdx + 1 + + # Update commit index + var matchIndices: seq[uint64] = @[node.lastLogIndex] + for p, idx in node.matchIndex: + matchIndices.add(idx) + matchIndices.sort() + + let medianIdx = matchIndices[matchIndices.len div 2] + if medianIdx > node.commitIndex: + if medianIdx <= node.lastLogIndex and + node.log[medianIdx - 1].term == node.currentTerm: + node.commitIndex = medianIdx + else: + if node.nextIndex[peerId] > 1: + dec node.nextIndex[peerId] + +proc state*(node: RaftNode): RaftState = node.state +proc isLeader*(node: RaftNode): bool = node.state == rsLeader +proc leaderId*(node: RaftNode): string = node.leaderId +proc logLen*(node: RaftNode): int = node.log.len diff --git a/src/barabadb/core/sharding.nim b/src/barabadb/core/sharding.nim new file mode 100644 index 0000000..531dbc8 --- /dev/null +++ b/src/barabadb/core/sharding.nim @@ -0,0 +1,133 @@ +## Sharding — hash-based and range-based data distribution +import std/tables +import std/hashes +import std/algorithm +import std/sets + +type + ShardStrategy* = enum + ssHash = "hash" + ssRange = "range" + ssConsistent = "consistent" + + Shard* = object + id*: int + name*: string + minKey*: string + maxKey*: string + nodeIds*: seq[string] # replica node ids + isActive*: bool + entryCount*: int + + ShardRouter* = ref object + strategy*: ShardStrategy + shards*: seq[Shard] + vnodeRing*: seq[(uint64, int)] # (hash, shard_id) sorted + replicas*: int + + ShardConfig* = object + numShards*: int + replicas*: int + strategy*: ShardStrategy + +proc defaultShardConfig*(): ShardConfig = + ShardConfig(numShards: 4, replicas: 1, strategy: ssHash) + +proc newShardRouter*(config: ShardConfig = defaultShardConfig()): ShardRouter = + result = ShardRouter( + strategy: config.strategy, + shards: @[], + vnodeRing: @[], + replicas: config.replicas, + ) + for i in 0..= shard.minKey and key <= shard.maxKey: + return i + return 0 + +proc getShardConsistent*(router: ShardRouter, key: string): int = + if router.vnodeRing.len == 0: + return getShardHash(router, key) + let h = hashKey(key) + for (ringHash, shardId) in router.vnodeRing: + if h <= ringHash: + return shardId + return router.vnodeRing[0][1] + +proc getShard*(router: ShardRouter, key: string): int = + case router.strategy + of ssHash: router.getShardHash(key) + of ssRange: router.getShardRange(key) + of ssConsistent: router.getShardConsistent(key) + +proc addVirtualNodes*(router: var ShardRouter, numVnodes: int = 100) = + for shard in router.shards: + for i in 0.. 5") + check ast.stmts[0].selHaving != nil + + test "Parse ORDER BY with direction": + let ast = parse("SELECT name FROM users ORDER BY age DESC") + check ast.stmts[0].selOrderBy.len == 1 + check ast.stmts[0].selOrderBy[0].orderByDir == sdDesc + + test "Parse ORDER BY multiple columns": + let ast = parse("SELECT * FROM t ORDER BY a ASC, b DESC") + check ast.stmts[0].selOrderBy.len == 2 + + test "Parse INNER JOIN": + let ast = parse("SELECT u.name, o.total FROM users u INNER JOIN orders o ON u.id = o.user_id") + check ast.stmts[0].selJoins.len == 1 + check ast.stmts[0].selJoins[0].joinKind == jkInner + + test "Parse LEFT JOIN": + let ast = parse("SELECT u.name FROM users u LEFT JOIN orders o ON u.id = o.user_id") + check ast.stmts[0].selJoins.len == 1 + check ast.stmts[0].selJoins[0].joinKind == jkLeft + + test "Parse multiple JOINs": + let ast = parse("SELECT * FROM a JOIN b ON a.id = b.aid JOIN c ON b.id = c.bid") + check ast.stmts[0].selJoins.len == 2 + + test "Parse CTE (WITH)": + let ast = parse("WITH active AS (SELECT * FROM users WHERE active = true) SELECT * FROM active") + check ast.stmts[0].selWith.len == 1 + check ast.stmts[0].selWith[0][0] == "active" + + test "Parse multiple CTEs": + let ast = parse("WITH a AS (SELECT id FROM t1), b AS (SELECT id FROM t2) SELECT * FROM a") + check ast.stmts[0].selWith.len == 2 + + test "Parse aggregate functions in SELECT": + let ast = parse("SELECT count(*), sum(amount), avg(price), min(age), max(score) FROM orders") + check ast.stmts[0].selResult.len == 5 + + test "Parse CASE expression": + let ast = parse("SELECT CASE WHEN age > 18 THEN 'adult' ELSE 'minor' END FROM users") + check ast.stmts[0].selResult.len == 1 + + test "Parse BETWEEN": + let ast = parse("SELECT * FROM products WHERE price BETWEEN 10 AND 100") + check ast.stmts[0].selWhere != nil + + test "Parse subquery in FROM": + let ast = parse("SELECT * FROM (SELECT id FROM users) AS sub") + check ast.stmts[0].selFrom != nil + + test "Parse UPDATE SET WHERE": + let ast = parse("UPDATE users SET name = 'Alice' WHERE id = 1") + check ast.stmts[0].updSet.len == 1 + + test "Parse DELETE WHERE": + let ast = parse("DELETE FROM users WHERE id = 1") + check ast.stmts[0].delWhere != nil + + test "Parse CREATE TYPE with properties": + let ast = parse("CREATE TYPE Person { name: str, age: int32 }") + check ast.stmts[0].ctName == "Person" + check ast.stmts[0].ctProperties.len == 2 + +suite "Raft Consensus": + test "Create cluster with nodes": + var cluster = newRaftCluster() + cluster.addNode("n1") + cluster.addNode("n2") + cluster.addNode("n3") + check cluster.nodes.len == 3 + check cluster.nodes["n1"].peers.len == 2 + + test "Initial state is follower": + var cluster = newRaftCluster() + cluster.addNode("n1") + check cluster.nodes["n1"].state == rsFollower + + test "Election — single node becomes leader": + var cluster = newRaftCluster() + cluster.addNode("n1") + let node = cluster.nodes["n1"] + node.becomeCandidate() + node.becomeLeader() + check node.isLeader + check node.leaderId == "n1" + + test "Log replication": + var cluster = newRaftCluster() + cluster.addNode("n1") + let node = cluster.nodes["n1"] + node.becomeCandidate() + node.becomeLeader() + let entry = node.appendLog("SET key1 value1") + check entry.index == 1 + check node.logLen == 1 + let entry2 = node.appendLog("SET key2 value2") + check entry2.index == 2 + check node.logLen == 2 + + test "RequestVote handling": + var cluster = newRaftCluster() + cluster.addNode("n1") + cluster.addNode("n2") + let n1 = cluster.nodes["n1"] + let n2 = cluster.nodes["n2"] + let req = RaftMessage(kind: rmkRequestVote, term: 1, senderId: "n2", + lastLogIndex: 0, lastLogTerm: 0) + let reply = n1.handleRequestVote(req) + check reply.success + check n1.votedFor == "n2" + + test "AppendEntries handling": + var cluster = newRaftCluster() + cluster.addNode("n1") + cluster.addNode("n2") + let n2 = cluster.nodes["n2"] + let msg = RaftMessage(kind: rmkAppendEntries, term: 1, senderId: "n1", + prevLogIndex: 0, prevLogTerm: 0, + entries: @[LogEntry(term: 1, index: 1, command: "SET x 1")], + leaderCommit: 0) + let reply = n2.handleAppendEntries(msg) + check reply.success + check n2.logLen == 1 + +suite "Sharding": + test "Hash-based sharding": + var router = newShardRouter(ShardConfig(numShards: 4, strategy: ssHash)) + check router.shardCount == 4 + let s1 = router.getShard("user_1") + let s2 = router.getShard("user_2") + check s1 >= 0 and s1 < 4 + check s2 >= 0 and s2 < 4 + + test "Consistent hashing": + var router = newShardRouter(ShardConfig(numShards: 4, strategy: ssConsistent)) + router.addVirtualNodes(50) + let s = router.getShard("some_key") + check s >= 0 and s < 4 + + test "Range-based sharding": + var router = newShardRouter(ShardConfig(numShards: 3, strategy: ssRange)) + router.setRangeBounds(@[("a", "f"), ("g", "n"), ("o", "z")]) + check router.getShardRange("apple") == 0 + check router.getShardRange("hello") == 1 + check router.getShardRange("top") == 2 + + test "Rebalance assigns nodes": + var router = newShardRouter(ShardConfig(numShards: 3, replicas: 2, strategy: ssHash)) + router.rebalance(@["node1", "node2", "node3"]) + for shard in router.shards: + check shard.nodeIds.len == 2 # 2 replicas + + test "Replicas of key": + var router = newShardRouter(ShardConfig(numShards: 2, replicas: 1, strategy: ssHash)) + router.rebalance(@["n1", "n2"]) + let replicas = router.replicasOf("test_key") + check replicas.len == 1 + + test "Active shard count": + var router = newShardRouter() + check router.activeShardCount == 4