DSA With AI
Week 11 ยท Heap / Priority Queue

Find Median from Data Stream

Solve Find Median from Data Stream by recognizing the Heap / Priority Queue pattern and turning the prompt into a small invariant before coding.

Hard Heap / Priority Queue Week 11 Practice runner

Frame the problem

  • Implement median_finder_session with the exact signature used by the interactive runner.
  • Use the visible tests to confirm the input and output shape before reading the final solution.
  • Treat challenge tests as edge-case pressure: empty inputs, repeated values, boundary shapes, or impossible states.
  • State the invariant before code, then dry-run one passing case and one failing-looking case.
1. Reveal example inputs and outputs

Example 1

Input:

median_finder_session([
  "addNum",
  "addNum",
  "findMedian",
  "addNum",
  "findMedian"
], [
  [
    1
  ],
  [
    2
  ],
  [],
  [
    3
  ],
  []
])

Output:

[
  null,
  null,
  1.5,
  null,
  2
]
2. Brute force first

What direct brute force would be correct for a tiny input? Name the exact repeated work that the target pattern removes.

3. Reveal the insight ladder
  1. Map the prompt to the Heap / Priority Queue pattern instead of starting from syntax.
  2. Balance a max-heap for the lower half and min-heap for the upper half.
  3. When the lower heap has one extra value, it is the median.
  4. Only reveal the final code after you can explain why each state update is safe.
4. Dry run before code
  1. median-basic: input [["addNum","addNum","findMedian","addNum","findMedian"],[[1],[2],[],[3],[]]] should produce [null,null,1.5,null,2]. Hint to check your state: Balance a max-heap for the lower half and min-heap for the upper half.
  2. median-single: input [["addNum","findMedian"],[[5],[]]] should produce [null,5]. Hint to check your state: When the lower heap has one extra value, it is the median.
5. Reveal final Python solution 
def median_finder_session(operations: list[str], arguments: list[list[int]]) -> list[Optional[float]]:
    small: list[int] = []
    large: list[int] = []
    output: list[Optional[float]] = []

    def add_num(value: int) -> None:
        heapq.heappush(small, -value)
        heapq.heappush(large, -heapq.heappop(small))
        if len(large) > len(small):
            heapq.heappush(small, -heapq.heappop(large))

    for op, args in zip(operations, arguments):
        if op == "addNum":
            add_num(args[0])
            output.append(None)
        elif op == "findMedian":
            if len(small) > len(large):
                output.append(float(-small[0]))
            else:
                output.append((-small[0] + large[0]) / 2)
    return output

Complexity: Derive the exact bounds from median_finder_session: count how often each input item is visited and the maximum size of the main state structure.

Interview narration

  • I will first describe the invariant in plain language.
  • Then I will explain what data structure carries that invariant across the traversal, loop, recursion, or DP transition.
  • Finally I will walk one edge case before writing the optimized version.

Common traps

  • Solving only the visible example instead of the invariant.
  • Forgetting empty input, singleton input, duplicate values, or impossible-state cases.
  • Revealing the solution before doing a dry run from the starter signature.

Follow-up drills

1. How do you turn this into a timed interview answer?

Start with the invariant, give the brute force in one sentence, name the optimized state, code the core loop or recursion, and run one visible test aloud before mentioning complexity.

2. How do you scale the same pattern to a larger input?

Track which state grows with the input: hash maps and sets grow with distinct values, queues grow with frontier width, recursion grows with depth, heaps grow with active candidates, and DP tables grow with state count.

3. What should you practice from blank tomorrow?

Rewrite median_finder_session without looking at the solution, then compare only the invariant and state updates before checking syntax.

Interactive runner

Write the required Python function. Your code runs locally in this browser. Hints reveal one failing case at a time.