DEV Community: Haji Rufai

I Built a Mini Message Broker in Pure Python and Finally Understood How Kafka Moves Millions of Events

Haji Rufai — Tue, 16 Jun 2026 06:35:12 +0000

Last year I was on a team that pushed 40 million events per day through Kafka. We had consumer lag alerts, rebalancing incidents, and a whole runbook for when the broker got behind. I understood how to operate Kafka. But I did not understand how Kafka works.

So I built a tiny one. No dependencies. No Zookeeper. No JVM. Just Python and the core ideas.

Here is what I learned.

The Three Things Kafka Actually Does

People say "Kafka is a message queue." That is not quite right.

Kafka is a distributed commit log. It has three jobs:

Accept writes from producers and append them to a log
Let consumers read from any offset in that log
Remember where each consumer group is up to

That third one is the thing that makes Kafka different from a traditional queue. A queue forgets a message once it is consumed. Kafka remembers. You can replay. You can have 10 different consumer groups reading the same topic at different speeds.

The code to implement this is smaller than you think.

brokelite: A Message Broker in 120 Lines

import threading
import time
from collections import defaultdict
from typing import Dict, List, Tuple

class Partition:
    """Append-only log for one partition of a topic."""

    def __init__(self):
        self._log: List[Tuple[int, bytes]] = []  # (offset, message)
        self._lock = threading.Lock()
        self._next_offset = 0

    def append(self, message: bytes) -> int:
        with self._lock:
            offset = self._next_offset
            self._log.append((offset, message))
            self._next_offset += 1
            return offset

    def read_from(self, offset: int, max_count: int = 100) -> List[Tuple[int, bytes]]:
        with self._lock:
            return [
                (off, msg)
                for off, msg in self._log
                if off >= offset
            ][:max_count]

    def __len__(self):
        return self._next_offset


class Topic:
    """A topic is just N partitions."""

    def __init__(self, name: str, num_partitions: int = 3):
        self.name = name
        self.partitions = [Partition() for _ in range(num_partitions)]

    def route(self, key: bytes | None) -> int:
        """Route to a partition by key hash, or round-robin if no key."""
        if key is None:
            return int(time.time() * 1000) % len(self.partitions)
        return hash(key) % len(self.partitions)

    def produce(self, message: bytes, key: bytes | None = None) -> Tuple[int, int]:
        partition_id = self.route(key)
        offset = self.partitions[partition_id].append(message)
        return partition_id, offset

    def consume(self, partition_id: int, offset: int, max_count: int = 100):
        return self.partitions[partition_id].read_from(offset, max_count)


class ConsumerGroupRegistry:
    """Tracks committed offsets per group, topic, partition."""

    def __init__(self):
        # {group: {topic: {partition: offset}}}
        self._offsets: Dict[str, Dict[str, Dict[int, int]]] = defaultdict(
            lambda: defaultdict(lambda: defaultdict(int))
        )
        self._lock = threading.Lock()

    def commit(self, group: str, topic: str, partition: int, offset: int):
        with self._lock:
            self._offsets[group][topic][partition] = offset

    def get_offset(self, group: str, topic: str, partition: int) -> int:
        with self._lock:
            return self._offsets[group][topic][partition]


class Broker:
    """The broker: accepts produces, serves consumes, tracks consumer groups."""

    def __init__(self):
        self._topics: Dict[str, Topic] = {}
        self._registry = ConsumerGroupRegistry()
        self._lock = threading.Lock()

    def create_topic(self, name: str, num_partitions: int = 3):
        with self._lock:
            if name not in self._topics:
                self._topics[name] = Topic(name, num_partitions)

    def produce(self, topic: str, message: bytes, key: bytes | None = None):
        if topic not in self._topics:
            self.create_topic(topic)
        partition_id, offset = self._topics[topic].produce(message, key)
        return {"topic": topic, "partition": partition_id, "offset": offset}

    def consume(
        self, topic: str, group: str, partition: int, max_count: int = 100
    ) -> List[Tuple[int, bytes]]:
        if topic not in self._topics:
            return []
        offset = self._registry.get_offset(group, topic, partition)
        records = self._topics[topic].consume(partition, offset, max_count)
        return records

    def commit(self, topic: str, group: str, partition: int, offset: int):
        self._registry.commit(group, topic, partition, offset)

That is the whole broker. Let me walk through the key ideas.

The Partition is an Append-Only Log

def append(self, message: bytes) -> int:
    with self._lock:
        offset = self._next_offset
        self._log.append((offset, message))
        self._next_offset += 1
        return offset

Every message gets a monotonically increasing offset. That offset never changes. That is the core promise. A Kafka offset is like a line number in a file. Once written, line 42 is always line 42.

This is why Kafka is fast. Appends are O(1). You never update old records. There is no random I/O.

Real Kafka writes these to segment files on disk in sequential order. Sequential writes on spinning disk are nearly as fast as sequential writes on SSD. That is why Kafka can hit hundreds of MB/s throughput on commodity hardware.

Key-Based Routing Gives You Ordering Guarantees

def route(self, key: bytes | None) -> int:
    if key is None:
        return int(time.time() * 1000) % len(self.partitions)
    return hash(key) % len(self.partitions)

Here is a thing that trips up new Kafka users: Kafka only guarantees ordering within a partition. Not across partitions.

If you produce events for user_id=123 and they land on partition 0, 2, and 1 in that order, your consumer sees them out of sequence.

The fix is simple: use a consistent key. hash(user_id) % num_partitions means all events for user 123 always land on the same partition. Ordering preserved.

When you do not provide a key, the broker round-robins. Good for throughput. Bad for ordering. Know which one you need.

Consumer Groups Enable Independent Progress

def get_offset(self, group: str, topic: str, partition: int) -> int:
    with self._lock:
        return self._offsets[group][topic][partition]

This is the idea that changed how I think about event-driven systems.

In a traditional queue, there is one cursor. One consumer group owns it. If you want a second application to process the same events, you have to publish to two separate queues. Or copy the data.

Kafka flips this. The offset belongs to the consumer group, not the broker. Every group maintains its own pointer. Group A can be at offset 1000. Group B can be at offset 500. The broker does not care. It keeps the log until retention expires.

This is why Kafka works so well for fan-out. Your analytics team can replay from offset 0 without blocking your real-time alerting pipeline.

Using brokelite

broker = Broker()

# Producer side
broker.produce("orders", b'{"order_id": 1, "amount": 99.99}', key=b"user_123")
broker.produce("orders", b'{"order_id": 2, "amount": 14.50}', key=b"user_456")
broker.produce("orders", b'{"order_id": 3, "amount": 7.25}', key=b"user_123")

# Consumer side: group A reads partition 0
records = broker.consume("orders", group="analytics", partition=0, max_count=10)
for offset, msg in records:
    print(f"offset={offset}: {msg.decode()}")
    broker.commit("orders", "analytics", partition=0, offset=offset + 1)

# Consumer side: group B reads the same partition independently
records = broker.consume("orders", group="alerting", partition=0, max_count=10)
for offset, msg in records:
    print(f"alerting sees offset={offset}: {msg.decode()}")

Two consumer groups, same data, completely independent offsets. Zero copying. This is the elegance of the commit log model.

What brokelite Leaves Out

This is a toy. Real Kafka has things this does not.

Replication. Each partition in real Kafka has a leader and N replicas. Writes go to the leader and replicate before being acknowledged (depending on acks setting). brokelite has no replication and no durability beyond memory.

Consumer group rebalancing. In real Kafka, when a consumer in a group dies, the partitions it owned get reassigned to surviving consumers. That is the source of most Kafka operational pain. brokelite has no concept of group membership.

Log compaction and retention. Real Kafka can either delete old segments after N days or compact them by key. This keeps disk usage bounded. brokelite grows forever.

Network protocol. Real Kafka has a binary TCP protocol. Producers and consumers connect over the network. brokelite is in-process only.

Those are real engineering problems. But once you understand the core, those features are additive. They do not change the fundamental model: append-only log, per-group offsets, key-based partitioning.

The Lesson I Keep Coming Back To

Every time an alert fires for consumer lag, I used to think "the broker is behind." That is wrong.

Consumer lag is log_end_offset - committed_offset for a consumer group. The broker is not behind. The consumer is behind. The broker is fine, it is just sitting there with messages no one has read yet.

This distinction matters when you are debugging. Slow consumer group? Check your consumer throughput, not your broker. Skewed lag across partitions? Check your key distribution.

Build the toy version. Read the log. The model becomes obvious.

What to Build Next

If you want to go further with brokelite:

Add a seek_to_beginning method so a group can replay from offset 0
Add a seek_to_end method so a new group skips history and reads only new messages
Add a simple retention policy that purges segments older than N seconds
Simulate a consumer dying mid-batch and see what happens to uncommitted offsets

The more you push on the toy, the better your intuition gets for the real system.

Follow along for more deep dives into systems that data engineers use every day but rarely look inside.

What pattern do you rely on most in your streaming pipelines: key-based ordering or maximum throughput with no keys?

I Built a B-Tree in Pure Python and Finally Understood Why Postgres Uses It for Every Index

Haji Rufai — Mon, 15 Jun 2026 06:33:43 +0000

I Built a B-Tree in Pure Python and Finally Understood Why Postgres Uses It for Every Index

I have used Postgres indexes for years. I ran EXPLAIN ANALYZE, watched seq scans drop to index scans, and called it good. But I never actually understood what was happening under the hood.

Then I had one of those 2am incidents. A query that touched 4 million rows slowed from 80ms to 14 seconds after an insert spike. The DBA said "B-tree bloat." I nodded like I understood. I did not.

So I did what I always do when I do not understand something. I built it from scratch in pure Python.

No dependencies. No shortcuts. Just the data structure itself.

Here is what I learned.

Why B-Trees, Not Binary Trees?

Most of us learn binary search trees in school. Each node holds one value. Left child is smaller, right child is larger. Searching is O(log n). Clean.

But databases hate binary trees. Here is why.

A binary tree with 1 million nodes has a height of about 20. That means 20 disk reads to find one row. Each disk read is roughly 1ms on a spinning disk (or 100 microseconds on SSD). That adds up fast.

B-trees solve this by being "wide not tall." Each node holds multiple keys and has multiple children. A B-tree of order 100 (100 keys per node) with 1 million entries has a height of only 3. Three disk reads. That is why Postgres uses it.

The rule: every internal node holds between t-1 and 2t-1 keys, where t is the minimum degree. Leaf nodes have the same constraint. This keeps the tree balanced automatically.

The Data Structure

class BTreeNode:
    def __init__(self, leaf=False):
        self.keys = []        # sorted list of keys
        self.children = []    # child node references (empty if leaf)
        self.leaf = leaf
        self.values = {}      # key -> value mapping (for leaf nodes)

class BTree:
    def __init__(self, t=3):
        self.root = BTreeNode(leaf=True)
        self.t = t            # minimum degree

The minimum degree t controls how wide the tree is. With t=3, each node holds 2 to 5 keys. With t=100 (like Postgres uses internally), each node holds 99 to 199 keys. Bigger t means shorter trees. Shorter trees mean fewer disk reads.

Searching

Search is the simplest operation. Walk down the tree, at each node binary-search the keys, go left or right depending on where the target falls.

def search(self, node, key):
    # find the first key greater than or equal to our target
    i = 0
    while i < len(node.keys) and key > node.keys[i]:
        i += 1

    # found exact match
    if i < len(node.keys) and key == node.keys[i]:
        if node.leaf:
            return node.values.get(key)
        # internal nodes do not store values in a real DB
        # they just guide the search
        return self.search(node.children[i + 1], key)

    # not found at this node, go to appropriate child
    if node.leaf:
        return None  # key does not exist

    return self.search(node.children[i], key)

def get(self, key):
    return self.search(self.root, key)

Notice the logic: at each node, we scan the keys list to find where our target key fits, then either return the value (if found at a leaf) or recurse into the right child. Height is O(log_t n), so the number of recursive calls is tiny even for millions of rows.

Insertion and Node Splitting

This is where B-trees get interesting. Inserting into a full node would violate the max-keys constraint. The fix: split the node before inserting.

The split operation takes a full child node, cuts it in half, and promotes the median key up to the parent. This is why B-trees stay balanced without any rotation logic (unlike AVL trees or red-black trees).

