Netflix Python Developer Interview Questions & Answers

netflix icon

Questions with Detailed ExplanationsWith Detailed Explanations

(Last Updated: August 28, 2026)

11. How would you minimize lock contention in a Python shared dictionary?Language SpecificHardNetflix

Question Details

Explain the correctness and performance tradeoffs of coarse-grained locking, lock striping, read-write locks, immutable snapshots, and process-based alternatives.

Short Interview Answer (30-60 seconds)

I would begin with one lock and keep every critical section very small. If profiling proves that lock waiting is a bottleneck, I would partition the data and use one lock per partition. This lock striping approach lets operations on unrelated keys use different locks. For data with many reads and rare updates, I would consider immutable snapshots. For CPU bound work or stronger isolation, I would consider processes that own separate state and communicate through messages.

Detailed Explanation

I would start with one lock because it is the easiest design to make correct. The lock must cover the complete logical operation, such as reading a value and then updating it. Slow calculation, logging, file access, and network access should stay outside the critical section.

Useful Questions to Ask the Interviewer
  1. Should I focus on Python language behavior, or also explain the runtime and standard library?
  2. Which Python version and execution environment should I assume?
  3. Would you like a small code example together with production tradeoffs and edge cases?

If profiling shows significant waiting, I would use lock striping. I would split the data into several dictionaries. Each dictionary has its own lock, and every key always maps to the same stripe. Operations on different stripes avoid waiting for the same application lock. Operations involving several keys must acquire all required stripe locks in a consistent order to prevent deadlock.

A read write lock may help when reads are long and greatly outnumber writes, but it often adds little value for very short dictionary reads. The threading module does not provide one, so another implementation adds overhead and may introduce fairness or starvation concerns.

Immutable snapshots fit data that is read often and changed rarely. A writer copies the current dictionary, applies changes, and publishes the new reference under a short lock. Readers obtain the reference under that lock, then read without holding it. Copying costs linear time and temporary linear memory.

Processes can remove shared thread state, but serialization and communication add cost.

How would you minimize lock contention in a Python shared dictionary? diagram
Where it is used

These approaches are useful in in memory caches, request counters, connection registries, routing tables, feature settings, and service state. One lock fits small dictionaries or light concurrency. Lock striping fits many independent keys with frequent concurrent access. Immutable snapshots fit configuration and routing data with many reads and rare updates. Process ownership fits CPU bound workloads or systems that need stronger isolation.

Why Interviewers Ask This

Interviewers ask this question to test whether a candidate can separate dictionary operations from synchronization guarantees. They want to see correct protection of compound operations and sound judgment about contention, deadlock risk, memory cost, portability, and production complexity.

Common interview mistakes

A common mistake is assuming that the Global Interpreter Lock makes a compound action such as check then update atomic. Another mistake is protecting only one step of a logical operation. Holding a lock during slow calculation, logging, file access, or network access creates unnecessary contention. In a striped design, changing the stripe calculation or using different locks for the same key breaks correctness. Acquiring several stripe locks in inconsistent orders can deadlock. Too many stripes increase lock objects, dictionaries, memory use, and maintenance complexity. Snapshot readers must never mutate a published snapshot, and writers must not modify the old dictionary in place. A process manager dictionary should not be assumed to remove contention because proxy calls still require communication and coordination.

Interview tip

Start with correctness and the simplest design. Explain one short lock first. Then say that profiling may justify lock striping, a read write lock, immutable snapshots, or process ownership. Mention compound operations, stable lock ordering, snapshot copying cost, and why the Global Interpreter Lock is not a complete synchronization strategy.

Interviewer may ask next
Does the Global Interpreter Lock make compound dictionary operations safe?

No. The Global Interpreter Lock does not make a compound operation such as check then update atomic. Another thread may run between the separate Python operations and change the dictionary. A dedicated lock must protect the complete logical operation. This matters for correctness and avoids depending on interpreter specific behavior. The tradeoff is lock waiting, so the protected work should remain small.

When is an immutable snapshot better than lock striping?

An immutable snapshot is better when reads are very frequent and updates are rare. A writer copies the current dictionary, applies the complete update to the copy, and publishes the new reference under a short lock. A reader obtains that reference under the same lock and then reads the selected snapshot without holding it. This reduces read lock duration. The tradeoff is linear copying time, temporary linear memory use, and readers that may continue using the previous complete snapshot.

12. Design a concurrent latency percentile tracker.Language SpecificHardNetflix

Question Details

Design a thread-safe LatencyTracker that records timestamped latency samples and returns requested percentiles while handling concurrency and bounded retention.

Short Interview Answer (30-60 seconds)

I would keep timestamped latency samples in a deque and protect the deque with a threading lock. Every record and query removes expired samples and enforces a maximum count. A percentile query copies the current values while holding the lock, then releases the lock before sorting. This gives the query a consistent snapshot while allowing new records to continue during the expensive sort.

Detailed Explanation

See the Code while reading this explanation.

I would store each sample as a monotonic timestamp and a latency value inside a deque. A threading.Lock protects every operation that reads or changes this shared deque. The Global Interpreter Lock is not enough because pruning, appending, checking length, and copying form a multi step operation.

Useful Questions to Ask the Interviewer
  1. Should I focus on Python language behavior, or also explain the runtime and standard library?
  2. Which Python version and execution environment should I assume?
  3. Would you like a small code example together with production tradeoffs and edge cases?

The record method validates the value, acquires the lock, records time with time.monotonic, appends the sample, removes expired entries, and removes the oldest entries when the count limit is exceeded. A monotonic clock is suitable because changes to the system clock do not affect elapsed time.

The percentile method validates a value from zero through one hundred. While holding the lock, it removes expired entries and copies retained latency values. It then releases the lock, sorts the private copy, and uses linear interpolation between neighboring values. With samples 10, 20, 30, and 40, P50 is 25 and P95 is 38.5.

An empty tracker raises ValueError. Recording is amortized constant time, although one call may remove several old entries. A query takes linear time to copy, n log n time to sort, and linear extra memory. This design suits bounded local metrics. A histogram is better for much larger workloads.

Design a concurrent latency percentile tracker. diagram
Example

LatencyTracker stores samples in recording order inside a deque. Each sample contains a time.monotonic timestamp and a latency value in milliseconds. max_age_seconds limits how long a sample is retained, while max_samples provides a second strict memory bound. The private cleanup method runs only while the lock is held. The percentile method removes expired data and copies the values under the lock, then sorts the private snapshot after releasing the lock. It maps the requested percentile to a position from zero through the final sorted index and uses linear interpolation when the position falls between two values. For the example values 10, 20, 30, and 40, the code returns 25 for P50 and 38.5 for P95.

Code
from __future__ import annotations

import math
import threading
import time
from collections import deque
from typing import Deque


class LatencyTracker:
    """Store recent latency samples and calculate percentiles safely."""

    def __init__(
        self,
        max_samples: int = 10_000,
        max_age_seconds: float = 300.0,
    ) -> None:
        # Both limits must be positive so retention stays bounded.
        if max_samples <= 0:
            raise ValueError("max_samples must be greater than zero")
        if max_age_seconds <= 0:
            raise ValueError("max_age_seconds must be greater than zero")

        self._max_samples = max_samples
        self._max_age_seconds = float(max_age_seconds)

        # Samples remain ordered by recording time.
        self._samples: Deque[tuple[float, float]] = deque()

        # One lock protects every compound operation on the shared deque.
        self._lock = threading.Lock()

    def record(self, latency_ms: float) -> None:
        """Record one finite, nonnegative latency value in milliseconds."""
        value = float(latency_ms)

        # Reject invalid values before changing shared state.
        if not math.isfinite(value) or value < 0:
            raise ValueError("latency_ms must be a finite nonnegative number")

        with self._lock:
            # Capture the timestamp while holding the lock so insertion and
            # cleanup use one consistent point in time.
            now = time.monotonic()
            self._samples.append((now, value))
            self._prune_locked(now)

    def percentile(self, requested_percentile: float) -> float:
        """Return a percentile from a consistent snapshot of retained samples."""
        percentile_value = float(requested_percentile)

        if not math.isfinite(percentile_value):
            raise ValueError("requested_percentile must be finite")
        if not 0.0 <= percentile_value <= 100.0:
            raise ValueError("requested_percentile must be between zero and one hundred")

        with self._lock:
            now = time.monotonic()

            # Expired values must not appear in the snapshot.
            self._prune_locked(now)

            if not self._samples:
                raise ValueError("no latency samples are available")

            # Copy only latency values while shared state is protected.
            snapshot = [latency for _, latency in self._samples]

        # Sorting the private snapshot does not block record calls.
        snapshot.sort()

        # Map the requested percentile to a zero based sorted position.
        position = (len(snapshot) - 1) * percentile_value / 100.0
        lower_index = math.floor(position)
        upper_index = math.ceil(position)

        # An exact position needs no interpolation.
        if lower_index == upper_index:
            return snapshot[lower_index]

        # Use linear interpolation between neighboring values.
        fraction = position - lower_index
        lower_value = snapshot[lower_index]
        upper_value = snapshot[upper_index]
        return lower_value + (upper_value - lower_value) * fraction

    def sample_count(self) -> int:
        """Return the number of samples that are currently retained."""
        with self._lock:
            now = time.monotonic()
            self._prune_locked(now)
            return len(self._samples)

    def _prune_locked(self, now: float) -> None:
        """Remove expired and excess samples while the caller holds the lock."""
        cutoff = now - self._max_age_seconds

        # Entries are ordered by time, so expiration starts at the left side.
        while self._samples and self._samples[0][0] < cutoff:
            self._samples.popleft()

        # Remove the oldest retained entries when the count limit is exceeded.
        while len(self._samples) > self._max_samples:
            self._samples.popleft()


if __name__ == "__main__":
    tracker = LatencyTracker(max_samples=100, max_age_seconds=60.0)

    # Record one consistent example set.
    for latency in (10.0, 20.0, 30.0, 40.0):
        tracker.record(latency)

    print("Sample count:", tracker.sample_count())
    print("P50:", tracker.percentile(50.0))
    print("P95:", tracker.percentile(95.0))
Where it is used

This tracker can be used inside a Python web service, worker process, database wrapper, API client, or background job. It can measure recent request latency, database query time, queue wait time, or external service response time. It is most useful when one Python process needs a bounded recent view for health checks, debugging, or local monitoring.

Why Interviewers Ask This

Interviewers ask this question to test whether a Python developer can protect shared mutable state, use locks correctly, choose a clock for elapsed time, bound memory, calculate percentiles consistently, and avoid holding a lock during expensive work. It also tests whether the candidate understands that the Global Interpreter Lock does not make a multi step operation logically atomic.

