Design duplicate detection over sliding windows

01The design, in the six-step framework

1. Requirements and use cases

A two-phase prompt from 2026 phone screens: flag repeats inside a sliding window of a stream, then the workload changes to read-heavy and the structure must change with it. A related report asks for duplicate files under bounded memory. The skill tested is evolving a design when the read-write ratio moves.

• A producer feeds items one at a time; each call reports whether the item repeats inside the window of the k most recent items, itself included.
• A consumer asks how many duplicate pairs or distinct items the current window holds.
• Phase two: a consumer asks, for an arbitrary range [l, r] of a fixed array, whether any duplicate exists, many times.
• Variant: a tool groups duplicate files by content under a memory limit.

Out of scope: fuzzy matching, and sharing across machines. Non-functional constraints: phase one is O(1) per item with memory O(k); phase two trades O(n) preprocessing for O(1) per query; the file variant reads as few bytes as possible.

2. Core entities and relationships

• WindowDuplicateDetector: the streaming structure; a deque of the window plus counts per item.
• RangeDuplicateIndex: the read-heavy structure; for each position, the nearest index where a duplicate pair ends.
• DuplicateFileFinder: groups files through cheaper-to-costlier fingerprints.
• Fingerprinter: a pluggable stage: size, partial hash, full hash.

These are separate classes because the two phases have different invariants; a single class with a mode flag would carry both sets of state. The index is built once from a snapshot and is immutable, so readers need no locks. See snapshots. What counts as "the same item" is decided by equality and hashing, which is why the detector works on anything hashable.

3. Interfaces and key methods

from typing import Generic, Hashable, List, Sequence, TypeVar

T = TypeVar("T", bound=Hashable)


class WindowDuplicateDetector(Generic[T]):
    def feed(self, item: T) -> bool:
        """True if item equals one of the previous k - 1 items; O(1)."""

    def distinct(self) -> int:
        """Distinct items in the window; O(1)."""

    def duplicate_pairs(self) -> int:
        """Equal pairs in the window; O(1)."""


class RangeDuplicateIndex(Generic[T]):
    def __init__(self, items: Sequence[T]) -> None:
        """O(n) build; immutable afterwards."""

    def has_duplicate(self, l: int, r: int) -> bool:
        """True if items[l..r] inclusive contains a repeat; O(1)."""

Invariants: the deque holds at most k items; counts matches the deque exactly with no zero entries; pairs equals the sum of c(c − 1) / 2 over counts. For the index, end_at[l] is the smallest r for which items[l..r] contains a repeat, and it never decreases as l grows.

adding a copy of an item with count c adds c pairs; removing one removes c − 1 pairs

4. Design patterns and the choices behind them

Counts, not a set, for the window

The force is an item that appears twice inside the window. With a set, when the older copy slides out, removing it also forgets the newer copy, and the next repeat goes undetected. A counter removes the item only at zero. That one bug is why a sliding window should keep counts. The running pair total updates in O(1) from the formula above, so no read rescans.

Precompute when reads dominate

The force is the phase change. Running the detector for each range query costs O(r − l), which is ruinous at millions of queries. Instead, compute once for every left end the first position that closes a duplicate. A right-to-left pass does this: end_at[i] = min(end_at[i + 1], next occurrence of items[i]). Then a query is one comparison. This is the "evolve towards precomputed summaries" the report describes, and the cost is that the data must be fixed or rebuilt. If updates continue, the honest position is that a segment tree over next-occurrence gives O(log n) for both; say it rather than building it, per trade-offs and over-engineering.

Cheap filters first for files

The force is that reading every byte is the expensive step. Two files of different sizes cannot be equal, so size is a free filter; a hash of the first 4 KB removes most of the rest; a full hash runs only on the survivors. The stages form a chain, each narrowing the candidate groups. Memory holds only group keys, since file contents are hashed in chunks. A hash match is not strictly a proof; a byte comparison settles it if a false positive would be costly.

5. Core flow and code

Flow for feed with k = 3 over a, b, a, c:

1. Feed a, then b: counts a:1, b:1, no duplicates.
2. Feed a: its count is 1, so it is a duplicate, and pairs rises by 1. The window is a, b, a.
3. Feed c: the window is full, so the oldest a leaves. Its count drops from 2 to 1 and pairs falls by 1. a stays known, which a set would have lost.

