Distributed database — partitioning, sharding আর CAP

১ · একটি যন্ত্র কখন যথেষ্ট নয় — distribute করার তিন কারণ

এখন পর্যন্ত পুরো কোর্সে ডেটাবেস ছিল একটি যন্ত্রে। পর্ব ৫২-এ (Backup, replication আর high availability) সেই ডেটার কয়েকটি কপি বানিয়েছি, কিন্তু প্রতিটি কপিতে ছিল পুরো ডেটা, আর সব লেখা যেত একটি primary-তে। Distributed database-এ ডেটা নিজেই কয়েকটি যন্ত্রে ভাগ হয়ে থাকে, আর যন্ত্রগুলো network দিয়ে মিলে একটি ডেটাবেসের মতো আচরণ করে। কেন এত ঝামেলা? কারণ তিনটি:

চাপ বাড়লে দুটি পথ। Scale-up (vertical scaling): আরও বড় যন্ত্র — বেশি CPU core, বেশি RAM, দ্রুত SSD. সরল, অ্যাপে কিছুই বদলাতে হয় না, কিন্তু একসময় আর বড় যন্ত্র নেই, আর দাম দ্রুত বাড়ে। Scale-out (horizontal scaling): আরও অনেক সাধারণ যন্ত্র, কাজ ভাগ করে। সীমা অনেক দূরে, কিন্তু এখন প্রতিটি কোয়েরি আর লেনদেনকে ভাবতে হয় “ডেটা কোথায়”.

২ · Partitioning — সারি ধরে ভাগ, column ধরে ভাগ

বড় টেবিলকে ছোট টুকরায় ভাগ করার নাম partitioning. ভাগের দুটি দিক:

horizontal ভাগ যখন আলাদা যন্ত্রে থাকে, তার নাম sharding, আর প্রতিটি টুকরা একটি shard. একই যন্ত্রের ভেতরে ভাগ — পর্ব ৫৩-এর table partitioning — পুরোনো ডেটা মোছা আর partition pruning-এ সাহায্য করে, কিন্তু যন্ত্রের সীমা বাড়ায় না। vertical ভাগের বড় রূপ হলো কাজ ধরে ভাগ: অর্ডারের ডেটাবেস এক সার্ভারে, ব্যবহারকারীর প্রোফাইল আরেকটিতে — microservice-এর যুগে খুব প্রচলিত, আর প্রায়ই sharding-এর আগের ধাপ।

PostgreSQL-এ একই যন্ত্রের ভেতরে দুই ধরনের ভাগ এভাবে লেখা হয় (চালানো যাবে না, SQLite-এ এই সুবিধা নেই):

-- range: মাস ধরে (partition key primary key-এর অংশ হতে হয়)
CREATE TABLE orders (
  order_id bigint, customer_id bigint NOT NULL,
  order_date date NOT NULL, status text,
  PRIMARY KEY (order_id, order_date)
) PARTITION BY RANGE (order_date);
CREATE TABLE orders_2025_09 PARTITION OF orders
  FOR VALUES FROM ('2025-09-01') TO ('2025-10-01');

-- hash: customer_id ধরে ৪ ভাগ (PostgreSQL 11 থেকে)
CREATE TABLE carts (customer_id bigint, items jsonb) PARTITION BY HASH (customer_id);
CREATE TABLE carts_p0 PARTITION OF carts FOR VALUES WITH (MODULUS 4, REMAINDER 0);

MySQL-এ PARTITION BY RANGE / LIST / HASH / KEY আছে। কয়েকটি যন্ত্রে ভাগ করতে লাগে বাড়তি স্তর — অ্যাপের নিজের কোড, Vitess বা Citus-এর মতো টুল (অংশ ৯), বা জন্ম থেকেই distributed কোনো ডেটাবেস।

৩ · কোন সারি কোন shard-এ — range, hash, list আর directory