Common interview mistakes

A common mistake is assuming the Global Interpreter Lock makes the tracker thread safe. It does not protect the complete sequence of pruning, appending, checking length, and copying. Another mistake is sorting shared data while holding the lock, which blocks writers for the full sorting time. Developers may also forget a count limit, use time.time for elapsed retention, accept NaN or infinity, calculate percentiles with an inconsistent formula, or return values from expired samples. Another error is claiming every record call is strictly constant time even though one call may remove many old entries.

Interview tip

Start with the concurrency rule. Explain that the lock protects the shared deque, but sorting uses a private snapshot outside the lock. Then describe the age limit, count limit, interpolation rule, empty state behavior, and exact time and memory costs.

Interviewer may ask next
What happens when records arrive during a percentile calculation?

The calculation returns the exact percentile for the snapshot copied while the lock was held. Records added after that copy are not part of the current result, but they remain safely stored for later queries. This behavior matters because the query gets a consistent view without blocking record calls during sorting.

How would you handle much higher sample volume?

I would replace exact snapshot sorting with a bounded histogram or an approximate quantile structure while keeping the same concurrency and retention rules. This change reduces query cost and gives more predictable memory use. The main tradeoff is accuracy because the current design returns an exact interpolated percentile for retained samples, while the alternative returns an estimate.

13. How would you implement an atomic counter in Python?Language SpecificHardNetflix

Question Details

Implement or describe a thread-safe counter with increment and read operations, then explain what guarantees Python locks provide.

Short Interview Answer (30-60 seconds)

I would store the counter value in a class and protect every read and increment with the same threading.Lock. The lock lets only one participating thread enter the protected section at a time. I would not rely on the Global Interpreter Lock because the complete read, add, and write sequence is not a safe application level synchronization contract.

Detailed Explanation

See the Code while reading this explanation.

I would implement the counter with one integer and one threading.Lock. The increment method acquires the lock, changes the value, and returns the new value. The read method acquires the same lock before returning the current value. This makes every completed increment and read behave as one protected operation for threads using that counter.

Useful Questions to Ask the Interviewer
  1. Should I focus on Python language behavior, or also explain the runtime and standard library?
  2. Which Python version and execution environment should I assume?
  3. Would you like a small code example together with production tradeoffs and edge cases?

A Python lock provides mutual exclusion. While one thread owns the lock, another thread that tries to acquire it must wait. Releasing the lock allows a waiting thread to continue. The with statement is useful because it releases the lock even if an exception leaves the protected block.

I would not depend on the Global Interpreter Lock. A counter update includes reading the current value, calculating a result, and storing that result. Explicit locking makes the required guarantee clear and remains correct across Python runtime changes.

Each read and increment uses constant work and no growing data structure. The counter object uses constant memory. Under heavy contention, waiting for the lock can reduce throughput. This design protects threads in one process only. Separate processes or servers need a process safe or external atomic counter.

How would you implement an atomic counter in Python? diagram
Example

The AtomicCounter class stores one integer and one threading.Lock. Both increment and read acquire the same lock with a with statement. Increment changes the protected value and returns the new value before the lock is released. Read returns the value while holding that same lock. Each operation uses constant work and the object uses constant memory. The example starts four threads, lets each thread perform ten thousand increments, waits for all threads to finish, and safely reads the final value of forty thousand.

Code
import threading


class AtomicCounter:
    def __init__(self, initial_value: int = 0) -> None:
        # Store the integer shared by all participating threads.
        self._value = initial_value

        # Use one lock to protect every read and update.
        self._lock = threading.Lock()

    def increment(self, amount: int = 1) -> int:
        # Only one participating thread can run this block at a time.
        with self._lock:
            self._value += amount
            return self._value

    def read(self) -> int:
        # Use the same lock to return a protected snapshot.
        with self._lock:
            return self._value


def worker(counter: AtomicCounter, repetitions: int) -> None:
    # Increment the shared counter once per loop iteration.
    for _ in range(repetitions):
        counter.increment()


def main() -> None:
    counter = AtomicCounter()
    thread_count = 4
    increments_per_thread = 10000

    # Create four threads that share the same counter object.
    threads = [
        threading.Thread(
            target=worker,
            args=(counter, increments_per_thread),
        )
        for _ in range(thread_count)
    ]

    # Start every worker thread.
    for thread in threads:
        thread.start()

    # Wait until every worker thread has completed.
    for thread in threads:
        thread.join()

    # Four threads times ten thousand increments gives forty thousand.
    print(counter.read())


if __name__ == "__main__":
    main()
Where it is used

This pattern is useful for request counts, completed task counts, retry statistics, generated sequence values within one process, and simple application metrics shared by worker threads. It works well when each protected operation is small and correctness matters more than avoiding lock contention. It is not sufficient for state shared across processes, containers, or servers.

Why Interviewers Ask This

Interviewers ask this question to check whether the candidate understands shared mutable state, thread synchronization, and the guarantees provided by Python locks. They also want to see whether the candidate knows that the Global Interpreter Lock is not a substitute for protecting a complete read, change, and write operation.

Common interview mistakes

A common mistake is using value += 1 without a lock and assuming the Global Interpreter Lock protects the complete update. Another mistake is locking increment while reading the value without the same lock. Creating a new lock inside each method call is also wrong because the calls would not share one synchronization object. Developers should not hold the lock during slow file access, network access, sleeping, or callbacks because this increases waiting and can create deadlock risks. A threading.Lock also does not coordinate separate processes or servers.

Interview tip

Start with the design: one value, one shared lock, and the same lock around every read and increment. Then explain that the lock provides mutual exclusion, while the Global Interpreter Lock does not replace explicit synchronization for the complete counter operation. Finish by stating the one process limitation and the contention cost.

Interviewer may ask next
What happens if a locked method tries to acquire the same lock again?

The thread can block itself when the counter uses threading.Lock because that lock is not reentrant. The second acquire cannot complete until the first acquire is released, but the same thread is waiting inside the protected call. This matters when one protected method calls another protected method. The preferred design is to avoid acquiring the same lock twice. If nested acquisition is required, threading.RLock permits the owning thread to acquire it again, with slightly more overhead and a risk of hiding unclear lock structure.

How should the counter change when multiple processes or servers update it?

The synchronization mechanism must change because threading.Lock only coordinates threads that share memory in one process. Multiple processes can use process shared state with a process safe lock. Multiple servers need an atomic database update or an external service that supports atomic increments. This matters because each process or server has independent memory. The tradeoff is greater communication cost and latency in exchange for correctness across runtime boundaries.

14. How would you process a large tree iteratively in Python to avoid recursion-depth failures?Language SpecificHardNetflix

Question Details

Explain an iterative depth-first traversal for computing subtree information, including stack representation, traversal order, and memory behavior.

Short Interview Answer (30-60 seconds)

I would replace recursion with an explicit Python list used as a stack. For subtree information, each stack item stores a node and a visited flag. On the first visit, I schedule the parent for later and then schedule its children. On the second visit, every child result is ready, so I compute the parent result. This avoids recursion depth failures, visits each tree node once for calculation, and uses linear total memory because the code stores results and validation state.

Detailed Explanation

See the Code while reading this explanation.

I would use a Python list as an explicit stack instead of making recursive calls. Each stack item contains a node and a visited flag. When a node is removed with visited set to false, I place it back with visited set to true, then place its children on the stack. This creates postorder traversal, which means children finish before their parent.

Useful Questions to Ask the Interviewer
  1. Should I focus on Python language behavior, or also explain the runtime and standard library?
  2. Which Python version and execution environment should I assume?
  3. Would you like a small code example together with production tradeoffs and edge cases?

When the node is removed again with visited set to true, every child result is available. The example computes subtree sizes, so the result is one plus the sizes of all direct child subtrees.

Python lists are suitable here because append and pop at the end are normally constant time. The traversal takes linear time for a valid tree because every node and parent to child connection is handled a constant number of times.

The explicit stack avoids Python recursion depth failures, but it does not remove memory costs. A deep or wide tree can make the stack large. The result dictionary and validation set also grow with the number of nodes. The code treats a missing dictionary entry as a leaf and rejects cycles or shared child nodes because the input must be a true tree.

How would you process a large tree iteratively in Python to avoid recursion-depth failures? diagram
Example

The tree is stored as a mapping from each node to its direct children. The stack stores a node and a visited flag. A false flag means the algorithm must schedule the children first. A true flag means all child results are complete and the node can be calculated. The validation set rejects cycles and shared child nodes, so the structure must be a true tree. A node with no mapping entry is treated as a leaf. For the sample tree, A has subtree size 5, B has subtree size 3, and C, D, and E each have subtree size 1.

Code
from collections.abc import Hashable, Mapping, Sequence
from typing import TypeVar

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


def compute_subtree_sizes(tree: Mapping[Node, Sequence[Node]], root: Node) -> dict[Node, int]:
    """Compute every subtree size with iterative postorder traversal."""

    # False means the node is entering the traversal.
    # True means all children are complete and the node can be calculated.
    stack: list[tuple[Node, bool]] = [(root, False)]

    # Store the completed subtree size for each node.
    subtree_sizes: dict[Node, int] = {}

    # A true tree contains each reachable node only once.
    # This set rejects cycles and shared child nodes.
    scheduled: set[Node] = {root}

    while stack:
        node, visited = stack.pop()

        if visited:
            # Every child result is available at this point.
            children = tree.get(node, ())
            subtree_sizes[node] = 1 + sum(subtree_sizes[child] for child in children)
            continue

        # Schedule the parent for calculation after its children.
        stack.append((node, True))

        # Reverse the sequence so children are processed in their original order.
        children = tree.get(node, ())
        for child in reversed(children):
            if child in scheduled:
                raise ValueError("Input must be a tree without cycles or shared child nodes")
            scheduled.add(child)
            stack.append((child, False))

    return subtree_sizes


if __name__ == "__main__":
    example_tree = {
        "A": ["B", "C"],
        "B": ["D", "E"],
        "C": [],
        "D": [],
        "E": [],
    }

    result = compute_subtree_sizes(example_tree, "A")
    print(result)

    assert result == {
        "D": 1,
        "E": 1,
        "B": 3,
        "C": 1,
        "A": 5,
    }
Where it is used

This pattern is useful for deeply nested file structures, syntax trees, category trees, organizational trees, dependency trees that are known to be true trees, and configuration trees. The same postorder pattern can compute subtree sizes, totals, heights, validation results, permissions, or aggregated metadata without depending on Python recursion depth.

Why Interviewers Ask This