from collections import Counter, deque
from dataclasses import dataclass
from typing import Callable, Deque, Dict, Generic, Hashable, List, Sequence, TypeVar

T = TypeVar("T", bound=Hashable)


class WindowDuplicateDetector(Generic[T]):
    def __init__(self, k: int) -> None:
        if k < 1:
            raise ValueError("k must be at least 1")
        self.k = k
        self.window: Deque[T] = deque()
        self.counts: Counter = Counter()
        self.pairs = 0

    def feed(self, item: T) -> bool:
        if len(self.window) == self.k:
            old = self.window.popleft()
            self.counts[old] -= 1
            self.pairs -= self.counts[old]
            if not self.counts[old]:
                del self.counts[old]
        seen = self.counts[item]
        self.pairs += seen
        self.counts[item] = seen + 1
        self.window.append(item)
        return seen > 0

    def distinct(self) -> int:
        return len(self.counts)

    def duplicate_pairs(self) -> int:
        return self.pairs


class RangeDuplicateIndex(Generic[T]):
    def __init__(self, items: Sequence[T]) -> None:
        n = len(items)
        self.n = n
        self.end_at: List[int] = [n] * (n + 1)
        next_seen: Dict[T, int] = {}
        for i in range(n - 1, -1, -1):
            nxt = next_seen.get(items[i], n)
            self.end_at[i] = min(self.end_at[i + 1], nxt)
            next_seen[items[i]] = i

    def has_duplicate(self, l: int, r: int) -> bool:
        if not 0 <= l <= r < self.n:
            raise IndexError("bad range")
        return self.end_at[l] <= r


@dataclass(frozen=True)
class FileInfo:
    path: str
    content: bytes


Fingerprinter = Callable[[FileInfo], Hashable]


class DuplicateFileFinder:
    def __init__(self, stages: List[Fingerprinter]) -> None:
        self.stages = stages

    def groups(self, files: List[FileInfo]) -> List[List[str]]:
        candidates = [files]
        for stage in self.stages:
            next_round: List[List[FileInfo]] = []
            for group in candidates:
                buckets: Dict[Hashable, List[FileInfo]] = {}
                for f in group:
                    buckets.setdefault(stage(f), []).append(f)
                next_round.extend(b for b in buckets.values() if len(b) > 1)
            candidates = next_round
        return [sorted(f.path for f in g) for g in candidates]


if __name__ == "__main__":
    import hashlib

    d: WindowDuplicateDetector[str] = WindowDuplicateDetector(3)
    assert [d.feed(x) for x in "abac"] == [False, False, True, False]
    assert d.distinct() == 3 and d.duplicate_pairs() == 0     # window is b, a, c
    assert d.feed("c") and d.duplicate_pairs() == 1

    r = RangeDuplicateIndex([1, 2, 3, 1, 2])
    assert r.has_duplicate(0, 3) and not r.has_duplicate(1, 3) and r.has_duplicate(1, 4)
    assert not r.has_duplicate(2, 3)

    finder = DuplicateFileFinder([
        lambda f: len(f.content),
        lambda f: f.content[:4],
        lambda f: hashlib.sha256(f.content).hexdigest(),
    ])
    files = [FileInfo("a", b"same bytes"), FileInfo("b", b"same bytes"),
             FileInfo("c", b"same bytez"), FileInfo("d", b"short")]
    assert finder.groups(files) == [["a", "b"]]

6. Extensions, concurrency and what interviewers probe

• Set versus counter. Have the failing example ready: a, b, a with the first a leaving. The set says a is gone.
• Read-heavy mode. Name the ratio that justifies the rebuild. If updates are one in a million, rebuilding on update is fine; if they are one in ten, use the segment tree.
• Memory-bounded streams. When distinct items are too many for a counter, a Bloom filter per time slice gives approximate detection with no false negatives, as in Bloom filters.
• File variant policy. Skip symlinks to avoid counting a file twice, set a minimum size, and treat hard links as the same inode.
• Thread safety. One lock for the detector, because feed changes three structures together. The built index needs none.
• Common mistake. Evicting after inserting, which makes the window off by one and treats an item exactly k positions back as a duplicate.

02Where this question was reported

• Land the Lab · Two-phase API: streaming windows then read-heavy query mode
• ReturnZero · Find duplicate files grouped by content under bounded memory; follow-ups: size pre-filter, symlink/min-size policy
• PracHub · Google SWE; evolve structure toward precomputed summaries