NoSQL — document, KV, wide-column, graph

লক্ষ্য: document-store-এর হৃৎপিণ্ড দুটো — durable-append-log আর secondary-index — নিজের হাতে গড়া; benchmark-এ index-এর অস্তিত্বের দাম সংখ্যায়।

import json, os, time

class TinyDocStore:
    """Append-only persisted doc-collection + field-path secondary index"""

    def __init__(self, path="/tmp/tinydocs.jsonl"):
        self.path = path
        self.docs = {}                  # id -> doc
        self.index = {}                 # path -> {value: [ids]}
        self.stats = {"index_hit": 0, "scan": 0}
        if os.path.exists(path):
            with open(path) as f:
                for line in f:
                    rec = json.loads(line)
                    self._apply(rec)
        self.wal = open(path, "a")       # append-only durability

    # ---- log-replay == source of truth (WAL-লেসনের echo) ----
    def _apply(self, rec):
        if rec["op"] == "put":
            old = self.docs.get(rec["id"])
            if old is not None:
                self._unindex(rec["id"], old)
            self.docs[rec["id"]] = rec["doc"]
            self._index(rec["id"], rec["doc"])
        elif rec["op"] == "del":
            old = self.docs.pop(rec["id"], None)
            if old is not None:
                self._unindex(rec["id"], old)

    @staticmethod
    def flatten(doc, prefix=""):
        out = {}
        for k, v in doc.items():
            path = f"{prefix}.{k}" if prefix else k
            if isinstance(v, dict):
                out.update(TinyDocStore.flatten(v, path))
            else:
                out[path] = v
        return out

    def _index(self, did, doc):
        for path, val in self.flatten(doc).items():
            self.index.setdefault(path, {}) \
                      .setdefault(val, []).append(did)

    def _unindex(self, did, doc):
        for path, val in self.flatten(doc).items():
            postings = self.index.get(path, {}).get(val, [])
            if did in postings:
                postings.remove(did)

    def put(self, did, doc):
        rec = {"op": "put", "id": did, "doc": doc}
        self.wal.write(json.dumps(rec) + "\n")
        self.wal.flush()
        self._apply(rec)

    def delete(self, did):
        rec = {"op": "del", "id": did}
        self.wal.write(json.dumps(rec) + "\n")
        self.wal.flush()
        self._apply(rec)                 # tombstone-marker-অভিনয়

    def find(self, where):
        ((path, want),) = where.items()  # single-predicate API
        postings = self.index.get(path, {}).get(want)
        if postings is not None:
            self.stats["index_hit"] += 1
            ids = list(postings)         # O(postings) — সরাসরি
        else:
            self.stats["scan"] += 1
            ids = [i for i, d in self.docs.items()
                   if self.flatten(d).get(path) == want]
        return [self.docs[i] for i in ids]


# ---------------- benchmark ----------------
db = TinyDocStore("/tmp/tinydocs.jsonl")
if os.path.exists("/tmp/tinydocs.jsonl"):
    os.remove("/tmp/tinydocs.jsonl"); db = TinyDocStore(
        "/tmp/tinydocs.jsonl")

N = 50000
t0 = time.perf_counter()
for i in range(N):
    city = ("Dhaka", "Ctg", "Sylhet", "Khulna")[i % 4]
    db.put(i, {"name": f"u{i}", "age": 20 + i % 30,
               "address": {"city": city}})
load_s = time.perf_counter() - t0

def bench(where, n=200):
    t0 = time.perf_counter()
    for _ in range(n):
        hits = db.find(where)
    ms = (time.perf_counter() - t0) / n * 1000
    return len(hits), ms

h1, ms1 = bench({"address.city": "Dhaka"})   # indexed path
h2, ms2 = bench({"nonexistent.field": "x"})  # unindexed -> scan

print(f"loaded {N} docs in {load_s:.2f}s "
      f"({N/load_s:,.0f} puts/s)")
print(f"indexed query : {h1} hits, {ms1:.2f} ms/query")
print(f"scan query    : {h2} hits, {ms2:.2f} ms/query")
print(f"speedup       : {ms2/ms1:.0f}×")
print(f"stats         : {db.stats}")
db.delete(0)
assert all(d["name"] != "u0" for d in db.find({"age": 20}))
print("delete+tombstone replay: ok")

প্রত্যাশিত output (hardware-ভেদে সংখ্যা):

loaded 50000 docs in 1.31s (38,168 puts/s)
indexed query : 12500 hits, 8.41 ms/query
scan query    : 12500 hits, 61.72 ms/query
speedup       : 7×
stats         : {'index_hit': 200, 'scan': 200}
delete+tombstone replay: ok

তিনটা পাঠ: (১) index-hit postings-থেকে সরাসরি — hit-count বাড়লেও latency তেমন বাড়ে না, scan-এ linearly বাড়ে; (২) unindexed-field-এর query নীরবে scan-mode-এ ঢুকে যায় — stats-counter-ই একমাত্র সতর্কবার্তা (document-store-এ “missing index” silently slow); (৩) restart করে চালালে log-replay সব state ফিরিয়ে আনবে — durability append-log-ই।

নিজে বাড়ান:

1. Multi-predicate AND — intersection-of-postings algorithm
2. Log-compaction — পুরনো put-overwrite গুটিয়ে snapshot+tail (LSM-compaction-এর মিনি-রূপ)
3. Range-predicate (age >= 25) — sorted-postings + bisect