Interviewers ask this question to check whether the candidate understands Python recursion limits, explicit stack traversal, postorder processing, and memory tradeoffs. It also tests whether the candidate can convert recursive control flow into reliable code for trees that may be too deep for Python recursion.

Common interview mistakes

A common mistake is calculating a parent during the first visit, before its child results exist. Another mistake is storing only nodes in the stack and losing the information that tells the algorithm whether a node is entering or leaving. Candidates may also use pop at the front of a Python list, which moves remaining elements and is slower. Other mistakes include forgetting that a wide tree can make the explicit stack large, claiming the iterative version uses constant memory, ignoring malformed input, and using this exact tree validation when shared nodes are valid in the real data model.

Interview tip

Start by saying that you replace Python call frames with a list based stack. Then explain the visited flag, why it creates postorder traversal, when the parent is calculated, and why total memory is still linear when results and validation state are stored.

Interviewer may ask next
What happens if the input contains a cycle or the same child under two parents?

The code raises ValueError when it sees a node that was already scheduled. This exact behavior rejects both cycles and shared child nodes because the function requires a true tree. It matters because a cycle could otherwise cause endless traversal, while a shared node would make subtree ownership ambiguous. The tradeoff is that the validation set uses memory proportional to the number of reachable nodes.

How does this iterative version compare with recursive postorder traversal?

The iterative version performs the same postorder calculation but stores traversal state in a Python list instead of Python call frames. This exact change avoids recursion depth failures and gives direct control over the stored state. Both approaches take linear time for a valid tree. The iterative code is more verbose, and it still needs memory for the stack, results, and validation set, but it is safer when tree depth is large or unpredictable.

15. Compute minimum task completion time.CodingMediumNetflix

Question Details

Given tasks with durations and directed dependencies, compute the minimum time required to complete all tasks when independent tasks can run in parallel.

Short Interview Answer (30-60 seconds)

I would model the tasks as a directed acyclic graph. I store outgoing edges in an adjacency list and count each task’s prerequisites with an indegree map. Then I use Kahn’s topological sort. For every task, I track its earliest start and finish time. A task starts after its latest prerequisite finishes. Independent source tasks can start together at time zero. The answer is the largest finish time. This takes O(V + E) time and O(V + E) auxiliary space.

Detailed Explanation

See the Code while reading this explanation.

The problem asks for the shortest total time needed to finish every task. Each task has a duration. A directed edge means one task must finish before another task can start. Because independent tasks may run at the same time, we do not add every duration. We find the longest required dependency chain by processing the graph in topological order.

Useful Questions to Ask the Interviewer
  1. What input sizes, value ranges, and edge cases should the solution handle?
  2. What output should be returned for empty, invalid, or duplicate input?
  3. Should I prioritize execution time or memory use, and may I use the standard library?
Compute minimum task completion time. diagram
How to Explain It in an Interview
1. Understand the input and output

The input has two parts.

The first part is a dictionary of task durations. In the example, the durations are A: 3, B: 2, C: 4, D: 3, and E: 1.

The second part is a list of directed dependencies. A pair such as A to C means A must finish before C can start.

The dependencies are A to C, B to C, B to D, C to E, and D to E.

The output is one integer. It is the minimum time needed to finish all tasks when independent tasks can run in parallel. For this example, the answer is 8.

2. Choose topological sorting and earliest finish times

I represent the dependencies with an adjacency list. For each task, the adjacency list stores the tasks that directly depend on it.

I also store an indegree count. The indegree of a task is the number of unfinished prerequisites it has.

I use Kahn’s algorithm for topological sorting. It starts with every task whose indegree is zero. These tasks have no prerequisites, so they can start at time zero.

For each task, I track two values. earliest_start is the first time the task may begin. earliest_finish is earliest_start plus the task duration.

The central invariant is this: when a task leaves the queue, all of its prerequisites have already been processed. Therefore, its earliest start and finish times are final.

3. Initialize the graph and queue

The adjacency list is:

A to [C] B to [C, D] C to [E] D to [E] E to []

The indegree values are:

A: 0 B: 0 C: 2 D: 1 E: 2

A and B have indegree zero. They enter the queue first.

Both start at time zero. A finishes at time 3. B finishes at time 2.

The initial queue is [A, B]. The initial earliest_start value is zero for every task.

4. Walk through the example

First, remove A from the queue.

A finishes at time 3. C depends on A, so update C’s earliest start from 0 to max(0, 3), which is 3. C’s indegree changes from 2 to 1. C still has one unfinished prerequisite, so it does not enter the queue.

The queue is now [B].

Next, remove B.

B finishes at time 2. C also depends on B. Update C’s earliest start to max(3, 2), which remains 3. C’s indegree changes from 1 to 0. All prerequisites of C are now complete. C finishes at 3 + 4, which is 7, and enters the queue.

D also depends on B. Update D’s earliest start to max(0, 2), which is 2. D’s indegree changes from 1 to 0. D finishes at 2 + 3, which is 5, and enters the queue.

The queue is now [C, D].

Next, remove C.

C finishes at time 7. E depends on C. Update E’s earliest start to max(0, 7), which is 7. E’s indegree changes from 2 to 1. E still waits for D.

The queue is now [D].

Next, remove D.

D finishes at time 5. E also depends on D. Update E’s earliest start to max(7, 5), which remains 7. E’s indegree changes from 1 to 0. E finishes at 7 + 1, which is 8, and enters the queue.

The queue is now [E].

Finally, remove E. It has no outgoing edges. The queue becomes empty.

The finish times are A: 3, B: 2, C: 7, D: 5, and E: 8. The largest finish time is 8.

5. Explain why the result is correct

Topological order processes every prerequisite before its dependent tasks.

For each edge from u to v, the algorithm gives v the value max(earliest_start[v], earliest_finish[u]). This means v waits for its latest-finishing prerequisite.

A dependent task enters the queue only when its indegree becomes zero. At that moment, every prerequisite has contributed its finish time. Therefore, the task’s earliest start and finish values are correct.

The maximum finish time is the minimum total project time because independent tasks run together whenever their dependencies allow it. The critical path is A to C to E. Its duration is 3 + 4 + 1, which is 8. The path B to D to E takes 2 + 3 + 1, which is 6, so it does not determine the total time.

6. Explain the Python implementation

The code first creates an empty adjacency list and an indegree value for every task.

It then reads each dependency. It adds the dependent task to the prerequisite’s adjacency list and increases the dependent task’s indegree.

Next, it creates the earliest_start map, the earliest_finish map, and a deque. Every zero-indegree task enters the deque and gets a finish time equal to its duration.

The main loop removes one task at a time. It updates the answer with that task’s finish time. It then updates every dependent task. When a dependent task’s indegree becomes zero, the code calculates its finish time and adds it to the queue.

The processed counter detects a cycle. If fewer tasks are processed than exist in the input, the dependency graph is not a DAG, so no valid schedule exists.

7. Explain complexity and edge cases

Let V be the number of tasks and E be the number of dependencies.

Each task enters and leaves the queue once. Each dependency edge is processed once. The time complexity is O(V + E).

The graph, indegree map, timing maps, and queue may all grow with the input. The auxiliary space complexity is O(V + E).

Important cases include multiple source tasks, disconnected groups of tasks, one task with many prerequisites, a single task, and a cycle. Multiple source tasks can start together. Disconnected groups are processed independently. A cycle is rejected because no topological schedule exists.

Key Insight / Why This Solution Works

The key idea is to combine Kahn’s topological sort with dynamic programming on a directed acyclic graph. The adjacency list stores each task’s direct dependents. The indegree map stores how many prerequisites each task still has. For every task v, earliest_start[v] is the largest finish time seen among its prerequisites. The invariant is that when v enters the queue, all of its prerequisites have been processed, so earliest_start[v] is final. We then calculate earliest_finish[v] as earliest_start[v] plus duration[v]. The largest finish time is the total completion time because independent tasks already overlap whenever possible.

Code
from collections import deque
from typing import Dict, List, Tuple


def minimum_completion_time(
    durations: Dict[str, int],
    dependencies: List[Tuple[str, str]],
) -> int:
    """Return the minimum time needed to complete all tasks in a DAG."""

    # Step 1: Create an adjacency list for outgoing dependency edges.
    # graph[task] contains the tasks that directly depend on task.
    graph: Dict[str, List[str]] = {task: [] for task in durations}

    # indegree[task] stores the number of prerequisites for that task.
    indegree: Dict[str, int] = {task: 0 for task in durations}

    # Step 2: Build the graph and count each task's prerequisites.
    for prerequisite, task in dependencies:
        graph[prerequisite].append(task)
        indegree[task] += 1

    # earliest_start[task] is the earliest time that task may begin.
    earliest_start: Dict[str, int] = {task: 0 for task in durations}

    # earliest_finish[task] is the earliest time that task may finish.
    earliest_finish: Dict[str, int] = {}

    # Kahn's algorithm uses a FIFO queue for zero-indegree tasks.
    queue = deque()

    # Step 3: Source tasks have no prerequisites.
    # They can start at time 0 and finish after their own duration.
    for task, degree in indegree.items():
        if degree == 0:
            queue.append(task)
            earliest_finish[task] = durations[task]

    processed = 0
    answer = 0

    # Step 4: Process tasks in topological order.
    while queue:
        task = queue.popleft()
        processed += 1

        # Track the latest finish time seen so far.
        answer = max(answer, earliest_finish[task])

        # Step 5: Update every task that depends on this task.
        for next_task in graph[task]:
            # A task must wait for its latest-finishing prerequisite.
            earliest_start[next_task] = max(
                earliest_start[next_task],
                earliest_finish[task],
            )

            # One prerequisite has now been completed.
            indegree[next_task] -= 1

            # When indegree becomes zero, all prerequisites are complete.
            if indegree[next_task] == 0:
                earliest_finish[next_task] = earliest_start[next_task] + durations[next_task]
                queue.append(next_task)

    # Defensive fallback: a cycle prevents a valid topological schedule.
    if processed != len(durations):
        raise ValueError("Dependencies contain a cycle")

    # Step 6: Return the latest earliest-finish time.
    return answer


if __name__ == "__main__":
    # Example from the diagram.
    example_durations = {
        "A": 3,
        "B": 2,
        "C": 4,
        "D": 3,
        "E": 1,
    }

    example_dependencies = [
        ("A", "C"),
        ("B", "C"),
        ("B", "D"),
        ("C", "E"),
        ("D", "E"),
    ]

    result = minimum_completion_time(
        example_durations,
        example_dependencies,
    )

    print(result)  # 8
Time & Space Complexity

