SWE · Coding

多个已排序区间流求并集:堆归并与在线合并

每个流按起点非降,只保存各流一个区间,输出规范化并集。

题目重点:每个流按起点非降,只保存各流一个区间,输出规范化并集。

题意与假设

这道练习题保留了面经中值得训练的核心约束,并将需要澄清的条件单独列出。现场拿到变体时,先和面试官确认这些条件。

  • 本文使用闭区间[a,b],a≤b;相同端点相接时合并。整数相邻如[1,2]和[3,4]默认不合并。
  • 流必须分别按起点排序;内部可重叠。输出生成器可能在后续发现非法输入前已经产出,生产接口要约定验证或部分结果语义。

解题路线

  1. 把各流头区间放进按起点排序的最小堆,每弹一个只从同一流补一个。所有未读元素不会早于该流头,因此全局弹出顺序正确。
  2. 保存当前合并区间。新起点≤当前终点时扩大终点;否则当前区间以后不可能再被连接,可安全输出。
  3. 双流可用双指针O(N),不必建堆;K流的堆将排序成本从O(NlogN)降为O(NlogK),前提是输入流已经排序。

手动走一遍

两流[(1,4),(8,9)]和[(3,6),(9,12)]按1、3、8、9弹出,先合成[1,6]再合成[8,12]。一个流内部[(1,10),(2,3)]也能正确处理;未排序[(5,6),(1,2)]需报错而非静默产生错并集。

Python 参考实现

下面的代码按上文假设实现,适合本地运行并追加自己的测试用例。

代码区可横向滚动;使用键盘时,先聚焦代码区再按方向键查看。

from heapq import heappush, heappop


def union_streams(streams):
    iters = [iter(s) for s in streams]
    heap = []
    for i, it in enumerate(iters):
        pair = next(it, None)
        if pair is not None:
            a, b = pair
            if a > b:
                raise ValueError("reversed interval")
            heappush(heap, (a, b, i))
    current = None
    while heap:
        a, b, i = heappop(heap)
        if current is None:
            current = (a, b)
        elif a <= current[1]:
            current = (current[0], max(current[1], b))
        else:
            yield current
            current = (a, b)
        pair = next(iters[i], None)
        if pair is not None:
            x, y = pair
            if x < a or x > y:
                raise ValueError("stream must be start-sorted with valid intervals")
            heappush(heap, (x, y, i))
    if current is not None:
        yield current

复杂度

O(NlogK)时间,K=1退化O(N);O(K)工作空间,另加输出缓冲/消费者存储。迭代器也会保留K个输入引用。

易错点与追问

容易失分的边界

  • 不要用a≤end+1代替闭区间重叠,除非约定整数集合邻接合并。
  • 外部数据仍需考虑读取I/O和背压;算法内存小不代表源迭代器无缓存。

继续深挖

  • 半开区间[a,b)接触时是否合并?零长区间是否忽略?
  • 如何输出被恰好一个区间覆盖的部分,而不是并集?
✓已复制微信号!