Database scaling &amp; optimization

import re
from collections import defaultdict


class SimpleTextIndex:
    def __init__(self):
        self.index = defaultdict(list)
        self.documents = {}

    def add_document(self, doc_id, text):
        self.documents[doc_id] = text
        words = set(re.findall(r'\w+', text.lower()))
        for word in wo…

12 0 Open

Composite index leftmost prefix in Python

Simulate a composite index in SQLite and check whether query columns match the leftmost prefix rule for index usage.

sqlite indexes database

import sqlite3


def get_indexed_columns(table_name):
    """Simulate a composite index by reading column names that start with 'idx_'."""
    conn = sqlite3.connect(":memory:")
    conn.execute(f"CREATE TABLE {table_name} (id INTEGER, idx_col1 TEXT, idx_col2 INTEGER, other TEXT)")
    conn.execute(f"CREATE INDEX idx_…

scatter-gather sharding distributed-systems

Cross Shard Query Scatter Gather Mock in Python

Simulate a distributed database cross-shard query using a scatter-gather pattern with a mock Python implementation.

from dataclasses import dataclass
from typing import List, Dict


@dataclass
class NodeResponse:
    node_id: int
    data: Dict[str, float]


def mock_query_shard(shard_id: int, shard_data: Dict[str, float], query: str) -> NodeResponse:
    """Simulate querying a single shard, returning matches whose value > 50."""
 …

13 0 Open

How to Build a Shard Map Mock Dict in Python

Implement a dictionary-like class that distributes keys across multiple shards using Python's hash() for realistic data partitioning.

dict sharding hash

class ShardMap:
    def __init__(self, shard_count):
        self.shards = {i: {} for i in range(shard_count)}
        self.shard_count = shard_count

    def _shard_for(self, key):
        return hash(key) % self.shard_count

    def __getitem__(self, key):
        return self.shards[self._shard_for(key)][key]

    d…

consistent-hashing distributed-systems sharding

How to Implement Consistent Hashing in Python

Build a consistent hash ring in Python that distributes keys across nodes and minimizes remapping when nodes are added or removed.

import hashlib
from bisect import bisect_right


class ConsistentHashRing:
    def __init__(self, nodes, replicas=3):
        self.replicas = replicas
        self.ring = {}
        self.sorted_keys = []
        for node in nodes:
            self.add_node(node)

    def _hash(self, key):
        return int(hashlib.md…

15 0 Open

saga sharding distributed-systems

How to Mock a Cross-Shard Saga in Python

Simulate a distributed saga with compensating transactions across multiple database shards using a lightweight Python class that tracks executed steps and rolls them back in reverse on failure.

import json


class SagaState:
    def __init__(self, saga_id):
        self.saga_id = saga_id
        self.executed_steps = []
        self.compensations = []

    def execute_step(self, shard, step_name, operation):
        self.executed_steps.append((shard, step_name))
        print(f"[Saga {self.saga_id}] Executin…

jsonb gin-index inverted-index

Simulate a GIN Index for JSONB in Python

Build a mock Generalized Inverted Index (GIN) that flattens JSON documents into key-value tokens for fast lookup queries, mimicking PostgreSQL JSONB indexing.

import json
import random
from collections import defaultdict

# Mock GIN (Generalized Inverted Index) for JSONB key-value pairs
class GINIndex:
    def __init__(self):
        self.posting_lists = defaultdict(list)  # token -> list of doc_ids
    
    def index(self, doc_id, json_obj):
        """Index a JSON documen…

snowflake id-generation clustering

Snowflake ID Generator with Cluster Index Mock in Python

A thread-safe Snowflake ID generator mock that creates unique 64-bit IDs across simulated cluster nodes and maintains a sorted in-memory index for range queries.

import time
import threading

class SnowflakeIDGenerator:
    def __init__(self, machine_id, datacenter_id):
        self.machine_id = machine_id
        self.datacenter_id = datacenter_id
        self.sequence = 0
        self.last_timestamp = -1
        self.machine_bits = 5
        self.datacenter_bits = 5
        …

13 0 Open

two-phase-commit distributed-systems transaction

Two Phase Commit Cross Shard Mock in Python

Simulates a two-phase commit across shards with failure handling to demonstrate distributed transaction coordination in Python.

"""Mock cross-shard two-phase commit with caution handling."""

class Shard:
    def __init__(self, name):
        self.name = name
        self.prepared = False
        self.committed = False
        self.aborted = False

    def prepare(self):
        # Simulate potential failure (1 in 3 chance on third shard)
     …