Reasoning Into CodeArunank Divya Sharan · Python edition← Arunank’s portfolio
Module 09 · Queues, Heaps, and Streaming Frontiers

085. Smallest K Values from a Stream

Bridge · Download this solution · All solutions and shared helpers

Problem

Consume an integer stream and return its smallest min(k,n) values in ascending order, retaining duplicates. k<=0 returns []. Use a heap of at most k candidates rather than sorting or storing the full stream.

Examples

Example 1

smallest_k([4,5,8,2,10], 3)
=> [2, 4, 5]

Large candidates are evicted.

Example 2

smallest_k([-3,2,-1], 2)
=> [-3, -1]

Negation works for negative original values too.

Example 3

smallest_k([], 3)
=> []

No candidate arrives.

Reveal the worked solution and reasoning

Why this mechanism works

For the smallest k, the eviction candidate is the largest retained value. Python's standard min-heap can act as a max-oriented heap by storing negated integers. Its smallest stored number then represents the largest original candidate.

State and boundaries

heap stores negatives of the smallest at-most-k consumed values. -heap[0] is the largest retained candidate.

Protocol

  1. Negate each incoming value before pushing.
  2. Remove the root when capacity is exceeded.
  3. Negate retained values back.
  4. Sort them ascending for output.

Visualization

original winners: 1,3,5
stored negatives: -5,-3,-1
root -5 represents the worst retained winner 5

Python implementation

import heapq

def smallest_k(values, k):
    if k <= 0:
        return []
    heap = []
    for value in values:
        heapq.heappush(heap, -value)
        if len(heap) > k:
            heapq.heappop(heap)
    return sorted(-value for value in heap)

Worked execution

Recorded state changes after statements; L numbers refer to the solution file. Long traces are explicitly condensed.

smallest_k([4,5,8,2,10], 3)

L4 smallest_k: if k <= 0:
values=[4, 5, 8, 2, 10]; k=3

L6 smallest_k: heap = []
values=[4, 5, 8, 2, 10]; k=3; heap=[]

L7 smallest_k: for value in values:
values=[4, 5, 8, 2, 10]; k=3; heap=[]; value=4

L8 smallest_k: heapq.heappush(heap, -value)
values=[4, 5, 8, 2, 10]; k=3; heap=[-4]; value=4

L7 smallest_k: for value in values:
values=[4, 5, 8, 2, 10]; k=3; heap=[-4]; value=5

L8 smallest_k: heapq.heappush(heap, -value)
values=[4, 5, 8, 2, 10]; k=3; heap=[-5, -4]; value=5

L7 smallest_k: for value in values:
values=[4, 5, 8, 2, 10]; k=3; heap=[-5, -4]; value=8

L8 smallest_k: heapq.heappush(heap, -value)
values=[4, 5, 8, 2, 10]; k=3; heap=[-8, -4, -5]; value=8

... 6 intermediate state changes omitted ...

L11 <genexpr>: return sorted(-value for value in heap)
.0=list_iterator; value=-5

L11 <genexpr>: return sorted(-value for value in heap)
.0=list_iterator; value=-4

L11 <genexpr>: return sorted(-value for value in heap)
.0=list_iterator; value=-2

L11 <genexpr>: return sorted(-value for value in heap)
.0=list_iterator; value=-2

Result: [2, 4, 5]. Large candidates are evicted.

smallest_k([-3,2,-1], 2)

L4 smallest_k: if k <= 0:
values=[-3, 2, -1]; k=2

L6 smallest_k: heap = []
values=[-3, 2, -1]; k=2; heap=[]

L7 smallest_k: for value in values:
values=[-3, 2, -1]; k=2; heap=[]; value=-3

L8 smallest_k: heapq.heappush(heap, -value)
values=[-3, 2, -1]; k=2; heap=[3]; value=-3

L7 smallest_k: for value in values:
values=[-3, 2, -1]; k=2; heap=[3]; value=2

L8 smallest_k: heapq.heappush(heap, -value)
values=[-3, 2, -1]; k=2; heap=[-2, 3]; value=2

L7 smallest_k: for value in values:
values=[-3, 2, -1]; k=2; heap=[-2, 3]; value=-1

L8 smallest_k: heapq.heappush(heap, -value)
values=[-3, 2, -1]; k=2; heap=[-2, 3, 1]; value=-1

L10 smallest_k: heapq.heappop(heap)
values=[-3, 2, -1]; k=2; heap=[1, 3]; value=-1

L11 <genexpr>: return sorted(-value for value in heap)
.0=list_iterator; value=1

L11 <genexpr>: return sorted(-value for value in heap)
.0=list_iterator; value=3

L11 <genexpr>: return sorted(-value for value in heap)
.0=list_iterator; value=3

Result: [-3, -1]. Negation works for negative original values too.

Why the solution is correct

Removing the minimum negated value removes the maximum original value from the candidate pool. The remaining k values are therefore the smallest among the previous winners and the new arrival. This preserves the desired pool inductively.

Time and space

O(n log(k+1)+k log k) time and O(k) working space.

Common mistakes

Negating only on insertion but not on output returns the wrong values. A plain min-heap would discard precisely the small values being sought.

Reasoning card

State: heap stores negatives of the smallest at-most-k consumed values. -heap[0] is the largest retained candidate.

Invariant: Heap contains the smallest min(k,n_seen) values.

Decision to rehearse: Incoming values compete with the largest current winner.

Finish: Sort them ascending for output

Transfer exercise

Abstract the winner-pool ordering so the same bounded-heap mechanism can retain either extreme.