Problem 78 · Object & System Design

In-Memory SQL-like Table

MediumClassic DSA & systems

01Problem

Implement Table(columns) with insert(row: dict), select(columns, where=None, order_by=None, limit=None) where where is a list of (column, op, value) conjuncts and order_by is a list of (column, 'asc'|'desc'). Optionally create_index(column) for equality lookups.

02Key insight

Separate the plan: pick an access path (hash index on an equality predicate if available, else full scan), filter, sort, project, limit. Multi-key mixed-direction ordering is done with successive stable sorts from the last key to the first.

03Solution

solution.pypython52 lines

import operator
from collections import defaultdict
from typing import Any

_OPS = {"=": operator.eq, "!=": operator.ne, "<": operator.lt,
        "<=": operator.le, ">": operator.gt, ">=": operator.ge}


class Table:
    def __init__(self, columns: list[str]) -> None:
        self.columns = list(columns)
        self.rows: list[dict[str, Any]] = []
        self.indexes: dict[str, defaultdict[Any, list[int]]] = {}

    def insert(self, row: dict[str, Any]) -> None:
        unknown = set(row) - set(self.columns)
        if unknown:
            raise KeyError(f"unknown columns: {sorted(unknown)}")
        rid = len(self.rows)
        self.rows.append({c: row.get(c) for c in self.columns})
        for col, idx in self.indexes.items():
            idx[self.rows[rid][col]].append(rid)

    def create_index(self, column: str) -> None:
        idx: defaultdict[Any, list[int]] = defaultdict(list)
        for rid, r in enumerate(self.rows):
            idx[r[column]].append(rid)
        self.indexes[column] = idx

    def select(self, columns: list[str] | None = None,
               where: list[tuple[str, str, Any]] | None = None,
               order_by: list[tuple[str, str]] | None = None,
               limit: int | None = None) -> list[dict[str, Any]]:
        where = where or []
        # access path: use an index for the first indexed equality predicate
        candidates = self.rows
        for col, op, val in where:
            if op == "=" and col in self.indexes:
                candidates = [self.rows[i] for i in self.indexes[col].get(val, [])]
                break
        result = [r for r in candidates
                  if all(r[c] is not None and _OPS[op](r[c], v) for c, op, v in where)]
        for col, direction in reversed(order_by or []):
            # NULLs sort last regardless of direction
            nonnull = [r for r in result if r[col] is not None]
            nulls = [r for r in result if r[col] is None]
            nonnull.sort(key=lambda r: r[col], reverse=direction.lower() == "desc")
            result = nonnull + nulls
        if limit is not None:
            result = result[:limit]
        cols = columns or self.columns
        return [{c: r[c] for c in cols} for r in result]

Try it first.

TimeO(n log n) scan+sort; O(k log k) with index hit

SpaceO(n)

tests.py · 3 passing

tests.pypython9 lines

t = Table(["id", "name", "dept", "salary"])
for row in [(1, "ann", "eng", 200), (2, "bob", "eng", 150), (3, "cat", "ops", 150), (4, "dan", "eng", None)]:
    t.insert(dict(zip(t.columns, row)))
q = t.select(["name"], where=[("dept", "=", "eng")], order_by=[("salary", "desc")])
assert q == [{"name": "ann"}, {"name": "bob"}, {"name": "dan"}]
q = t.select(["id"], order_by=[("salary", "asc"), ("name", "desc")], limit=2)
assert q == [{"id": 3}, {"id": 2}]
t.create_index("dept")
assert t.select(["id"], where=[("dept", "=", "ops"), ("salary", ">=", 100)]) == [{"id": 3}]

04Staff-level follow-up

How do columnar engines (Snowflake, Databricks/Photon) differ? Store columns contiguously, prune with per-micro-partition min/max zone maps, evaluate predicates vectorized over column batches, and late-materialize projected columns.