Common Interview Problem Patterns Questions

Familiarize yourself with typical problem patterns: string manipulation (finding substrings, counting characters), array operations (finding duplicates, sorting, searching), two-pointer techniques, basic recursion, and simple dynamic programming. Focus on problems that are LeetCode Easy to Easy-Medium difficulty.

EasyTechnical

75 practiced

As part of schema validation you need to know how deeply nested incoming JSON records are. Given a Python object parsed from JSON (dictionaries, lists, primitives), implement max_depth(obj) -> int using recursion to compute the maximum nesting depth. Discuss recursion limits and iterative alternatives for very deep inputs.

Sample Answer

To compute maximum nesting depth of a Python object parsed from JSON, recursively treat dicts and lists as containers that add one level; primitives (str, int, float, bool, None) contribute depth 0. Base case returns 0 for primitives; for containers, return 1 + max(child depths) or 1 for empty containers.

python

def max_depth(obj):
    """
    Return maximum nesting depth for a JSON-parsed Python object.
    Primitives -> 0, [] or {} -> 1, nested containers add levels.
    """
    # primitives: not list or dict
    if not isinstance(obj, (list, dict)):
        return 0

    # get children iterator
    if isinstance(obj, dict):
        children = obj.values()
    else:
        children = obj

    if not children:  # empty container
        return 1

    # recursively compute max depth among children
    return 1 + max(max_depth(child) for child in children)

Key points:- Time: O(n) visiting each node once. Space: O(d) recursion stack where d is depth.- Edge cases: empty list/dict -> depth 1; homogeneous primitives -> 0.- Recursion limits: Python's recursion limit (~1000) may be exceeded for very deep JSON. For extreme depth, use an iterative DFS with an explicit stack:

python

def max_depth_iter(obj):
    if not isinstance(obj, (list, dict)):
        return 0
    stack = [(obj, 1)]
    maxd = 0
    while stack:
        cur, d = stack.pop()
        maxd = max(maxd, d)
        if isinstance(cur, dict):
            items = cur.values()
        else:
            items = cur
        for child in items:
            if isinstance(child, (list, dict)):
                stack.append((child, d + 1))
    return maxd

Use the iterative version in production pipelines that may receive adversarial or very deep inputs.

MediumTechnical

53 practiced

Design a MedianFinder class in Python with methods add_num(num: int) and find_median() to maintain the median of an incoming stream of integers. Use two heaps (max-heap for lower half, min-heap for upper half) to support O(log n) insertion and O(1) median query. Explain how you'd use this in monitoring streaming metrics.

Sample Answer

Approach: maintain two heaps — a max-heap for the lower half and a min-heap for the upper half. Keep sizes balanced (difference ≤ 1). Insertion is O(log n); median read is O(1).

python

import heapq

class MedianFinder:
    def __init__(self):
        # max_heap stores negatives to simulate max-heap
        self.max_heap = []  # lower half
        self.min_heap = []  # upper half

    def add_num(self, num: int) -> None:
        # Push into max_heap first (as negative)
        heapq.heappush(self.max_heap, -num)
        # Ensure every element in max_heap <= every element in min_heap
        if self.min_heap and -self.max_heap[0] > self.min_heap[0]:
            val = -heapq.heappop(self.max_heap)
            heapq.heappush(self.min_heap, val)
        # Rebalance sizes so difference <= 1
        if len(self.max_heap) > len(self.min_heap) + 1:
            heapq.heappush(self.min_heap, -heapq.heappop(self.max_heap))
        elif len(self.min_heap) > len(self.max_heap) + 1:
            heapq.heappush(self.max_heap, -heapq.heappop(self.min_heap))

    def find_median(self) -> float:
        if len(self.max_heap) > len(self.min_heap):
            return float(-self.max_heap[0])
        if len(self.min_heap) > len(self.max_heap):
            return float(self.min_heap[0])
        # even count
        return (-self.max_heap[0] + self.min_heap[0]) / 2.0

Key points:- Use negatives to simulate max-heap in Python.- After each insert, enforce order invariant and balance sizes.

Complexity:- add_num: O(log n)- find_median: O(1)- Space: O(n)

Edge cases:- Single element, duplicates, negative numbers, streaming large volume (keeps only heaps).

Usage in monitoring streaming metrics:- Keep running median latency, request size, or error rates without storing full history.- Integrate with streaming frameworks (Kafka/Flink/Spark Streaming) to maintain per-key MedianFinder instances (e.g., per API endpoint). Periodically emit median as an aggregated metric to monitoring systems (Prometheus/Grafana) for robust central tendency resistant to outliers.

MediumTechnical

78 practiced

Table transactions(transaction_id PK, user_id INT, amount DECIMAL, occurred_at TIMESTAMP). Write a SQL query that computes a running total per user ordered by occurred_at and flags transactions where the running total within the last 24 hours exceeds threshold T. Use window functions/range framing and discuss timezone and late-arriving data implications.

Sample Answer

Approach: compute, per user, a 24-hour rolling sum of amount ordered by event time and flag rows where that sum > T. Use a RANGE frame on occurred_at so each row sums amounts from occurred_at - 24 hours through current row. Store timestamps in UTC (timestamptz) and break ties with transaction_id for deterministic ordering.

SQL (Postgres / modern engines that support RANGE with INTERVAL):

sql

