System requirements

Functional:

1. Transaction Processing: Support ACID (Atomicity, Consistency, Isolation, Durability) compliant transactions for operations like Create, Read, Update, Delete (CRUD).
2. SQL Interface: Provide a SQL-like query language for data manipulation and definition (DML, DDL).
3. Data Integrity: Enforce data integrity constraints (e.g., primary keys, foreign keys if supported, data types).
4. Concurrency Control: Manage simultaneous access to data by multiple users/transactions without compromising consistency.
5. Data Schema Management: Allow creation, alteration, and dropping of tables and indexes.
6. Backup and Restore: Provide mechanisms for backing up and restoring the database.

Non-Functional:

• • Scalability:Horizontal Scalability: Scale out by adding more nodes to handle increasing load (TPS) and data volume.
  • Read/Write Scalability: Scale both read and write operations.

1. High Availability: The system should remain operational even if some nodes or a datacenter (within a region) fail. Aim for minimal downtime (e.g., 99.99% availability).
2. Fault Tolerance: Tolerate node failures, network partitions within a datacenter.
3. Strong Consistency: Ensure all clients see a consistent view of the data, typically linearizability for OLTP.
4. Durability: Once a transaction is committed, it must persist even in the event of system failures (power loss, crashes).
5. Low Latency: Provide low-latency responses for transactions (e.g., p99 under 50ms for typical operations).
6. Maintainability: System should be easy to monitor, manage, and upgrade.
7. Security: Protect data from unauthorized access (authentication, authorization, encryption at rest/transit).

Capacity estimation

Transactions Per Second (TPS): Start from a few thousand, scaling to hundreds of thousands or millions TPS.

Data Storage: Scale from terabytes to petabytes.

Concurrent Users: Tens of thousands to hundreds of thousands.

Data Model: Primarily structured, relational data.

Read/Write Ratio: Mixed, e.g., 60% reads, 40% writes.

Average Transaction Size: Small, typical OLTP (e.g., a few KBs).

Data Growth: Assume significant data growth over time, requiring elastic scaling of storage.

API design

• Client Connection API:connect(host, port, user, password, database_name): Establishes a connection.
• disconnect(): Closes a connection.
• Transaction API:beginTransaction(): Starts a new transaction.
• commit(): Commits the current transaction.
• rollback(): Aborts the current transaction.
• Query Execution API:executeQuery(sql_statement, parameters): Executes a DML (SELECT, INSERT, UPDATE, DELETE) or DDL statement. Returns result sets for SELECTs, affected row counts for others.
• prepareStatement(sql_template): Creates a prepared statement.
• executePrepared(statement_handle, parameters): Executes a prepared statement.
• Metadata API (less common for end-users, more for tools):getTableSchema(table_name)
• listTables()

Database design

Data Model: Relational model (tables, rows, columns).

• Partitioning (Sharding):Data will be sharded across multiple nodes.
• Strategy: Primarily hash-based partitioning on a designated partition key (often the primary key or part of it) to ensure even data distribution.
• Co-location: Related data (e.g., orders and their line items) should be co-located on the same shard by using the same partition key to minimize cross-shard transactions.
• Indexes:Local Indexes: Each shard will maintain local indexes for the data it stores (e.g., on primary keys, frequently queried columns).
• Global Indexes (Optional/Complex): May be considered for columns not part of the partition key, but add significant complexity.

Metadata Storage: A separate, highly available key-value store (e.g., ZooKeeper/etcd) will store the shard metadata:

• Mapping of partition key ranges/hashes to shard IDs.
• Mapping of shard IDs to the set of nodes (replicas) hosting that shard.
• Location of the current leader/primary for each shard's replica set.

High-level design