Let V be the number of tasks and E be the number of directed dependencies. Building the graph takes O(V + E) time. During topological sorting, each task is removed from the queue once, and each dependency edge is examined once. Therefore, the total time is O(V + E). The adjacency list uses O(V + E) memory. The indegree map, timing maps, and queue use O(V) more memory. The total auxiliary space is O(V + E).

Where it is used

This pattern is useful for project scheduling, build systems, job pipelines, course prerequisites, data-processing workflows, and deployment steps. It works when tasks have directed prerequisites and independent tasks may run at the same time. It also helps calculate the earliest possible completion time of a dependency graph.

Why Interviewers Ask This

This question tests whether you can recognize a directed acyclic graph scheduling problem. The interviewer wants to see if you can build an adjacency list, maintain indegree counts, and use topological order correctly. It also tests whether you understand parallel execution. You must take the maximum prerequisite finish time instead of summing unrelated work. The question also checks your ability to explain an invariant, detect cycles, write clean Python, and give accurate O(V + E) time and space complexity.

Common interview mistakes

A common mistake is adding every task duration, which ignores parallel work. Another mistake is using the first prerequisite finish time instead of the maximum finish time across all prerequisites. Some candidates enqueue a task before its indegree reaches zero, even though another prerequisite is still unfinished. Others reverse the dependency edge and build the adjacency list in the wrong direction. It is also easy to forget disconnected source tasks or omit cycle detection. Finally, claiming O(V) time is incorrect because every dependency edge must also be processed.

Interview tip

State the invariant before writing code: when a task enters the queue, all of its prerequisites are complete, so its earliest start time is final. This makes the update rule and correctness argument much easier to explain.

Interviewer may ask next
How would you return the actual critical path instead of only its total completion time?

Store a predecessor for each task. When earliest_finish[u] is greater than the current earliest_start[v], update earliest_start[v] and set predecessor[v] to u. If two prerequisites have the same finish time, either one can represent a valid critical path. After processing the graph, find a task with the largest earliest_finish value. Follow predecessor links backward, then reverse the collected tasks. For this example, the path is A, C, E. The time complexity remains O(V + E), and the predecessor map uses O(V) additional space.

What changes if the dependency graph contains a cycle?

A cycle means at least one group of tasks waits on itself, so no valid completion schedule exists. Kahn’s algorithm detects this when the queue becomes empty before all tasks are processed. The function compares processed with the number of tasks and raises an error when they differ. The time complexity remains O(V + E), and the auxiliary space remains O(V + E).

16. Compute subtree sums with tree DFS.CodingHardNetflix

Question Details

Given a rooted tree and an integer value for each node, compute the sum of values in every node's subtree using depth-first search.

Short Interview Answer (30-60 seconds)

I would use post-order depth-first search. For each node, I start with its own value. I recursively compute each child’s subtree sum and add the returned value to the current total. After every child is finished, I store the total for the node and return it to the parent. This works because each parent uses complete child results. The time complexity is O(n). The recursion stack uses O(h) auxiliary space, where h is the tree height.

Detailed Explanation

See the Code while reading this explanation.

The input contains a root node, a children adjacency list, and one integer value for every node. We must return a dictionary that maps each node to the sum of its own value and every descendant value. Post-order DFS fits this problem because a parent’s result depends on the completed results of its children.

Useful Questions to Ask the Interviewer
  1. What input sizes, value ranges, and edge cases should the solution handle?
  2. What output should be returned for empty, invalid, or duplicate input?
  3. Should I prioritize execution time or memory use, and may I use the standard library?
Compute subtree sums with tree DFS. diagram
How to Explain It in an Interview
1. Understand the input and required output

The root tells us where traversal begins.

The children dictionary stores each node and its direct children. The edges are directed from parent to child.

The values dictionary stores the integer value of each node.

The output is a dictionary that maps each node to its complete subtree sum.

In the example, the edges are 1 to 2, 1 to 3, 2 to 4, 2 to 5, and 3 to 6. The node values are 1 to 5, 2 to 2, 3 to -3, 4 to 4, 5 to 1, and 6 to 6.

2. Choose post-order DFS

I use recursive depth-first search.

Each call solves one node’s complete subtree. The call returns one integer, which is that subtree’s sum.

The central invariant is: when dfs(node) returns, it has computed and stored the correct sum for node and every descendant in that node’s subtree.

The children must finish before their parent. This processing order is called post-order traversal.

3. Initialize the state

I create an empty dictionary named sums. It stores node to completed subtree sum.

Inside dfs(node), I set total to values[node]. This first includes the current node’s own value.

Then I visit every child. Each child returns its complete subtree sum, which I add to total.

A leaf has no children. Its loop does not run, so it stores and returns its own value.

4. Walk through the example

Start with dfs(1). Node 1 begins with total 5 and visits node 2.

Node 2 begins with total 2 and visits node 4.

Node 4 is a leaf. It stores sums[4] = 4 and returns 4.

Back at node 2, total changes from 2 to 2 + 4 = 6. Node 2 then visits node 5.

Node 5 is a leaf. It stores sums[5] = 1 and returns 1.

Node 2 now computes 6 + 1 = 7. It stores sums[2] = 7 and returns 7.

Back at node 1, total changes from 5 to 5 + 7 = 12. Node 1 then visits node 3.

Node 3 begins with total -3 and visits node 6.

Node 6 is a leaf. It stores sums[6] = 6 and returns 6.

Node 3 computes -3 + 6 = 3. It stores sums[3] = 3 and returns 3.

Finally, node 1 computes 12 + 3 = 15. It stores sums[1] = 15 and returns 15.

The post-order return order is 4, 5, 2, 6, 3, 1.

The final mapping is {1: 15, 2: 7, 3: 3, 4: 4, 5: 1, 6: 6}.

5. Explain why the result is correct

Every child call returns the exact sum of that child’s subtree.

The parent starts with its own value and adds the returned sum from every child.

Therefore, after all children finish, total contains the parent’s value and every descendant value exactly once.

Storing sums[node] after the children finish gives the correct answer for that node.

6. Explain the Python implementation

The outer function creates the sums dictionary.

The inner dfs function computes one subtree. It starts total with the current node’s value.

It reads the current node’s children with children.get(node, []). This also works when a leaf is missing from the dictionary.

For each child, it calls dfs(child) and adds the returned subtree sum to total.

After the loop, the subtree is complete. The function stores sums[node] = total and returns total to the parent.

The outer function calls dfs(root) and returns the completed sums dictionary.

7. Explain complexity and edge cases

Let n be the number of nodes and h be the tree height.

The time complexity is O(n). Each node is visited once, and each parent-to-child edge is processed once.

The recursion stack uses O(h) auxiliary space. A balanced tree has a smaller height. A skewed tree can have h equal to n.

The output dictionary uses O(n) space because it stores one result for every node.

Important edge cases are a leaf node, a single-node tree, negative or zero values, and a skewed tree with deep recursion.

Key Insight / Why This Solution Works

The key idea is to compute child subtree sums before computing the parent’s sum. Each recursive DFS call returns one integer: the complete subtree sum for that node. The invariant is that when dfs(node) returns, sums[node] has already been stored and equals values[node] plus the completed sums returned by all children. Post-order DFS is suitable because the parent cannot be finished until all child results are available.

Code
from typing import Dict, List


def compute_subtree_sums(
    root: int,
    children: Dict[int, List[int]],
    values: Dict[int, int],
) -> Dict[int, int]:
    # Store the completed subtree sum for every node.
    sums: Dict[int, int] = {}

    def dfs(node: int) -> int:
        # Begin with the current node's own value.
        total = values[node]

        # Process every child before finishing the current node.
        for child in children.get(node, []):
            # The recursive call returns the child's complete subtree sum.
            total += dfs(child)

        # All child subtrees are complete, so this total is final.
        sums[node] = total

        # Return the completed subtree sum to the parent.
        return total

    # Start the post-order DFS from the root.
    dfs(root)

    # Return one subtree sum for every node.
    return sums


if __name__ == "__main__":
    root = 1

    children = {
        1: [2, 3],
        2: [4, 5],
        3: [6],
        4: [],
        5: [],
        6: [],
    }

    values = {
        1: 5,
        2: 2,
        3: -3,
        4: 4,
        5: 1,
        6: 6,
    }

    result = compute_subtree_sums(root, children, values)

    # Display the mapping in node order, matching the diagram.
    ordered_result = {node: result[node] for node in sorted(result)}
    print(ordered_result)
    # Expected: {1: 15, 2: 7, 3: 3, 4: 4, 5: 1, 6: 6}
Time & Space Complexity

Let n be the number of nodes and h be the height of the tree. The time complexity is O(n) because every node is visited once and every parent-to-child edge is processed once. The recursion stack uses O(h) auxiliary space because the active calls form one root-to-leaf path. In a skewed tree, h can equal n. The output dictionary uses O(n) space because it stores one subtree sum for each node.

Where it is used

This pattern is useful when a parent’s result depends on completed results from its children. Examples include calculating folder sizes, organization totals, category totals, expression-tree values, and other tree dynamic programming problems.

Why Interviewers Ask This

This question tests whether you recognize post-order tree processing. The interviewer wants to see whether you can define what each recursive call returns, combine child results correctly, and maintain a clear invariant. It also checks whether you can represent a rooted tree with an adjacency list, handle leaves and negative values, write clean recursive Python, and explain O(n) time and O(h) recursion-stack space accurately.

Common interview mistakes

A common mistake is storing sums[node] before visiting the children. That stores an incomplete total. Another mistake is forgetting to return total to the parent. Candidates may forget that a leaf still stores and returns its own value. They may incorrectly use one shared total across all calls instead of one local total per call. They may also claim O(1) auxiliary space and forget the O(h) recursion stack.

Interview tip

State the invariant before writing code: when dfs(node) returns, it has stored and returned the correct sum for that complete subtree. This explains both the recursion order and why sums[node] is stored after the child loop.

Interviewer may ask next
How would you handle a tree so deep that recursive DFS may exceed Python’s recursion limit?

I would simulate post-order traversal with an explicit stack. I could store pairs such as (node, processed). On the first visit, I push the node again as processed and then push its children. On the processed visit, every child sum is already available, so I calculate and store the node’s sum. The time remains O(n). The explicit stack can use O(n) space in the worst case, and the output dictionary uses O(n). The tradeoff is more code, but it avoids recursion-limit errors.

How would the solution change if the tree edges were undirected?

I would pass the parent into each DFS call or maintain a visited set. While visiting a node’s neighbors, I would skip the parent so the traversal does not move back along the same edge. The subtree calculation remains the same because the chosen root defines the parent-child direction. The time complexity stays O(n). The recursion stack uses O(h), and the output dictionary uses O(n).

17. Find pairs of disjoint strings.CodingMediumNetflix