প্রতিটি সারির জন্য একটি নিয়ম লাগে: এটি কোন shard-এ যাবে? নিয়মটি চলে একটি column বা column-এর দলের ওপর, যার নাম shard key (partition key). প্রচলিত চারটি নিয়ম:

এবার দোকানের নমুনা ডেটাবেসে হাতে-কলমে। এর টেবিল: customers (customer_id, name, city, phone, joined_on), categories, products (product_id, name, category_id, price, stock), orders (order_id, customer_id, order_date, status) আর order_items (order_id, product_id, quantity, unit_price). ধরুন তিনটি shard, আর নিয়ম shard = customer_id % 3:

SELECT customer_id, name, city, customer_id % 3 AS shard
FROM customers
ORDER BY shard, customer_id;
SELECT c.customer_id % 3 AS shard,
       COUNT(DISTINCT c.customer_id) AS customers,
       COUNT(o.order_id) AS orders
FROM customers c
LEFT JOIN orders o ON o.customer_id = c.customer_id
GROUP BY shard
ORDER BY shard;

এবার range: মাস ধরে অর্ডার ভাগ। order_parts টেবিলে প্রতিটি ভাগের নাম (part) আর সীমা — lo থেকে শুরু, hi-এর ঠিক আগে শেষ (half-open, পর্ব ১৬)। নভেম্বরের ভাগটি আগেই বানানো, এখনো খালি:

SELECT p.part, p.lo, p.hi, COUNT(o.order_id) AS orders,
       MAX(o.order_date) AS newest
FROM order_parts p
LEFT JOIN orders o ON o.order_date >= p.lo AND o.order_date < p.hi
GROUP BY p.part
ORDER BY p.part;
-- ১০ থেকে ২০ সেপ্টেম্বরের অর্ডার কোন ভাগে খুঁজতে হবে?
SELECT part FROM order_parts
WHERE lo < '2025-09-21' AND hi > '2025-09-10';

shard key বাছাই — সবচেয়ে দামি সিদ্ধান্ত

shard key পরে বদলানো মানে প্রায় সব ডেটা নতুন জায়গায় সরানো, তাই এটি শুরুতেই ভাবতে হয়। দোকানের orders-এর দুই প্রার্থী তুলনা করুন:

সাধারণ নিয়ম: (১) অনেক আলাদা মান আর সমান ছড়ানো; (২) সবচেয়ে বেশি চলা কোয়েরির WHERE-এ key থাকে, যাতে সেগুলো এক shard-এ যায়; (৩) একসাথে লাগা ডেটা একই key পায় — orders আর order_items দুটিই customer_id ধরে; (৪) key-এর মান কখনো বদলায় না, নইলে সারিটিকে অন্য যন্ত্রে সরাতে হয়। দোকানের লেনদেনের কাজে (OLTP) তাই customer_id সাধারণত ভালো; মাসভিত্তিক বিক্রির হিসাব চলে পর্ব ৫৫-এর data warehouse-এ। আর সময়-ক্রমের id (auto-increment বা timestamp) range-এর key হলে hot spot অনিবার্য — Google Spanner-এর ডকুমেন্টেশনও primary key-তে এমন ক্রমাগত বাড়া মান এড়াতে বলে।

৪ · Consistent hashing আর rebalancing

তিনটি shard আর পারছে না, চতুর্থটি যোগ করলেন। key % 3 থেকে key % 4: কতগুলো key-কে নতুন জায়গায় যেতে হবে? SQL-এ ১২,০০০টি key গুনে দেখি (recursive CTE, পর্ব ২৫):

WITH RECURSIVE k(id) AS (
  SELECT 1 UNION ALL SELECT id + 1 FROM k WHERE id < 12000)
SELECT SUM(id % 3 <> id % 4) AS moved_3_to_4,
       SUM(id % 4 <> id % 5) AS moved_4_to_5,
       SUM(id % 3 <> id % 6) AS moved_3_to_6,
       COUNT(*) AS keys
FROM k;