+-----------------+ +----------------------+ +---------------------+ +---------------------+ | Client |<--->| Load Balancer / |<--->| Gateway / Router |<--->| Coordination | | (App + Driver) | | Connection Pool | | (SQL Parser, | | Service (ZK/etcd) | +-----------------+ +----------------------+ | Query Planner, | | (Metadata, Leader | | Txn Coordinator) | | Discovery) | +---------------------+ +---------------------+ | ^ (Shard Info, Leader Addr) | (Query to specific Shard Leader) | +----------------------|----------------------+ | | | +-------------------+ +-------------------+ +-------------------+ | Shard 1 (Raft Grp)| | Shard 2 (Raft Grp)| | Shard N (Raft Grp)| | - Leader | | - Leader | | - Leader | | - Follower(s) | | - Follower(s) | | - Follower(s) | | (Data + WAL) | | (Data + WAL) | | (Data + WAL) | +-------------------+ +-------------------+ +-------------------+

Client: Applications with a database driver.

Load Balancer/Connection Pool: Distributes client connections to Gateways.

• Gateway/Router:Handles SQL parsing, query planning/optimization.
• Acts as Transaction Coordinator for distributed transactions.
• Consults Coordination Service for shard metadata and leader locations.
• Routes requests to the appropriate shard leader.
• Coordination Service (e.g., ZooKeeper/etcd):Stores shard metadata (partitioning scheme, replica locations).
• Facilitates leader discovery for each shard's Raft group.
• May assist in other cluster-wide coordination tasks (e.g., schema changes, node liveness).
• Shards (Raft Groups):Each shard is a replica set (e.g., 3 or 5 nodes) forming a Raft consensus group.
• One node in each group is elected Leader, handling writes and strongly consistent reads for that shard.
• Data is replicated via Raft protocol for fault tolerance and strong consistency within the shard.
• Each node has its own storage (WAL, data files).

Request flows

Single-Shard Write:

1. Client sends SQL to Gateway (via LB).
2. Gateway parses SQL, consults Coordination Service (ZK) to find the leader of the target shard.
3. Gateway forwards request to the shard leader.
4. Shard leader appends to its Raft log, replicates to followers, gets majority ACK, applies write, ACKs Gateway.
5. Gateway ACKs client.

• Multi-Shard Write (Distributed Transaction):
  1. Client sends transaction (BEGIN, ops, COMMIT) to Gateway. Gateway acts as Transaction Coordinator (TC).
  2. TC identifies involved shards, gets their leaders from ZK. Designates one shard as "anchor."
  3. Prepare Phase: TC sends "prepare" (with transaction ID - GTID) to leaders of involved shards. Each shard leader writes a "write intent" to its local Raft log, locks resources, ACKs TC.
  4. Commit Decision: If all prepare, TC instructs anchor shard leader to write "GTID committed" to its Raft log. Once this is Raft-committed on anchor, transaction is globally committed. Anchor ACKs TC.
  5. TC ACKs client.
  6. Apply Phase (Async): TC informs all participating shard leaders to apply their prepared intents (by writing "apply GTID" to their local Raft logs). If TC fails, shards can query anchor shard for GTID status.

Detailed component design

Transaction Coordinator (within Gateway):

• Responsibilities: Manage lifecycle of distributed transactions (prepare, commit/abort), interact with ZK for shard leader discovery, interact with shard leaders.
• Algorithm: Implements a 2PC-like protocol using an anchor shard's Raft log for the commit decision point.
  • Assigns unique Global Transaction ID (GTID).
  • Tracks participant shards and their states (prepared, committed, aborted).
  • Handles recovery by allowing participants to query the anchor shard for transaction status.
• Scalability: Gateways are stateless and can be horizontally scaled. Load balancers distribute TC workload. The anchor shard selection helps distribute the commit load.
• Data Structure: In-memory map of GTID -> TransactionState (participants, status, anchor_shard_id).

Shard (Raft Group):

• Responsibilities: Store a subset of data, replicate data within its group, elect a leader, handle writes/reads for its data, participate in distributed transactions (write intents, apply commits).
• Algorithm: Implements Raft consensus protocol.
  • Leader Election: Standard Raft leader election.
  • Log Replication: Leader replicates log entries (writes, intents, apply commands) to followers; commits upon majority ACK.
  • Snapshotting/Log Compaction: Periodically takes snapshots of its state machine and truncates the Raft log to manage disk space and speed up recovery.
  • Concurrency Control (within shard leader): Uses Multi-Version Concurrency Control (MVCC) to allow concurrent reads and writes, assigning timestamps to transactions/versions. Read operations access the appropriate version based on their snapshot timestamp.
