Partitioning আর sharding — range, hash, hot spot, rebalancing আর unique ID

১ · কেন ভাগ করবেন — একটি মেশিনে যখন আর আঁটে না

একটি কাল্পনিক chat অ্যাপের কথা ভাবুন। শুরুতে সব message একটি database সার্ভারে। ব্যবহারকারী বাড়তে বাড়তে একদিন তিনটি দেয়ালে ধাক্কা: disk ভরে আসছে (table কয়েক TB); প্রতি সেকেন্ডে হাজারো নতুন message, সব লেখা একটিমাত্র সার্ভারে; আর backup ও restore-এ এখন ঘণ্টার পর ঘণ্টা।

প্রথম চেষ্টা সাধারণত বড় মেশিন — vertical scaling, পর্ব ১৩ — Vertical বনাম horizontal scaling-এ। তারপর কয়েকটি read replica — পর্ব ২২ — Replication-এ। কিন্তু replica শুধু পড়া ভাগ করে; প্রতিটিতে পুরো ডেটা, আর প্রতিটি লেখা সবগুলোতে পৌঁছায়। ডেটার আকার আর লেখার চাপ যখন একটি মেশিনের সীমা ছাড়ায়, তখন একটাই পথ: ডেটাকেই ভাগ করা।

Partitioning মানে ডেটাকে এমনভাবে টুকরো করা যাতে প্রতিটি record (সারি, document বা key) ঠিক একটি টুকরোর অংশ হয়। টুকরোগুলো আলাদা মেশিনে বসলে সাধারণত তাকে sharding বলে। এখানে ভাগ হচ্ছে সারি ধরে (horizontal partitioning); column ধরে ভাগ (vertical partitioning) আর database তত্ত্বের fragmentation আছে DBMS কোর্সের Distributed database — partitioning, sharding আর CAP পর্বে।

বাস্তবে partitioning আর replication একসাথে থাকে: ডেটা ভাগ হয় কয়েকটি shard-এ, আর প্রতিটি shard-এর থাকে একটি leader আর কয়েকটি replica. একটি মেশিনে সাধারণত কয়েকটি shard-এর কপি থাকে — কোনোটির leader, কোনোটির replica. নিচের ছবিতে পুরো চিত্রটি; এই পর্বের প্রায় প্রতিটি অংশ এর কোনো একটি টুকরো নিয়ে।

২ · Shard key — সবচেয়ে বড় সিদ্ধান্ত

কোন record কোন partition-এ যাবে, তা ঠিক করে record-এর এক বা একাধিক field — এটিই shard key, অনেক সিস্টেমে partition key. ভার সমান হবে কি না, query কয়টি shard ছোঁবে, hot spot হবে কি না — সবই এর ফল; আর বদলাতে হলে পুরো ডেটা নতুন করে ভাগ (অংশ ৮)। চারটি শর্ত:

নিচের stepper-এ chat অ্যাপের messages table-এর জন্য চারটি প্রার্থী key একে একে যাচাই করা হলো — আগে খারাপগুলো, কেন খারাপ, তারপর ভালোটি।

শেষ ধাপের key-টির দুটি অংশ: partition ঠিক করে প্রথম অংশ, আর partition-এর ভেতরে সারির ক্রম ঠিক করে দ্বিতীয় অংশ (clustering key). Cassandra-ধাঁচের database-এ লেখাটি এমন:

CREATE TABLE messages (
  conversation_id bigint,
  bucket          int,       -- time window, splits very large conversations
  message_id      bigint,    -- Snowflake-style id, grows with time
  sender_id       bigint,
  body            text,
  PRIMARY KEY ((conversation_id, bucket), message_id)
) WITH CLUSTERING ORDER BY (message_id DESC);

ভেতরের বন্ধনী (conversation_id, bucket) হলো partition key — এর hash ঠিক করে কোন node. বাইরের message_id ভেতরে নতুন থেকে পুরোনো ক্রমে সাজায় — একটি composite key-এর দুই অংশের দুই কাজ।

৩ · ভাগের নিয়ম — range, hash, compound key, directory আর geo

Shard key ঠিক হলো; এবার নিয়ম — key-এর কোন মান কোন partition-এ। পাঁচটি প্রচলিত নিয়ম আছে, আর প্রতিটির সুবিধা অন্যটির দুর্বলতা।

Range partitioning — অভিধানের খণ্ডের মতো

Range partitioning-এ key-গুলো সাজানো ক্রমে পরপর পরিসরে ভাগ হয় — অভিধানের খণ্ডের মতো: “ক–ঘ” এক খণ্ডে, “ঙ–ত” আরেকটিতে। পাশাপাশি key একই partition-এ, তাই “১ থেকে ৭ মার্চের সব অর্ডার”-এর মতো range scan অল্প কয়েকটি partition-এই শেষ। সীমা ঠিক করতে হয় ডেটা দেখে। দুর্বলতা: ক্রমাগত বাড়তে থাকা key-এ সব নতুন লেখা শেষ partition-এ — একটি hot spot.

