Design an extensible spreadsheet aggregation API

01The design, in the six-step framework

1. Requirements and use cases

A spreadsheet supports SUM and MAX over a rectangular range. A developer must be able to add a new aggregate, say AVG or COUNTIF, without copying the loop that walks the range. Formulas look up aggregates by name. Results for an unchanged range should not be recomputed on every read, and a large range should be computable in chunks that can later be split across threads.

Out of scope: parsing full formulas with nesting (see the spreadsheet with formulas), cell-to-cell dependency graphs, and persistence. The reported interview focused on the aggregate interface, caching and merging partial results.

Constraints: one traversal per aggregate call; adding an aggregate touches one new class and one registration line; the cache must be invalidated exactly for ranges that contain a changed cell; the merge of partial states must give the same answer as a single pass.

2. Core entities and relationships

• Aggregate: the contract with four hooks: init makes an empty state, accumulate folds one value in, finalize turns state into the result, merge combines two states.
• Sum, Max, Average: implementations whose only knowledge is their own state shape; Average keeps (sum, count) rather than a running mean so that merge is exact.
• Range: an immutable rectangle that knows how to enumerate its cells, split itself into row chunks and test containment.
• Sheet: holds values, the aggregate registry and the result cache, and owns the single traversal.

Nothing is composed here except values inside the sheet; aggregates are stateless and shared, ranges are values. That is why the cache key can be the pair (range, aggregate name).

3. Interfaces and key methods

class Aggregate(Protocol[S, R]):
    """Three hooks; the sheet owns the traversal."""
    def init(self) -> S: ...
    def accumulate(self, state: S, value: int) -> S: ...
    def finalize(self, state: S) -> R: ...
    def merge(self, a: S, b: S) -> S: ...        # combine partial states from chunks

class Sheet:
    def aggregate(self, rng: "Range", agg: Aggregate[S, R]) -> R:
        """One traversal of the range; cached until a cell in it changes."""
    def register(self, name: str, agg: Aggregate) -> None:
        """Make an aggregate available to formulas as NAME(A1:B9)."""

Invariants: finalize(merge(a, b)) equals finalizing a single pass over the union of the chunks that produced a and b; a cache entry exists only for a range whose cells have not changed since it was computed; the registry maps upper-case names to one aggregate instance each.

4. Design patterns and the choices behind them

Template method inverted into a strategy

What varies is the folding rule; what stays fixed is the walk over the range. A template method would put the walk in a base class and let subclasses override a step; that works but couples every aggregate to the sheet's traversal class. Passing the aggregate as a strategy object with init, accumulate and finalize keeps aggregates free of inheritance and lets the sheet choose plain or chunked traversal. The rejected alternative, one method per aggregate on the sheet (sum_range, max_range), duplicates the loop and is exactly what the interviewer said not to do.

Open for new aggregates, closed for traversal

Adding Median means one class and register("MEDIAN", Median()); the sheet, the cache and the formula lookup do not change. That is the open-closed principle as a concrete test: name the next feature and count the files it touches.

Mergeable state, not just a running value

The fourth hook, merge, is what makes chunking correct. A running mean cannot be merged; (sum, count) can. Max merges trivially; a distinct-count would need a set or a sketch. Asking "is your state mergeable?" is how to tell which aggregates can be parallelised, and it is the question the reported round ended on.

Cache keyed by range and invalidated by containment

Results are cached per (range, aggregate). A cell write drops every cached entry whose range contains the cell. This is linear in the number of cached ranges; an interval index would make it logarithmic and is the usual follow-up.

5. Core flow and code

1. Cells (0,0) to (2,1) are set. A formula SUM(0,0:2,1) arrives at evaluate.
2. The name is looked up in the registry, the range is parsed into a Range value, and aggregate is called.
3. Cache miss: init gives 0, accumulate runs once per cell, finalize returns the total, which is cached under (range, "Sum").
4. A second SUM over the same range returns from the cache without a traversal; a MAX over it misses, because the key includes the aggregate name.
5. set(1, 1, 99) drops both cached entries because the range contains (1,1).
6. With chunk_rows=1, the same sum is computed as three partial states merged with merge, yielding the identical result; those three folds could run on three threads.

from dataclasses import dataclass
from typing import Generic, Protocol, TypeVar

S = TypeVar("S"); R = TypeVar("R")

class Aggregate(Protocol[S, R]):
    def init(self) -> S: ...
    def accumulate(self, state: S, value: int) -> S: ...
    def finalize(self, state: S) -> R: ...
    def merge(self, a: S, b: S) -> S: ...