• Scalability: Number of shards can be increased (re-sharding). Each Raft group scales independently to some extent (usually 3 or 5 nodes). Write throughput is limited by the single leader, but many shards distribute overall load.
• Data Structure: Raft Log (persistent), State Machine (e.g., B-trees or LSM-trees for actual data, indexed by primary key and other indexed columns), MVCC version information.

Trade offs/Tech choices

Strong Consistency vs. Availability/Latency: Chose strong consistency (linearizability) using Raft and a 2PC-like distributed transaction protocol. This can increase latency and reduce availability during network partitions compared to eventually consistent systems (CAP theorem: aiming for CP).

Raft for Shard Replication: Provides strong consistency and fault tolerance within a shard. Simpler than Paxos to understand and implement generally.

ZooKeeper for Coordination: Mature, reliable for metadata, leader discovery. Offloads these tasks from the database core. Alternative: etcd.

Hash Partitioning: Good for even data distribution and point lookups. Less ideal for range queries unless combined with other strategies or specific indexing.

Anchor Shard for Distributed Tx Commit: Avoids ZK becoming a bottleneck for every transaction commit, leveraging the data shards' Raft logs for high throughput.

SQL Interface: Wider adoption and existing ecosystem, but more complex to implement in a distributed manner than NoSQL interfaces.

Failure scenarios/bottlenecks

Coordination Service (ZK) Failure: If ZK cluster goes down, new connections can't find shard leaders, metadata changes are impossible. Mitigate: Robust ZK deployment (e.g., 5 nodes across AZs).

Shard Leader Failure: Raft handles leader re-election within the shard. Brief unavailability during election.

Multiple Node Failures in a Shard: If > (N-1)/2 nodes fail in a Raft group, shard becomes unavailable for writes (loses majority). Mitigate: Deploy replicas across AZs, aim for 3 or 5 replicas.

• Network Partition:Within a shard: Raft partition with majority remains active. Minority partition becomes unavailable.
• Between client/gateway and shards: Requests fail.
• Isolating ZK: Critical.
• Transaction Coordinator (Gateway) Failure:If before anchor commit: Transaction eventually times out and aborts. Participants release locks.
• If after anchor commit: Transaction is committed. Other participants can query anchor for status and apply.

Hot Shards: Uneven data access patterns can overload a specific shard leader. Mitigate: Good partition key choice, adaptive re-sharding, read replicas (if consistency allows relaxed reads for some queries).

Distributed Transaction Latency: Multi-round protocol inherently adds latency. Minimize by co-locating frequently accessed data.

Anchor Shard Bottleneck (for commits): While distributed, if many distributed transactions involve the same potential anchor shards, those can become write-limited.

Cross-Shard Scans/Joins: Can be very expensive, requiring scatter-gather operations.

Future improvements

Distributed Query Optimizer: Enhance the query planner to better optimize queries that span multiple shards (e.g., pushdowns, smarter join strategies).

Global Secondary Indexes: Implement efficient global secondary indexes for queries on non-partition key columns. This is complex, often requiring separate indexing structures and coordination.

Online Schema Changes (Zero-Downtime): Support for ALTER TABLE without locking tables for extended periods or requiring downtime. Often involves shadow tables or phased rollouts.

Cross-Region Replication (Asynchronous): For disaster recovery, implement asynchronous replication of committed data to a different geographical region.

Tunable Consistency Levels: Allow applications to choose relaxed consistency levels (e.g., eventual consistency for reads on certain data) for better performance/availability where strong consistency isn't strictly needed.

Automated Re-sharding/Elasticity: More dynamic and automated re-sharding based on load or data size to improve load balancing and elasticity.

Time Travel Queries / Point-in-Time Recovery (PITR): Leverage MVCC and transaction logs to allow querying data as of a past timestamp and to restore the database to a specific point in time.

Enhanced Security Features: Row-level security, more granular access controls, integration with external KMS.

Columnar Storage Option: For workloads that might benefit from analytical queries on OLTP data (HTAP), explore options for columnar representations or integration.