Consistent hashing (Karger ও সহকর্মীরা, ১৯৯৭; শুরুতে web cache-এর জন্য) এই অপচয় দূর করে। hash-এর সব মানকে একটি বৃত্ত ভাবুন — hash ring. প্রতিটি যন্ত্রকেও hash করে বৃত্তের একটি বিন্দুতে বসানো হয়। একটি key-এর মালিক হলো key-এর অবস্থান থেকে ঘড়ির কাঁটার দিকে প্রথম যে যন্ত্র পড়ে। নতুন যন্ত্র D বসলে শুধু D আর তার আগের যন্ত্রের মাঝের key-গুলো D-তে যায়; বাকি সবাই যেখানে ছিল সেখানেই থাকে।

একটি সমস্যা: প্রতিটি যন্ত্র বৃত্তে একটি মাত্র বিন্দু হলে ভাগ্যক্রমে কারও ভাগে বিশাল চাপ পড়তে পারে। সমাধান virtual node: প্রতিটি যন্ত্রকে বৃত্তের অনেকগুলো বিন্দুতে বসানো (যেমন A#0, A#1, …)। Amazon-এর Dynamo (২০০৭-এর গবেষণাপত্র) আর তার অনুপ্রেরণায় তৈরি Cassandra এই কৌশল ব্যবহার করে; শক্তিশালী যন্ত্রকে বেশি virtual node দিয়ে বেশি ভাগও দেওয়া যায়। নিচের প্রোগ্রাম ১০,০০০ গ্রাহক-key নিয়ে দুইভাবে হিসাব করে:

import bisect, hashlib
def h(s):  # stable 32-bit hash (Python's built-in hash() changes per run)
    return int(hashlib.md5(s.encode()).hexdigest()[:8], 16)
def build_ring(nodes, vnodes):
    points = sorted((h(f"{n}#{v}"), n) for n in nodes for v in range(vnodes))
    return [p for p, _ in points], [n for _, n in points]
def owner(ring, key):  # first point clockwise from the key's position
    pos, names = ring
    return names[bisect.bisect(pos, h(key)) % len(pos)]
keys = [f"customer:{i}" for i in range(1, 10001)]
for v in (1, 100):
    r3, r4 = build_ring("ABC", v), build_ring("ABCD", v)
    before = {k: owner(r3, k) for k in keys}
    after = {k: owner(r4, k) for k in keys}
    load = [sum(1 for k in keys if before[k] == n) for n in "ABC"]
    moved = [k for k in keys if before[k] != after[k]]
    only_to_d = all(after[k] == "D" for k in moved)
    print(f"vnodes={v:<3} load A,B,C = {load}  add D: moved {len(moved)}"
          f" ({len(moved) / 100:.1f}%), all moved keys went to D: {only_to_d}")
mod = sum(1 for k in keys if h(k) % 3 != h(k) % 4)
print(f"hash % N, 3 -> 4 shards: moved {mod} ({mod / 100:.1f}%)")

vnodes=1   load A,B,C = [284, 5750, 3966]  add D: moved 5615 (56.1%), all moved keys went to D: True
vnodes=100 load A,B,C = [3180, 3447, 3373]  add D: moved 2419 (24.2%), all moved keys went to D: True
hash % N, 3 -> 4 shards: moved 7521 (75.2%)

Rebalancing — যন্ত্র যোগ বা বাদের পর ডেটা নতুন করে সাজানো — বাস্তবে আরও কয়েকভাবে হয়। এক, শুরুতেই যন্ত্রের চেয়ে অনেক বেশি “যৌক্তিক” ভাগ বানানো (যেমন ১,০২৪টি), আর নতুন যন্ত্র এলে কয়েকটি পুরো ভাগ তাকে দেওয়া — key-এর হিসাব কখনো বদলায় না, বদলায় শুধু ভাগ→যন্ত্রের তালিকা। দুই, range ভাগ বড় হলে মাঝখান থেকে দুই টুকরা করা (Bigtable/HBase-এর tablet বা region, MongoDB-র chunk এই ধাঁচের)। যেকোনো পথে সরানো চলে চালু অবস্থায়: আগে কপি, তারপর কপির সময়ের পরিবর্তন (CDC) ধরে ফেলা, শেষে router-এর তালিকা বদল।

৫ · কয়েক shard জুড়ে কোয়েরি — scatter–gather আর global index

অ্যাপ আর shard-গুলোর মাঝে থাকে একটি router (অ্যাপের লাইব্রেরি বা আলাদা proxy). কোয়েরির WHERE-এ shard key থাকলে router সেটি সরাসরি এক shard-এ পাঠায়। না থাকলে পথ একটাই — scatter–gather: কোয়েরি সব shard-এ ছড়িয়ে দেওয়া (scatter), প্রতিটির আংশিক ফল জোগাড় করে মেলানো (gather). এমন cross-shard query-র দাম দুটি: প্রতিটি shard কাজ করে, আর উত্তর আসে সবচেয়ে ধীর shard-এর গতিতে — shard যত বেশি, কোনো একটির হঠাৎ ধীর হওয়ার সম্ভাবনা তত বেশি (tail latency).

নিচের এডিটরে তিনটি shard টেবিল আগেই বানানো: orders_s0, orders_s1, orders_s2 — প্রতিটিতে customer_id % 3 অনুযায়ী অর্ডার, সাথে একটি বাড়তি column total (অর্ডারের মোট টাকা)। gather-এর অংশটি একটি view (পর্ব ২৪), আর ভেতরে UNION ALL (পর্ব ২২):

CREATE VIEW partial AS
  SELECT 0 AS shard, COUNT(*) AS n, SUM(total) AS s, AVG(total) AS a FROM orders_s0
  UNION ALL SELECT 1, COUNT(*), SUM(total), AVG(total) FROM orders_s1
  UNION ALL SELECT 2, COUNT(*), SUM(total), AVG(total) FROM orders_s2;
SELECT shard, n, s, ROUND(a, 1) AS a FROM partial;
SELECT SUM(n) AS orders, SUM(s) AS revenue,
       ROUND(SUM(s) / SUM(n), 1) AS true_avg,
       ROUND(AVG(a), 1) AS avg_of_avgs
FROM partial;

Join আর global secondary index

দুটি টেবিলের join-এ সারিগুলো আলাদা যন্ত্রে থাকলে ডেটা network দিয়ে সরাতে হয় — সবচেয়ে দামি কাজ। এড়ানোর তিনটি কৌশল। Co-location: যে টেবিলগুলো একসাথে join হয়, সেগুলো একই key-তে ভাগ — customers, orders, order_items তিনটিই customer_id ধরে, তাই একজন গ্রাহকের join পুরোটা এক shard-এর ভেতরে। Reference table: ছোট, কম বদলানো টেবিল (categories) প্রতিটি shard-এ পুরো কপি। ডেটা সরানো: বড় বিশ্লেষণে দুই দিকের সারি join key ধরে নতুন করে ভাগ করে পাঠানো (shuffle) — data warehouse-এর কাজ।

আরেক সমস্যা: গ্রাহক shard হয়েছে customer_id ধরে, কিন্তু login-এ খোঁজা হয় ফোন নম্বর দিয়ে। প্রতিটি shard-এর নিজের index (local index) আছে, তবু কোন shard-এ খুঁজতে হবে জানা নেই। সমাধান global secondary index: phone থেকে customer_id আর shard-এর একটি আলাদা টেবিল, যা নিজেও phone ধরে ভাগ করা যায়। এডিটরে customers_s0..s2 আর phone_idx (phone, customer_id, shard) আগেই বানানো:

-- ধাপ ১: index থেকে shard জানা, ধাপ ২: শুধু সেই shard-এ যাওয়া
SELECT * FROM phone_idx WHERE phone = '01811000002';
SELECT customer_id, name, city FROM customers_s2 WHERE customer_id = 2;
-- index না থাকলে: তিনটি shard-কেই জিজ্ঞেস করতে হয়
SELECT 's0' AS asked, COUNT(*) AS found FROM customers_s0 WHERE phone = '01811000002'
UNION ALL
SELECT 's1', COUNT(*) FROM customers_s1 WHERE phone = '01811000002'
UNION ALL
SELECT 's2', COUNT(*) FROM customers_s2 WHERE phone = '01811000002';

৬ · কয়েক shard জুড়ে লেনদেন — two-phase commit আর saga

Rahim-এর wallet shard A-তে, Karima-র shard B-তে। Rahim ৫০০ টাকা পাঠালেন: A-তে −৫০০, B-তে +৫০০. দুটি commit হতে হবে একসাথে, নয়তো একটিও না — পর্ব ৪৪-এর atomicity, এবার দুই যন্ত্রে। এমন লেনদেনের নাম distributed transaction, আর প্রচলিত প্রোটোকল two-phase commit (2PC), যার ধারণা ১৯৭০-এর দশকের শেষের দিকে Jim Gray-সহ গবেষকদের লেখায় আসে। একটি coordinator সিদ্ধান্ত নেয়; প্রতিটি shard একজন participant.

2PC-এর খরচ স্পষ্ট: অন্তত দুটি round trip, প্রতিটি participant-এ বাড়তি fsync, আর পুরো সময় lock ধরা — একটি ধীর participant সবাইকে ধীর করে। blocking এড়াতে three-phase commit-এর প্রস্তাব আছে, কিন্তু তা network-এর সময় নিয়ে কড়া অনুমান করে, তাই প্রায় কোথাও ব্যবহৃত হয় না। বাস্তব সমাধান হলো coordinator-কে নিজেই replicated করা: Google Spanner-এ প্রতিটি participant একটি Paxos দল (অংশ ৮), তাই একটি যন্ত্র মরলেও সিদ্ধান্ত হারায় না। PostgreSQL আর MySQL-এ 2PC-এর participant হওয়ার কমান্ড আছে (XA standard বা নিজস্ব), coordinator-টি বাইরের কোনো transaction manager:

-- shard A-তে (max_prepared_transactions > 0 হতে হবে; default 0)
BEGIN;
UPDATE wallets SET balance = balance - 500 WHERE owner = 'Rahim';
PREPARE TRANSACTION 't42';     -- vote YES: ডিস্কে টেকসই, lock ধরা থাকে
-- coordinator-এর সিদ্ধান্তের পর, যেকোনো সংযোগ থেকে:
COMMIT PREPARED 't42';         -- অথবা ROLLBACK PREPARED 't42'
SELECT gid, prepared FROM pg_prepared_xacts;  -- ঝুলে থাকা in-doubt লেনদেন

MySQL-এ একই কাজ XA START, XA END, XA PREPARE, XA COMMIT দিয়ে। সতর্কতা: ভুলে যাওয়া prepared লেনদেন সার্ভার রিস্টার্টের পরেও lock ধরে রাখে, আর PostgreSQL-এ vacuum-কেও আটকায় — তাই এগুলোর monitoring জরুরি।

Saga — ছোট ছোট local transaction আর compensation

অনলাইন দোকানে অর্ডার একটি লম্বা কাজ: অর্ডার তৈরি, bKash থেকে টাকা কাটা, মজুদ আটকে রাখা, courier বুক করা। প্রতিটি হয়তো আলাদা সেবা, আলাদা ডেটাবেসে; courier কোম্পানির সিস্টেম তো বাইরের, সেটি 2PC-তে অংশই নেবে না। আর কয়েক মিনিট lock ধরে রাখা অসম্ভব। Hector Garcia-Molina আর Kenneth Salem ১৯৮৭ সালের একটি গবেষণাপত্রে এর বিকল্প দেন: saga — একটি লম্বা কাজকে কয়েকটি local transaction-এ ভাগ করা, প্রতিটি সঙ্গে সঙ্গে commit; আর প্রতিটির জন্য একটি compensating transaction, যা তার প্রভাব অর্থের দিক থেকে উল্টে দেয়। কোনো ধাপ ব্যর্থ হলে আগের সফল ধাপগুলোর compensation চলে উল্টো ক্রমে.

saga চালানোর দুটি ধাঁচ: orchestration — একটি কেন্দ্রীয় সেবা ধাপগুলো ডাকে আর অবস্থা লিখে রাখে; choreography — প্রতিটি সেবা event শুনে নিজের কাজ করে পরের event ছাড়ে। orchestrator-এর log দেখতে কেমন, আর ব্যর্থতার পর কোন compensation চলবে — SQL-এ। saga_steps টেবিলে প্রতিটি ধাপের saga_id, step_no, service (কোন সেবা), step (কাজ), compensation (উল্টানোর কাজ) আর status:

SELECT step_no, service, step, status
FROM saga_steps WHERE saga_id = 7 ORDER BY step_no;
-- ব্যর্থতার পর: সফল ধাপগুলোর compensation, উল্টো ক্রমে
SELECT step_no, service, compensation AS run_now
FROM saga_steps
WHERE saga_id = 7 AND status = 'done'
ORDER BY step_no DESC;

৭ · CAP theorem — নেটওয়ার্ক ভাগ হলে consistency না availability

২০০০ সালে PODC সম্মেলনের একটি বক্তৃতায় Eric Brewer একটি অনুমান (conjecture) দেন, আর ২০০২ সালে Seth Gilbert ও Nancy Lynch সেটিকে নির্দিষ্ট সংজ্ঞায় প্রমাণ করেন। তিনটি শব্দ, প্রমাণের অর্থে:

• C — consistency: এখানে মানে linearizability — যেন ডেটার একটিমাত্র কপি আছে, আর প্রতিটি পড়া সর্বশেষ সম্পন্ন লেখা দেখে। (ACID-এর C, অর্থাৎ নিয়ম না ভাঙা, সম্পূর্ণ আলাদা জিনিস।)
• A — availability: চালু থাকা প্রতিটি নোডে আসা প্রতিটি অনুরোধ একটি (ত্রুটি নয় এমন) উত্তর পায়।
• P — partition tolerance: নোডগুলোর মধ্যে বার্তা হারালেও — network partition — ব্যবস্থা কাজ চালিয়ে যায়।

যুক্তিটি সহজ। দুটি replica, মাঝের লিংক কাটা। একটিতে লেখা হলো, অন্যটিতে পড়ার অনুরোধ এল। দ্বিতীয়টি উত্তর দিলে পুরোনো মান দেবে (C ভাঙল); নতুন মান জানতে অপেক্ষা করলে বা ত্রুটি দিলে উত্তর নেই (A ভাঙল)। তৃতীয় পথ নেই, কারণ খবর পৌঁছানোর কোনো উপায়ই নেই। ধাপে ধাপে:

PACELC — partition না থাকলেও একটি বাছাই আছে

CAP শুধু partition-এর সময়ের কথা বলে, অথচ বেশিরভাগ সময় network ঠিকই থাকে। Daniel Abadi ২০১২ সালের একটি প্রবন্ধে বিষয়টি গুছিয়ে দেন — PACELC: Partition হলে A না C; Else (স্বাভাবিক সময়ে) Latency না Consistency. কারণ শক্ত consistency মানে প্রতিটি লেখায় (অনেক সময় পড়াতেও) অন্য replica-র সম্মতির জন্য অপেক্ষা — পর্ব ৫২-এর sync replication-এর round trip. Abadi-র নিজের উদাহরণে Dynamo আর Cassandra-র মতো ব্যবস্থা default সেটিংয়ে PA/EL (দ্রুত উত্তরকে প্রাধান্য), আর পুরোপুরি consistent ব্যবস্থাগুলো PC/EC. তবে অনেক ব্যবস্থায় এটি সেটিং দিয়ে বদলানো যায় — তাই কোনো ডেটাবেসকে একটি লেবেলে বাঁধার আগে তার কনফিগারেশন দেখুন।

৮ · Consistency model, quorum আর consensus

“consistent” কথাটি একটি মাত্রা, হ্যাঁ-না নয়। replica-রা কতটা একমত দেখাবে, তার প্রতিশ্রুতিকে বলে consistency model. শক্তিশালী থেকে দুর্বল, মোটামুটি এই ক্রমে:

দুই replica-র একটি ছোট গল্পে পার্থক্যগুলো দেখুন। R1 ঢাকায়, R2 চট্টগ্রামে, replication async; অর্ডার 1009-এর status শুরুতে pending:

Quorum — N, W আর R

Dynamo-ধাঁচের ব্যবস্থায় (Cassandra, Riak) প্রতিটি ডেটা থাকে N টি replica-তে। লেখা সফল হয় W টি replica “পেয়েছি” বললে, পড়া জিজ্ঞেস করে R টি replica-কে আর সবচেয়ে নতুন সংস্করণটি নেয়। নিয়ম: W + R > N হলে প্রতিটি পড়ার দল প্রতিটি সফল লেখার দলের সাথে অন্তত একটি replica-তে মেলে। প্রমাণ: দুটি দলের মোট সদস্য W + R, অথচ আলাদা replica আছে মাত্র N টি; W + R > N হলে pigeonhole নীতিতে অন্তত একজন দুই দলেই আছে। সব সম্ভাব্য দল গুনে দেখি:

from itertools import combinations
N = 3                      # replicas 0, 1, 2; all start with version 1
def check(W, R):
    stale = total = 0
    for wrote in combinations(range(N), W):        # replicas that acked version 2
        for asked in combinations(range(N), R):    # replicas a later read contacts
            total += 1
            newest = max(2 if r in wrote else 1 for r in asked)
            stale += (newest == 1)                 # read returned the old version
    return stale, total
print("W R  W+R>N  stale reads  writes ok with  reads ok with")
for W, R in [(1, 1), (2, 1), (1, 2), (2, 2), (3, 1), (1, 3)]:
    s, t = check(W, R)
    print(f"{W} {R}  {str(W + R > N):6} {s:>3} of {t:<6}  {N - W} down         {N - R} down")

W R  W+R>N  stale reads  writes ok with  reads ok with
1 1  False    6 of 9       2 down         2 down
2 1  False    3 of 9       1 down         2 down
1 2  False    3 of 9       2 down         1 down
2 2  True     0 of 9       1 down         1 down
3 1  True     0 of 3       0 down         2 down
1 3  True     0 of 3       2 down         0 down

Consensus — এক পাতায়

পর্ব ৫২-এ দেখেছি, কে primary তা ঠিক করতে consensus লাগে। সাধারণ রূপে সমস্যাটি: কয়েকটি যন্ত্র, যাদের কিছু ব্যর্থ হতে পারে আর বার্তা দেরিতে আসতে পারে, একটি মানে একমত হবে আর সিদ্ধান্ত কখনো উল্টাবে না। Leslie Lamport-এর Paxos (১৯৯৮ সালে প্রকাশিত) এর ক্লাসিক সমাধান, কিন্তু বোঝা আর বানানো কঠিন বলে পরিচিত। Diego Ongaro ও John Ousterhout ২০১৪ সালে Raft দেন, বোঝার সুবিধাকে প্রধান লক্ষ্য করে। দুটিরই মূল কৌশল majority quorum: একজন নেতা বাছাই হয় (leader election), নেতা প্রতিটি কাজ একটি replicated log-এ যোগ করে, আর majority-র ডিস্কে পৌঁছালেই সেটি commit. ২N + ১ যন্ত্রে N টি ব্যর্থতা সয়। etcd (Raft) আর ZooKeeper (নিজস্ব Zab প্রোটোকল) এই সেবা দেয় — shard-এর তালিকা, নেতা কে, কনফিগারেশন রাখতে; আর distributed SQL ডেটাবেসগুলো প্রতিটি ডেটা-খণ্ডের লেখাতেই এটি চালায়।

৯ · NewSQL ও distributed SQL — আর কখন distribute করবেন না

২০১০-এর দশকে একটি ধারা আসে, যাকে NewSQL বা distributed SQL বলা হয়: SQL আর ACID লেনদেন রেখে নিজে থেকেই sharding, replication আর failover. প্রচলিত কয়েকটি, সংক্ষেপে (খুঁটিনাটি সংস্করণভেদে বদলায়):

-- order_items-এও customer_id column থাকতে হবে, যাতে একই worker-এ থাকে
SELECT create_distributed_table('orders', 'customer_id');
SELECT create_distributed_table('order_items', 'customer_id', colocate_with => 'orders');
SELECT create_reference_table('categories');   -- প্রতিটি worker-এ পুরো কপি

এগুলো sharding-এর অনেক কষ্ট লুকায়, কিন্তু পদার্থবিজ্ঞান নয়: কয়েক খণ্ড জোড়া লেনদেনে এখনো consensus আর 2PC-এর round trip, আর দূরের region-এ replica মানে ধীর commit. shard key-এর মতো সিদ্ধান্ত (primary key-এর গঠন, কোন টেবিল একসাথে) এখনো আপনার।

কখন distribute করবেন না

সবচেয়ে গুরুত্বপূর্ণ কথাটি শেষে: বেশিরভাগ অ্যাপের sharding লাগে না। আজকের একটি সার্ভারে কয়েকশো CPU core আর কয়েক TB RAM পর্যন্ত পাওয়া যায়, আর ভালো index ও কোয়েরিসহ একটি PostgreSQL বা MySQL বিপুল কাজ সামলায়। sharding-এর আগে সাধারণত এই ক্রমে ভাবুন: ধীর কোয়েরি আর index ঠিক করা (পর্ব ৪১, ৫৩) → connection pool → পড়ার জন্য read replica (পর্ব ৫২) → cache (পর্ব ৫৮) → বড় যন্ত্র → পুরোনো ডেটা archive আর একই যন্ত্রে partitioning → কাজ ধরে আলাদা ডেটাবেস → তারপর sharding. প্রতিটি ধাপের পর আবার মাপুন। sharding-এর পর প্রতিটি নতুন feature-এ প্রশ্ন আসে “এটি কয়টি shard ছোঁবে?”, foreign key আর unique constraint shard পেরিয়ে কাজ করে না, আর ব্যর্থতার নতুন নতুন ধরন আসে। এই দাম তখনই দেবেন, যখন মাপা সংখ্যা বলছে আর কোনো পথ নেই।

১০ · AI-এর যুগে distributed ডিজাইন

“আমার অ্যাপ scale করতে চাই” লিখলে AI প্রায়ই সঙ্গে সঙ্গে sharding, Kafka আর microservice-এর বড় নকশা দেয় — যা দেখতে চমৎকার, কিন্তু অনেক সময় অপ্রয়োজনীয় বা ভুল। সাধারণ ভুল:

AI-এর নকশা মানার আগে — চেকলিস্ট

• মাপা সংখ্যা দিয়ে দেখেছি যে একটি যন্ত্র (replica আর cache-সহ) সত্যিই যথেষ্ট নয়
• প্রধান কোয়েরিগুলোর তালিকা করে দেখেছি কোনটি এক shard-এ আর কোনটি সব shard-এ যাবে
• নমুনা ডেটায় shard-এর টেবিলে আর একক টেবিলে একই কোয়েরি চালিয়ে সারির সংখ্যা আর যোগফল মিলিয়েছি
• প্রতিটি কয়েক-shard লেনদেনের জন্য ব্যর্থতার পথ (coordinator মরা, compensation দুবার চলা) লিখে পরীক্ষা করেছি
• ব্যবহার করা ডেটাবেসের consistency সেটিং (W, R বা isolation) নিজের সংস্করণের ডকুমেন্টেশনে মিলিয়েছি