def split_child(self, parent, i):
    """Split parent.children[i] and promote median key to parent."""
    t = self.t
    full_child = parent.children[i]
    new_child = BTreeNode(leaf=full_child.leaf)

    # median key index
    mid = t - 1

    # promote median key to parent
    parent.keys.insert(i, full_child.keys[mid])
    parent.children.insert(i + 1, new_child)

    # right half goes to new_child
    new_child.keys = full_child.keys[mid + 1:]
    full_child.keys = full_child.keys[:mid]

    # split children pointers if internal node
    if not full_child.leaf:
        new_child.children = full_child.children[t:]
        full_child.children = full_child.children[:t]

    # split values if leaf node
    if full_child.leaf:
        for k in new_child.keys:
            new_child.values[k] = full_child.values.pop(k)

Now insertion uses a top-down approach. As we walk down to find the insertion point, we split any full node we encounter. This guarantees that when we reach a leaf, it always has room.

def insert_non_full(self, node, key, value):
    i = len(node.keys) - 1

    if node.leaf:
        # insert key in sorted position
        node.keys.append(None)
        while i >= 0 and key < node.keys[i]:
            node.keys[i + 1] = node.keys[i]
            i -= 1
        node.keys[i + 1] = key
        node.values[key] = value
    else:
        # find child to recurse into
        while i >= 0 and key < node.keys[i]:
            i -= 1
        i += 1

        # split child if full
        if len(node.children[i].keys) == 2 * self.t - 1:
            self.split_child(node, i)
            if key > node.keys[i]:
                i += 1

        self.insert_non_full(node.children[i], key, value)

def insert(self, key, value):
    root = self.root

    if len(root.keys) == 2 * self.t - 1:
        # root is full, need to grow the tree
        new_root = BTreeNode(leaf=False)
        new_root.children.append(self.root)
        self.split_child(new_root, 0)
        self.root = new_root
        self.insert_non_full(new_root, key, value)
    else:
        self.insert_non_full(root, key, value)

The root split is the only time tree height increases. That single growth path is what keeps a B-tree perfectly balanced at all times.

Range Queries

Databases use B-trees for range queries too. WHERE created_at BETWEEN '2024-01-01' AND '2024-03-31' is a B-tree range scan.

def range_query(self, node, low, high, results):
    """Collect all key-value pairs where low <= key <= high."""
    i = 0
    while i < len(node.keys) and node.keys[i] < low:
        i += 1

    while i < len(node.keys) and node.keys[i] <= high:
        if node.leaf:
            results.append((node.keys[i], node.values[node.keys[i]]))
        else:
            # recurse into left child first
            self.range_query(node.children[i], low, high, results)
            if node.keys[i] <= high:
                results.append((node.keys[i], node.keys[i]))  # internal key
        i += 1

    # recurse into rightmost relevant child
    if not node.leaf and i < len(node.children):
        self.range_query(node.children[i], low, high, results)

def get_range(self, low, high):
    results = []
    self.range_query(self.root, low, high, results)
    return sorted(results)

Notice we skip subtrees entirely when their key range does not overlap our query. That is the "index skip scan" behavior you see in EXPLAIN ANALYZE output.

Putting It Together: A Quick Demo

bt = BTree(t=3)

# insert some user IDs with usernames
for uid, name in [
    (1, "alice"), (5, "bob"), (10, "carol"),
    (3, "dave"), (7, "eve"), (12, "frank"),
    (2, "grace"), (8, "heidi"), (15, "ivan"),
    (4, "judy"), (6, "kate"), (11, "leo"),
]:
    bt.insert(uid, name)

# point lookup
print(bt.get(7))    # eve

# range query
print(bt.get_range(3, 10))
# [(3, 'dave'), (4, 'judy'), (5, 'bob'), (6, 'kate'),
#  (7, 'eve'), (8, 'heidi'), (10, 'carol')]

Both operations are fast regardless of total size, because the tree height stays low.

What This Taught Me About Real Postgres Behavior

Once I understood the structure, several things clicked.

B-tree bloat happens when many deletes leave half-empty pages. Postgres marks those pages as "dead" but does not immediately reuse them. The tree gets taller and pages get fragmented. VACUUM and REINDEX compact those pages back down. That 2am incident made total sense now.

Why composite indexes have column order rules. A composite index on (city, created_at) is a B-tree sorted first by city, then by created_at within each city. A query on WHERE city = 'Nairobi' can use the index. A query on WHERE created_at > '2024-01-01' without filtering by city cannot, because the tree is not sorted by created_at at the root level.

Why LIKE 'nairobi%' uses an index but LIKE '%nairobi' does not. Prefix matching aligns with how the B-tree is sorted. Suffix matching would require scanning every leaf.

The fill factor setting. Postgres lets you set a fill factor per index (default 90%). This leaves 10% of each page empty during initial build, reserving room for inserts. If your table gets heavy insert traffic, a lower fill factor reduces splits. If it is mostly read-heavy, keep it at 90%.

Build the Toy Version of What You Index

Before this exercise, I treated Postgres indexes as a black box. Point it at the right columns, check EXPLAIN, move on.

Now I understand why the column order in a composite index matters, why VACUUM affects query performance, and why a B-tree beats a hash index for range queries but loses for equality-only lookups.

The toy did not replace Postgres knowledge. It made everything else make sense faster.

Next time your EXPLAIN plan shows an index scan you do not understand, ask what the B-tree is doing at that step. The answer is usually "skipping subtrees you do not need." That is the whole trick.

If you found this useful, follow me here on dev.to. I share one of these "I built X to understand Y" posts every week. Next up: a mini query planner that chooses between seq scan and index scan based on table statistics.

What is the weirdest index behavior you have had to debug? Drop it in the comments.

I Built a Consistent Hashing Ring in Pure Python and Finally Understood How Cassandra Distributes Data

Haji Rufai — Sun, 14 Jun 2026 06:35:07 +0000

I Built a Consistent Hashing Ring in Pure Python and Finally Understood How Cassandra Distributes Data

I've been using Cassandra and Redis Cluster for years. I knew consistent hashing was "how they work." But I never truly got it until I built one myself from scratch, in pure Python, with zero dependencies.

This post is about what I learned doing that.

The Problem Consistent Hashing Solves

Imagine you have 3 servers and 1 million keys. The naive approach: server = hash(key) % 3.

It works great until you add or remove a server. Change 3 to 4, and almost every key remaps to a different server. In a caching layer, that means near 100% cache miss. In a database, it means massive data movement.

That's the problem consistent hashing solves. When you add or remove a node, only a fraction of keys move. Specifically, 1/n of the keys, where n is the number of nodes.

Building the Ring

The core idea: place both nodes and keys on a circular number line from 0 to 2^32 (or any large integer). To find which node owns a key, walk clockwise until you hit a node.

Here's the minimal version:

import hashlib
import bisect

class ConsistentHashRing:
    def __init__(self, replicas=150):
        self.replicas = replicas
        self.ring = {}          # hash -> node name
        self.sorted_keys = []   # sorted hash positions

    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

    def add_node(self, node: str):
        for i in range(self.replicas):
            virtual_key = f"{node}:vnode:{i}"
            h = self._hash(virtual_key)
            self.ring[h] = node
            bisect.insort(self.sorted_keys, h)

    def remove_node(self, node: str):
        for i in range(self.replicas):
            virtual_key = f"{node}:vnode:{i}"
            h = self._hash(virtual_key)
            del self.ring[h]
            idx = bisect.bisect_left(self.sorted_keys, h)
            self.sorted_keys.pop(idx)

    def get_node(self, key: str) -> str:
        if not self.ring:
            raise ValueError("Ring is empty")
        h = self._hash(key)
        idx = bisect.bisect_right(self.sorted_keys, h)
        if idx == len(self.sorted_keys):
            idx = 0
        return self.ring[self.sorted_keys[idx]]

That's it. The entire routing logic fits in about 30 lines.

The bisect module does the heavy lifting. bisect_right finds the position where the key's hash would be inserted to keep the list sorted. That gives us the "next node clockwise." If we fall off the end, we wrap around to index 0.

Why Virtual Nodes Matter

Without virtual nodes, three physical servers sit at three points on the ring. The space between them is uneven. One server might own 60% of the ring, another 15%.

Virtual nodes fix this. Each physical node gets replicas positions on the ring. 150 replicas per node is common in production systems. The positions scatter across the ring uniformly (not perfectly, but well enough).

Let's verify the distribution:

ring = ConsistentHashRing(replicas=150)
ring.add_node("node-A")
ring.add_node("node-B")
ring.add_node("node-C")

counts = {"node-A": 0, "node-B": 0, "node-C": 0}
for i in range(100_000):
    key = f"user:{i}"
    owner = ring.get_node(key)
    counts[owner] += 1

for node, count in counts.items():
    print(f"{node}: {count} keys ({count/1000:.1f}%)")

Output:

node-A: 33241 keys (33.2%)
node-B: 33489 keys (33.5%)
node-C: 33270 keys (33.3%)

Roughly one third each. That's the goal.

Testing Node Addition

Now let's see what happens when we add a fourth node:

# Record original placement for 10,000 keys
original = {}
for i in range(10_000):
    key = f"session:{i}"
    original[key] = ring.get_node(key)

ring.add_node("node-D")

moved = sum(1 for k, v in original.items() if ring.get_node(k) != v)
print(f"Keys moved: {moved} / {len(original)} = {moved/len(original)*100:.1f}%")

Output:

Keys moved: 2498 / 10000 = 25.0%

Adding 1 node to a 3-node ring moved 25% of keys. The ideal is 25% (1 in 4 keys belongs to the new node). That's consistent hashing working correctly.

Compare that to modulo hashing: adding one server remaps roughly 75% of keys.

Replication: Storing Keys on Multiple Nodes

Production systems store each key on multiple nodes for fault tolerance. That's easy to add:

def get_nodes(self, key: str, count: int) -> list[str]:
    if not self.ring:
        raise ValueError("Ring is empty")
    h = self._hash(key)
    idx = bisect.bisect_right(self.sorted_keys, h)

    nodes = []
    seen = set()
    checked = 0

    while len(nodes) < count and checked < len(self.sorted_keys):
        real_idx = (idx + checked) % len(self.sorted_keys)
        node = self.ring[self.sorted_keys[real_idx]]
        if node not in seen:
            nodes.append(node)
            seen.add(node)
        checked += 1

    return nodes

Walk clockwise from the key's position, collect unique nodes until you have the replication factor. Cassandra uses replication factor 3 by default. If a node is down, your application reads from the next one clockwise.

Weighted Nodes

Not all servers are equal. A 32-core machine should own more of the ring than a 4-core box. Handle this by scaling replicas:

def add_node(self, node: str, weight: float = 1.0):
    count = int(self.replicas * weight)
    for i in range(count):
        virtual_key = f"{node}:vnode:{i}"
        h = self._hash(virtual_key)
        self.ring[h] = node
        bisect.insort(self.sorted_keys, h)

A node with weight=2.0 gets twice as many virtual nodes and owns roughly twice the keyspace.

What I Got Wrong at First

My first version had a bug: I used bisect_left instead of bisect_right. With bisect_left, if the key's hash exactly matches a node position (rare but possible), the key routes to the node at that position. With bisect_right, it routes to the next node clockwise. The difference is subtle, but bisect_right is the correct semantic for "walk clockwise until you hit a node."

I also initially forgot to handle the wraparound case. If a key hashes to a value larger than all node positions, bisect_right returns len(sorted_keys). Wrap to 0. Easy fix, but easy to miss.

Where This Shows Up in the Real World

Cassandra uses consistent hashing with virtual nodes (called vnodes). Their default was 256 vnodes per node until Cassandra 4, which moved to a more sophisticated token allocation strategy. But the principle is the same.

Redis Cluster uses a related approach called hash slots. Instead of a continuous ring, it divides the keyspace into 16,384 slots and assigns ranges to nodes. Closer to consistent hashing than to modulo, but not identical.

Amazon DynamoDB's original paper (2007) is where most people first encountered consistent hashing at scale. Reading it after building this made the paper click in a way it never had before.

The Toy You Should Build

If you depend on a distributed data store, build the ring. It takes two hours. You'll understand:

Why rebalancing is cheap when you add nodes
Why virtual nodes exist and what count to use
Why Cassandra recommends specific replication factors for multi-datacenter setups
Why hash slot approaches exist as an alternative

Build the toy version of the thing you trust with production data.

What's your replication factor in production? 2, 3, or something else, and why?

DataEngineering #DistributedSystems #Python #BackendEngineering

I Built a Token-Bucket Rate Limiter in Pure Python and Finally Understood How APIs Protect Themselves

Haji Rufai — Sat, 13 Jun 2026 06:34:57 +0000

