Netflix Python Developer Interview Questions & Answers

netflix icon

Questions with Detailed ExplanationsWith Detailed Explanations

(Last Updated: September 3, 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. 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.

17. Calculate employee levels, balanced employees, and a level histogram from an org chart.CodingMediumNetflix

Question Details

Given an organization chart represented as a tree, calculate each employee's level, identify employees with equal numbers of nodes above and below, and produce a histogram by level.

Short Interview Answer (30-60 seconds)

I would use one depth-first search starting from the CEO at level 0. When I enter an employee, I record the current depth as the level and increase that level’s histogram count. When the recursive calls return, I calculate the employee’s subtree size. The number below is subtree size minus one. If it equals the depth, the employee is balanced. Every employee is visited once, so the time complexity is O(n), and the auxiliary space complexity is O(n).

Detailed Explanation

See the Code while reading this explanation.

The input is a rooted organization tree. We need to calculate each employee’s level, find employees whose number of managers above equals their number of descendants below, and count how many employees appear at each level. A depth-first search fits this problem because depth is available while moving down the tree, while subtree size is available when the recursive calls return.

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?
Calculate employee levels, balanced employees, and a level histogram from an org chart. diagram
How to Explain It in an Interview
1. Define the input and outputs

The organization chart is stored as an adjacency list. Each employee maps to a list of direct reports.

For the example:

CEO -> [CTO, CFO, COO] CTO -> [Dev1] CFO -> [Fin1] COO -> [Ops1] Dev1 -> [] Fin1 -> [] Ops1 -> []

The CEO is the root.

The level of an employee is the number of managers above that employee. The CEO has level

  1. CTO, CFO, and COO have level
  2. Dev1, Fin1, and Ops1 have level 2.

The number below an employee means the number of descendants in that employee’s subtree. If subtree_size includes the employee, then descendants equals subtree_size minus one.

An employee is balanced when:

depth == subtree_size - 1

The function returns a map of employee levels, a list of balanced employees, and a histogram that maps each level to its employee count.

2. Choose one depth-first search

I call DFS with the CEO and depth 0.

When DFS enters an employee, it records the employee’s level and increases the histogram count for that level.

It then visits every direct report with depth plus one.

When all child calls return, DFS adds their subtree sizes and includes the current employee. This gives the exact subtree size for the current employee.

3. Initialize the state

The levels map starts as an empty dictionary. It stores employee name to level.

The histogram starts empty. It stores level to employee count.

The balanced list starts empty. It stores employees that pass the balance condition.

The traversal starts at CEO with depth 0.

The main invariant is: after DFS returns from an employee, the returned subtree size is correct for that employee, and every employee in that subtree has already been processed.

4. Walk through the exact example

Step 1: Enter CEO at depth 0. Record CEO: 0. The histogram becomes {0: 1}.

Step 2: Enter CTO at depth 1. Record CTO: 1. The histogram becomes {0: 1, 1: 1}.

Step 3: Enter Dev1 at depth 2. Record Dev1: 2. The histogram becomes {0: 1, 1: 1, 2: 1}. Dev1 has no reports, so its subtree size is 1. Its descendant count is 0. Depth 2 does not equal 0, so Dev1 is not balanced.

Step 4: Return to CTO. CTO’s subtree contains CTO and Dev1, so its subtree size is 2. Its descendant count is 1. Its depth is also 1, so CTO is balanced.

Step 5: Enter CFO at depth 1. Record CFO: 1. The histogram becomes {0: 1, 1: 2, 2: 1}.

Step 6: Enter Fin1 at depth 2. Record Fin1: 2. The histogram becomes {0: 1, 1: 2, 2: 2}. Fin1 has subtree size 1 and 0 descendants. Depth 2 does not equal 0, so Fin1 is not balanced.

Step 7: Return to CFO. CFO has subtree size 2 and 1 descendant. Its depth is 1, so CFO is balanced.

Step 8: Enter COO at depth 1. Record COO: 1. The histogram becomes {0: 1, 1: 3, 2: 2}.

Step 9: Enter Ops1 at depth 2. Record Ops1: 2. The histogram becomes {0: 1, 1: 3, 2: 3}. Ops1 has subtree size 1 and 0 descendants. Depth 2 does not equal 0, so Ops1 is not balanced.

Step 10: Return to COO. COO has subtree size 2 and 1 descendant. Its depth is 1, so COO is balanced.

Step 11: Return to CEO. The CEO’s subtree contains all 7 employees. The CEO therefore has 6 descendants. Its depth is 0, so the CEO is not balanced.

The final result is:

levels = {CEO: 0, CTO: 1, CFO: 1, COO: 1, Dev1: 2, Fin1: 2, Ops1: 2}

balanced = [CTO, CFO, COO]

histogram = {0: 1, 1: 3, 2: 3}

5. Explain why the result is correct

The depth passed into DFS is exactly the number of managers above the current employee.

The subtree size returned by DFS includes the current employee and every descendant. Subtracting one removes the current employee and gives the exact number of descendants below.

Therefore, depth == subtree_size - 1 correctly identifies balanced employees.

The histogram is correct because every employee increases exactly one bucket for the employee’s level.

6. Explain the Python implementation

The outer function creates the levels dictionary, histogram, and balanced list.

The nested dfs function receives an employee and that employee’s depth. It records the level and histogram count before visiting direct reports.

It starts subtree_size at 1 because the subtree contains the current employee. Each recursive child call returns a child subtree size, which is added to the total.

After all children return, the code calculates descendants as subtree_size - 1. It adds the employee to balanced when descendants equals depth. It then returns the subtree size to the parent.

After DFS finishes, the code returns the levels map, balanced list, and histogram sorted by level.

7. Explain complexity and edge cases

Let n be the number of employees. Each employee is entered once and returned from once. Therefore, the time complexity is O(n).

The auxiliary space complexity is O(n). The levels map, histogram, balanced list, and recursion stack can grow with the number of employees.

Relevant edge cases are a single-employee tree, a skewed tree, employees with no reports, and employees represented by empty child lists. A single CEO has depth 0 and 0 descendants, so that CEO is balanced under the same rule.

Key Insight / Why This Solution Works

The key insight is that the two values used by the balance test become available at different parts of one DFS. The employee’s depth is known when DFS enters the node, so it gives the number of managers above. The employee’s subtree size is known after all child calls return, so subtree_size - 1 gives the number of descendants below. The invariant is that when DFS returns from a node, its complete subtree has been processed and its returned subtree size is correct. This lets one traversal produce all three required outputs.

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


def analyze_org_chart(
    org: Dict[str, List[str]], root: str
) -> Tuple[Dict[str, int], List[str], Dict[int, int]]:
    """Return employee levels, balanced employees, and a level histogram."""

    # employee name -> level from the root
    levels: Dict[str, int] = {}

    # level -> number of employees at that level
    histogram = defaultdict(int)

    # Employees whose managers above equal descendants below
    balanced: List[str] = []

    def dfs(employee: str, depth: int) -> int:
        # Record the employee's level when entering the node.
        levels[employee] = depth

        # Count this employee in the correct level bucket.
        histogram[depth] += 1

        # The subtree contains at least the current employee.
        subtree_size = 1

        # Visit each direct report with the next depth.
        for report in org.get(employee, []):
            subtree_size += dfs(report, depth + 1)

        # Remove the current employee to count only descendants.
        descendants = subtree_size - 1

        # depth is the number of managers above this employee.
        if depth == descendants:
            balanced.append(employee)

        # Give the complete subtree size to the parent call.
        return subtree_size

    # The root starts at level 0.
    dfs(root, 0)

    # Return the histogram in increasing level order.
    return levels, balanced, dict(sorted(histogram.items()))


if __name__ == "__main__":
    org_chart = {
        "CEO": ["CTO", "CFO", "COO"],
        "CTO": ["Dev1"],
        "CFO": ["Fin1"],
        "COO": ["Ops1"],
        "Dev1": [],
        "Fin1": [],
        "Ops1": [],
    }

    levels, balanced, histogram = analyze_org_chart(org_chart, "CEO")

    print("levels =", levels)
    print("balanced =", balanced)
    print("histogram =", histogram)
Time & Space Complexity

Let n be the number of employees. The time complexity is O(n) because DFS visits each employee once and processes each reporting relationship once. The auxiliary space complexity is O(n). The levels dictionary can hold n entries. The balanced list can contain up to n employees. The histogram can contain up to n level entries. The recursion stack can also grow to n when the organization tree is completely skewed.

Where it is used

This pattern is useful for organization charts, file-system trees, category trees, reporting hierarchies, and other rooted trees. It is helpful when a program needs information from above a node, such as depth, and information from below a node, such as subtree size or descendant count, during the same traversal.

Why Interviewers Ask This

This problem tests whether a candidate can combine top-down and bottom-up information in a tree. The level comes from the path from the root, while the descendant count comes from recursive return values. The interviewer is also checking recursion, adjacency-list traversal, invariant reasoning, management of several result structures, correct execution order, and accurate time and space analysis.

Common interview mistakes

A common mistake is counting only direct reports instead of all descendants. Another mistake is comparing depth with subtree size without subtracting the current employee. Some candidates check the balance condition before all child calls return, when the final subtree size is still unknown. Others forget to update the histogram on every node entry. It is also incorrect to claim O(1) auxiliary space because the result structures and recursion stack can grow with the tree.

Interview tip

Explain the solution with one sentence before coding: depth gives the number above, and subtree size minus one gives the number below. Then show that DFS provides both values in one traversal.

Interviewer may ask next
How would you handle an organization tree that is too deep for Python recursion?

I would replace recursive DFS with an explicit stack. Each stack item would store the employee, depth, and whether the node is being entered or returned from. On entry, I would record the level and histogram. On return, I would combine the child subtree sizes and run the same balance test. Correctness is preserved because the explicit stack follows the same entry and postorder-return order. The time remains O(n), and the auxiliary space remains O(n). The main tradeoff is more implementation detail, but it avoids Python recursion-depth errors.

Can the algorithm work if the organization chart is not a valid tree?

The original algorithm assumes one rooted tree. If reporting links can contain cycles, I would add a visit-state map and reject a cycle because subtree size is not well-defined for a cycle. If one employee can have multiple managers, the employee can be reached through more than one path, so levels and descendant subtrees may no longer have the same tree meaning. Validation takes O(n + e) time and O(n) space, where e is the number of reporting links. The tradeoff is extra validation and a need to define the desired graph semantics.

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. 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.

20. Find the k closest points to the origin.CodingMediumNetflix

Question Details

Given points in a plane and an integer k, return the k points closest to the origin and explain the data structure, edge cases, and complexity.

Short Interview Answer (30-60 seconds)

I use a size-limited max heap to keep the k closest points seen so far. Python heapq is a min heap, so I store each squared distance as a negative value. For every point, I compute x * x + y * y, push the negative distance with the point, and pop when the heap grows beyond k. This removes the farthest candidate. The solution takes O(n log k) time and O(k) auxiliary space.

Detailed Explanation

See the Code while reading this explanation.

The problem gives a list of points in a two-dimensional plane and an integer k. We must return the k points closest to the origin, which is (0, 0). We compare squared distances, x² + y², because taking a square root would not change their order. A size-limited heap lets us keep only the best k points while processing the input.

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 the k closest points to the origin. diagram
How to Explain It in an Interview
1. Understand the input and output

The input contains points in the form [x, y] and an integer k. The output contains point values, not indices.

For the example:

points = [[1, 3], [-2, 2], [4, 0], [0, 1]] k = 2

One valid output is:

[[-2, 2], [0, 1]]

The returned order does not matter.

2. Choose a size-limited max heap

We want to remove the farthest candidate whenever we have more than k points. A max heap is useful because its root represents the largest distance among the kept points.

Python heapq implements a min heap. To simulate a max heap, we store each squared distance as a negative value.

Each heap item has this form:

(-distance_sq, point)

A larger real distance becomes a smaller negative number. Therefore, heapq removes the farthest point first.

The invariant is that after each insertion and optional pop, the heap contains the closest min(k, processed points) points seen so far. When the heap contains k points, its root represents the farthest kept point.

3. Initialize and process each point

Start with an empty heap:

max_heap = []

For each point [x, y], calculate:

distance_sq = x * x + y * y

Push (-distance_sq, [x, y]) into the heap.

If the heap size becomes greater than k, pop one item. The popped item is the farthest point among the current candidates.

4. Walk through the verified example

For [1, 3], the squared distance is 1² + 3² = 10. Push (-10, [1, 3]). The heap contains one point.

For [-2, 2], the squared distance is (-2)² + 2² = 8. Push (-8, [-2, 2]). The heap now contains two points, so nothing is removed.

For [4, 0], the squared distance is 4² + 0² = 16. Push (-16, [4, 0]). The heap size becomes 3, which is greater than k. Pop once. The point [4, 0] is removed because its squared distance, 16, is the largest among the three candidates. The heap keeps [1, 3] and [-2, 2].

For [0, 1], the squared distance is 0² + 1² = 1. Push (-1, [0, 1]). The heap size again becomes 3. Pop once. The farthest kept point is [1, 3], with squared distance 10, so it is removed.

The final heap contains [-2, 2] and [0, 1]. Their squared distances are 8 and 1. These are the two smallest distances in the example.

5. Explain why the algorithm is correct

After every insertion and optional pop, the heap contains the closest min(k, processed points) points seen so far.

If a new point is farther than the current kept set, that new point becomes the farthest candidate and is removed. If the new point is closer, the previous farthest kept point is removed instead.

After all points are processed, the heap contains the k closest points from the input.

6. Explain the Python implementation

The function returns an empty list when k is zero or negative. It then creates an empty heap and processes every input point.

For each point, it calculates the squared distance and pushes the negative distance with the point. If the heap grows beyond k, it pops once.

After the loop, it extracts and returns the point from every heap item. A heap is not a fully sorted structure, so the returned order may vary.

7. Explain complexity and edge cases

Each of the n points is pushed into a heap whose size is kept near k. A push or pop costs O(log k), so the total time is O(n log k).

The heap stores at most k points after trimming, so the auxiliary space is O(k).

Relevant edge cases include k = 0, k equal to or greater than the number of points, negative coordinates, duplicate points, and equal distances. If k is at least the number of points, all points are returned. When distances tie, any valid set of k closest points may be returned.

Key Insight / Why This Solution Works

The key idea is to keep only the best k points instead of sorting all n points. The algorithm uses a size-limited max heap. Each entry stores (-distance_sq, point), where distance_sq is x * x + y * y. Python heapq is a min heap, so the negative distance makes the farthest real point appear at the root. After each push, the algorithm pops once if the heap size is greater than k. The invariant is that the heap contains the closest min(k, processed points) points seen so far.

Code
from typing import List
import heapq


class Solution:
    def kClosest(self, points: List[List[int]], k: int) -> List[List[int]]:
        # Return no points when k is zero or negative.
        if k <= 0:
            return []

        # Python heapq is a min heap.
        # Negative distances make it behave like a max heap.
        max_heap: list[tuple[int, List[int]]] = []

        # Process each point.
        for x, y in points:
            # Squared distance is enough for comparison.
            distance_sq = x * x + y * y

            # Store the negative distance and the point.
            heapq.heappush(max_heap, (-distance_sq, [x, y]))

            # Keep only the k closest points seen so far.
            # This removes the current farthest candidate.
            if len(max_heap) > k:
                heapq.heappop(max_heap)

        # Return the points left in the heap.
        # Any output order is acceptable.
        return [point for _, point in max_heap]


if __name__ == "__main__":
    points = [[1, 3], [-2, 2], [4, 0], [0, 1]]
    k = 2

    result = Solution().kClosest(points, k)
    print(result)  # One valid output: [[-2, 2], [0, 1]]
Time & Space Complexity

Let n be the number of input points. The algorithm pushes every point into the heap and may pop one point after each push. Because the heap stays near size k, each push or pop takes O(log k) time. The total time is O(n log k). The heap stores at most k points after trimming, so the auxiliary space is O(k). Computing x * x + y * y takes O(1) time, and no square root is needed.

Where it is used

This top-k heap pattern is useful when a program must keep only the best k items from a large input or stream. Examples include finding nearby locations, selecting the lowest prices, keeping the highest scores, and tracking the most important events without sorting every candidate.

Why Interviewers Ask This

The interviewer is checking whether you recognize a top-k problem and choose a suitable heap. They want to see that you understand Python heapq is a min heap and can simulate a max heap with negative distances. The question also tests whether you can maintain a clear invariant, compare distances without unnecessary square roots, handle ties and duplicates, write correct Python, and explain O(n log k) time with O(k) auxiliary space.

Common interview mistakes

A common mistake is pushing positive distances into heapq. That creates a normal min heap and causes the closest point to be removed when the size exceeds k. Another mistake is forgetting to pop when the heap grows beyond k. Some candidates calculate square roots even though squared distances are enough. Others claim the heap output is sorted, but heap order is not fully sorted. It is also incorrect to claim that the displayed output order is the only valid order.

Interview tip

Explain the invariant before writing code: after every push and optional pop, the heap contains the closest min(k, processed points) points seen so far. This makes the reason for using negative distances and removing one point easy to defend.

Interviewer may ask next
How would the solution work if the points arrived as a continuous stream?

Keep the same size-limited heap between arrivals. For each new point, calculate its squared distance, push the negative distance and point, and pop once if the heap size becomes greater than k. The invariant remains unchanged, so the heap contains the closest min(k, points seen) points. Processing m streamed points takes O(m log k) total time and O(k) auxiliary space. The main benefit is that the full stream does not need to be stored.

What changes if the returned points must be ordered from closest to farthest?

Use the same heap process to select the k closest points. Then sort only those k points by x * x + y * y before returning them. The heap phase takes O(n log k), and sorting the result adds O(k log k). The total time is O(n log k + k log k), and the heap still uses O(k) auxiliary space. The tradeoff is extra work to guarantee the output order.

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.

Content Accuracy and Verification: To the fullest extent permitted by applicable law, we do not represent or warrant that interview guides, questions, answers, examples, or diagrams are accurate, complete, current, error-free, or suitable for any particular purpose. You are responsible for independently reviewing and verifying the information before relying on it.