Question Details

Given a collection of strings, find pairs that do not share characters and explain the chosen representation and complexity.

Short Interview Answer (30-60 seconds)

I would convert each string into a set of its unique characters. Then I would compare every unordered pair once by using indices i and j, with j greater than i. For each pair, I call isdisjoint on the two cached sets. If it returns true, I add the original strings to the result. This works because isdisjoint is true exactly when the strings share no characters. The expected time is O(T + sum of min(u_i, u_j)), with O(sum of u_i) auxiliary space.

Detailed Explanation

See the Code while reading this explanation.

The problem asks us to return every pair of strings that does not share any character. A useful representation is a set of characters for each string. A set keeps each character only once. Python's isdisjoint method then gives us a direct way to test whether two strings share a character. We build the sets once and reuse them while checking every unordered pair.

Useful Questions to Ask the Interviewer
  1. What input sizes, value ranges, and edge cases should the solution handle?
  2. What output should be returned for empty, invalid, or duplicate input?
  3. Should I prioritize execution time or memory use, and may I use the standard library?
Find pairs of disjoint strings. diagram
How to Explain It in an Interview
1. Understand the input and required output

The input is a list of strings. The output is a list of pairs containing the original string values.

For the example, the input is ["ab", "cd", "ac", "ef"].

The returned result is [("ab", "cd"), ("ab", "ef"), ("cd", "ef"), ("ac", "ef")].

A pair is valid only when the two strings have no common character. We return string pairs, not index pairs and not character sets.

2. Choose sets as the representation

For each string, I create a set containing its unique characters.

The cached sets are:

Index 0: "ab" becomes {'a', 'b'}.

Index 1: "cd" becomes {'c', 'd'}.

Index 2: "ac" becomes {'a', 'c'}.

Index 3: "ef" becomes {'e', 'f'}.

Then I use set.isdisjoint. It returns true when the two sets have no shared element.

The central invariant is: after each processed pair, result contains exactly the disjoint pairs among all pairs checked so far.

3. Initialize the state and traversal order

I build char_sets with one cached set for each input string.

I also create an empty result list.

The outer loop selects index i. The inner loop starts at i + 1, so j is always greater than i.

This skips self-pairs such as comparing "ab" with itself. It also prevents reverse duplicates such as returning both ("ab", "cd") and ("cd", "ab").

4. Walk through the example

Step 1 checks indices (0, 1), which contain "ab" and "cd".

The result is initially empty. {'a', 'b'}.isdisjoint({'c', 'd'}) is true. We append ("ab", "cd"). The result becomes [("ab", "cd")].

Step 2 checks indices (0, 2), which contain "ab" and "ac".

The state before the check is [("ab", "cd")]. Both sets contain 'a', so isdisjoint returns false. We skip this pair. The result stays [("ab", "cd")].

Step 3 checks indices (0, 3), which contain "ab" and "ef".

The sets have no common character, so isdisjoint returns true. We append ("ab", "ef"). The result becomes [("ab", "cd"), ("ab", "ef")].

Step 4 checks indices (1, 2), which contain "cd" and "ac".

Both sets contain 'c', so isdisjoint returns false. We skip the pair. The result remains [("ab", "cd"), ("ab", "ef")].

Step 5 checks indices (1, 3), which contain "cd" and "ef".

The sets are disjoint, so we append ("cd", "ef"). The result becomes [("ab", "cd"), ("ab", "ef"), ("cd", "ef")].

Step 6 checks indices (2, 3), which contain "ac" and "ef".

The sets are disjoint, so we append ("ac", "ef"). The final result is [("ab", "cd"), ("ab", "ef"), ("cd", "ef"), ("ac", "ef")].

All six unordered pairs have now been checked, so the function returns the result.

5. Explain why the result is correct

Every unordered pair is checked exactly once because the inner loop uses j greater than i.

A pair is added if and only if isdisjoint returns true.

Therefore, every returned pair shares no character. Every omitted pair shares at least one character. This means the result contains exactly all valid disjoint pairs.

6. Explain the Python implementation

The list comprehension builds one character set for every input string.

The nested loops generate every unordered pair in this order: (0, 1), (0, 2), (0, 3), (1, 2), (1, 3), and (2, 3).

The condition char_sets[i].isdisjoint(char_sets[j]) checks the cached sets.

When the condition is true, the code appends the original strings. It does not append the sets or the indices.

After every pair has been processed, the function returns the complete result list.

7. Explain complexity and edge cases

Let T be the total number of characters across all strings. Building all sets costs O(T) time.

Let u_i be the number of unique characters in string i. For one pair, isdisjoint takes expected O(min(u_i, u_j)) time because Python sets use hashing and membership checks are O(1) on average.

The full expected time is O(T + sum over all i < j of min(u_i, u_j)). If each string has at most k unique characters, this can be written as O(T + n^2 * k).

The cached sets use O(sum of u_i) auxiliary space. The returned output uses O(p) space for p valid pairs.

An empty collection or a collection with one string returns no pairs. An empty string is disjoint with every other string. Repeated characters inside one string do not change its set. Duplicate strings at different indices are still treated as separate input elements.

Key Insight / Why This Solution Works

The key idea is to separate character extraction from pair comparison. We convert every string once into a cached set of unique characters. Then we compare each unordered pair with set.isdisjoint. The invariant is that after each processed pair, result contains exactly the disjoint pairs among all pairs checked so far. Using j > i makes each unordered pair appear once and avoids self-pairs. This approach is easier to read and avoids rebuilding character collections during every comparison.

Code
from typing import List, Set, Tuple


def find_disjoint_pairs(strings: List[str]) -> List[Tuple[str, str]]:
    # Step 1: Cache the unique characters for every string.
    char_sets: List[Set[str]] = [set(s) for s in strings]

    # Step 2: Store every valid disjoint string pair here.
    result: List[Tuple[str, str]] = []

    # Step 3: Choose the first index of each unordered pair.
    for i in range(len(strings)):
        # Start at i + 1 to skip self-pairs and reverse duplicates.
        for j in range(i + 1, len(strings)):
            # Step 4: True means the two strings share no characters.
            if char_sets[i].isdisjoint(char_sets[j]):
                # Step 5: Append the original string values.
                result.append((strings[i], strings[j]))

    # Step 6: Return all valid pairs after every pair is checked.
    return result


if __name__ == "__main__":
    example = ["ab", "cd", "ac", "ef"]
    answer = find_disjoint_pairs(example)
    print(answer)
    # Expected output:
    # [('ab', 'cd'), ('ab', 'ef'), ('cd', 'ef'), ('ac', 'ef')]
Time & Space Complexity

Let T be the total number of characters in all strings. Building the character sets takes O(T) time. Let u_i be the number of unique characters in string i. Python sets use hashing, so membership checks are O(1) on average. The expected cost of isdisjoint for one pair is O(min(u_i, u_j)). Therefore, the total expected time is O(T + sum over i < j of min(u_i, u_j)). If every string has at most k unique characters, this is O(T + n^2 * k). The cached sets use O(sum of u_i) auxiliary space. The output uses O(p) space for p returned pairs.

Where it is used

This pattern is useful when software must compare many collections for overlap. Examples include checking whether tag sets conflict, finding records with no shared labels, comparing permission scopes, or grouping items that have separate feature sets. Caching sets is especially helpful when the same strings or collections are compared more than once.

Why Interviewers Ask This

The interviewer is checking whether you can choose a suitable representation for repeated comparisons. They want to see that you understand sets, unordered-pair traversal, and the meaning of isdisjoint. They also test whether you avoid duplicate work, return the required string values, explain a correctness invariant, and give accurate expected-time complexity for Python hash-based structures. Edge cases such as empty strings, repeated characters, and duplicate input elements show careful reasoning.

Common interview mistakes

A common mistake is checking both (i, j) and (j, i), which returns duplicate pair orders. Another mistake is starting j at i, which compares a string with itself. Some candidates rebuild sets inside the nested loops and repeat unnecessary work. Another error is appending sets or indices when the required output contains the original string values. Candidates may also claim that Python set operations are guaranteed O(1), even though that is average behavior. Repeated characters inside one string should not be counted more than once because the representation is a set.

Interview tip

State the invariant before coding: after each checked pair, the result contains exactly the valid disjoint pairs seen so far. Then explain why j starts at i + 1. This makes both the traversal and the correctness argument easy to defend.

Interviewer may ask next
How would the solution change if the alphabet were small and fixed?

I could represent each string with a bitmask instead of a Python set. Each bit would represent one possible character. Two strings would be disjoint when mask_a & mask_b equals zero. Building the masks would still take O(T) time. Each pair check would become O(1), so checking all pairs would take O(n^2) time. The masks would use O(n) auxiliary space. The tradeoff is that this method requires a known, bounded alphabet.

How would you handle a very large result without storing every pair in memory?

I would keep the same cached sets and pair traversal, but I would yield each valid pair from a generator instead of appending it to a list. Correctness stays the same because every unordered pair is still checked once and a pair is yielded only when isdisjoint returns true. The expected processing time remains O(T + sum over i < j of min(u_i, u_j)). Cached-set space remains O(sum of u_i), while extra stored output space becomes O(1) beyond the current yielded pair. The tradeoff is that the caller receives a stream instead of a reusable list.

18. Return the longest contiguous subarray with all distinct values.CodingMediumNetflix

Question Details

Given an integer array, return a longest contiguous subarray in which no value repeats.

Short Interview Answer (30-60 seconds)

I use a sliding window and a hash map. The window runs from left to right, and the map stores each value’s latest index. When the current value already appears inside the window, I move left to one position after its previous index. After each step, I compare the valid window with the best one found so far. This returns a longest distinct contiguous subarray in O(n) expected time with O(n) auxiliary space.

Detailed Explanation

See the Code while reading this explanation.

The input is an integer array. We must return one longest contiguous part of the array that contains no repeated value. A sliding window fits this problem because it keeps one valid range while moving from left to right. A hash map lets us move the start of the window directly past an earlier duplicate.

Useful Questions to Ask the Interviewer
  1. What input sizes, value ranges, and edge cases should the solution handle?
  2. What output should be returned for empty, invalid, or duplicate input?
  3. Should I prioritize execution time or memory use, and may I use the standard library?
Return the longest contiguous subarray with all distinct values. diagram
How to Explain It in an Interview
1. Understand the input and required output

The input is an integer array called nums. The output is the values from one longest contiguous subarray with no duplicates.

Contiguous means the values must come from neighboring positions in the original array. We cannot skip elements.

For nums = [5, 1, 3, 5, 2, 3, 4, 1], the returned subarray is [5, 2, 3, 4, 1]. It comes from indices 3 through 7.