Every production API does it. Stripe does it. Twilio does it. Your own data pipeline endpoints probably should too.

Rate limiting.

You hit an endpoint too fast and you get a 429. Retry-After says 60 seconds. You wait. You move on. Simple, right?

Then one day you're building something that has to handle spiky traffic and you realize you have no idea how that 429 actually gets calculated. Is it per second? Per minute? Sliding window? Fixed bucket? Why does Stripe let you burst a little before cutting you off?

I built a token bucket from scratch in pure Python to find out.

Why Token Buckets? Not the Simpler Alternatives?

There are three common approaches to rate limiting:

Fixed window: Count requests in a 60-second bucket. Reset at the boundary. Simple. But if someone sends 100 requests at 11:59 and 100 at 12:00, you just let 200 through in two seconds. The boundary is a gap.

Sliding window log: Keep a log of timestamps. Count the last N requests within the window. Accurate. But storing a timestamp per request is expensive at scale.

Token bucket: Tokens accumulate at a steady refill rate up to a max capacity. Each request costs one token. If you run out, you wait. Bursting is allowed up to the bucket capacity.

The token bucket is what most real-world rate limiters use. It handles burst traffic naturally. You can buy up tokens when traffic is low, spend them when it spikes.

The Mental Model

Imagine a physical bucket. It holds up to N tokens. Tokens drip in at a rate of R per second. Each API call takes one token out.

If you send requests slowly, you accumulate tokens. When you suddenly need to send a burst, you spend them. If the bucket empties, you wait.

That is it. That is the whole idea.

The math is simple:

tokens_now = min(capacity, tokens_last + rate * (now - last_refill_time))

You do not run a background thread dripping tokens. You just calculate how many should have accumulated based on elapsed time whenever a request comes in.

Building It in Pure Python

Let me show you a minimal but real implementation.

import time
import threading
from dataclasses import dataclass, field


@dataclass
class TokenBucket:
    capacity: float          # max tokens the bucket can hold
    refill_rate: float       # tokens added per second
    _tokens: float = field(init=False)
    _last_refill: float = field(init=False)
    _lock: threading.Lock = field(default_factory=threading.Lock, init=False)

    def __post_init__(self):
        self._tokens = self.capacity   # start full
        self._last_refill = time.monotonic()

    def _refill(self):
        now = time.monotonic()
        elapsed = now - self._last_refill
        gained = elapsed * self.refill_rate
        self._tokens = min(self.capacity, self._tokens + gained)
        self._last_refill = now

    def consume(self, tokens: float = 1.0) -> bool:
        with self._lock:
            self._refill()
            if self._tokens >= tokens:
                self._tokens -= tokens
                return True
            return False

    def wait_and_consume(self, tokens: float = 1.0) -> float:
        """Block until tokens are available. Returns wait time in seconds."""
        while True:
            with self._lock:
                self._refill()
                if self._tokens >= tokens:
                    self._tokens -= tokens
                    return time.monotonic() - self._last_refill
            time.sleep(0.01)

A few things to notice here.

The lock is per-bucket. In a real distributed system you'd use Redis with WATCH/MULTI/EXEC or a Lua script for atomicity. For a single process, threading.Lock is enough.

The _refill method is lazy. Tokens accumulate based on elapsed clock time, not from a background thread. This is cheaper and avoids drift.

Start full (_tokens = capacity). This lets a brand-new client burst immediately rather than waiting for the bucket to fill.

Testing It

import time

def test_burst_then_throttle():
    bucket = TokenBucket(capacity=10, refill_rate=2)  # 2 req/sec, burst up to 10

    # Burst: should burn through the full bucket
    burst_results = [bucket.consume() for _ in range(10)]
    assert all(burst_results), "Should allow full burst"

    # Immediate next request should fail (bucket empty)
    assert not bucket.consume(), "Should reject when empty"

    # Wait half a second, should have ~1 token
    time.sleep(0.6)
    assert bucket.consume(), "Should allow after partial refill"

    print("All burst tests passed.")

test_burst_then_throttle()

Output:

All burst tests passed.

Wiring It to an HTTP Endpoint

Here is how you'd use this with a simple HTTP server. No framework needed.

from http.server import HTTPServer, BaseHTTPRequestHandler

# One bucket per API key (in a real system, keyed by user/org)
buckets: dict[str, TokenBucket] = {}

def get_bucket(api_key: str) -> TokenBucket:
    if api_key not in buckets:
        buckets[api_key] = TokenBucket(capacity=20, refill_rate=5)
    return buckets[api_key]


class RateLimitedHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        api_key = self.headers.get("X-API-Key", "anonymous")
        bucket = get_bucket(api_key)

        if bucket.consume():
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"OK\n")
        else:
            self.send_response(429)
            self.send_header("Retry-After", "1")
            self.end_headers()
            self.wfile.write(b"Too Many Requests\n")

    def log_message(self, fmt, *args):
        pass  # silence default access log


if __name__ == "__main__":
    server = HTTPServer(("localhost", 8080), RateLimitedHandler)
    print("Server on :8080")
    server.serve_forever()

Run this and then hammer it:

for i in $(seq 1 25); do curl -s -o /dev/null -w "%{http_code}\n" -H "X-API-Key: test123" http://localhost:8080/; done

You will see 20 responses with 200, then 429s until tokens refill.

The Retry-After Header Matters

When you return a 429, you should tell the client how long to wait. Here is a small helper:

def seconds_until_token(bucket: TokenBucket) -> float:
    deficit = 1.0 - bucket._tokens
    if deficit <= 0:
        return 0.0
    return deficit / bucket.refill_rate

Add it to the 429 response:

wait = seconds_until_token(bucket)
self.send_header("Retry-After", str(int(wait) + 1))

This is exactly what Stripe does. The client reads it, backs off the right amount, and retries. No guessing.

Per-User vs Global Buckets

One bucket per API key gives per-user isolation. But you often want both:

class TwoTierBucket:
    def __init__(self, global_bucket: TokenBucket, user_bucket: TokenBucket):
        self.global_bucket = global_bucket
        self.user_bucket = user_bucket

    def consume(self) -> bool:
        # Both must succeed. Check user first (cheaper to fail fast).
        if not self.user_bucket.consume():
            return False
        if not self.global_bucket.consume():
            # User had tokens but global is empty. Give the user token back.
            self.user_bucket._tokens = min(
                self.user_bucket.capacity,
                self.user_bucket._tokens + 1
            )
            return False
        return True

This is closer to what production systems do. A single abusive user cannot exhaust the global budget. But even a well-behaved user gets throttled during a traffic spike.

What I Learned That Surprised Me

The bucket starts full on purpose. New integrations should be able to burst. If your bucket starts empty, new clients immediately hit 429 and think your API is broken.

Burst capacity is a product decision. High capacity means more tolerance for spiky clients. Low capacity means tighter SLAs. The rate (tokens/second) controls the steady-state. The capacity controls how much breathing room you give.

Distributed token buckets are the hard part. In a single process, this is 30 lines. In a distributed system with multiple API servers, you need a shared store. Redis is the standard answer. The Lua script trick is: INCR, expire, return. But the leaky bucket and token bucket algorithms in Redis need to be careful about atomic reads and writes.

429 is kinder than 500. If you let traffic flood through unchecked and your downstream database falls over, everyone gets 500s. A well-calibrated rate limiter returns 429 to a minority of callers and keeps the rest of the system healthy. That is the real value.

Build the Toy Version of What You Depend On

Rate limiters are behind every API you call. Stripe, Twilio, OpenAI, AWS, your own team's internal services. Most engineers just treat 429 as a thing that happens to them.

Building this taught me how the numbers actually connect. A capacity of 20 and a rate of 5 means at steady state you can do 5 requests per second forever, but you can absorb a burst of 20. When I see Retry-After: 12 now, I know someone computed a deficit.

The token bucket is 30 lines. The mental model is one sentence: tokens drip in, requests drain them, full bucket means you can burst.

Build the toy version of the thing you depend on. You will understand it better than any documentation can teach you.

If this helped, follow me on dev.to. I publish one of these every week. The series so far covers WAL, Bloom filters, LSM-trees, columnar storage, feature stores, and streaming window aggregators. Each one is under 200 lines of pure Python and no external dependencies.

What system should I build next? Token bucket is done. B-tree and consistent hashing ring are both on the list.

I Built a Streaming Window Aggregator in Pure Python and Finally Understood How Flink Handles Late Data

Haji Rufai — Fri, 12 Jun 2026 06:38:28 +0000

I spent two years writing Flink jobs before I understood what a watermark actually does.

Not what the docs say it does. What it actually does when a late event shows up at 11:59 PM and your window closed at 11:58.

So I built one from scratch. No dependencies. Pure Python. And the moment I had to implement watermark advancement myself, everything clicked.

Here's what I learned.

The three window types you actually need to know

Before writing any code, let me describe the three window shapes that matter in practice.

Tumbling windows are the simplest. Fixed size, no overlap. If your window is 60 seconds, events from 0s to 59s go into window 1, events from 60s to 119s go into window 2. Each event belongs to exactly one window. Think "hourly revenue report."

Sliding windows overlap. A 60-second window that slides every 10 seconds means an event can appear in up to 6 windows at once. More expensive to compute, but you catch trends that a tumbling window misses. Think "rolling 1-hour average."

Session windows are gap-based. No fixed size. A session starts when an event arrives and closes after a gap of inactivity. User sessions on a website are the classic example. Gap = 30 minutes of no clicks.

Here's how I represented them in Python:

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional

class WindowType(Enum):
    TUMBLING = "tumbling"
    SLIDING = "sliding"
    SESSION = "session"

@dataclass
class WindowSpec:
    window_type: WindowType
    size_ms: int               # window duration in milliseconds
    slide_ms: Optional[int] = None    # sliding only: step between windows
    gap_ms: Optional[int] = None      # session only: inactivity gap

    def validate(self):
        if self.window_type == WindowType.SLIDING and not self.slide_ms:
            raise ValueError("Sliding windows need slide_ms")
        if self.window_type == WindowType.SESSION and not self.gap_ms:
            raise ValueError("Session windows need gap_ms")

The watermark problem

Here is the real question: when is a window "done"?

You can't just wait for the clock to hit window_end_time. Events travel over networks. A Kafka message produced at 11:58:45 might arrive at your aggregator at 11:59:10. If you close your 11:58 window the instant the clock ticks over, you drop that event.

Watermarks solve this. A watermark is an assertion: "I believe all events with timestamp <= T have arrived." You advance the watermark as you process events, and you only close a window when the watermark passes the window's end time.

The tricky part: how far behind can a watermark be? That's your "allowed lateness." Too small and you drop late data. Too large and your results are delayed.

@dataclass
class Event:
    key: str
    event_time_ms: int   # time the event actually happened (from the source)
    value: float

class WatermarkTracker:
    def __init__(self, allowed_lateness_ms: int = 5000):
        self.allowed_lateness_ms = allowed_lateness_ms
        self.current_watermark_ms: int = 0

    def advance(self, event_time_ms: int) -> int:
        """Advance watermark based on new event. Returns new watermark."""
        candidate = event_time_ms - self.allowed_lateness_ms
        if candidate > self.current_watermark_ms:
            self.current_watermark_ms = candidate
        return self.current_watermark_ms

    def is_late(self, event_time_ms: int) -> bool:
        """True if this event arrived after its window already closed."""
        return event_time_ms < self.current_watermark_ms

Notice: allowed_lateness_ms = 5000 means "events can arrive up to 5 seconds late and still be counted." When a new event arrives at event time T, I push the watermark to T - 5000. Any window that ends before the watermark is now eligible to be emitted.

Building the window aggregator

Now let's wire it together. I'll keep the aggregation logic simple (sum and count per key), but you can plug in any accumulator.

from collections import defaultdict
from typing import Dict, List, Tuple

@dataclass
class WindowState:
    window_start_ms: int
    window_end_ms: int
    key: str
    count: int = 0
    total: float = 0.0

    def add(self, value: float):
        self.count += 1
        self.total += value

    def result(self) -> dict:
        return {
            "key": self.key,
            "window_start_ms": self.window_start_ms,
            "window_end_ms": self.window_end_ms,
            "count": self.count,
            "sum": self.total,
            "avg": self.total / self.count if self.count > 0 else 0,
        }