class Sum:
    def init(self) -> int: return 0
    def accumulate(self, state: int, value: int) -> int: return state + value
    def finalize(self, state: int) -> int: return state
    def merge(self, a: int, b: int) -> int: return a + b

class Max:
    def init(self) -> int | None: return None
    def accumulate(self, state: int | None, value: int) -> int | None:
        return value if state is None or value > state else state
    def finalize(self, state: int | None) -> int | None: return state
    def merge(self, a: int | None, b: int | None) -> int | None:
        return b if a is None else a if b is None else max(a, b)

class Average:
    """State is (sum, count): mergeable, unlike a running mean."""
    def init(self) -> tuple[int, int]: return (0, 0)
    def accumulate(self, s: tuple[int, int], v: int) -> tuple[int, int]: return (s[0] + v, s[1] + 1)
    def finalize(self, s: tuple[int, int]) -> float | None: return None if s[1] == 0 else s[0] / s[1]
    def merge(self, a: tuple[int, int], b: tuple[int, int]) -> tuple[int, int]: return (a[0] + b[0], a[1] + b[1])

@dataclass(frozen=True)
class Range:
    top: int; left: int; bottom: int; right: int          # inclusive, zero-based
    def cells(self):
        for r in range(self.top, self.bottom + 1):
            for c in range(self.left, self.right + 1):
                yield (r, c)
    def chunks(self, rows: int):
        for t in range(self.top, self.bottom + 1, rows):
            yield Range(t, self.left, min(t + rows - 1, self.bottom), self.right)
    def contains(self, r: int, c: int) -> bool:
        return self.top <= r <= self.bottom and self.left <= c <= self.right

class Sheet:
    def __init__(self) -> None:
        self._values: dict[tuple[int, int], int] = {}
        self._registry: dict[str, Aggregate] = {"SUM": Sum(), "MAX": Max(), "AVG": Average()}
        self._cache: dict[tuple[Range, str], object] = {}

    def register(self, name: str, agg: Aggregate) -> None:
        self._registry[name.upper()] = agg

    def set(self, r: int, c: int, value: int) -> None:
        self._values[(r, c)] = value
        self._cache = {k: v for k, v in self._cache.items() if not k[0].contains(r, c)}

    def get(self, r: int, c: int) -> int:
        return self._values.get((r, c), 0)

    def _fold(self, rng: Range, agg: Aggregate[S, R]) -> S:
        state = agg.init()
        for (r, c) in rng.cells():
            state = agg.accumulate(state, self.get(r, c))
        return state

    def aggregate(self, rng: Range, agg: Aggregate[S, R], chunk_rows: int = 0) -> R:
        key = (rng, type(agg).__name__)
        if key in self._cache:
            return self._cache[key]  # type: ignore[return-value]
        if chunk_rows > 0:                      # partial states per chunk, merged: parallel-ready
            state = agg.init()
            for part in rng.chunks(chunk_rows):
                state = agg.merge(state, self._fold(part, agg))
        else:
            state = self._fold(rng, agg)
        result = agg.finalize(state)
        self._cache[key] = result
        return result

    def evaluate(self, formula: str) -> object:
        """'SUM(0,0:2,1)' -> registry lookup, parse the range, one traversal."""
        name, _, rest = formula.partition("(")
        a, b = rest.rstrip(")").split(":")
        t, l = map(int, a.split(",")); bt, rt = map(int, b.split(","))
        return self.aggregate(Range(t, l, bt, rt), self._registry[name.strip().upper()])

6. Extensions, concurrency and what interviewers probe

• Aggregates with parameters, such as COUNTIF(range, ">3"): build the aggregate instance with its predicate at lookup time (a small factory per name) and include the parameter in the cache key.
• Cycles: if an aggregate's range contains the cell holding the formula, evaluation loops. Detect it at formula set time with the dependency check from the spreadsheet problem; ranges simply expand to many dependencies.
• Parallel chunks: hand _fold of each chunk to a pool and merge the futures' results; correctness rests entirely on the merge invariant, so test it first.
• Cache invalidation cost: thousands of cached ranges make the containment scan slow; an R-tree or a per-row index of cached ranges fixes it, and saying when it is needed matters more than building it.
• Test first: Sum, Max and Average over a known range, then the same values computed with chunking, asserting equality; then a set inside and outside the range checking which cache entries survive.
• Common mistake: making Average keep a running mean so merging two chunks gives the wrong answer, and keying the cache on the range alone so MAX returns SUM's result.

02Where this question was reported

• PracHub · Google: add new aggregate without duplicating traversal; aggregate init/accumulate/finalise interface; cycles, caching, chunk merging