2. Choose the algorithm and data structure

Use a sliding window. The current window is nums[left:right + 1]. The left variable marks its first index. The right variable marks its last index.

Use a hash map called last_seen. It stores each value and the latest processed index where that value appeared. For example, last_seen[5] = 3 means the latest processed 5 was at index 3.

The central invariant is that, after any required movement of left, nums[left:right + 1] contains no repeated values.

3. Initialize the state

Set left to 0. Start with an empty last_seen map.

Set best_start to 0 and best_len to 0. These variables describe the longest valid window found so far.

Then process the array from left to right using right and value.

4. Walk through the example

At right = 0, value = 5. It has not appeared inside the current window. Store last_seen[5] = 0. The window is [5], so the best becomes [5].

At right = 1, value = 1. It is new in the current window. Store last_seen[1] = 1. The window becomes [5, 1], so the best becomes [5, 1].

At right = 2, value = 3. It is new in the current window. Store last_seen[3] = 2. The window becomes [5, 1, 3], so the best becomes [5, 1, 3].

At right = 3, value = 5. Its previous index is 0, and index 0 is still inside the current window because 0 >= left. Move left from 0 to 1. Then store last_seen[5] = 3. The valid window is now [1, 3, 5]. The best remains [5, 1, 3].

At right = 4, value = 2. It is new in the current window. Store last_seen[2] = 4. The window becomes [1, 3, 5, 2]. Its length is 4, so it becomes the new best.

At right = 5, value = 3. Its previous index is 2, and index 2 is inside the current window because 2 >= left. Move left from 1 to 3. Then store last_seen[3] = 5. The valid window becomes [5, 2, 3]. The best remains [1, 3, 5, 2].

At right = 6, value = 4. It is new in the current window. Store last_seen[4] = 6. The window becomes [5, 2, 3, 4]. Its length ties the best length. The code updates only for a strictly longer window, so it keeps [1, 3, 5, 2].

At right = 7, value = 1. Its previous index is 1, but left is already 3. The old 1 is outside the current window because 1 < left, so left stays 3. Store last_seen[1] = 7. The window becomes [5, 2, 3, 4, 1]. Its length is 5, so it becomes the final best window.

The function returns nums[3:8], which is [5, 2, 3, 4, 1].

5. Explain why the result is correct

After duplicate handling, the current window contains only distinct values. If a duplicate lies inside the window, moving left to one position after its earlier index removes the earlier copy and keeps the range contiguous.

For each right index, the algorithm keeps the longest distinct window that can end at that index. It compares that valid window with the best one found earlier. Therefore, after the final index is processed, best_start and best_len describe a longest distinct contiguous subarray.

6. Explain the Python implementation

The for loop gives the current index and value. The duplicate condition checks two facts: the value appeared before, and its previous index is greater than or equal to left. Only then is the earlier copy still inside the current window.

After moving left when needed, the code records the current index in last_seen. It calculates the current length as right - left + 1. It updates best_start and best_len only when the current window is strictly longer.

Finally, it returns nums[best_start:best_start + best_len].

7. Explain complexity and edge cases

The expected running time is O(n). The right index processes every array position once. The left index only moves forward. Python dictionary lookup and insertion are O(1) on average.

The auxiliary space is O(n) because last_seen may store one entry for every distinct value. The returned slice uses O(k) output space, where k is the returned subarray length.

For an empty input, the function returns an empty list. For one element, it returns that one-element subarray. If all values are distinct, it returns the whole array. If all values are equal, any one-element subarray is longest, and this implementation keeps the earliest one. Negative values and zero work without special handling.

Key Insight / Why This Solution Works

The key idea is to keep one valid sliding window instead of checking every possible subarray. The window is nums[left:right + 1]. The hash map last_seen stores each processed value and its latest index. The invariant is that, after any required left adjustment, the current window contains no repeated values. When a repeated value has a previous index inside the window, left jumps to last_seen[value] + 1. This removes the earlier copy without moving left one position at a time. For every right index, the resulting window is the longest distinct window ending there. The algorithm records it when it is strictly longer than the current best.

Code
from typing import List


def longest_distinct_subarray(nums: List[int]) -> List[int]:
    # left is the first index of the current distinct window.
    left = 0

    # Map each value to its latest processed index.
    last_seen: dict[int, int] = {}

    # Store the start and length of the best window found so far.
    best_start = 0
    best_len = 0

    # Expand the window by processing each value from left to right.
    for right, value in enumerate(nums):
        # Move left only when the earlier copy is inside the window.
        if value in last_seen and last_seen[value] >= left:
            left = last_seen[value] + 1

        # Record the current value's latest index.
        last_seen[value] = right

        # Measure the current valid window.
        current_len = right - left + 1

        # Save the window only when it is strictly longer.
        if current_len > best_len:
            best_start = left
            best_len = current_len

    # Return the longest distinct contiguous subarray.
    return nums[best_start : best_start + best_len]


if __name__ == "__main__":
    example = [5, 1, 3, 5, 2, 3, 4, 1]
    result = longest_distinct_subarray(example)

    print("Input:", example)
    print("Longest distinct contiguous subarray:", result)
    # Expected output: [5, 2, 3, 4, 1]
Time & Space Complexity

The expected time is O(n), where n is the number of values in nums. The right index visits each array position once. The left index only moves forward. Python dictionary lookup and insertion take O(1) time on average, so the full algorithm takes O(n) expected time. The auxiliary space is O(n) because last_seen may contain one entry for each distinct value. The returned list also uses O(k) output space, where k is the length of the returned subarray.

Where it is used

This sliding-window pattern is useful when software must find the longest continuous range that follows a rule. Similar examples include finding a longest substring without repeated characters, tracking a recent event range with unique identifiers, or maintaining a valid range while new sequence values arrive.

Why Interviewers Ask This

This question tests whether a candidate recognizes the sliding-window pattern and chooses a useful hash-map representation. It also checks duplicate handling, forward-only pointer movement, and the ability to maintain a clear invariant. The interviewer can see whether the candidate distinguishes a contiguous subarray from a subsequence, updates the answer at the correct time, writes valid Python, handles tied and empty cases, and explains expected hash-map time and auxiliary space accurately.

Common interview mistakes

A common mistake is moving left whenever a value appeared before. Left should move only when the previous index is still inside the current window. Another mistake is moving left backward when an older duplicate is already outside the window. Candidates may measure or save the window before removing an active duplicate, which can record an invalid range. Some people confuse a contiguous subarray with a subsequence and skip values. Another mistake is updating the best window on equal length even though this implementation keeps the earliest tied result. It is also incorrect to describe Python dictionary operations as guaranteed O(1); they are O(1) on average.

Interview tip

State the invariant before writing code: after duplicate handling, nums[left:right + 1] contains only distinct values. Then explain the condition last_seen[value] >= left. This shows exactly when left must move and prevents the main duplicate-handling bug.

Interviewer may ask next
How would you return the start and end indices instead of the subarray values?

Use the same sliding-window algorithm and keep best_start and best_len. After the loop, return [best_start, best_start + best_len - 1]. For the example, the result is [3, 7]. These indices describe the same saved window, so the correctness argument does not change. The expected time remains O(n), and the auxiliary space remains O(n). Returning two indices also avoids copying the final slice.

How would the solution change if the values arrived one at a time as a stream?

Keep last_seen, left, the current index, best_start, and best_len between arrivals. For each new value, perform the same duplicate check, move left when needed, update the latest index, and compare the current window length with the best length. This uses O(1) average work per arriving value and up to O(n) map space. To return the actual best subarray values later, the system must also retain the stream values or store a copy of the best window. That extra storage is the main tradeoff.

19. Plan eviction and cleanup for a production TTL cache.CodingMediumNetflix

Question Details

Explain and implement an eviction or compaction strategy so expired entries do not accumulate indefinitely, while preserving correct get and put behavior.

Short Interview Answer (30-60 seconds)

I would store the latest live entry for each key in a hash map and keep expiration records in a min-heap. Every successful positive-TTL put gets a globally increasing version. Cleanup pops records whose expiration time is at or before now. It removes a map entry only when the key, version, and expiration time still match, so stale records cannot delete newer values. Put costs O(log h) plus amortized cleanup, get has expected O(1) lookup plus amortized cleanup, and compacted space is O(n).

Detailed Explanation

See the Code while reading this explanation.

The cache must return a value only while that value is live. It must also remove expired and stale data so memory does not grow forever. The solution uses a hash map for the newest live entry and a min-heap for expiration records. A globally increasing version separates the current entry from older heap records for the same key.

Useful Questions to Ask the Interviewer
  1. What input sizes, value ranges, and edge cases should the solution handle?
  2. What output should be returned for empty, invalid, or duplicate input?
  3. Should I prioritize execution time or memory use, and may I use the standard library?
Plan eviction and cleanup for a production TTL cache. diagram
How to Explain It in an Interview
1. Define the operations and stored data

The cache supports three operations:

  • put(key, value, ttl, now) stores or replaces a value.
  • get(key, now) returns a value only when it has not expired.
  • cleanup(now) removes records whose expiration time has arrived.

The hash map stores:

key -> (value, expires_at, version)

The min-heap stores:

(expires_at, key, version)

An entry is expired when expires_at <= now. A missing or expired key returns None.

2. Use the map as the source of truth

The map contains only the newest live entry for each key. The heap may still contain records created by older writes.

Every successful put with ttl > 0 receives a new globally increasing version. Versions are never reused.

Cleanup deletes a key only when the popped record has the same key, version, and expiration time as the current map entry. A record that does not match is stale, so cleanup skips it.

This prevents an old heap record from deleting a newer value or a later reinsertion of the same key.

3. Process cleanup before reads and writes

Both put and get call cleanup(now) first.

Cleanup repeatedly examines the smallest expiration time in the min-heap. While that time is less than or equal to now, cleanup pops the record.

If the key is no longer in the map, the record is stale and can be ignored.

If the current map entry has the same version and expiration time, cleanup deletes it. Otherwise, the popped record belongs to an older write and is skipped.

For ttl <= 0, put removes the current key and returns without creating a live entry.

For ttl > 0, put increments the version counter, calculates expires_at = now + ttl, updates the map, and pushes the new expiration record.

After cleanup, get returns None when the key is missing. Otherwise, it returns the stored value.

4. Walk through the verified example

The consistent example uses these operations:

  1. put("A", 100, ttl=5, now=0)
  2. put("B", 200, ttl=3, now=1)
  3. put("A", 101, ttl=4, now=3)
  4. get("B", now=4)
  5. cleanup(now=5)
  6. get("A", now=6)
  7. cleanup(now=8)
  8. get("A", now=8)