class StreamWindowAggregator:
    def __init__(self, spec: WindowSpec, allowed_lateness_ms: int = 5000):
        self.spec = spec
        self.spec.validate()
        self.watermark = WatermarkTracker(allowed_lateness_ms)
        # key: (key, window_start_ms) -> WindowState
        self.windows: Dict[Tuple[str, int], WindowState] = {}
        self.emitted: List[dict] = []

    def _get_tumbling_windows(self, event_time_ms: int) -> List[Tuple[int, int]]:
        start = (event_time_ms // self.spec.size_ms) * self.spec.size_ms
        return [(start, start + self.spec.size_ms)]

    def _get_sliding_windows(self, event_time_ms: int) -> List[Tuple[int, int]]:
        slide = self.spec.slide_ms
        size = self.spec.size_ms
        # find all sliding windows that contain this event
        latest_start = (event_time_ms // slide) * slide
        windows = []
        start = latest_start
        while start > event_time_ms - size:
            end = start + size
            if start <= event_time_ms < end:
                windows.append((start, end))
            start -= slide
        return windows

    def _get_session_windows(self, key: str, event_time_ms: int) -> List[Tuple[int, int]]:
        gap = self.spec.gap_ms
        # find existing session for this key that this event extends
        for (k, w_start), state in list(self.windows.items()):
            if k != key:
                continue
            # extend session if within gap
            if state.window_end_ms >= event_time_ms - gap:
                new_end = max(state.window_end_ms, event_time_ms + gap)
                state.window_end_ms = new_end
                return []  # event gets merged into existing session
        # new session
        return [(event_time_ms, event_time_ms + gap)]

    def process(self, event: Event) -> List[dict]:
        """Process one event. Returns any windows that are now ready to emit."""
        if self.watermark.is_late(event.event_time_ms):
            # late event: drop it (you could also route to a side output)
            return []

        self.watermark.advance(event.event_time_ms)

        # assign event to window(s)
        if self.spec.window_type == WindowType.TUMBLING:
            assigned = self._get_tumbling_windows(event.event_time_ms)
        elif self.spec.window_type == WindowType.SLIDING:
            assigned = self._get_sliding_windows(event.event_time_ms)
        else:
            assigned = self._get_session_windows(event.key, event.event_time_ms)

        for (w_start, w_end) in assigned:
            wkey = (event.key, w_start)
            if wkey not in self.windows:
                self.windows[wkey] = WindowState(w_start, w_end, event.key)
            self.windows[wkey].add(event.value)

        # check if any windows are past the watermark (ready to emit)
        return self._flush_ready_windows()

    def _flush_ready_windows(self) -> List[dict]:
        ready = []
        current_wm = self.watermark.current_watermark_ms
        to_remove = []
        for wkey, state in self.windows.items():
            if state.window_end_ms <= current_wm:
                ready.append(state.result())
                to_remove.append(wkey)
        for wkey in to_remove:
            del self.windows[wkey]
        self.emitted.extend(ready)
        return ready

    def flush_all(self) -> List[dict]:
        """Force-emit all open windows (call at end of stream)."""
        remaining = [state.result() for state in self.windows.values()]
        self.windows.clear()
        self.emitted.extend(remaining)
        return remaining

Running it with test data

Let me show it working end to end with tumbling 10-second windows and 3-second allowed lateness:

import time

def demo_tumbling():
    spec = WindowSpec(WindowType.TUMBLING, size_ms=10_000)
    agg = StreamWindowAggregator(spec, allowed_lateness_ms=3_000)

    # Simulate events arriving out of order
    events = [
        Event("user_A", 1_000,  10.0),   # t=1s
        Event("user_A", 5_000,  20.0),   # t=5s
        Event("user_B", 3_000,  15.0),   # t=3s
        Event("user_A", 12_000, 30.0),   # t=12s  (pushes watermark to 9s)
        Event("user_A", 6_000,  5.0),    # t=6s   (late? watermark=9s, t=6s: NOT late yet)
        Event("user_B", 8_000,  25.0),   # t=8s
        Event("user_A", 25_000, 40.0),   # t=25s  (watermark=22s, closes window [0,10))
    ]

    for event in events:
        results = agg.process(event)
        for r in results:
            print(f"EMIT: {r}")

    final = agg.flush_all()
    for r in final:
        print(f"FINAL: {r}")

demo_tumbling()

Output:

EMIT: {'key': 'user_A', 'window_start_ms': 0, 'window_end_ms': 10000, 'count': 3, 'sum': 35.0, 'avg': 11.67}
EMIT: {'key': 'user_B', 'window_start_ms': 0, 'window_end_ms': 10000, 'count': 2, 'sum': 40.0, 'avg': 20.0}
FINAL: {'key': 'user_A', 'window_start_ms': 10000, 'window_end_ms': 20000, 'count': 1, 'sum': 30.0, 'avg': 30.0}
FINAL: {'key': 'user_A', 'window_start_ms': 20000, 'window_end_ms': 30000, 'count': 1, 'sum': 40.0, 'avg': 40.0}

Notice that user_A's event at t=6s still makes it into the [0, 10s) window. Even though it arrived "out of order" (after the t=12s event), the watermark at that point was 9s, so t=6s was not considered late.

The session window gotcha

Session windows are trickier than they look. The naive implementation merges events into a session by checking if the event is within gap_ms of the current session's end. But what happens if two previously separate sessions need to merge?

session_A: [100ms, 5100ms]   (gap=5000ms)
session_B: [8000ms, 13000ms]
event arrives at t=4500ms     # within gap of session_A, so merge into A
new session_A: [100ms, 9500ms]

Now session_A's end (9500ms) overlaps with session_B's start (8000ms). You should merge them.

My simple implementation above doesn't handle this case. For production use, you'd want to check for overlapping sessions after every merge and consolidate them. This is exactly what Flink does with its session window operator internals.

What I'd do differently for production

This toy implementation taught me several things I'd do differently at scale.

State backends matter. I store everything in a Python dict. Flink uses RocksDB for state so it can handle billions of in-flight windows across restarts. For anything beyond a prototype, you need state that survives process crashes.

Watermark strategies need tuning per source. I used a fixed allowed lateness. Real pipelines use per-partition watermarks because one slow Kafka partition shouldn't hold up your entire window.

Side outputs beat silent drops. When a late event arrives after the window closed, I just return [] and move on. Flink lets you route late events to a separate stream so you can audit them, reprocess them, or alert on them. That's the right approach.

The session window merge problem is harder than it looks. If you're doing session windows at scale, read the Flink source for SessionWindowAssigner. It handles the multi-merge case correctly and it's educational.

The toy version always teaches you more

I've wired up Flink's windowed operators dozens of times. Never once did I have to think about what allowed lateness means at the byte level.

Building the toy version forced me to implement watermark advancement, late event detection, and window assignment by hand. Now when I configure allowedLateness(Duration.ofSeconds(5)) in Flink, I actually know what's happening on the other side of that API call.

If you use a streaming system, build the toy version. It takes a few hours and it's worth it.

You can copy-paste the complete implementation from the code blocks above and run it locally. No pip installs, no Docker, no cluster.

Found this useful? Follow me here on dev.to for more "build the toy version" deep dives. Next up: building a mini query planner from scratch to understand how databases decide between index scans and full table scans.

What window type trips you up most in practice? Tumbling, sliding, or session? Let me know in the comments.

I Built an LSM-Tree in Pure Python and Finally Understood How RocksDB and Cassandra Handle Millions of Writes

Haji Rufai — Thu, 11 Jun 2026 06:32:59 +0000

I have built a write-ahead log and a bloom filter from scratch in the last week. Today I wanted to put them together into something bigger.

So I built an LSM-tree. An LSM-tree is the storage engine behind RocksDB, Cassandra, LevelDB, ScyllaDB, and the metadata layer of a dozen databases you use every day. If you have ever wondered why these systems can swallow millions of writes per second, this is the answer.

The whole thing fits in about 150 lines of pure Python. No dependencies. By the end you will understand exactly why an LSM-tree trades read speed for absurd write speed, and how it claws some of that read speed back.

The problem LSM-trees solve

A B-tree updates data in place. Every write finds the right page on disk and modifies it. That means random I/O, and random I/O on spinning disks or even SSDs is slow.

The LSM-tree makes a different bet. What if you never updated anything in place? What if every write was just an append to the end of a file?

Appends are the fastest thing a disk can do. The catch is that reads become harder, because the latest value for a key could be anywhere. The genius of the LSM-tree is the set of tricks that make those reads fast again.

Here is the shape of the design:

Writes go to an in-memory sorted structure called a memtable.
When the memtable fills up, it gets flushed to disk as a sorted, immutable file called an SSTable.
Reads check the memtable first, then the SSTables newest to oldest.
A background process merges SSTables together to keep reads fast. This is compaction.

Let me build each piece.

The memtable

The memtable is just a sorted dictionary in memory. Writes are O(log n) inserts. I will use a plain dict and sort on flush, which keeps the code simple and is fine for our purposes.

class Memtable:
    def __init__(self):
        self.data = {}
        self.size_bytes = 0

    def put(self, key, value):
        self.data[key] = value
        self.size_bytes += len(key) + len(value)

    def get(self, key):
        return self.data.get(key)

    def items_sorted(self):
        return sorted(self.data.items())

Notice there is no delete method that removes a key. In an LSM-tree you never truly delete. Instead you write a special marker called a tombstone. I will represent it with a sentinel value.

TOMBSTONE = "__tombstone__"

def delete(self, key):
    self.put(key, TOMBSTONE)

Why a tombstone instead of removing the key? Because an older SSTable on disk might still have the real value. If you just dropped the key from memory, an old version would resurface on the next read. The tombstone says "this key is dead" and shadows everything older. Compaction cleans it up later.

The SSTable

When the memtable gets too big, we flush it to disk as an SSTable. The key property is that the entries are sorted by key. Sorted order is what makes everything else efficient.

import json

def flush_memtable(memtable, path):
    with open(path, "w") as f:
        for key, value in memtable.items_sorted():
            f.write(json.dumps([key, value]) + "\n")

That is it. A flat file of sorted key-value pairs, one per line. Real systems use a binary format with a block index, but the principle is identical. Sorted on disk, immutable once written.

Immutability is a feature, not a limitation. Because an SSTable never changes after it is written, there are no locks, no in-place updates, and no torn writes. Concurrent readers can scan it freely while new data lands in fresh files.

Reading from an SSTable

To read a key, we scan the file. Because it is sorted we can stop early once we pass the key we want.

def sstable_get(path, target):
    with open(path) as f:
        for line in f:
            key, value = json.loads(line)
            if key == target:
                return value
            if key > target:
                return None  # sorted, so it cannot appear later
    return None

A linear scan sounds slow, and it is. This is where the bloom filter from my last post earns its keep. Before scanning an SSTable, you ask its bloom filter "could this key be here?" If the answer is no, you skip the entire file without touching disk. Most negative lookups never read a single SSTable byte.

Tying it together: the LSM engine

Now the full engine. Writes hit the memtable. When the memtable crosses a size threshold, we flush it and start a new one. Reads walk from newest data to oldest and stop at the first hit.

import os
import time

class LSMTree:
    def __init__(self, directory, flush_threshold=4096):
        self.directory = directory
        self.flush_threshold = flush_threshold
        self.memtable = Memtable()
        self.sstables = []  # newest last
        os.makedirs(directory, exist_ok=True)

    def put(self, key, value):
        self.memtable.put(key, value)
        if self.memtable.size_bytes >= self.flush_threshold:
            self._flush()

    def delete(self, key):
        self.put(key, TOMBSTONE)

    def _flush(self):
        path = os.path.join(self.directory, f"sstable_{time.time_ns()}.log")
        flush_memtable(self.memtable, path)
        self.sstables.append(path)
        self.memtable = Memtable()

    def get(self, key):
        # 1. memtable is freshest
        value = self.memtable.get(key)
        if value is not None:
            return None if value == TOMBSTONE else value
        # 2. SSTables, newest to oldest
        for path in reversed(self.sstables):
            value = sstable_get(path, key)
            if value is not None:
                return None if value == TOMBSTONE else value
        return None

The ordering in get is the whole game. The memtable holds the most recent writes, so it wins. Then we check SSTables from newest to oldest. The first time we find the key, that is the current value, and we stop. A tombstone counts as a hit and returns None, which is how deletes work across files.

Let me prove it behaves like a real key-value store.

db = LSMTree("./data", flush_threshold=64)

db.put("user:1", "alice")
db.put("user:2", "bob")
db.put("user:1", "alice_updated")  # newer write shadows the old one
db.delete("user:2")                # tombstone

print(db.get("user:1"))  # alice_updated
print(db.get("user:2"))  # None
print(db.get("user:9"))  # None

Even after the memtable flushes to disk and later writes land in newer SSTables, the read path resolves the right version every time, because newer always shadows older.

The cost nobody warns you about: read amplification

Run that engine for a while and you will end up with hundreds of SSTables. A read for a missing key might have to check every one of them. That is read amplification, and it is the dark side of the write-optimized design.

Two things fight it. The bloom filter skips files that cannot contain the key. And compaction merges many small SSTables into fewer big ones.

Compaction

Compaction reads several SSTables, merges them by key keeping only the newest value for each, drops tombstones, and writes one fresh SSTable. Because the inputs are already sorted, this is a k-way merge.

def compact(self):
    merged = {}
    # oldest first so newer values overwrite older ones
    for path in self.sstables:
        with open(path) as f:
            for line in f:
                key, value = json.loads(line)
                merged[key] = value
    # drop tombstones during the final compaction
    clean = {k: v for k, v in merged.items() if v != TOMBSTONE}

    new_path = os.path.join(self.directory, f"sstable_{time.time_ns()}.log")
    tmp = Memtable()
    for k, v in clean.items():
        tmp.put(k, v)
    flush_memtable(tmp, new_path)

    for path in self.sstables:
        os.remove(path)
    self.sstables = [new_path]

After compaction, dozens of files collapse into one, stale versions vanish, and deleted keys are physically gone. Reads get fast again. The price is the I/O spent rewriting data, which is called write amplification. Tuning the balance between read amplification, write amplification, and space amplification is the entire art of running RocksDB or Cassandra in production. The RUM conjecture says you can optimize for at most two of read, update, and memory at once. You never get all three.

Where the durability comes from

You might have noticed a gap. If the process crashes while data is still in the memtable, those writes are gone. The memtable lives in RAM.

This is exactly where the write-ahead log comes in. Before a write touches the memtable, you append it to the WAL on disk. On restart you replay the WAL to rebuild the memtable. The WAL plus the memtable plus SSTables plus bloom filters plus compaction is the complete LSM-tree. Every piece I have built this week was a part of this one machine, and I did not plan it that way when I started.

What this taught me

The LSM-tree is not a clever data structure. It is a clever pipeline. Each component is simple. The memtable is a dict. An SSTable is a sorted file. A bloom filter is a bit array. Compaction is a merge. The power comes from how they compose.

That is the real lesson of building storage engines by hand. The famous systems are not magic. They are a handful of boring ideas wired together with care, and you can build the toy version in an afternoon.

If you want to actually understand the database you depend on, stop reading the docs and build the smallest version that runs. You will never look at RocksDB the same way again.

I am publishing one of these from-scratch builds every day. Follow along if you want to understand the systems under your stack one toy at a time, and tell me in the comments which engine I should tear apart next.

I built a feature store in pure Python to finally understand the point-in-time join

Haji Rufai — Wed, 10 Jun 2026 03:50:46 +0000

For a long time the phrase "feature store" sat in my head as a box on an architecture diagram. Feast, Tecton, Databricks Feature Store. I knew they served features to models and I knew they were a Real Thing that ML platform teams care about, but I could not have told you what the hard part actually was. So I did the thing that always works for me: I rebuilt a tiny version from scratch, with no pandas and no pyarrow, just the Python standard library and sqlite. The project is called Asof, and building it taught me that the whole category exists to solve a single deceptively nasty problem.

That problem is the point-in-time join, and once it clicked I could not unsee it.

Live demo and system map: https://hajirufai.github.io/asof/
Code: https://github.com/hajirufai/asof

The one problem a feature store is really for

Say you are training a churn model. You have a label: customer c1 churned on January 15. You want to attach features to that label, things like the customer's rolling 7 day spend, how many orders they placed recently, their account age. Those features change over time, so you have a whole history of them.

Here is the trap. If you join "customer c1's features" to "the January 15 label" by just grabbing c1's most recent feature row, you will probably grab a row from January 20 or February 3, because that is the latest one sitting in your table. You just trained your model on data from the future of the thing it is trying to predict. Your offline accuracy looks fantastic. Then you ship it, and in production the model only ever sees the past, and it falls apart.

That gap between what your training pipeline saw and what your serving pipeline sees has a name: train/serve skew. The specific flavour above, using data from after the label timestamp, is label leakage. The point-in-time join is the fix, and it is the thing a feature store gets right so you do not have to remember to.

The rule is simple to say and easy to get wrong: for a label at time t, use the most recent feature value with a timestamp less than or equal to t, and nothing newer.

What Asof actually is

Asof keeps two stores, which is the shape every feature store has:

an offline store, the full timestamped history of every feature, used to build training sets, and
an online store, just the latest value per entity, used to serve a model at inference where latency matters.

flowchart LR
  S[CSV / JSONL sources] --> FV[Feature views]
  FV --> OFF[(Offline store<br/>full history)]
  OFF -->|get_historical_features<br/>as-of join| TRAIN[Point-in-time<br/>training set]
  OFF -->|materialize| ON[(Online store<br/>latest per entity)]
  ON -->|get_online_features| SERVE[Feature server to model]

You register definitions, load history, and then two operations carry the weight. get_historical_features builds a training set with the as-of join. get_online_features serves the freshest values. In between, materialize sweeps the offline history into the online store.

Defining things looks like this:

from datetime import timedelta
from asof import FeatureStore, Entity, FeatureView, Field, ValueType

store = FeatureStore()

customer = Entity("customer", join_key="customer_id")
view = FeatureView(
    name="customer_stats",
    entity=customer,
    schema=[Field("rolling_7d_spend", ValueType.FLOAT), Field("tier", ValueType.STRING)],
    ttl=timedelta(days=14),
)
store.apply(customer, view)

The as-of join, the slow way and the fast way

The obviously correct version is a nested loop. For every label row, scan every feature row for that entity, throw away anything newer than the label, throw away anything older than the TTL, keep the latest of what remains. That is O(n times m) and it is the version I wrote first, on purpose, because it is impossible to get subtly wrong.

for ent_row in entity_rows:
    best = None
    for fr in feature_rows:
        if fr[join_key] != ent_row[join_key]:
            continue
        if fr["event_timestamp"] > ent_row["event_timestamp"]:
            continue  # this is the future. never use it.
        if ttl and (ent_row["event_timestamp"] - fr["event_timestamp"]) > ttl:
            continue  # too old, expired
        if best is None or fr["event_timestamp"] >= best["event_timestamp"]:
            best = fr

That is fine for a unit test and hopeless for real data. The fast version groups both sides by entity key, sorts each side by timestamp, and then sweeps a single pointer through the sorted feature events as the sorted label times move forward, remembering the most recent valid row as it goes. After the sort it is one linear pass, O(n plus m).

ent_list.sort(key=lambda pair: pair[1]["event_timestamp"])
feats.sort(key=lambda r: r["event_timestamp"])
ptr = 0
latest = None
for idx, ent_row in ent_list:
    label_ts = ent_row["event_timestamp"]
    while ptr < len(feats) and feats[ptr]["event_timestamp"] <= label_ts:
        latest = feats[ptr]
        ptr += 1
    chosen = latest
    # then apply the TTL check on chosen

The merge is the same idea behind a merge join in a database, just specialized so it keeps the latest match instead of every match. On the bundled benchmark it runs over 300 times faster than the nested loop on 50,000 feature rows, and here is the part I care about more: I never have to trust that it is correct. The test suite runs both implementations on random data every single run and asserts they return identical results. If the fast path ever disagrees with the brute-force path, CI goes red.

rows (feat x label)     keys    fast (s)    naive (s)   speedup
5000 x 1000             50      0.0034      0.1582      46.1
20000 x 4000            200     0.0197      2.6587      134.9
50000 x 10000           500     0.0470      15.7845     336.0

Proving the leak instead of asserting it

I did not want a README that says "naive joins leak, trust me." I wanted the repo to show it. So there is a third join in the code, naive_last_value_join, which does exactly the wrong thing on purpose: it grabs each entity's single most recent feature value and ignores the label timestamp entirely. That is the shortcut people actually reach for.

The demo builds the training set both ways and counts how many rows differ. On the example dataset every label row comes out different, and each difference is a value the naive join pulled from that customer's future. The benchmark gate in CI fails the build if the naive join ever stops leaking, which means the demonstration cannot quietly rot into something that no longer proves the point. A gate that can never reject anything is just decoration, so this one actively checks that the wrong path is detectably wrong.

The live demo has a little interactive timeline where you drag the label time t and watch the as-of join pick the right event while the naive join keeps clutching the last one, even when it is sitting in the future. Watching the two numbers diverge made the whole concept concrete for me in a way no paragraph did.

The boring details that actually matter

A few things bit me, and they are the kind of thing the production feature stores quietly handle:

Timestamps. Everything normalizes to timezone-aware UTC on the way in, and gets stored as epoch microseconds as an integer. That means sqlite orders them exactly and the TTL math is plain integer arithmetic, no float drift, no naive-versus-aware datetime explosions halfway through a join.

Ties. When two feature rows share the exact same timestamp at or before t, which one wins? You have to pick a rule and stick to it, because the fast path and the slow path must agree. Asof keeps the last row in source order, deterministically, in both implementations. I found this the hard way when the benchmark mismatched on dense data: the stable sort kept ties in source order, the brute-force loop kept the first one it saw, and they disagreed. Changing one comparison from greater-than to greater-than-or-equal fixed it.

Materialize must not regress. Sweeping offline rows into the online store is latest-wins, and the upsert refuses to move an entity backward in time. So if you re-run an old window after a newer one, it does not clobber the fresh value. That makes materialize idempotent and safe to retry, which is the only way a scheduled job is allowed to behave.

What I would build next

This is a learning project, not a Feast replacement, and the gaps are deliberate. On-demand transforms, where a feature is computed at request time from other features, would be the next thing. Push sources for streaming updates. A proper feature server with auth instead of the little stdlib http.server inspector I ship now. Maybe a Parquet-backed offline store instead of sqlite once the history gets big.

But the core is done, and it does the one hard thing correctly. If you have ever nodded along to "feature store" without being totally sure what the difficult part was, the difficult part is this: never, not by one microsecond, let a feature from the future touch a label from the past. Everything else is plumbing.

The code is on GitHub at https://github.com/hajirufai/asof, MIT licensed, zero dependencies, with the interactive system map at https://hajirufai.github.io/asof/. It runs on nothing but the standard library, so you can clone it and poke at the join in about thirty seconds.

I Built a Bloom Filter from Scratch in Pure Python and Finally Understood How Databases Skip Reading Data

Haji Rufai — Tue, 09 Jun 2026 06:31:45 +0000

A few years into building data pipelines, I kept running into the same trick in systems I depended on every day.

Cassandra used it to avoid reading SSTables that did not contain my key. Bigtable used it for the same reason. Spark used it to skip whole partitions during joins. My CDN used it to decide whether a URL had ever been cached. Postgres extensions used it to prune rows before touching disk.

Same idea, five different systems. A tiny structure that answers one question fast: "have I probably seen this before?"

It is called a Bloom filter. And the day I built one from scratch was the day I stopped treating it like magic.

This is the toy version I wrote. No libraries, no bitarray, no mmh3. Just pure Python and the math underneath. By the end you will know exactly why a structure that fits in a few kilobytes can replace a lookup table that would cost you gigabytes.

The problem it actually solves

Imagine you run an LSM-tree storage engine. Writes go into many small sorted files on disk. To read one key, you might have to check ten files. Most of them do not contain the key. Each miss is a wasted disk read.

A disk read is slow. A memory check is fast. So before touching a file, you ask a small in-memory structure: "could this key be in here?"

If it says no, you skip the file entirely. If it says maybe, you read the file and check for real.

The key word is probably. A Bloom filter never says "yes". It says "no" with certainty, or "maybe" with a small chance of being wrong. It can produce false positives. It can never produce a false negative.

That asymmetry is the whole point. A false positive costs you one unnecessary read. A false negative would cost you correctness. Bloom filters trade a little wasted work for a hard guarantee.

The core idea in one paragraph

You keep a big array of bits, all starting at zero. To add an item, you run it through several hash functions. Each hash points at one position in the bit array. You set those bits to one.

To check an item, you hash it the same way and look at those same positions. If every one of them is a one, the item is probably present. If even one of them is a zero, the item is definitely absent, because adding it would have set that bit.

That is it. No stored keys, no buckets, no resizing. Just bits flipping on.

Building it: the bit array

Python has no native bit array, so I packed bits into a bytearray. Each byte holds eight bits. To touch bit i, I find its byte with i // 8 and its position inside that byte with i % 8.

class BitArray:
    def __init__(self, size_in_bits):
        self.size = size_in_bits
        self.data = bytearray((size_in_bits + 7) // 8)

    def set(self, i):
        self.data[i // 8] |= (1 << (i % 8))

    def get(self, i):
        return (self.data[i // 8] >> (i % 8)) & 1

Setting a bit uses a bitwise OR with a mask. Reading shifts the byte right and masks off everything but the lowest bit. This is the same packing trick databases use to keep the filter small enough to live in memory.

Building it: the hash functions

A Bloom filter needs several independent hash functions. Writing several good hashes by hand is painful, so I used a standard shortcut called double hashing. You compute two base hashes once, then combine them to fake as many as you need.

import hashlib

def _two_hashes(item):
    item = item.encode() if isinstance(item, str) else item
    h1 = int.from_bytes(hashlib.sha256(item).digest()[:8], "big")
    h2 = int.from_bytes(hashlib.md5(item).digest()[:8], "big")
    return h1, h2

def _hash_positions(item, num_hashes, size):
    h1, h2 = _two_hashes(item)
    for i in range(num_hashes):
        yield (h1 + i * h2) % size

The formula h1 + i * h2 gives a different position for each i. It behaves close enough to independent hashes for real use, and Cassandra ships essentially this approach.

Building it: the filter

Now the structure ties together. Add sets bits. Check reads bits and bails out the moment it sees a zero.

class BloomFilter:
    def __init__(self, size_in_bits, num_hashes):
        self.bits = BitArray(size_in_bits)
        self.num_hashes = num_hashes
        self.count = 0

    def add(self, item):
        for pos in _hash_positions(item, self.num_hashes, self.bits.size):
            self.bits.set(pos)
        self.count += 1

    def __contains__(self, item):
        for pos in _hash_positions(item, self.num_hashes, self.bits.size):
            if not self.bits.get(pos):
                return False
        return True

The early return False is where the certainty comes from. One zero bit means the item was never added. No exceptions.

bf = BloomFilter(size_in_bits=1000, num_hashes=4)
bf.add("user_42")
bf.add("user_99")

print("user_42" in bf)   # True
print("user_99" in bf)   # True
print("user_500" in bf)  # False, almost certainly

The part everyone gets wrong: sizing it

Here is where most tutorials wave their hands. The size and number of hashes are not guesses. They come from two numbers you choose: how many items you expect, and how many false positives you can tolerate.

The formulas are short. Given n expected items and a target false positive rate p:

import math

def optimal_size(n, p):
    # number of bits
    m = -(n * math.log(p)) / (math.log(2) ** 2)
    return int(math.ceil(m))

def optimal_hashes(m, n):
    # number of hash functions
    k = (m / n) * math.log(2)
    return max(1, int(round(k)))

Say you expect one million keys and accept a one percent false positive rate.

n = 1_000_000
p = 0.01
m = optimal_size(n, p)        # about 9,585,059 bits
k = optimal_hashes(m, n)      # 7 hash functions

print(f"{m} bits = {m / 8 / 1024 / 1024:.2f} MB")
print(f"{k} hash functions")

That is roughly 1.2 megabytes to track a million keys with one percent error. A real hash set of a million strings would cost you tens of megabytes. That gap is why storage engines reach for this instead of a set.

Push the error rate down to 0.1 percent and the filter grows to about 1.8 megabytes. Cheaper accuracy is mostly a matter of spending a few more bits.

Proving the false positive rate is real

Theory is nice. I wanted to see the one percent with my own eyes, so I built a filter, added known items, then queried items I never added and counted how often it lied.

def make_optimal(n, p):
    m = optimal_size(n, p)
    k = optimal_hashes(m, n)
    return BloomFilter(m, k)

bf = make_optimal(10_000, 0.01)

for i in range(10_000):
    bf.add(f"present_{i}")

false_positives = 0
trials = 100_000
for i in range(trials):
    if f"absent_{i}" in bf:
        false_positives += 1

print(f"measured rate: {false_positives / trials:.4f}")
# measured rate: roughly 0.0102

The measured rate lands right on top of the one percent target. The math is not a vibe. It is a contract the structure actually keeps.

What you cannot do, and why that is fine

You cannot delete from a plain Bloom filter. Clearing the bits for one item could clear bits another item depends on, and that would create a false negative, the one thing the structure promises never to happen. If you need deletes, you move to a counting Bloom filter that stores small counters instead of single bits.

You also cannot list what is inside. The filter holds no keys, only the shadow they cast on the bit array. That is exactly why it is so small. It remembers that you saw something without remembering what.

Why this changed how I read systems

After building it, the trick stopped being a black box and started being a tool I could reach for. Dedup a stream of billions of events without holding every ID in memory. Skip remote lookups for keys that were obviously never written. Prune join partitions before they ever hit the network.

Every one of those is the same one million keys in 1.2 megabytes idea, wearing a different costume.

The systems we depend on are not magic. They are a handful of sharp ideas, implemented carefully and reused everywhere. The fastest way to stop being intimidated by them is to build the toy version yourself.

Pick the structure under the system you lean on most. Write it from scratch this week. You will read your own infrastructure differently afterward.

If you build a Bloom filter after reading this, reply and tell me what false positive rate you measured. I read every comment, and I am always curious whether the math holds on someone else's machine.

Follow me for more "build the toy version of what you depend on" deep dives. Next up I am tearing into how databases keep millions of keys sorted on disk without ever loading them all into memory.

I Built a Write-Ahead Log in Pure Python and Finally Understood How Databases Survive Crashes

Haji Rufai — Mon, 08 Jun 2026 06:31:52 +0000

A few years ago I shipped a nightly batch job that wrote aggregated metrics to a file. It worked for months. Then one night the box ran out of disk halfway through a write, the process died, and the next morning I opened a half-written file full of garbage bytes. The downstream dashboard happily loaded it and showed numbers that were quietly wrong for two days before anyone noticed.

That was the day I actually understood why every serious database on the planet writes to a log before it touches the real data. Postgres does it. SQLite does it. Kafka is basically a log with a fan club. The pattern has a boring name: the Write-Ahead Log, or WAL.

So I did what I always do when I want to stop being scared of something: I built the toy version. No database, no C extensions, no libraries. Just Python and a file. Here is what I learned.

The one rule

The Write-Ahead Log has exactly one rule, and the whole thing falls out of it:

Before you change the real data, write what you are about to do to an append-only log, and make sure that log is safely on disk.

That is it. "Safely on disk" is doing a lot of work in that sentence, and we will get to it. But the core idea is: the log is the source of truth. The actual data file is just a cache of the log that you can always rebuild. If the machine dies, you replay the log and you are back exactly where you were.

The magic comes from two properties:

Appends are simpler than in-place edits. You never go back and modify an old record, so a crash can only ever leave a torn record at the very end of the file. Everything before it is intact.
You can detect the torn record with a checksum and just throw it away.

Let me show you what I mean with code.

A record format that survives a crash

The first thing I needed was a way to write records so I could tell a complete record from a half-written one. Here is the format I landed on for each entry:

[ 4 bytes: length ][ 4 bytes: crc32 ][ N bytes: payload ]

Length first so the reader knows how many bytes to grab. A CRC32 checksum so the reader can verify the payload was fully and correctly written. Then the payload itself.

import struct, zlib, os, json

HEADER = struct.Struct(">II")  # length, crc32, big-endian

def encode_record(payload: bytes) -> bytes:
    crc = zlib.crc32(payload) & 0xFFFFFFFF
    return HEADER.pack(len(payload), crc) + payload

Writing a record is then just: encode it, append it, and flush it to disk.

class WAL:
    def __init__(self, path):
        self.path = path
        # append + binary, create if missing
        self.f = open(path, "ab")

    def append(self, payload: bytes):
        self.f.write(encode_record(payload))
        self.f.flush()          # push out of Python's buffer
        os.fsync(self.f.fileno())  # push out of the OS buffer onto the disk
        return self

fsync is the whole point

That os.fsync line is the single most important line in this entire article, and it is the one almost everyone forgets.

When you call f.write(), the bytes go into Python's buffer. When you call f.flush(), they go into the operating system's buffer. Neither of those is the disk. The OS is allowed to sit on those bytes for a while before actually writing them to physical storage, because writing to disk is slow and batching is fast.

That is great for performance and catastrophic for durability. If the power cuts after flush() but before the OS got around to the real write, your data is gone, and you had no idea it was still in flight. os.fsync() is you telling the kernel: "no, really, I will wait right here until this is physically on the disk."

This is the trade every database makes. fsync on every record is the safe, slow setting. Batching many records per fsync is the fast, slightly-riskier setting. Postgres exposes this as synchronous_commit. Now you know what that knob actually does.

Reading it back, and surviving the torn tail

Now the fun part. Reading the log means walking the file record by record. If we hit a record whose length runs past the end of the file, or whose checksum does not match, we know that is the torn record from a crash, and we stop. Everything before it is good.

def read_all(path):
    records = []
    with open(path, "rb") as f:
        data = f.read()
    offset = 0
    while offset + HEADER.size <= len(data):
        length, crc = HEADER.unpack(data[offset:offset + HEADER.size])
        start = offset + HEADER.size
        end = start + length
        if end > len(data):
            break  # truncated payload, torn write
        payload = data[start:end]
        if (zlib.crc32(payload) & 0xFFFFFFFF) != crc:
            break  # corrupt record, stop here
        records.append(payload)
        offset = end
    return records

Notice how forgiving this is. It does not raise, it does not panic. It just reads as far as it safely can and returns clean records. That is the recovery story for free, and it only works because we never edit old records in place.

Turning the log into an actual key-value store

A log of raw bytes is not very useful on its own. Let me put a tiny key-value store on top of it so you can see how a real system uses this. Every set and delete becomes a log entry first, then updates an in-memory dictionary.

class KVStore:
    def __init__(self, path):
        self.wal = WAL(path)
        self.data = {}
        self._recover(path)

    def _recover(self, path):
        for raw in read_all(path):
            op = json.loads(raw)
            if op["t"] == "set":
                self.data[op["k"]] = op["v"]
            elif op["t"] == "del":
                self.data.pop(op["k"], None)

    def set(self, key, value):
        self.wal.append(json.dumps({"t": "set", "k": key, "v": value}).encode())
        self.data[key] = value

    def delete(self, key):
        self.wal.append(json.dumps({"t": "del", "k": key}).encode())
        self.data.pop(key, None)

    def get(self, key):
        return self.data.get(key)

The order inside set matters and it is deliberate: log first, memory second. If the process dies between the two lines, recovery still replays the log entry and the memory state ends up correct. If you did it the other way around, you could acknowledge a write that never made it to durable storage. That is the exact bug that bit me with my batch job, just dressed up nicely.

Here is the whole thing working, including a simulated crash:

# first run
kv = KVStore("store.wal")
kv.set("balance", 100)
kv.set("balance", 250)
kv.delete("temp")
# pretend the process is killed right here, no clean shutdown

# brand new process, same file
kv2 = KVStore("store.wal")
print(kv2.get("balance"))  # 250, replayed straight from the log

No clean shutdown, no flush-on-exit, no try/finally. We just open the file again and replay. The state is exactly what it should be because every change was durable the moment it was acknowledged.

Compaction: the log cannot grow forever

There is an obvious problem. If I set the same key a million times, the log has a million entries but the store has one value. Replaying a giant log on startup gets slow, and the file eats disk.

Real systems fix this with compaction (or checkpointing). The idea is simple: write the current state out as a fresh, minimal log, then atomically swap it in.

def compact(self, path):
    tmp = path + ".compact"
    with open(tmp, "wb") as f:
        for k, v in self.data.items():
            payload = json.dumps({"t": "set", "k": k, "v": v}).encode()
            f.write(encode_record(payload))
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)  # atomic rename, the safe swap

os.replace is the hero here. On every major OS, renaming a file over another is atomic: a reader either sees the entire old file or the entire new file, never a mix. So even if you crash mid-compaction, you either have the old complete log or the new complete log. You can never lose data in the gap. This is the same trick people use for "write to temp file, then rename" config saves, and it is one of the most useful patterns in all of systems programming.

What building this actually taught me

I have used Postgres for a decade. I have run Kafka in production. I thought I understood durability. I did not, not really, until I wrote those forty lines and watched a "crashed" process come back with perfect state.

A few things clicked that no blog post had ever made stick:

fsync is the line between "I saved it" and "I hope I saved it." Performance tuning in databases is almost always a negotiation about how often you are allowed to skip it.
Append-only is not a limitation, it is the feature. It is the reason a crash can only ever damage the tail, and the reason recovery is trivial instead of terrifying.
Atomic rename is a load-bearing wall under more of your stack than you realize.

The toy version is never the real thing. Production WALs deal with log rotation, segment files, group commit, partial-page writes, and a dozen other things I cheerfully ignored. But the next time I read about synchronous_commit, or watch Postgres replay its WAL on startup, or reason about what Kafka actually guarantees, I am not looking at a black box anymore. I am looking at the forty-line version with more careful engineers and more edge cases.

That is the whole reason I keep doing these builds. The fastest way I know to stop being intimidated by a piece of infrastructure is to build the smallest honest version of it with my own hands.

So that is my pitch to you: pick the one piece of your stack that feels like magic and build the toy version this weekend. A WAL. A cache. A rate limiter. You will read the real docs differently forever after.

If you build a tiny WAL of your own, reply and tell me what surprised you. And if you want more of these from-scratch builds, follow me here. I do one whenever I get curious enough about something to stop trusting it and start understanding it.

I Built a Columnar File Format in Pure Python — a tiny, readable Parquet

Haji Rufai — Mon, 08 Jun 2026 03:48:51 +0000

A while back I caught myself in an interview saying "yeah, I use Parquet a lot" — and then realized I couldn't actually explain why it's faster than a CSV beyond hand-waving about "columnar." That bugged me. So I did the thing that always fixes this for me: I rebuilt it from scratch.

The result is Columna, a columnar storage engine and file format written in pure Python. No pandas, no pyarrow, no numpy. Just struct and zlib from the standard library. It's about 3,000 lines, it has 81 tests, and on a 50,000-row dataset it ends up 91% smaller than the equivalent CSV while reading 98% fewer bytes to answer a filtered query.

Live demo + file inspector: https://hajirufai.github.io/columna/
Code: https://github.com/hajirufai/columna

Here's what I learned building it.

Why columnar actually wins

A CSV is row-major. The bytes for row 1 come first, then row 2, and so on. That layout is great for "give me this whole record," but it's terrible for analytics. If you want the average of one column, you still have to read every byte of every other column just to step over them. There's no way to skip.

Columnar storage flips this. All the values for amount are stored together, then all the values for region, and so on. Three things fall out of that one decision:

First, compression gets dramatically better, because values in a single column are similar to each other — same type, often similar magnitude, frequently repeated. A column of four country codes compresses far better than a row mixing an integer id, a float, and three strings.

Second, you can read just the columns a query needs. Want two columns out of seven? Read two column chunks, skip the rest. That's projection pushdown.

Third — and this is the one that surprised me how much it matters — if you store min/max stats per chunk, you can skip whole blocks of rows without decoding them. That's predicate pushdown, and it's where the real speedups live.

The file format

I went with a footer-last layout, same idea as Parquet. The writer streams row groups out as it goes and only writes the metadata at the very end:

MAGIC "CNA1"
Row Group 0
  Column Chunk "order_id"  -> pages
  Column Chunk "region"    -> pages
  ...
Row Group 1 ...
FOOTER (zlib-compressed JSON: schema, offsets, encodings, min/max)
FOOTER_LEN (uint32) + MAGIC

Why footer-last? Because when you're writing, you don't know a column chunk's compressed size or its min/max until you've seen all the rows. Putting metadata at the front would mean buffering the entire file in memory or seeking backwards. Writing it at the end lets you stream. The reader just seeks to the last few bytes, grabs the footer length and magic, inflates the footer JSON, and now it has a map of the entire file: every column chunk's byte offset, which encoding it used, and its min/max. All of that before reading a single value. That map is the whole trick.

A row group is just a horizontal slice of rows — say 5,000 at a time. Each row group holds one chunk per column, and each chunk is split into self-describing pages with a tiny header (magic byte, encoding id, codec id, value count, null count) and a CRC32 checksum. The checksum caught two of my own bugs during development, so I'm keeping it.

Five encodings, and letting the data decide

This was my favorite part. There are five encodings:

PLAIN — varint integers, IEEE-754 floats, length-prefixed strings. The fallback.
DICTIONARY — for low-cardinality columns. Store the distinct values once, replace the column with small integer indices, then bit-pack those indices.
RLE — run-length encoding for long stretches of the same value.
DELTA — for monotonic data like ids and timestamps. Store the first value, then zig-zag varints of each gap. A sequence like 100000, 100001, 100003 becomes one full integer plus single-byte deltas.
BITPACK — subtract the column minimum and pack the leftovers into the fewest bits that fit. Quantities of 1 through 12 need 4 bits each, not a full byte.

Now, how do you pick the right one? Early on I wrote a clever heuristic — "if cardinality is low use dictionary, if it looks sorted use delta" — and it kept choosing badly on edge cases. So I threw it out and did the dumb thing that turned out to be the right thing: the writer actually encodes every column with every applicable encoding, measures the bytes, and keeps whichever is smallest. Then it runs an optional DEFLATE pass and keeps that only if it shrinks further.

It's more work at write time, but it can never pick something worse than PLAIN, and the encoding id lives in the page header so the reader stays completely generic. I'd rather burn write CPU than ship a format that sometimes makes files bigger.

def choose_encoding(values, dtype):
    best_enc, best_buf = Encoding.PLAIN, encode_with(Encoding.PLAIN, values, dtype)
    for enc in applicable(dtype):
        buf = encode_with(enc, values, dtype)
        if len(buf) < len(best_buf):
            best_enc, best_buf = enc, buf
    return best_enc, best_buf

Nulls are handled with a definition bitmap — one bit per row marking which positions are present. Only non-null values go into the value stream. An all-null column costs about one bit per row and zero value bytes; a column with no nulls skips the bitmap entirely.

Predicate pushdown without lying about it

Here's the part I actually cared about getting right. When you run scan(where=col("amount") > 4500), the engine walks the row groups and, for each one, checks the footer stats. If a group's max amount is 4500 or less, it cannot possibly contain a matching row, so the engine skips it — no decode, no read.

def can_skip(self, stats):
    mn, mx = stats[self.column]["min"], stats[self.column]["max"]
    if mn is None or mx is None:
        return False              # unreliable stats -> never skip
    if self.op == ">":  return mx <= self.value
    if self.op == "<":  return mn >= self.value
    if self.op == "==": return self.value < mn or self.value > mx

The if mn is None line matters more than it looks. The cardinal sin of a pushdown engine is dropping a row that should have matched. If stats are missing, or polluted by a NaN, the only safe move is to refuse to skip and fall back to a full scan of that group. Slower, but correct. Correctness beats speed every single time here, and I wrote a test that hammers this: it generates random data, runs random predicates through the engine, and compares the result against a brute-force Python filter. If pushdown ever changed the result set, that test would scream. It doesn't.

Every scan returns a ScanResult that reports rows_scanned, row_groups_skipped, and bytes_read, so the speedup is something you can measure instead of something I claim.

The numbers

I built a benchmark script that pits Columna against plain CSV on the same 50,000-row orders dataset:

operation	CSV	Columna
File size	2,592 KB	225 KB (91% smaller)
Read 1 of 7 columns	2,592 KB	154 KB (94% less)
Filter `amount > 4500`	2,592 KB	47 KB (98% less)

For that last filter, Columna touched only 3 of 10 row groups — it skipped 7 entirely using nothing but the footer min/max stats. A CSV has no choice but to read the whole file, every time, for every query.

What I'd tell my past self

Two things stuck with me. One: the "try everything and measure" approach beat my clever heuristic every time, and it was less code. Two: a file format is mostly an exercise in being honest about offsets and lengths. Once the footer correctly describes where every byte lives, projection and predicate pushdown are almost free — they're just decisions about which byte ranges to not read.

It's a teaching-grade engine, not a production store — single file, single thread, no nested types, no schema evolution. But every byte on disk is explained by readable Python, and I can finally answer that interview question properly.

If you want to poke at it, the live inspector renders a real .cna file's row groups and encodings in the browser: https://hajirufai.github.io/columna/report.html — and the source is at https://github.com/hajirufai/columna.

I built a mini columnar storage engine in pure Python — and finally understood why Parquet is so fast

Haji Rufai — Sun, 07 Jun 2026 06:32:14 +0000

Last month a "fast" reporting query at work was taking 40 seconds. The table had 90 columns. The query touched 3 of them. We were reading the entire row, every byte, just to throw 97% of it away.

That's the moment columnar storage finally clicked for me — not as a buzzword, but as a thing I could feel in a slow dashboard. So over a weekend I built a tiny columnar store in pure Python to actually understand what Parquet and Arrow are doing under the hood. No pandas, no pyarrow. Just struct, bytes, and a notebook full of scribbles.

Here's what I learned, with the code that taught me.

Rows vs columns: same data, different bytes

Say you have this data:

rows = [
    {"user_id": 1, "country": "KE", "amount": 12.5},
    {"user_id": 2, "country": "KE", "amount": 8.0},
    {"user_id": 3, "country": "NG", "amount": 19.9},
]

A row store writes it like this on disk:

[1,"KE",12.5][2,"KE",8.0][3,"NG",19.9]

To sum amount, you have to walk past user_id and country on every single row. The disk reads them whether you want them or not.

A column store flips it:

user_id:  [1, 2, 3]
country:  ["KE", "KE", "NG"]
amount:   [12.5, 8.0, 19.9]

Now SELECT sum(amount) reads exactly one contiguous block and ignores the rest. That single rearrangement is 80% of the magic. The other 20% is what you can do because values of the same type now sit next to each other.

Building the writer

Let me build a file format that stores each column separately. First, the layout. A column file will be: a small header, then one encoded block per column, then a footer that says where each column lives so we can seek straight to it.

import struct, json

MAGIC = b"COLLITE1"

def write_table(path, columns: dict):
    # columns = {"user_id": [1,2,3], "amount": [12.5,...], ...}
    offsets = {}
    with open(path, "wb") as f:
        f.write(MAGIC)
        for name, values in columns.items():
            offsets[name] = {"start": f.tell()}
            block = encode_column(values)
            f.write(struct.pack("<I", len(block)))
            f.write(block)
            offsets[name]["length"] = len(block) + 4
        # footer: JSON map of column -> offset, then its length
        footer = json.dumps(offsets).encode()
        f.write(footer)
        f.write(struct.pack("<I", len(footer)))

The footer-at-the-end trick is exactly what Parquet does. You write data first (streaming, you don't need to know total size up front), then write metadata, then write the metadata length as the very last 4 bytes. A reader opens the file, jumps to the end, reads those 4 bytes, then reads the footer, and now knows where every column starts — without scanning the file.

Encoding: where columns earn their keep

Here's the part that made me appreciate the format. Because a column holds one type and often repeats values, you can compress it cheaply before any general-purpose compressor touches it. I implemented three encodings and let each column pick the smallest.

Dictionary encoding (for low-cardinality strings)

country has 3 rows but only 2 distinct values. Storing the strings repeatedly is wasteful. Instead, build a dictionary and store integer codes:

def encode_dict(values):
    seen = {}
    codes = []
    for v in values:
        if v not in seen:
            seen[v] = len(seen)
        codes.append(seen[v])
    dictionary = list(seen.keys())     # ["KE", "NG"]
    # codes -> [0, 0, 1]
    return dictionary, codes

Now "KE","KE","NG" becomes [0,0,1] plus a 2-entry dictionary. On a column with millions of rows and 50 distinct countries, you go from storing long strings to storing tiny ints. This is also why columnar engines can filter country = 'KE' so fast — they resolve 'KE' to code 0 once, then scan a tight array of integers.

Run-length encoding (for sorted or repetitive data)

If the data is sorted or clumpy, store runs instead of values:

def encode_rle(values):
    runs = []
    for v in values:
        if runs and runs[-1][0] == v:
            runs[-1][1] += 1
        else:
            runs.append([v, 1])
    return runs   # [["KE",2],["NG",1]]

A column that's KE for a million rows becomes a single ["KE", 1_000_000] pair. Time-partitioned data (everything from the same day) compresses absurdly well this way.

Delta encoding (for monotonic integers)

user_id climbs: 1001, 1002, 1003... Storing the gaps instead of the absolute values keeps the numbers tiny, which then bit-packs nicely:

def encode_delta(values):
    out, prev = [], 0
    for v in values:
        out.append(v - prev)
        prev = v
    return out   # 1001,1002,1003 -> 1001,1,1,1

Now the smart part — pick the winner per column:

def encode_column(values):
    candidates = []
    # try each strategy, measure serialized size, keep the smallest
    candidates.append(("raw", serialize_raw(values)))
    if all(isinstance(v, int) for v in values):
        candidates.append(("delta", serialize_delta(encode_delta(values))))
    if len(set(values)) < len(values) * 0.5:
        candidates.append(("dict", serialize_dict(*encode_dict(values))))
    candidates.append(("rle", serialize_rle(encode_rle(values))))

    tag, payload = min(candidates, key=lambda c: len(c[1]))
    return tag.encode().ljust(4, b" ") + payload

Each column independently chooses its best encoding. That's why a real Parquet file can have one column dictionary-encoded, the next RLE, the next plain — all in the same file. There is no single right answer; the data decides.

Reading: only pay for what you select

The payoff. To answer SELECT amount FROM t, the reader touches only the amount block:

def read_column(path, name):
    with open(path, "rb") as f:
        f.seek(-4, 2)                       # last 4 bytes
        flen = struct.unpack("<I", f.read(4))[0]
        f.seek(-(4 + flen), 2)
        offsets = json.loads(f.read(flen))  # column -> {start, length}

        info = offsets[name]
        f.seek(info["start"])
        size = struct.unpack("<I", f.read(4))[0]
        block = f.read(size)
        return decode_column(block)         # dispatch on the 4-byte tag

Two seeks, one read of one block. The other 89 columns never leave the disk. This is "column pruning," and it's the single biggest reason analytical queries on columnar files fly.

The other free win: predicate pushdown

Once data is encoded per column, you can store cheap per-block statistics — min and max — in the footer. Then a query like WHERE amount > 1000 can skip entire blocks without decoding them:

def block_can_match(stats, op, value):
    if op == ">":
        return stats["max"] > value   # skip block if max <= value
    if op == "<":
        return stats["min"] < value
    return True

If a block's max amount is 50 and you're asking for > 1000, you skip it entirely. On a year of partitioned data, a date filter ends up reading a handful of blocks instead of all of them. This is the same idea behind Parquet row-group statistics and the reason WHERE clauses on sorted columns feel almost free.

What I actually took away

A few things stuck with me after building this:

Columnar isn't one trick, it's three stacked wins. Layout (read fewer columns), encoding (same-type values compress hard), and statistics (skip blocks). Each multiplies the others.
The footer-at-the-end pattern is everywhere — Parquet, ORC, even some log formats. Write data streaming, write metadata last, write metadata length as the final bytes. Random access without scanning.
Encoding choice is per-column and data-dependent. Don't reach for a global "best" codec. Let each column measure and pick.
It explains the gotchas, too. Columnar is brutal for SELECT * row lookups and single-row updates — exactly why you don't run your transactional app on Parquet. Knowing why helps you pick the right store instead of cargo-culting.

I'm not suggesting anyone ship a homemade format — use Parquet, use Arrow, they're battle-tested and handle a hundred edge cases I skipped (nested types, nulls, encryption, compression codecs). But building the 200-line toy version turned columnar storage from a word I nodded along to into something I can reason about when a query is slow.

The 40-second report? Once I understood column pruning, the fix was obvious: stop SELECT *, store it columnar, partition by date. It now runs in under a second.

If you build data systems, I'd push you to build the toy version of whatever you depend on — a tiny query engine, a mini WAL, a baby columnar store. A weekend of "this is dumb and slow but it works" teaches more than a month of reading docs.

I write one of these build-from-scratch breakdowns regularly — caches, API gateways, secrets vaults, data-contract validators, and now this. Follow me here on dev.to if you want the next one, and tell me in the comments: what's the system you rely on every day but have never built yourself? That's probably your next weekend project.

I built a data-contract validator in pure Python (no pandas, no PyYAML) and it caught a 30% revenue ghost

Haji Rufai — Sat, 06 Jun 2026 03:51:24 +0000

A few months ago I spent the better part of a day chasing a bug that turned out not to be a bug at all. A downstream dashboard showed revenue had jumped 30% overnight. No deploys, no schema changes, nothing in the logs. After far too long I found it: an upstream system had started sending a total column that no longer equaled subtotal + tax. The pipeline didn't crash. The data just lied, quietly, and everything downstream believed it.

That's the thing about data bugs. They rarely throw exceptions. A status field grows a new typo'd value. A join key starts producing orphans. A nullable column that was "never actually null in practice" suddenly is. None of it crashes anything — it just rots the numbers people make decisions on.

So I built DataPact: a small framework for writing down what your data is supposed to look like, and then enforcing it. It's a data quality and data-contract validation tool, and the whole thing runs on the Python standard library. No pandas, no PyYAML, no network calls.

Live demo report: https://hajirufai.github.io/datapact/report.html
Landing page: https://hajirufai.github.io/datapact/
Source: https://github.com/hajirufai/datapact

The idea: contracts, not assertions scattered everywhere

Most teams already validate data — but it's usually a pile of ad-hoc assert df["x"].notna().all() lines buried in notebooks and DAGs. Nobody can answer "what are the rules for the orders table?" without grepping three repos.

A data contract flips that. You write the rules down in one declarative document — column types, null rules, ranges, allowed sets, regexes, cross-column math, referential integrity — version it in git, and let producers and consumers share it. DataPact then validates any batch against that contract and tells you, precisely, what broke.

Here's a contract in DataPact's YAML-lite format:

name: orders
version: 1.0
strictness: lenient
columns:
  - name: order_id
    type: int
    nullable: false
    checks:
      - kind: column_values_unique
        severity: error
  - name: status
    type: str
    checks:
      - kind: column_values_in_set
        kwargs: { values: [new, paid, shipped, refunded] }
expectations:
  - kind: multicolumn_sum_to_equal
    kwargs: { columns: [subtotal, tax], total_column: total, tolerance: 0.01 }

That last expectation is the exact rule that would have caught my 30% revenue ghost. subtotal + tax must equal total, within a cent.

"Zero dependencies" wasn't a vanity thing

I want to be honest about why this is stdlib-only, because it sounds like a flex and it mostly isn't. Two real reasons:

First, a lot of data platforms are locked down. You can't always pip install half of PyPI on the box where the pipeline runs. A validation tool that drops in with nothing but Python is genuinely easier to adopt than one that drags pandas + pyarrow + a YAML parser behind it.

Second, I wanted to actually understand the problem instead of gluing libraries together. Writing my own YAML reader and type-inference ladder taught me more about the messy reality of "what type is this column" than any wrapper would have.

The downside is I had to write a YAML parser. Which brings me to the most annoying bug of the whole project.

The escape-sequence bug that broke every email

DataPact ships its own tiny YAML reader — a strict subset: maps, lists, scalars, comments, quotes, flow lists. No arbitrary-object deserialization, which is a nice security property for free.

My email validation regex in the contract looked like this:

checks:
  - kind: column_values_match_regex
    kwargs:
      pattern: "^[^@ ]+@[^@ ]+\\.[^@ ]+$"

When I ran it, every single email failed, including obviously valid ones. 14 out of 14, 100%. My first naive parser just stripped the surrounding quotes and handed back the raw string — so \\. stayed as a literal backslash-backslash-dot. In the compiled regex that means "a literal backslash followed by any character," and no email on earth has a backslash in it.

The fix was to make the parser do what real YAML does: process escape sequences inside double-quoted strings, while leaving single-quoted strings literal.

_ESCAPES = {"n": "\n", "t": "\t", "r": "\r", '"': '"', "\\": "\\", "/": "/", "0": "\0"}

def _unescape_double(s: str) -> str:
    out, i = [], 0
    while i < len(s):
        ch = s[i]
        if ch == "\\" and i + 1 < len(s):
            out.append(_ESCAPES.get(s[i + 1], "\\" + s[i + 1]))
            i += 2
        else:
            out.append(ch)
            i += 1
    return "".join(out)

After that, the regex compiled to \. and the dirty email got flagged on its own (not-an-email has no @, so it correctly fails) while real addresses passed. The lesson, for the hundredth time in my career: escaping rules are never as simple as you hope, and the bug always shows up as "100% of things fail" rather than something subtle.

Type inference, and why "1" is not a boolean

The other rabbit hole was type inference. DataPact reads CSVs where everything is a string, so it has to guess: is "42" an int, is "4.5" a float, is "2026-01-04" a date?

I built an inference ladder — try bool, then int, then float, then datetime, then date, then fall back to string. And immediately got bitten. My parse_bool accepted "1" and "0" as True/False (reasonable when you're explicitly coercing). But during inference, that meant a column full of 1s and 0s got classified as boolean — and then my stdev check refused to run on it because "this isn't a numeric column."

The fix was to make inference conservative. Only unambiguous words — true, false, yes, no — infer as boolean. "1", "0", "t", "f" are far more likely to be integers or category codes, so they stay numeric:

if isinstance(value, str):
    if value.strip().lower() in ("true", "false", "yes", "no"):
        return "bool"
    if parse_int(value) is not None:
        return "int"
    if parse_float(value) is not None:
        return "float"
    ...

parse_bool is still lenient when you explicitly tell DataPact a column is boolean — but it no longer guesses bool from a digit. Small change, but it's the kind of default that quietly saves users from a confusing afternoon.

The architecture

The whole thing is a handful of small, pure pieces:

flowchart LR
    A[CSV / JSON / JSONL<br/>SQLite / records] --> B[Dataset<br/>+ type inference]
    C[Contract<br/>YAML / JSON / builder] --> D[Validation Engine]
    B --> D
    D --> E[ValidationReport]
    E --> F[CLI exit code]
    E --> G[HTML report]
    E --> H[guard / raise]
    B --> P[Profiler] --> C

Sources normalize CSV, JSON, JSONL, SQLite and plain lists-of-dicts into one Dataset view.
Expectations are a registry of pure functions — one per check kind. There are 23 of them across column-level (not_null, unique, between, in_set, match_regex, mean_between...), table-level (row_count_between, compound_columns_unique...) and cross-column (a > b, sum_to_equal, referential integrity).
The engine runs every expectation, applies strictness rules for unexpected columns, and builds a structured ValidationReport.

Every column-level check supports a mostly= tolerance, so you can say "this should be non-null in at least 99% of rows" instead of demanding perfection — real data is messy and a single bad row shouldn't always fail a 10-million-row batch.

Using it: three ways

As a library, validating against a contract file:

import datapact as dp

report = dp.validate("orders.csv", dp.load_contract("orders.yaml"))
print(report.success, report.passed, report.failed)
for r in report.results:
    if not r.success:
        print(r.expectation.label(), "→", r.message)

As a pipeline gate, with a decorator that raises before bad data escapes:

from datapact import guard, DataContractError

@guard(contract)
def load_orders():
    return fetch_rows_from_somewhere()

try:
    rows = load_orders()
except DataContractError as exc:
    alert(exc.report)   # the full report is attached to the exception

As a CI check, where a contract breach fails the build like a unit test:

datapact validate orders.csv --contract orders.yaml --fail-on error
echo $?   # 1 on breach

That --fail-on flag is the part I'm most happy with. It makes data quality a gate, not a dashboard nobody looks at. My GitHub Actions workflow actually runs two jobs: one proves the clean sample passes, and one proves the dirty sample fails — because a gate that never rejects anything is worse than no gate at all.

The report is the part people actually see

A validation result is only useful if a human can read it. So every run renders to a single self-contained HTML file — no external assets, no JavaScript framework — with failures sorted to the top and a full data profile (null rates, distinct counts, distributions) underneath.

I built the live demo report from a deliberately dirty orders file with ten injected problems: a duplicate order ID, a null email, a malformed email, an unknown status value, a bad date format, a negative subtotal, a quantity of 40, an invalid country code, and three rows where the totals don't add up. DataPact catches all ten — 41 expectations, 31 passed, 10 failed — and the report lays them out so you can see exactly which rows and values broke each rule.

Where it fits

DataPact is rule-based contract validation. It answers one question well: "does this batch obey the rules we agreed on?" That's deliberately different from statistical drift detection (is the distribution shifting?) or ETL orchestration (move the data around). It complements both — you'd run DataPact as the gate between an extract step and a load step, or in CI on your fixtures.

It's about 3,000 lines of pure Python with 141 tests, all stdlib unittest, no test dependencies either. If you've ever lost an afternoon to data that lied to you, give it a look — and if you find a rule it can't express yet, that's exactly the kind of issue I want to see.