Hash partitioning — সমান ছড়ানো, ক্রম হারানো

Hash partitioning-এ key-এর hash দেখে partition ঠিক হয় — hash(key) mod P, বা hash-এর পরিসরকে ভাগ করে। ভালো hash পাশাপাশি key-কেও দূরে ছড়ায়, তাই ক্রমিক key-তেও hot spot নেই। দাম: range scan হারায় — “১ থেকে ৭ মার্চ” এখন সব partition-এ। Hash থেকে node বাছাইয়ের নিয়ম (ring, virtual node, rendezvous) পর্ব ১৬ — Consistent hashing-এ; এখানে P মানে partition-এর সংখ্যা, node-এর নয় (অংশ ৫)।

Compound key আর salting — দুটির মাঝামাঝি

আগের CQL উদাহরণটি মাঝামাঝি পথ: প্রথম অংশের hash-এ partition, দ্বিতীয় অংশে ভেতরের ক্রম। আরেকটি কৌশল salting: key-এর সাথে একটি ছোট লেজ (যেমন #0 থেকে #7) জুড়ে একটি key-কে কয়েকটি key বানানো, যাতে সেগুলো আলাদা partition-এ যায়। লেখা ছড়ায়, কিন্তু পড়ার সময় সবগুলো লেজের key পড়ে মেলাতে হয় — দাম অংশ ৪-এ।

Directory (lookup table) — নমনীয়, কিন্তু আরেকটি সার্ভিস

Directory-ভিত্তিক sharding-এ সূত্র নেই; একটি table লিখে রাখে কোন key কোথায় — যেমন tenant 17 → shard 3. বড় গ্রাহককে নিজস্ব shard দেওয়া বা একটি tenant সরানো খুব সহজ। দাম: প্রতিটি অনুরোধে বাড়তি lookup, আর table-টি বন্ধ হলে পুরো সিস্টেম বন্ধ — তাই একে replicate আর cache করতে হয়। প্রায়ই মিশ্র রূপ: বেশিরভাগ hash দিয়ে, directory-তে শুধু ব্যতিক্রম।

Geo-partitioning — ব্যবহারকারীর কাছাকাছি, আইনের ভেতরে

Geo-partitioning-এ ডেটা ভাগ হয় এলাকা ধরে — বাংলাদেশের ব্যবহারকারীর ডেটা কাছের একটি region-এ, ইউরোপেরটা ইউরোপে। লাভ: latency কম, আর data residency-র শর্ত পূরণ — নির্দিষ্ট ডেটা নির্দিষ্ট দেশের ভেতরে রাখার আইনি বা চুক্তির বাধ্যবাধকতা। ইউরোপীয় ইউনিয়নের GDPR (২০১৮ থেকে কার্যকর) এলাকার বাইরে ব্যক্তিগত ডেটা পাঠানোয় শর্ত দেয়; বাংলাদেশসহ বহু দেশে এমন নিয়ম তৈরি ও বদলের প্রক্রিয়ায়, আর তা দেশ, খাত ও সময়ভেদে আলাদা — নকশার আগে সর্বশেষ সরকারি নির্দেশনা আর আইনি পরামর্শ নিন। প্রকৌশলের প্রশ্ন থেকেই যায়: ব্যবহারকারী দেশ বদলালে ডেটা সরবে কি না, আর দুই এলাকার মানুষের chat কোথায় থাকবে।

৪ · Hot spot আর skew — সমান ভাগেও কেন একটি shard ডোবে

অসম ভাগের (skew) দুটি রূপ। ডেটার skew: একটি partition-এ অন্যগুলোর চেয়ে অনেক বেশি ডেটা — যেমন একটি বিশাল tenant. চাপের skew: ডেটা সমান, কিন্তু একটি partition-এ অনেক বেশি অনুরোধ — যেমন একজন তারকা খেলোয়াড়ের profile, বা ভাইরাল হওয়া একটি পোস্টের like-count. Hash partitioning প্রথমটি অনেকটা সামলায়, কিন্তু দ্বিতীয়টি নয়: একটি key যত অনুরোধই পাক, সেটি এক partition-এই থাকে।

জনপ্রিয়তা প্রায়ই Zipf distribution-এর কাছাকাছি: ক্রমে k-তম key চাওয়ার সম্ভাবনা 1/k-এর সমানুপাতিক। তখন সবচেয়ে জনপ্রিয় key পায় মোট অনুরোধের 1/H_n ভাগ, যেখানে H_n = 1 + 1/2 + … + 1/n. n = ১,০০০-এ H_n ≈ ৭.৪৯, অর্থাৎ একটিমাত্র key একাই প্রায় ১৩.৪% — ৮টি shard-এর গড় ভাগ ১২.৫%-এর চেয়েও বেশি!

নিচের প্রোগ্রাম ৮টি shard-এ দুটি পরীক্ষা চালায়: (A) আজকের ৮,০০০টি message — range-এ (৬৪ দিনের ইতিহাস দেখে প্রতিটি shard ৮ দিনের, শেষটি খোলা) আর hash-এ; (B) ১,০০০ জন ব্যবহারকারীর ওপর Zipf-ধাঁচের ১৬,০০০টি অনুরোধ, তারপর শুধু user:1-কে ৮ ভাগে salting. Hash পর্ব ১৬-এর মতোই FNV-1a, আর random সংখ্যা নির্দিষ্ট seed থেকে।

// 8 shards. (A) time-ordered keys: range vs hash. (B) a Zipf hot key, then salting.
function h32(s) {                // FNV-1a 32-bit, then MurmurHash3's fmix32 to mix bits
  let h = 0x811c9dc5;
  for (let i = 0; i < s.length; i++) { h ^= s.charCodeAt(i); h = Math.imul(h, 0x01000193); }
  h ^= h >>> 16; h = Math.imul(h, 0x85ebca6b); h ^= h >>> 13; h = Math.imul(h, 0xc2b2ae35);
  return (h ^ (h >>> 16)) >>> 0;
}
let seed = 23; function rand() { // mulberry32: small seeded PRNG
  seed = (seed + 0x6D2B79F5) | 0; let t = Math.imul(seed ^ (seed >>> 15), 1 | seed);
  t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t; return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
}
const S = 8, DAY = 86400000, T0 = Date.UTC(2026, 0, 1), SALT = 8;
const show = (name, c) => console.log(name.padEnd(12) + c.map((x) => String(x).padStart(6)).join(""));
const byRange = (t) => Math.min(S - 1, Math.floor((t - T0) / (8 * DAY)));  // shard i = days 8i..8i+7
const range = new Array(S).fill(0), hash = new Array(S).fill(0);
for (let i = 0; i < 8000; i++) {              // 8,000 messages written today (day 64)
  const t = T0 + 64 * DAY + Math.floor(rand() * DAY);
  range[byRange(t)]++; hash[h32("msg:" + t + ":" + i) % S]++;
}
show("shard", [0, 1, 2, 3, 4, 5, 6, 7]); show("range(time)", range); show("hash(id)", hash);
const U = 1000, cum = []; let tot = 0;        // Zipf: user k is requested with weight 1/k
for (let k = 1; k <= U; k++) { tot += 1 / k; cum.push(tot); }
function zipfUser() {                        // binary search in the cumulative weights
  const x = rand() * tot; let lo = 0, hi = U - 1;
  while (lo < hi) { const m = (lo + hi) >> 1; if (cum[m] < x) lo = m + 1; else hi = m; } return lo + 1;
}
const plain = new Array(S).fill(0), salted = new Array(S).fill(0); let hot = 0;
for (let i = 0; i < 16000; i++) {             // 16,000 requests
  const u = zipfUser(), suffix = Math.floor(rand() * SALT); if (u === 1) hot++;
  plain[h32("user:" + u) % S]++;
  salted[h32(u === 1 ? "user:1#" + suffix : "user:" + u) % S]++;   // salt only the hot key
}
show("zipf", plain); show("salted x" + SALT, salted);
console.log("user:1 = " + hot + " of 16000 requests; busiest shard / average: " +
  (Math.max(...plain) / 2000).toFixed(2) + " -> " + (Math.max(...salted) / 2000).toFixed(2));

shard            0     1     2     3     4     5     6     7
range(time)      0     0     0     0     0     0     0  8000
hash(id)       996  1013  1005  1016  1028   978  1051   913
zipf          1317  1336  2271  2203  1847  1284  2345  3397
salted x8     1572  1630  2271  2728  2629  1559  2345  1266
user:1 = 2131 of 16000 requests; busiest shard / average: 1.70 -> 1.36

প্রতিকার — প্রতিটির দাম আছে

• Salting: লেখার সময় গরম key-এর সাথে লেজ (user:1#0 … #7); পড়ার সময় r-টি লেজের সবগুলো পড়ে মেলাতে হয় — পড়া r গুণ। তাই এটি মানায় লেখা-ভারী, চেনা গরম key-এ (like বা view গোনা); sub-key-গুলো সত্যিই আলাদা shard-এ পড়ছে কি না যাচাই করুন।
• Cache: গরম key-এর পড়া database-এর আগেই থামানো — অল্প TTL-এর local cache, কয়েকটি কপি, request coalescing; বিস্তারিত পর্ব ৩০ — Distributed cache-এ।
• Hot partition ভেঙে সরানো: গরম পরিসরের partition দুই ভাগ করে একটি অংশ অন্য node-এ — নিচের stepper. একটিমাত্র key-কে অবশ্য ভাঙা যায় না।
• নকশা বদল: বিখ্যাত অ্যাকাউন্টের পোস্ট follower-দের feed-এ লেখার বদলে পড়ার সময় আনা — পর্ব ৫৬ — Case study: news feed-এ।

client (map v7) → N2:  GET shop:1700
N2 → client:  MOVED  P2b → N4  (map v9)
client: নতুন map v9 আনল → N4:  GET shop:1700
N4 → client:  OK

৫ · Rebalancing — ডেটা সরানোর তিন কৌশল

Node যোগ হয়, বিকল হয়, ডেটা বাড়ে, partition গরম হয় — তখন ডেটা নতুন করে ভাগ করে সরাতে হয়; এটিই rebalancing. ভালো rebalancing-এর তিনটি চাওয়া: শেষে ভার সমান; সরানোর সময়েও পড়া-লেখা চলে; আর যতটুকু না সরালেই নয়, তার বেশি ডেটা নড়ে না।

যা করবেন না — key থেকে সরাসরি hash mod N, যেখানে N = node-এর সংখ্যা। পর্ব ১৬-এ প্রমাণ করা হয়েছে, N থেকে N + 1 করলে N/(N + 1) ভাগ key-এর node বদলায় — ৪ থেকে ৫-এ ৮০%. Cache-এ এর মানে হঠাৎ miss-এর ঢল; database-এ মানে প্রায় পুরো ডেটা নেটওয়ার্কে কপি।

কৌশল ১ — নির্দিষ্ট সংখ্যক partition, node-এর চেয়ে অনেক বেশি

শুরুতেই node-এর চেয়ে অনেক বেশি partition — ধরুন ৪টি node-এ ২৫৬টি (fixed partitions). ঠিকানা দুই ধাপে: partition = hash(key) mod 256, যা কখনো বদলায় না; আর node = map[partition], যা rebalancing-এ বদলায়। নতুন node এলে সে অন্যদের কাছ থেকে কয়েকটি পুরো partition নেয়; কোনো key-এর partition বদলায় না, বদলায় শুধু partition-এর মালিক। Redis Cluster-এর ১৬,৩৮৪টি hash slot ঠিক এই চিন্তা (পর্ব ১৬)।

// 256 fixed partitions on 4 nodes; add a 5th node. Compare with hash mod N.
function h32(s) {                // FNV-1a 32-bit, then MurmurHash3's fmix32 to mix bits
  let h = 0x811c9dc5;
  for (let i = 0; i < s.length; i++) { h ^= s.charCodeAt(i); h = Math.imul(h, 0x01000193); }
  h ^= h >>> 16; h = Math.imul(h, 0x85ebca6b); h ^= h >>> 13; h = Math.imul(h, 0xc2b2ae35);
  return (h ^ (h >>> 16)) >>> 0;
}
let seed = 5; function rand() {  // mulberry32: small seeded PRNG
  seed = (seed + 0x6D2B79F5) | 0; let t = Math.imul(seed ^ (seed >>> 15), 1 | seed);
  t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t; return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
}
const P = 256, K = 100000, TB = 8;              // 8 TB of data in total
const owner = Array.from({ length: P }, (_, p) => p % 4);   // partition map: owner[p] = node
const before = owner.slice();
const perNode = (n) => Array.from({ length: n }, (_, i) => owner.filter((o) => o === i).length);
console.log("before: partitions per node " + perNode(4).join(" "));
let moved = 0;
while (perNode(5)[4] < Math.floor(P / 5)) {     // node 4 takes whole partitions, busiest donor first
  const c = perNode(4), donor = c.indexOf(Math.max(...c));
  owner[owner.lastIndexOf(donor)] = 4; moved++;
}
console.log("after:  partitions per node " + perNode(5).join(" ") + "  (moved " + moved + " of " + P + ")");
let fixedMoved = 0, modMoved = 0;
for (let i = 0; i < K; i++) {
  const h = h32("order:" + Math.floor(rand() * 1e9));
  if (before[h % P] !== owner[h % P]) fixedMoved++;   // key -> partition never changes
  if (h % 4 !== h % 5) modMoved++;                    // key -> node directly by mod N
}
const pct = (x) => (100 * x / K).toFixed(1) + "%";
console.log("keys moved: fixed partitions " + pct(fixedMoved) + ", hash mod N " + pct(modMoved) + ", ideal 1/5 = 20.0%");
console.log("data copied (8 TB total): fixed " + (TB * 1024 * moved / P).toFixed(0) + " GB, mod N about " +
  (TB * 1024 * modMoved / K).toFixed(0) + " GB");

before: partitions per node 64 64 64 64
after:  partitions per node 51 51 51 52 51  (moved 51 of 256)
keys moved: fixed partitions 19.8%, hash mod N 80.2%, ideal 1/5 = 20.0%
data copied (8 TB total): fixed 1632 GB, mod N about 6571 GB

P কত হবে? P-ই node-এর সর্বোচ্চ সংখ্যা — ২৫৬টি partition-এ ২৫৬-এর বেশি node কাজে আসে না; আবার P খুব বড় হলে প্রতিটি partition-এর metadata আর replication-এর খরচ জমে। ডেটা বাড়লে partition-ও বড় হয়, আর বড় partition সরাতে সময় বেশি — বাছাইয়ের হিসাব ভাবনার প্রশ্ন ২-এ।

কৌশল ২ — চাহিদামতো ভাঙা আর জোড়া (dynamic partitioning)

Range partitioning-এ সীমা আগে ঠিক করা কঠিন, তাই অনেক সিস্টেম dynamic partitioning করে: partition নির্দিষ্ট মাপ বা চাপ ছাড়ালে ভেঙে দুটি (partition split), খুব ছোট হলে পাশেরটির সাথে জোড়া — Bigtable-এর tablet আর HBase-এর region এভাবে বাড়ে। ফাঁদ: খালি database শুরু হয় একটি partition দিয়ে, তাই আগে থেকে কয়েকটি সীমা বসানো হয় (pre-splitting).

কৌশল ৩ — consistent hashing আর virtual node

প্রতিটি node ring-এর কয়েকটি বৃত্তাংশের মালিক; নতুন node এলে গড়ে মাত্র প্রায় 1/(N + 1) ভাগ সরে, আর partition-এর সংখ্যা node-এর সমানুপাতিক — Dynamo-ধাঁচের store আর Cassandra-র token ring এই পথে (পর্ব ১৬ — Consistent hashing-এ)।

একটি partition কীভাবে সরে: snapshot কপি, কপির সময়ের পরিবর্তন replication log থেকে মিলিয়ে নেওয়া, শেষে অল্প সময় লেখা থামিয়ে মালিকানা বদল আর map-এর সংস্করণ বাড়ানো; কপির গতি বাঁধা থাকে। পুরোপুরি স্বয়ংক্রিয় rebalancing ঝুঁকির: একটি node সাময়িক ধীর হলে তাকে “মৃত” ভেবে ডেটা সরাতে শুরু করলে বাড়তি চাপে অন্যরাও ধীর হয় — একটি cascading failure. তাই অনেক দল প্রস্তাব স্বয়ংক্রিয় রাখে, অনুমোদন দেন মানুষ।

৬ · Request routing — কোন key কোন node-এ, সে খবর কে রাখে

Partition সরে, node বদলায় — অনুরোধ ঠিক node-এ পৌঁছাবে কীভাবে? এই কাজের নাম request routing, আর এর কেন্দ্রে থাকে partition map: কোন partition এখন কোন node-এ, তার সংস্করণ-নম্বরসহ তালিকা। তিনটি প্রচলিত নকশা:

Map-এর সত্যের উৎস কে? সব router আর client-কে একই সময়ে একই map দেখাতে হলে একটি নির্ভরযোগ্য উৎস লাগে। প্রায়ই সেটি একটি coordination service — যেমন etcd বা ZooKeeper — যা consensus দিয়ে নিজের ডেটা কয়েকটি মেশিনে মিলিয়ে রাখে আর বদল হলে সবাইকে জানায় (watch). Lease আর fencing token দিয়ে “দুজন একসাথে মালিক” পরিস্থিতি ঠেকানোর বিস্তারিত পর্ব ৪৩ — Coordination — distributed lock, lease আর fencing token-এ।

তবু কিছু client কিছুক্ষণ পুরোনো map দেখবেই। তাই map-এ সংস্করণ নম্বর (epoch), আর প্রতিটি node যাচাই করে partition-টি এখনো তার কি না — না হলে redirect (অংশ ৪-এর stepper, ধাপ ৫)। সবচেয়ে বিপজ্জনক partition সরানোর মুহূর্ত: পুরোনো আর নতুন মালিক দুজনেই লেখা নিলে split-brain.

৭ · Secondary index আর cross-shard query

Messages ভাগ হয়েছে conversation_id দিয়ে; নতুন feature চায় “একজন ব্যবহারকারীর পাঠানো সব message” — শর্তে sender_id, shard key নয়। Secondary index (shard key ছাড়া অন্য field-এর index; একটি database-এর ভেতরে এর মূল কথা পর্ব ২১ — Index আর query performance-এ) shard-এ ভাগ করা ডেটায় দুই রকম হতে পারে:

Local index-এ লেখা সহজ, পড়া ছড়ায়; global index-এ পড়া এক জায়গায়, লেখা ছড়ায়। Global index-কে সারির সাথে একই মুহূর্তে হালনাগাদ রাখতে দুই shard জুড়ে transaction লাগে (পর্ব ২৫), তাই বেশিরভাগ সিস্টেম একে async-ভাবে হালনাগাদ করে — বিনিময়ে eventual consistency. নামগুলো Martin Kleppmann-এর “Designing Data-Intensive Applications” (২০১৭) বইয়ের।

Scatter-gather-এর আসল দাম — সবচেয়ে ধীর shard

Local index-এ sender_id দিয়ে খোঁজা মানে প্রতিটি shard-এ একটি অনুরোধ (scatter), তারপর সব উত্তর জোড়া (gather). মোট সময় ঠিক করে সবচেয়ে ধীর উত্তর। পর্ব ৪ (Latency, throughput আর percentile)-এর fan-out সূত্র: প্রতিটি shard স্বাধীনভাবে p সম্ভাবনায় ধীর হলে N-টির অন্তত একটি ধীর হওয়ার সম্ভাবনা 1 − (1 − p)^N. নিচের simulation ১৬টি shard দিয়ে তা যাচাই করে; প্রতিটি shard-এর latency ধরে নেওয়া — সাধারণত ৫ থেকে ১৫ ms, কিন্তু ০.৫% অনুরোধে ৪০ থেকে ১০০ ms-এর একটি বিরতি (যেমন garbage collection).

// Scatter-gather: a query that needs 1 shard vs one that must wait for all 16 shards.
let seed = 1; function rand() {  // mulberry32: small seeded PRNG
  seed = (seed + 0x6D2B79F5) | 0; let t = Math.imul(seed ^ (seed >>> 15), 1 | seed);
  t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t; return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
}
function shardMs() {             // assumed: 5 ms + exponential (mean 3 ms); 0.5% of calls hit a 40-100 ms pause
  const base = 5 - 3 * Math.log(1 - rand());
  return rand() < 0.005 ? base + 40 + 60 * rand() : base;
}
const pctl = (a, q) => a.slice().sort((x, y) => x - y)[Math.ceil(q * a.length) - 1];  // nearest rank
const Q = 20000, one = [], all = [];
for (let i = 0; i < Q; i++) {
  one.push(shardMs());
  let slowest = 0;
  for (let s = 0; s < 16; s++) slowest = Math.max(slowest, shardMs());   // wait for every shard
  all.push(slowest);
}
const row = (name, a) => console.log(name.padEnd(18) + pctl(a, 0.5).toFixed(1).padStart(7) +
  pctl(a, 0.99).toFixed(1).padStart(8) + pctl(a, 0.999).toFixed(1).padStart(8));
console.log("query              p50 ms  p99 ms p99.9ms");
row("1 shard", one); row("all 16 shards", all);
const cut = pctl(one, 0.99), hit = all.filter((x) => x > cut).length / Q;
console.log("slower than the 1-shard p99: simulated " + (100 * hit).toFixed(1) + "%, formula 1 - 0.99^16 = " +
  (100 * (1 - 0.99 ** 16)).toFixed(1) + "%");
console.log("for a 16-shard p99, each shard must be fast at its p" + (100 * 0.99 ** (1 / 16)).toFixed(2));

query              p50 ms  p99 ms p99.9ms
1 shard               7.1    20.8    93.8
all 16 shards        14.9   100.5   108.0
slower than the 1-shard p99: simulated 15.0%, formula 1 - 0.99^16 = 14.9%
for a 16-shard p99, each shard must be fast at its p99.94

Cross-shard join আর transaction

দুই table-এর join আলাদা shard-এ পড়লে নেটওয়ার্কে সারি আনা-নেওয়া লাগে। প্রচলিত কৌশল:

• একসাথে রাখা (co-location): যে table-গুলো একসাথে join হয়, তাদের একই shard key — যেমন orders আর order_items দুটোই customer_id দিয়ে ভাগ, তাই একজন গ্রাহকের সব এক shard-এ।
• ছোট table সবখানে: দেশ বা পণ্যের ধরনের মতো ছোট, কম-বদলানো table-এর কপি প্রতিটি shard-এ (reference table).
• আগে থেকে জুড়ে রাখা: দরকারি তথ্যের কপি সারির ভেতরেই — denormalization (DBMS কোর্সের Denormalization পর্ব); বিনিময়ে কপি মিলিয়ে রাখার দায়।
• বিশ্লেষণ আলাদা জায়গায়: বড় aggregate আর report চলুক একটি data warehouse-এ, চলতি shard-গুলোতে নয় (DBMS কোর্সের Data warehousing পর্ব)।

দুই shard-এর ডেটা একসাথে বদলাতে — যেমন দুই shard-এর দুই অ্যাকাউন্টে টাকা পাঠানো — লাগে two-phase commit বা saga: পর্ব ২৫ — Distributed transaction; এক shard-এর ভেতরের isolation পর্ব ২৪ — Transaction আর isolation level-এ। নিয়ম: যা একসাথে atomic হতে হবে, তার ডেটা একই shard key-তে রাখুন।

৮ · চালু সিস্টেমে resharding — dual write আর backfill

কখনো পুরো ভাগটাই নতুন করে করতে হয় — shard key ভুল ছিল, বা fixed partition না থাকায় node বাড়ানো যাচ্ছে না। একে বলে resharding, আর কঠিন অংশ সিস্টেম না থামিয়ে করা — চলন্ত বাসের চাকা বদলানোর মতো। ধারাটি দুটি কৌশলে দাঁড়িয়ে: dual write (নতুন লেখা দুই জায়গাতেই) আর backfill (পুরোনো ডেটা পেছনে কপি)। নিচের ধাপগুলো একটি সাধারণ রূপরেখা; বাস্তব tool আর সিস্টেমভেদে ধাপ ও নাম আলাদা।

এই ধারা schema বদলের expand–contract পদ্ধতির আত্মীয় (পর্ব ৫৩ — Deployment আর release): নতুনটি পাশে দাঁড় করানো, দুটো একসাথে চালানো, যাচাই, তারপর পুরোনোটি সরানো — প্রতিটি ধাপে ফেরার পথ খোলা রেখে।

৯ · কেন্দ্রীয় counter ছাড়া unique ID

একটি database-এ নতুন সারির ID দেয় auto-increment — ১, ২, ৩ … কিন্তু ১৬টি shard-এর প্রত্যেকের নিজের counter থাকলে সবাই একটি করে “ID ৪২” দেবে। আর সবাই একটি কেন্দ্রীয় counter-কে জিজ্ঞেস করলে প্রতিটি লেখায় বাড়তি নেটওয়ার্ক অনুরোধ, আর counter-টি একটি single point of failure. ভালো ID হবে অনন্য, বানাতে কারও সাথে সমন্বয় লাগবে না, সম্ভব হলে সময়ক্রমে সাজানো (index আর “নতুন আগে” query সস্তা) আর ছোট।

UUID v4 — পুরোপুরি এলোমেলো

128-bit UUID-এর version 4-এ ১২২টি bit এলোমেলো; যেকোনো মেশিন একাই বানায়। মিলে যাওয়ার ভয় নগণ্য: birthday সমস্যার হিসাবে ৫০% সম্ভাবনায় একটি মিল পেতে লাগে প্রায় √(2 ln 2 · 2¹²²) ≈ ২.৭ × ১০^১৮টি ID — সেকেন্ডে ১০০ কোটি করে বানালেও প্রায় ৮৬ বছর। আসল সমস্যা index-এ: ID এলোমেলো বলে B-tree-তে প্রতিটি নতুন সারি পড়ে কোনো এলোমেলো পাতায় — memory-র পাতাগুলো কাজে লাগে না, পাতা ভাঙে বেশি। যেসব database primary key-এর ক্রমে সারি রাখে (যেমন MySQL-এর InnoDB), সেখানে ক্ষতি আরও বেশি; B-tree-র ভেতরের কথা পর্ব ১৯ — Storage engine-এ।

UUID v7 — সময়ক্রমে সাজানো UUID

২০২৪ সালে প্রকাশিত RFC 9562 (পুরোনো RFC 4122-এর জায়গায়) UUID v7 সংজ্ঞায়িত করে: প্রথম 48 bit-এ Unix সময় millisecond-এ, বাকিটা মূলত এলোমেলো। নতুন ID প্রায় সবসময় আগেরগুলোর চেয়ে বড়, তাই B-tree-র ডান প্রান্তে বসে — অথচ বানাতে সমন্বয় লাগে না। দাম: ID থেকে সৃষ্টির সময় পড়া যায়, আর আকার 128 bit; ভাষা আর database-এ সমর্থন সংস্করণভেদে আলাদা। সাবধান: সময়ক্রমের ID দিয়ে range partitioning করলে আবার অংশ ৩-এর hot spot — এমন ID hash করে ভাগ করুন।

Snowflake-ধাঁচের 64-bit ID

প্রকাশিত engineering blog অনুযায়ী Twitter ২০১০ সালে Snowflake নামের ID-সার্ভিস ঘোষণা ও open-source করে। Snowflake-ধাঁচের ID-র গঠন: ওপরের 1 bit ফাঁকা (সংখ্যা ধনাত্মক রাখতে), 41 bit সময় — নিজের বাছা epoch থেকে millisecond, 10 bit worker-এর নম্বর, 12 bit ক্রমিক সংখ্যা।

সময় সবচেয়ে ওপরের bit-গুলোতে, তাই ID-গুলো মোটামুটি সময়ক্রমে সাজানো — k-sorted. নিচের প্রোগ্রাম একটি generator বানায় (নির্ভুল হিসাবে BigInt), ID ভেঙে দেখায়, ক্রম যাচাই করে, এক millisecond-এ ৪,০৯৭টি ID চায়, আর ক্ষমতা ও শেষ হওয়ার তারিখ হিসাব করে।

// Snowflake-style 64-bit ID = 41-bit ms timestamp | 10-bit worker | 12-bit sequence
const EPOCH = BigInt(Date.UTC(2024, 0, 1));            // custom epoch: 2024-01-01 00:00 UTC
const make = (ms, w, seq) => ((BigInt(ms) - EPOCH) << 22n) | (BigInt(w) << 12n) | BigInt(seq);
const parse = (id) => ({ ms: Number((id >> 22n) + EPOCH), worker: Number((id >> 12n) & 1023n), seq: Number(id & 4095n) });
function generator(worker) {     // one per machine: no network call, no central counter
  let last = -1, seq = 0;
  return (ms) => {               // ms = this machine's clock reading
    if (ms <= last) {            // same ms (or clock behind): next sequence number
      ms = last; seq = (seq + 1) & 4095;
      if (seq === 0) ms = last + 1;     // all 4,096 used: move to the next ms
    } else seq = 0;
    last = ms; return make(ms, worker, seq);
  };
}
const t = Date.UTC(2026, 9, 28, 10, 0, 0);              // a fixed clock reading for the demo
const w7 = generator(7), w300 = generator(300);
const made = [["w300", w300(t)], ["w7", w7(t)], ["w7", w7(t)], ["w300", w300(t + 1)], ["w7", w7(t + 1)]];
made.forEach(([who, id], i) => { const p = parse(id);
  console.log("#" + i + " " + who.padEnd(5) + String(id).padStart(19) + "  +" + (p.ms - t) + "ms worker=" + p.worker + " seq=" + p.seq); });
const order = made.map((m, i) => [m[1], i]).sort((a, b) => (a[0] < b[0] ? -1 : 1)).map((x) => "#" + x[1]);
console.log("sorted by ID: " + order.join(" ") + "   (exact order only between different ms)");
let id; for (let i = 0; i < 4097; i++) id = w7(t + 5);  // burst: 4,097 IDs asked in one ms
const q = parse(id); console.log("4,097th ID asked at +5ms: +" + (q.ms - t) + "ms seq=" + q.seq + " (sequence was full)");
console.log("capacity: " + 4096 * 1000 + " IDs/s per worker, " +
  (4096 * 1000 * 1024 / 1e9).toFixed(2) + " billion/s for 1,024 workers");
const years = 2 ** 41 / (365.25 * 86400000);
console.log("41-bit ms timestamp lasts " + years.toFixed(1) + " years -> overflows on " +
  new Date(Number(EPOCH) + 2 ** 41).toISOString().slice(0, 10));

#0 w300  373772884378828800  +0ms worker=300 seq=0
#1 w7    373772884377628672  +0ms worker=7 seq=0
#2 w7    373772884377628673  +0ms worker=7 seq=1
#3 w300  373772884383023104  +1ms worker=300 seq=0
#4 w7    373772884381822976  +1ms worker=7 seq=0
sorted by ID: #1 #2 #0 #4 #3   (exact order only between different ms)
4,097th ID asked at +5ms: +6ms seq=0 (sequence was full)
capacity: 4096000 IDs/s per worker, 4.19 billion/s for 1,024 workers
41-bit ms timestamp lasts 69.7 years -> overflows on 2093-09-06

দুটি দুর্বল জায়গা। ঘড়ি: NTP ঘড়ি পেছালে একই সময় আবার আসে, ID মিলে যেতে পারে — তাই generator শেষ ব্যবহৃত সময় মনে রাখে আর ঘড়ি পেছালে অপেক্ষা করে বা ID দিতে অস্বীকার করে; clock skew-এর গভীরে পর্ব ৩৯ — সময় আর ক্রম-এ। Worker নম্বর: দুটি মেশিন একই নম্বর পেলে ID মিলবে; নম্বর আসে config থেকে বা coordination service-এর lease থেকে (পর্ব ৪৩) — container আসা-যাওয়ার যুগে এটিই সবচেয়ে সাধারণ ভুল।

Ticket server আর ID-র ব্লক

আরেকটি সরল পথ: কেন্দ্রীয় counter একবারে একটি ব্লক দেয় — একটি app server ১,০০০টি ID নিয়ে (৫০,০০১ থেকে ৫১,০০০) নিজের memory থেকে বিলোয়। কেন্দ্রীয় অনুরোধ হাজার গুণ কম; দাম restart-এ ফাঁক, আর সার্ভারভেদে সময়ক্রম নেই। Flickr-এর প্রকাশিত engineering blog (২০১০) অনুযায়ী তাঁরা দুটি MySQL ticket server চালাতেন — একটি বিজোড়, অন্যটি জোড় সংখ্যা দেয়। ছোট আর অনুমান-অযোগ্য ID-র নকশা পর্ব ৫৪ — Case study: URL shortener-এ।

১০ · AI-এর যুগে — sharding-এর নকশায় AI-কে যাচাই

“Database কীভাবে shard করব?” — AI কয়েক সেকেন্ডে উত্তর দেয়, প্রায়ই “user_id দিয়ে hash, সাথে Kafka আর Redis”. আপনার access pattern না জেনে দেওয়া এমন উত্তরের ভুলগুলো নির্দিষ্ট:

AI-এর sharding নকশা হাতে পেলে — যাচাইয়ের চেকলিস্ট

• ডেটার আকার, লেখার QPS আর বৃদ্ধির হিসাব নিজে আবার করেছি — sharding সত্যিই দরকার কি না দেখেছি
• শীর্ষ query-গুলোর প্রতিটি কয়টি shard ছোঁয়, লিখে মিলিয়েছি
• Node যোগ-বাদে কত partition আর কত ডেটা সরবে, simulation বা হাতে হিসাব করেছি
• সবচেয়ে গরম key-এর চাপ (Zipf-এর শীর্ষ ভাগ) অনুমান করে hot spot-এর পরিকল্পনা রেখেছি
• ID-র ক্ষমতা আর আয়ু হিসাব করেছি; ঘড়ি পেছানো আর worker নম্বরের সংঘর্ষ নিয়ে failure-mode পর্যালোচনা করেছি
• Index-এর consistency আর product-এর প্রতিটি সীমা অফিসিয়াল documentation-এ মিলিয়েছি