The initial state is:

store = {} expiry_heap = [] next_version = 0

Step 1 stores A.

expires_at = 0 + 5 = 5 version = 1

The state becomes:

store = {A: (100, 5, 1)} heap = [(5, A, 1)]

Step 2 stores B.

expires_at = 1 + 3 = 4 version = 2

The state becomes:

store = {A: (100, 5, 1), B: (200, 4, 2)} heap = [(4, B, 2), (5, A, 1)]

Step 3 updates A.

expires_at = 3 + 4 = 7 version = 3

The map replaces A v1 with A v3:

store = {A: (101, 7, 3), B: (200, 4, 2)} heap = [(4, B, 2), (5, A, 1), (7, A, 3)]

The old record (5, A, 1) remains in the heap, but it is now stale.

Step 4 calls get("B", now=4).

Cleanup pops (4, B, 2). It matches the current B entry, so B is deleted. The later lookup misses, so get returns None.

The state is:

store = {A: (101, 7, 3)} heap = [(5, A, 1), (7, A, 3)]

Step 5 calls cleanup(now=5).

Cleanup pops (5, A, 1). The current A entry is (101, 7, 3). Its version and expiration time do not match the popped record. The record is stale, so cleanup skips it.

The state is:

store = {A: (101, 7, 3)} heap = [(7, A, 3)]

Step 6 calls get("A", now=6).

The earliest expiration time is 7, which is later than 6. A is still live, so get returns 101.

Step 7 calls cleanup(now=8).

Cleanup pops (7, A, 3). It matches the current A entry, so A is deleted.

The state becomes:

store = {} heap = []

Step 8 calls get("A", now=8).

The key is missing, so get returns None.

The three get results are None, 101, and None.

5. Compact the heap when stale records grow

Repeated updates can leave stale records inside the heap. Those records are safe because of the version check, but they still use memory.

The implementation rebuilds the heap when:

len(expiry_heap) > 2 * len(store) + 32

The rebuild creates one heap record for each current map entry and then calls heapq.heapify.

This prevents stale records from accumulating indefinitely. The threshold avoids rebuilding after every update.

6. Explain why the solution is correct

The map always stores the newest live entry for each key.

The min-heap exposes the earliest record that may have expired.

A popped record can delete a map entry only when its key, version, and expiration time all match the current entry.

Globally increasing versions are never reused. Therefore, a stale record cannot match a later reinsertion of the same key.

Because get runs cleanup first, it does not return an entry whose expiration time is at or before now.

7. Explain complexity and edge cases

Let h be the heap size, n be the number of live entries, and k be the number of records popped during one cleanup call.

A positive-TTL put performs one heap push. This costs O(log h), plus cleanup work.

A get performs an expected O(1) Python dictionary lookup, plus cleanup work.

Cleanup costs O(k log h) when it pops k expired or stale records.

A heap rebuild costs O(n). This work is amortized because rebuilding happens only after the heap becomes much larger than the live map.

The map uses O(n) space. The compacted heap also uses O(n) space, with a fixed extra allowance from the +32 threshold.

Important edge cases include updating a key before its old TTL expires, ttl <= 0, a missing key, a request exactly at the expiration time, reinserting a removed key, and repeatedly rewriting one key.

Key Insight / Why This Solution Works

Use two data structures with different jobs. The hash map stores the latest live entry as key -> (value, expires_at, version). The min-heap stores expiration candidates as (expires_at, key, version), so its top record is the next one that may need removal. The central invariant is that a heap record may delete an entry only when its key, version, and expiration time all match the current map entry. Every successful positive-TTL put receives a globally increasing version that is never reused. This makes lazy deletion safe. Periodic heap rebuilding removes stale records that have not yet reached the top.

Code
import heapq
from typing import Any, Optional


class TTLCache:
    def __init__(self) -> None:
        # Source of truth for each latest live entry.
        # key -> (value, expires_at, version)
        self.store: dict[str, tuple[Any, int, int]] = {}

        # Min-heap of possible expirations.
        # Each item is (expires_at, key, version).
        self.expiry_heap: list[tuple[int, str, int]] = []

        # Successful positive-TTL writes receive unique versions.
        self.next_version = 0

    def cleanup(self, now: int) -> None:
        # Remove every heap record whose expiration time has arrived.
        while self.expiry_heap and self.expiry_heap[0][0] <= now:
            expires_at, key, version = heapq.heappop(self.expiry_heap)

            # The key may already be missing.
            current = self.store.get(key)
            if current is None:
                continue

            _, current_expires_at, current_version = current

            # Delete only when the popped record still describes
            # the latest map entry. Otherwise, it is stale.
            if current_version == version and current_expires_at == expires_at:
                del self.store[key]

        # Rebuild when stale records make the heap much larger
        # than the number of live entries.
        if len(self.expiry_heap) > 2 * len(self.store) + 32:
            self.expiry_heap = [
                (expires_at, key, version) for key, (_, expires_at, version) in self.store.items()
            ]
            heapq.heapify(self.expiry_heap)

    def put(
        self,
        key: str,
        value: Any,
        ttl: int,
        now: int,
    ) -> None:
        # Remove expired entries before changing the cache.
        self.cleanup(now)

        # A non-positive TTL means the key must not remain live.
        if ttl <= 0:
            self.store.pop(key, None)
            return

        # Allocate a globally increasing version.
        self.next_version += 1
        version = self.next_version

        # Convert the TTL into an absolute expiration time.
        expires_at = now + ttl

        # Store the latest live value.
        self.store[key] = (value, expires_at, version)

        # Add this version's expiration record to the min-heap.
        heapq.heappush(
            self.expiry_heap,
            (expires_at, key, version),
        )

    def get(self, key: str, now: int) -> Optional[Any]:
        # Remove expired entries before reading.
        self.cleanup(now)

        current = self.store.get(key)
        if current is None:
            return None

        value, expires_at, _ = current

        # Defensive check. cleanup should already remove this entry.
        if expires_at <= now:
            del self.store[key]
            return None

        return value


if __name__ == "__main__":
    cache = TTLCache()

    # 1) A v1 expires at time 5.
    cache.put("A", 100, ttl=5, now=0)

    # 2) B v2 expires at time 4.
    cache.put("B", 200, ttl=3, now=1)

    # 3) A v3 replaces A v1 and expires at time 7.
    cache.put("A", 101, ttl=4, now=3)

    # 4) B is expired at time 4.
    print(cache.get("B", now=4))  # None

    # 5) The stale A v1 record is popped and ignored.
    cache.cleanup(now=5)

    # 6) A v3 is still live at time 6.
    print(cache.get("A", now=6))  # 101

    # 7) A v3 is expired and removed by time 8.
    cache.cleanup(now=8)

    # 8) A is now missing.
    print(cache.get("A", now=8))  # None
Time & Space Complexity

Let h be the heap size, n be the number of live keys, and k be the number of records removed during cleanup. A positive-TTL put pushes one heap record, so that part costs O(log h), plus amortized cleanup work. A get uses an expected O(1) Python dictionary lookup, plus amortized cleanup work. Cleanup costs O(k log h) when it pops k records. Rebuilding the heap costs O(n), but it happens only when the heap is much larger than the live map. The map and compacted heap use O(n) auxiliary space, with a fixed extra allowance from the +32 threshold.

Where it is used

This pattern is useful in in-memory caches, session stores, API-token caches, temporary authorization data, rate-limit state, and other systems where keys expire at different times. The map gives fast access by key. The heap finds the next possible expiration without scanning every live entry. Version checks make repeated writes safe, and compaction controls stale-record memory growth.

Why Interviewers Ask This

This question tests whether the candidate can combine a hash map and a heap for different responsibilities. The interviewer is checking fast lookup, ordered expiration, safe lazy deletion, repeated updates, expiration boundaries, stale-record memory growth, and amortized complexity. It also tests whether the candidate notices that versions derived only from the current entry can be reused after deletion. The candidate must explain a practical compaction tradeoff and keep the walkthrough, invariant, and Python code consistent.

Common interview mistakes

A common mistake is deleting a key whenever any heap record for that key expires. An older record could then delete a newer value. Another mistake is deriving a version only from the current map entry. Removing and later reinserting a key could reuse an old version. Candidates may also forget to call cleanup before get, use expires_at < now instead of expires_at <= now, keep a live entry when ttl <= 0, omit heap compaction, or claim that every get is strictly O(1) even when it performs cleanup.

Interview tip

State the invariant before coding: the map owns the current value, and a heap record may delete it only when the key, version, and expiration time all match.

Interviewer may ask next
How would you make this cache safe when many threads call get and put at the same time?

Protect the store, heap, and version counter with the same lock. cleanup, get, and put must observe and update those structures atomically. In Python, an RLock is convenient because get and put call cleanup while already holding the lock. The map-and-heap invariant stays the same. Heap work remains O(log h), and dictionary lookup remains expected O(1), but lock contention can reduce throughput. Sharding the cache across several independent locks can improve concurrency at the cost of more complexity.

What changes if expired entries must be removed even when no get or put calls arrive?

Add a background worker that waits until the earliest heap expiration time. When that time arrives, it acquires the same lock and runs cleanup. A put that adds an earlier expiration must wake the worker so it can shorten its wait. The map, heap records, and version-matching rule do not change. Heap operations remain O(log h), and compacted space remains O(n). The tradeoff is extra thread, timer, wake-up, and shutdown logic in exchange for prompt cleanup during idle periods.

20. Implement a TTL cache.CodingMediumNetflix

Question Details

Implement an in-memory cache where every key expires after a time-to-live duration, and define how expired entries are detected and removed.

Short Interview Answer (30-60 seconds)

I would use a dictionary for the current record of each key and a min-heap for expiration records. Each heap item stores the expiration time, version, and key. Before every put or get, I pop records whose expiration time is less than or equal to now. A version check prevents an old heap record from deleting a newer value. Dictionary operations are O(1) on average, heap operations are O(log h), and auxiliary space is O(k + h).

Detailed Explanation

See the Code while reading this explanation.

The cache stores values in memory and makes each value unavailable after a fixed TTL. Scanning every key on each operation would be wasteful. Instead, the solution uses a dictionary for direct key lookup and a min-heap for expiration order. A version number makes overwrites safe when an older expiration record is still inside the heap.

Useful Questions to Ask the Interviewer
  1. What input sizes, value ranges, and edge cases should the solution handle?
  2. What output should be returned for empty, invalid, or duplicate input?
  3. Should I prioritize execution time or memory use, and may I use the standard library?
Implement a TTL cache. diagram
How to Explain It in an Interview
1. Define the input and output

The constructor receives ttl_seconds.

