Multiples stream

Two related tasks hide under this title. Ask which one, then offer the other as a follow-up.

1. Multiples of a set. Given positive integers bases = [3, 5], yield every positive number divisible by at least one base, in increasing order, without duplicates: 3, 5, 6, 9, 10, 12, 15, 18, …
2. Numbers built from a set of primes (Hamming numbers for [2, 3, 5]): every number whose prime factors all come from the set: 1, 2, 3, 4, 5, 6, 8, 9, 10, 12, …

Both streams are infinite, so the answer is a generator that does a small amount of work per value. Also ask whether "the n-th value" queries come without iterating (part 2 below).

Multiples: a heap of next multiples. Each base b contributes the arithmetic sequence b, 2b, 3b, …. Merging k sorted sequences is a heap problem: keep (next_multiple, base) for each base, pop the smallest, push next_multiple + base. Skip a popped value equal to the last yielded one (15 comes from both 3 and 5). Each output costs O(log k), and memory is O(k).

Hamming numbers: one pointer per prime. Every Hamming number other than 1 is a smaller Hamming number times one of the primes. Keep the output list h and, for each prime p, an index i_p into h. The next value is min(p × h[i_p]); advance every pointer that produced it (advancing all of them is what removes duplicates like 6 = 2 × 3 = 3 × 2). O(k) per value; it does keep the generated values in memory, but only from the smallest pointer onward is needed, so trim the list's front.

The n-th multiple directly (part 2). Count the multiples up to x with inclusion and exclusion over subsets of the bases: count(x) = Σ over non-empty subsets S of (−1)^(|S|+1) · floor(x / lcm(S)). count is non-decreasing in x, so binary-search the smallest x with count(x) >= n. That is O(2^k · log(n · min(bases))): instant for a handful of bases, even for n = 10^18.

import heapq
from math import gcd
from itertools import combinations
from typing import Dict, Iterator, List, Sequence


def multiples(bases: Sequence[int]) -> Iterator[int]:
    """Every positive number divisible by some base, increasing, no duplicates."""
    if not bases or any(b <= 0 for b in bases):
        raise ValueError("bases must be positive")
    heap = [(b, b) for b in sorted(set(bases))]
    heapq.heapify(heap)
    last = 0
    while True:
        value, base = heapq.heappop(heap)
        heapq.heappush(heap, (value + base, base))
        if value != last:
            last = value
            yield value


def smooth_numbers(primes: Sequence[int]) -> Iterator[int]:
    """Numbers whose prime factors all lie in `primes`, increasing (1 first)."""
    primes = sorted(set(primes))
    h: List[int] = [1]
    offset = 0                    # h[0] is the value with global index `offset`
    idx: Dict[int, int] = {p: 0 for p in primes}
    yield 1
    while True:
        nxt = min(p * h[idx[p] - offset] for p in primes)
        for p in primes:
            if p * h[idx[p] - offset] == nxt:
                idx[p] += 1
        h.append(nxt)
        yield nxt
        lowest = min(idx.values())
        if lowest - offset > 1024:        # drop values no pointer can reach again
            del h[: lowest - offset]
            offset = lowest


def _lcm(a: int, b: int) -> int:
    return a // gcd(a, b) * b


def count_multiples(x: int, bases: Sequence[int]) -> int:
    """How many numbers in [1, x] are divisible by at least one base."""
    bases = sorted(set(bases))
    total = 0
    for r in range(1, len(bases) + 1):
        for subset in combinations(bases, r):
            l = 1
            for b in subset:
                l = _lcm(l, b)
                if l > x:
                    break
            if l <= x:
                total += (x // l) * (1 if r % 2 else -1)
    return total


def nth_multiple(n: int, bases: Sequence[int]) -> int:
    lo, hi = 1, n * min(bases)    # the n-th multiple of the smallest base is an upper bound
    while lo < hi:
        mid = (lo + hi) // 2
        if count_multiples(mid, bases) >= n:
            hi = mid
        else:
            lo = mid + 1
    return lo

For [2, 3, 5] the gap between pointers grows slowly: generating the millionth Hamming number keeps only a few thousand values once the front is trimmed.

1. Many bases. Inclusion-exclusion is 2^k. With 30 bases, prune subsets whose lcm already exceeds x (the code breaks early), and drop any base that is a multiple of another base first: it adds nothing.
2. Merging streams in general. The multiples generator is the k-way merge of sorted streams. The same heap pattern merges sorted event logs, paginated API results or sorted files.
3. Concurrency. Two consumers sharing one infinite generator need a lock around next, or each consumer gets its own generator. Generators are not thread-safe.
4. Overflow in other languages. In Java or C++, p * h[i] and value + base overflow 64 bits for large indexes; bound the stream or use 128-bit math.
5. Inside an interview loop. Interviewers often start with "multiples of 3 or 5 up to N" and ask for a stream, then the n-th term. Recognizing the shift from enumeration to counting is what they are looking for.

Mid-level (L4)

• A working generator for multiples with the heap or a per-base counter scan, without duplicates.
• Yields lazily and handles duplicate bases.
• Tests against a brute-force filter.

Senior (L5)

• O(log k) per value with a heap and a clear statement of memory use.
• Solves the Hamming variant with one pointer per prime and explains why advancing all matching pointers removes duplicates.
• Validates input and keeps memory bounded on long runs.

Staff (L6+)

• Gets the n-th value without iterating, with inclusion-exclusion and binary search, and states its cost.
• Connects the stream to k-way merges of real data and discusses thread safety and overflow.
• Chooses enumeration or counting based on the queries the caller makes.