11. Implement a One-Pass Reservoir-Sampling Solution in Python
Given values with tied maximums, return a random maximum index in one pass using Python's random facilities and constant extra space.
I would scan the values once while keeping the current maximum, one selected index, and the number of maximum values seen. A larger value resets the selected index and count. A tied value replaces the selected index with probability one divided by the updated count. This gives every maximum index an equal chance and uses constant extra space.
See the Code while reading this explanation.
The practical solution is to apply reservoir sampling only to indexes whose values equal the largest value seen so far. During one scan, keep the current maximum, the selected index, the number of tied maximums seen, and a flag that records whether the iterable contained an item.
- Should I focus on Python language behavior, or also explain the runtime and standard library?
- Which Python version and execution environment should I assume?
- Would you like a small code example together with production tradeoffs and edge cases?
The first value becomes the current maximum. When a later value is larger, it becomes the new maximum, its index becomes the selected index, and the count resets to one. When a value equals the current maximum, increase the count and replace the selected index with probability one divided by that count. In Python, calling randrange with the count and checking whether the result is zero gives exactly that probability.
For values [4, 9, 2, 9, 9], indexes 1, 3, and 4 each have probability one third. The function works with lists, generators, and other one pass iterables because it never reads an item twice. Empty input raises ValueError. Values must support consistent greater than and equality comparisons. Floating point NaN needs an explicit policy because its normal comparisons do not define a usable maximum. The algorithm takes linear time and constant extra memory.
The function uses enumerate to obtain each value and its zero based index during one iteration. The first item initializes the current maximum, selected index, and maximum count. A larger value replaces the current maximum and resets the count to one because all earlier candidates are no longer maximum values. An equal value increases the count. The new index replaces the previous selection only when randrange returns zero. After k tied maximum values have been processed, each matching index has probability one divided by k of being selected. Only a fixed number of variables are stored, so the extra memory cost remains constant.
import random
from collections.abc import Iterable
from typing import Any
def random_max_index(
values: Iterable[Any],
rng: random.Random | None = None,
) -> int:
"""Return a uniformly random index among all maximum values."""
# Use the supplied generator for repeatable tests.
# Otherwise create a local generator for this call.
random_source = rng if rng is not None else random.Random()
# Store only constant extra state.
has_value = False
current_max: Any = None
chosen_index = 0
maximum_count = 0
# Read every input value exactly once.
for index, value in enumerate(values):
if not has_value:
# The first value is the first maximum candidate.
current_max = value
chosen_index = index
maximum_count = 1
has_value = True
elif value > current_max:
# A larger value removes all earlier candidates.
current_max = value
chosen_index = index
maximum_count = 1
elif value == current_max:
# This index is another maximum candidate.
maximum_count += 1
# Select this index with probability 1 / maximum_count.
if random_source.randrange(maximum_count) == 0:
chosen_index = index
if not has_value:
raise ValueError("values must contain at least one item")
return chosen_index
if __name__ == "__main__":
sample = [4, 9, 2, 9, 9]
# A fixed seed makes this example repeatable.
seeded_rng = random.Random(7)
selected_index = random_max_index(sample, seeded_rng)
print("Selected index:", selected_index)
print("Selected value:", sample[selected_index])This pattern is useful for large files, database result streams, generators, event streams, and telemetry pipelines where storing every matching index would waste memory. It can select one representative record uniformly from all records that share the largest score. In production, passing a dedicated random generator makes tests repeatable and prevents test code from changing shared random state.
Interviewers ask this question to test whether a candidate can process an iterable exactly once, use Python random facilities correctly, keep constant extra state, and explain why every maximum index has the same selection probability. It also tests careful handling of iterators, empty input, tied values, comparison behavior, and testable randomness.
A common mistake is storing every maximum index in a list. That is correct for selection fairness but can use linear extra memory. Another mistake is replacing the chosen index with probability one half for every tie. That makes later indexes more likely. Candidates may also forget to reset the count when a larger value appears, scan the iterable twice, return the maximum value instead of its index, or ignore empty input. Another mistake is assuming NaN follows ordinary maximum comparison rules.
State the invariant clearly. After processing any prefix with k occurrences of its maximum value, each of those k indexes has probability one divided by k of being selected. Then explain the reset for a larger value, the replacement rule for a tie, and the linear time with constant extra memory.