put(key, value, now) stores the value under the key. Its expiration time is now + ttl_seconds.

get(key, now) returns the stored value when the current record has not expired. It returns None when the key is missing or expired.

A record is expired when expires_at <= now.

2. Choose the data structures

The dictionary maps each key to:

(value, expires_at, version)

It stores the newest record for that key.

The min-heap stores:

(expires_at, version, key)

Python heapq is a min-heap. Therefore, the record with the smallest expiration time is always at the top.

The version number identifies one specific write. It prevents an old heap record from deleting a newer value written under the same key.

3. Initialize and maintain the invariant

The dictionary starts empty. The heap starts empty. next_version starts at zero.

Before every put and get, the cache runs _evict_expired(now).

After this cleanup, every record remaining in the dictionary has expires_at greater than now.

The heap may still contain stale records. A popped record may delete a dictionary entry only when both its expiration time and version match the current dictionary record.

4. Walk through the exact example

The TTL is 5 seconds.

At t=0, put A=100.

The cache creates version 1 and calculates expires_at = 0 + 5 = 5. The dictionary becomes A -> (100, 5, 1). The heap becomes [(5, 1, A)].

At t=1, put B=200.

The cache creates version 2 and calculates expires_at = 1 + 5 = 6. The dictionary contains A -> (100, 5, 1) and B -> (200, 6, 2). The heap contains (5, 1, A) and (6, 2, B).

At t=3, put A=150.

No record has expired. The cache creates version 3 and calculates expires_at = 3 + 5 = 8. The dictionary replaces A with A -> (150, 8, 3). The heap receives (8, 3, A). The older record (5, 1, A) remains in the heap, but it is stale.

At t=5, get A.

Cleanup pops (5, 1, A). The current dictionary record for A is (150, 8, 3). The expiration time and version do not match, so the popped record is stale and cannot delete A. The cache returns 150. The dictionary remains unchanged. The heap contains (6, 2, B) and (8, 3, A).

At t=6, get B.

Cleanup pops (6, 2, B). It matches the current dictionary record for B, so B is deleted. The following lookup does not find B, so get returns None. The dictionary contains only A -> (150, 8, 3). The heap contains (8, 3, A).

At t=9, get A.

Cleanup pops (8, 3, A). It matches the current dictionary record for A, so A is deleted. The following lookup does not find A, so get returns None.

The returned values are [150, None, None]. The final dictionary is empty, and the final heap is empty.

5. Explain why the solution is correct

The heap always exposes the next possible expiration in time order.

Cleanup removes every heap record whose expiration time is less than or equal to now.

A matching expiration time and version proves that the popped record still represents the current dictionary record. It is therefore safe to delete that key.

A mismatch proves that the heap record belongs to an older write. Ignoring it prevents a stale record from deleting a newer value.

After cleanup, every dictionary record is valid at the supplied time. Therefore, get returns a value only when that value has not expired.

6. Explain the Python implementation

The constructor validates the TTL and creates the dictionary, heap, and version counter.

_evict_expired repeatedly examines the heap top. It pops every record with expires_at <= now. It deletes a key only when the popped expiration time and version match the current dictionary record.

put runs cleanup first. It creates a new version, calculates the expiration time, updates the dictionary, and pushes one heap record.

get also runs cleanup first. It then returns the value from the current dictionary record or None when the key is absent.

7. Explain complexity and edge cases

Let h be the number of heap records before cleanup. Let r be the number of expired or stale records popped during one operation.

A put takes O(r log h + log h) time for cleanup and the new heap push, plus O(1) average dictionary work.

A get takes O(r log h) cleanup time plus O(1) average dictionary lookup.

Each heap record is pushed once and popped at most once. Cleanup work is therefore amortized across the sequence of operations.

Auxiliary space is O(k + h), where k is the number of current dictionary records. Repeated overwrites can make h larger than k because stale heap records remain until they reach the top.

A negative TTL is rejected. A TTL of zero makes an entry expire at the same timestamp. The next operation at that time or later removes it.

Key Insight / Why This Solution Works

The key insight is to separate direct lookup from expiration order. The dictionary provides O(1) average access to the newest record for a key. The min-heap exposes the next possible expiration without scanning all dictionary entries. The central invariant is that after _evict_expired(now), every dictionary record has expires_at greater than now. Version numbers make lazy cleanup safe. A heap record can delete a key only when its expires_at and version still match the current dictionary record. Any mismatch means the heap record belongs to an older overwrite and must be ignored.

Code
from __future__ import annotations

import heapq
from typing import Any, Optional


class TTLCache:
    def __init__(self, ttl_seconds: int) -> None:
        # Reject a TTL that would expire entries before they are written.
        if ttl_seconds < 0:
            raise ValueError("ttl_seconds must be non-negative")

        # Every write expires after this fixed number of seconds.
        self.ttl_seconds = ttl_seconds

        # key -> (value, expires_at, version)
        # The dictionary stores the newest record for each key.
        self.store: dict[str, tuple[Any, int, int]] = {}

        # Each heap record is (expires_at, version, key).
        # heapq keeps the smallest expires_at at index 0.
        self.expiry_heap: list[tuple[int, int, str]] = []

        # Each write receives a unique increasing version.
        self.next_version = 0

    def _evict_expired(self, now: int) -> None:
        # Pop every record whose expiration time has arrived.
        while self.expiry_heap and self.expiry_heap[0][0] <= now:
            expires_at, version, key = heapq.heappop(self.expiry_heap)

            # The key may already have been removed.
            current = self.store.get(key)
            if current is None:
                continue

            _, current_expires_at, current_version = current

            # Delete only when the popped heap record still represents
            # the current dictionary record for this key.
            if current_expires_at == expires_at and current_version == version:
                del self.store[key]

    def put(self, key: str, value: Any, now: int) -> None:
        # Remove expired current records and stale heap records
        # that have reached the top.
        self._evict_expired(now)

        # Give this write a new version.
        self.next_version += 1
        version = self.next_version

        # Calculate the exact expiration time.
        expires_at = now + self.ttl_seconds

        # Store the newest record for the key.
        self.store[key] = (value, expires_at, version)

        # Add its expiration record to the min-heap.
        heapq.heappush(
            self.expiry_heap,
            (expires_at, version, key),
        )

    def get(self, key: str, now: int) -> Optional[Any]:
        # Remove records that are expired at this time.
        self._evict_expired(now)

        # Read the newest remaining record for the key.
        current = self.store.get(key)
        if current is None:
            return None

        # The first tuple item is the cached value.
        return current[0]


if __name__ == "__main__":
    cache = TTLCache(ttl_seconds=5)

    # Exact example from the diagram.
    cache.put("A", 100, now=0)
    cache.put("B", 200, now=1)
    cache.put("A", 150, now=3)

    results = [
        cache.get("A", now=5),
        cache.get("B", now=6),
        cache.get("A", now=9),
    ]

    print(results)  # [150, None, None]
    print(cache.store)  # {}
    print(cache.expiry_heap)  # []
Time & Space Complexity

Let h be the number of heap records before cleanup, and let r be the number of expired or stale records popped during the operation. A put takes O(r log h + log h) time for cleanup and the new heap push, plus O(1) average dictionary work. A get takes O(r log h) time for cleanup, plus O(1) average dictionary lookup. Each heap record is pushed once and popped at most once, so cleanup work is amortized across many operations. Auxiliary space is O(k + h), where k is the number of current dictionary records. Repeated overwrites can make h larger than k.

Where it is used

This pattern is useful for in-memory caches, temporary authentication tokens, sessions, rate-limit state, deduplication records, and other data that becomes invalid after a fixed time. The dictionary supports fast access by key. The min-heap supports ordered lazy expiration without scanning every stored key during each operation.

Why Interviewers Ask This

This question tests whether you can combine data structures with different strengths. The interviewer is checking whether you choose fast dictionary lookup, use a min-heap for expiration order, handle overwrites without deleting newer values, define the expiration boundary correctly, maintain a clear invariant, and explain amortized heap cleanup accurately. It also tests whether your Python code and complexity claims match the behavior you describe.

Common interview mistakes

One mistake is deleting a key whenever any old heap record expires. That can remove a newer value written under the same key. Compare both expires_at and version before deleting. Another mistake is running cleanup only during get. put must also clean expired records before writing. A third mistake is using expires_at < now instead of expires_at <= now. In this design, an entry is expired exactly at its expiration timestamp. Candidates may also claim every operation is O(1), even though heap pushes and pops cost O(log h). Finally, repeated overwrites can leave stale records in the heap, so space can grow beyond the number of current keys.

Interview tip

Say the invariant before coding: after cleanup at time now, every dictionary record expires after now, and a popped heap record may delete a key only when both its expiration time and version match the current record.

Interviewer may ask next
How would you reduce stale heap records after many overwrites?

The current solution uses lazy cleanup, so old heap records remain until they reach the top. I could rebuild the heap when its size becomes much larger than the number of current dictionary records. I would create one heap record for each current dictionary entry and call heapq.heapify. Rebuilding takes O(k) time and O(k) temporary or replacement space for k current keys. The tradeoff is an occasional O(k) pause in exchange for lower long-term heap memory.

How would the design change if each put could have a different TTL?

I would pass ttl_seconds to put instead of using one shared TTL from the constructor. Each write would calculate expires_at = now + ttl_seconds and store that expiration time in both the dictionary and heap. The version check and cleanup logic would remain unchanged, so correctness is preserved. The time complexity would still be O(r log h + log h) for put and O(r log h) plus O(1) average lookup for get. Space would remain O(k + h).

More questions load as you scroll

Disclaimer: This interview guide is for educational and informational purposes only. It is designed to help readers prepare, but it does not guarantee any interview result, hiring decision, offer, or outcome. Interview questions, hiring criteria, and preferred answers can vary by employer, interviewer, industry, location, and time. The examples and explanations reflect the authors' research and judgment, are provided without warranties of any kind, and should not be treated as the only correct approach. Diagrams are simplified illustrations intended to highlight the main components and their interactions; actual systems and implementations may be more complex. Alternative approaches may be equally valid or better suited to a particular question, context, or interviewer. To the fullest extent permitted by applicable law, the author, contributors, and publisher are not liable for decisions made, actions taken, or losses incurred based on this guide.

Company Notice: This guide is an independent educational resource and is not affiliated with, endorsed by, sponsored by, or approved by the company named in this guide. Company names are used only to identify interview experiences commonly reported by candidates. Interview practices can change without notice, and inclusion of company-specific content does not mean these questions are official, complete, or guaranteed to be asked. To the fullest extent permitted by law, the author, contributors, and publisher are not responsible for outcomes related to use of this material.