WITH base AS (
  SELECT
    transaction_id,
    user_id,
    amount,
    occurred_at::timestamptz AS occurred_at_utc
  FROM transactions
)
SELECT
  transaction_id,
  user_id,
  amount,
  occurred_at_utc,
  SUM(amount) OVER (
    PARTITION BY user_id
    ORDER BY occurred_at_utc, transaction_id
    RANGE BETWEEN INTERVAL '24 hours' PRECEDING AND CURRENT ROW
  ) AS rolling_24h_sum,
  CASE WHEN
    SUM(amount) OVER (
      PARTITION BY user_id
      ORDER BY occurred_at_utc, transaction_id
      RANGE BETWEEN INTERVAL '24 hours' PRECEDING AND CURRENT ROW
    ) > /* threshold */ 1000 THEN true ELSE false END AS exceeds_threshold
FROM base
ORDER BY user_id, occurred_at_utc, transaction_id;

Notes / reasoning:- Use timestamptz/UTC to avoid timezone ambiguity; convert to local zones at presentation.- Tie-breaker transaction_id ensures deterministic order when occurred_at duplicates.- Some engines (older Postgres, Redshift, BigQuery) differ in RANGE/INTERVAL support — alternative is a correlated subquery or a SUM(...) FILTER with window on rows between unbounded and current and a JOIN on time bounds, or use LAG/lead buckets in Spark/Beam streaming.

Late-arriving data and correctness:- Late / out-of-order events can change prior rolling sums. In batch pipelines, reprocess affected windows (partition by date ranges) or perform incremental upserts with windowed recomputation.- In streaming, use watermarks and allowed-lateness to balance correctness vs latency; keep state long enough to accept expected lateness and emit corrected alerts on late updates.- Deduplicate on a unique key (transaction_id) and maintain idempotent writes.- For large partitions, add indexes on (user_id, occurred_at) and consider summarizing with time-buckets if exact per-transaction granularity is unnecessary.

HardTechnical

75 practiced

Implement first_unique_at_each_index(stream: Iterable[str]) -> List[Optional[str]] in Python that returns the first non-repeating character at each prefix of the stream (or None if none). Use O(1) amortized operations per character with a queue and counts. Explain memory implications for large alphabets and how this helps streaming token uniqueness checks.

Sample Answer

Approach: Maintain a FIFO queue of characters seen in order and a hashmap (collections.Counter or dict) of counts. For each incoming character c: increment count[c]; append c to queue. Then pop from queue head while it has count > 1. The head (or None if empty) is the first unique for that prefix. This is amortized O(1) per char because each character is pushed once and popped at most once.

python

from collections import deque, defaultdict
from typing import Iterable, List, Optional

def first_unique_at_each_index(stream: Iterable[str]) -> List[Optional[str]]:
    q = deque()                      # candidate uniques in order
    count = defaultdict(int)        # counts per token
    result: List[Optional[str]] = []

    for ch in stream:
        count[ch] += 1
        q.append(ch)

        # drop from queue any items no longer unique
        while q and count[q[0]] > 1:
            q.popleft()

        result.append(q[0] if q else None)

    return result

Key concepts:- Each char enqueued once and dequeued at most once → amortized O(1) per char.- Space: O(k + m) where k = number of distinct tokens tracked (size of count) and m = current queue size (≤ number of distinct tokens seen). For small fixed alphabets (ASCII) this is bounded; for large/unbounded token sets (UUIDs, long strings, or byte sequences) memory grows with distinct tokens.

Memory implications for large alphabets:- Counts dict can become large if many distinct tokens appear; use approximate structures (Count-Min Sketch) or eviction policies (time/window-based sliding) when strict per-token correctness isn't required.- For streaming pipelines, prefer hashing tokens to fixed-size bins, or maintain uniqueness only within a sliding window to bound memory.

How this helps in data engineering:- Enables online detection of first-unique tokens per prefix useful for deduplication, watermarking, or identifying earliest non-repeated event keys in ingestion streams.- Amortized O(1) makes it suitable as a lightweight per-record check inside high-throughput pipelines; for very high cardinality, combine with windowing, sampling, or probabilistic counters to control memory.

MediumTechnical

69 practiced

Implement longest_unique_substring(s: str) -> int in Python to return the length of the longest substring without repeating characters. This sliding-window pattern is useful when evaluating token entropy or unique consecutive sequences in logs. Aim for O(n) time using a hash map of last-seen indices.

Sample Answer

To solve this in O(n) time, use a sliding window with two pointers (left, right) and a hash map that stores the last-seen index of each character. When a repeated character is found inside the current window, advance left to one past that character's last index. Update max length as you expand right.

python

def longest_unique_substring(s: str) -> int:
    """
    Return length of longest substring without repeating characters.
    O(n) time, O(min(n, charset)) space.
    """
    last_seen = {}          # char -> last index seen
    left = 0                # start of current window
    max_len = 0

    for right, ch in enumerate(s):
        if ch in last_seen and last_seen[ch] >= left:
            # repeated inside window -> move left just past last occurrence
            left = last_seen[ch] + 1
        last_seen[ch] = right
        max_len = max(max_len, right - left + 1)

    return max_len

Key points:- last_seen ensures we only move left forward (never back), giving linear time.- Works for arbitrary unicode characters.

Time Complexity: O(n) — each index visited once.Space Complexity: O(min(n, alphabet_size)) — map stores distinct chars.

Edge cases:- Empty string -> 0- All-unique characters -> len(s)- All-same characters -> 1

Alternative: Use an array of size 256/128 for ASCII to reduce dict overhead if charset is known and small.

Unlock Full Question Bank

Get access to hundreds of Common Interview Problem Patterns interview questions and detailed answers.

Join thousands of developers preparing for their dream job.