완전 이진 트리의 배열 표현과 힙 불변식, 삽입·삭제·heapify의 원리를 설명하고 동률·갱신·취소·상위 k개·다중 병합·최단 거리 예제를 검증한다.
개요: 완전 이진 트리에 우선순위를 유지한다
이진 힙(binary heap)은 완전 이진 트리의 모양과 부모·자식 간 우선순위 규칙을 함께 유지하는 자료구조다. 최소 힙에서는 부모가 자식보다 크지 않고, 최대 힙에서는 부모가 자식보다 작지 않다. 이 국소 규칙 덕분에 루트에서 전체의 최솟값 또는 최댓값을 얻을 수 있다.
작업 중 마감 시각이 가장 빠른 것을 반복해서 고르거나, 여러 정렬된 입력에서 다음으로 작은 값을 고를 때 사용할 수 있다. 모든 원소를 늘 정렬해 두는 대신 다음 극값을 효율적으로 선택하는 데 초점을 맞춘다.
우선순위 큐는 우선순위에 따라 꺼내는 연산 계약이고 이진 힙은 그 구현 방법 중 하나다. 다른 힙 구조나 정렬된 컨테이너로도 우선순위 큐를 구현할 수 있다. 메모리 할당 영역을 뜻하는 heap과도 다른 개념이다.
이 문서는 배열 인덱스 표현, 삽입과 루트 제거, bottom-up heapify의 비용, 동률 처리·갱신·취소·상위 k개·다중 병합을 설명한다. Python 3.9.6에서 실행했으며 모든 코드가 최신 버전 전용 API를 요구하지 않도록 기본 최소 힙 연산을 사용한다.
완전 이진 트리와 배열 인덱스
완전 이진 트리는 마지막 층을 제외한 층이 가득 차고 마지막 층은 왼쪽부터 채워진 형태다. 모든 노드가 자식 둘을 가져야 한다는 뜻이 아니며 마지막 노드에 형제나 자식이 없을 수 있다. 이 형태를 유지하면 높이가 O(log n)으로 제한된다.
노드를 너비 우선 순서로 배열에 놓으면 인덱스 i의 왼쪽 자식은 2i+1, 오른쪽 자식은 2i+2다. 루트가 아닌 노드의 부모는 (i−1)//2다. 자식 인덱스가 길이 이상이면 해당 자식이 없다.
아래 길이 6에서 루트 0의 자식은 1,2이고 노드 2의 자식은 5 하나다. 자식이 항상 둘이라는 가정으로 오른쪽을 읽으면 범위를 벗어날 수 있다. 마지막 내부 노드는 n//2−1이고 그 뒤는 잎이다.
포인터로 실제 트리 노드를 만들 필요가 없다는 점이 이진 힙 배열 표현의 장점이다. 다만 배열의 평탄한 순서가 우선순위 전체 순서와 같은 것은 아니다. 인덱스는 트리의 모양을 나타낸다.
예제 코드
size = 6
for index in range(size):
children = [c for c in (2 * index + 1, 2 * index + 2) if c < size]
parent = None if index == 0 else (index - 1) // 2
print(index, parent, children)
실행 결과
0 None [1, 2]
1 0 [3, 4]
2 0 [5]
3 1 []
4 1 []
5 2 []힙은 정렬된 배열이 아니다
최소 힙은 각 부모가 자식보다 크지 않다는 관계만 보장한다. 서로 다른 하위 트리의 노드 사이에는 전체 순서가 정해지지 않는다. [1,4,2,7,5,3]은 유효한 최소 힙이지만 배열 자체는 오름차순이 아니다.
루트에서 임의 노드로 내려가는 경로의 값은 감소하지 않는다. 따라서 루트가 모든 노드보다 크지 않다는 결론은 얻을 수 있다. 그러나 두 번째·세 번째로 작은 값의 배열 위치가 고정된다는 결론은 얻을 수 없다.
힙을 단순 순회해서 출력하면 정렬 결과를 얻는 것이 아니다. 값을 하나씩 pop하거나 별도로 정렬해야 한다. Java PriorityQueue의 반복 순서 같은 라이브러리 계약에서도 이 점을 확인해야 한다.
아래는 모든 자식과 부모 관계를 검사하고 sorted와 비교한다. 힙 조건은 True, 배열 정렬 여부는 False다. “힙 불변식 유지”와 “전체 배열 정렬”을 구분하는 가장 작은 실용적인 확인 방법이다.
예제 코드
heap = [1, 4, 2, 7, 5, 3]
print(all(heap[(i - 1) // 2] <= heap[i] for i in range(1, len(heap))))
print(heap == sorted(heap))
print(heap[0], min(heap))
실행 결과
True
False
1 1삽입: 끝에 붙인 뒤 위로 올린다
완전 이진 트리의 모양을 유지하려면 새 원소를 배열 끝에 추가한다. 기존 노드의 부모·자식 관계는 그대로이고 새 노드와 그 부모 사이에서만 힙 조건이 깨질 수 있다. 새 값이 부모보다 작으면 교환하고 부모 위치에서 같은 검사를 반복한다.
이 과정을 sift-up 또는 상향 이동이라고 한다. 한 번 교환할 때마다 한 층 올라가므로 교환 횟수는 높이 O(log n)을 넘지 않는다. 동등한 값에서는 이동을 멈추어도 최소 힙 조건을 만족한다.
예를 들어 [2,5,4,9,7]에 1을 붙이면 인덱스 5의 부모 인덱스 2 값 4와 교환하고, 이어서 루트 2와 교환한다. 최종 배열은 [1,5,2,9,7,4]다. 전체를 정렬하지 않고 한 경로만 복구한다.
배열 append가 용량 확장을 일으킬 때의 복사 비용은 힙 비교·교환 비용과 별개다. 이진 힙의 상향 경로 O(log n)과 기반 동적 배열의 특정 재할당 지연을 구분해야 한다. 비교 연산이 큰 객체를 읽는다면 그 비용도 포함한다.
예제 코드
data = [2, 5, 4, 9, 7]
data.append(1)
index = len(data) - 1
trace = []
while index > 0:
parent = (index - 1) // 2
if data[parent] <= data[index]:
break
data[parent], data[index] = data[index], data[parent]
trace.append((index, parent))
index = parent
print(trace)
print(data)
실행 결과
[(5, 2), (2, 0)]
[1, 5, 2, 9, 7, 4]루트 제거: 마지막 값으로 메운 뒤 내려보낸다
루트를 제거한 뒤 배열 첫 원소를 단순 삭제하면 나머지 인덱스가 모두 당겨져 트리 구조가 달라진다. 대신 마지막 원소를 꺼내 루트 자리에 놓으면 완전 이진 트리의 모양을 유지하면서 루트에서만 힙 조건을 복구하면 된다.
두 자식이 있으면 더 작은 자식을 고른다. 부모가 그 자식보다 크면 교환하고 아래로 내려간다. 더 큰 자식을 선택하면 교환 후 부모가 다른 작은 자식보다 커져 조건이 깨질 수 있다. 오른쪽 자식이 없는 경우도 별도로 처리한다.
한 원소만 있을 때는 마지막 값을 꺼내면 이미 빈 배열이므로 루트에 다시 쓰지 않는다. 빈 힙 pop은 오류다. 이 두 경계는 일반적인 여러 원소 사례만 검사하면 놓치기 쉽다.
내려가는 경로 역시 O(log n) 길이다. 조회 peek는 루트만 읽지만 pop은 구조를 바꾸고 복구하므로 비용이 다르다. 극값을 읽는 연산과 제거하는 연산을 같은 것으로 설명하지 않는다.
최소 이진 힙 전체 구현
아래 MinHeap은 생성 시 입력을 복사하고 bottom-up 방식으로 힙을 만든다. peek, push, pop을 제공하며 빈 조회·제거는 IndexError다. 비교에는 <만 사용해 자식이 더 작을 때 교환한다.
_down은 더 작은 자식과 비교하여 필요할 때 내려간다. push는 새 노드부터 부모와 비교한다. pop은 루트 값을 저장하고 마지막 값으로 메운 뒤 _down을 호출한다. 데이터 값들은 일관된 순서 비교가 가능해야 한다.
data를 공개한 것은 교육과 검증 편의를 위한 것이다. 외부에서 임의로 바꾸면 힙 조건이 깨질 수 있으므로 서비스 API에서는 직접 수정을 제한하는 것이 적절하다. 사용자 정의 비교가 중간에 예외를 던지는 상황의 강한 예외 안전성까지 구현한 코드는 아니다.
아래 값들을 모두 꺼내면 오름차순 결과가 된다. 같은 값 1도 두 번 보존되며 음수도 정상적으로 처리된다. 힙은 집합이 아니므로 중복 값을 자동 제거하지 않는다.
예제 코드
class MinHeap:
def __init__(self, values=()):
self.data = list(values)
for index in range(len(self.data) // 2 - 1, -1, -1):
self._down(index)
def __len__(self):
return len(self.data)
def peek(self):
if not self.data:
raise IndexError('heap empty')
return self.data[0]
def push(self, value):
self.data.append(value)
index = len(self.data) - 1
while index > 0:
parent = (index - 1) // 2
if not self.data[index] < self.data[parent]:
break
self.data[index], self.data[parent] = self.data[parent], self.data[index]
index = parent
def _down(self, index):
size = len(self.data)
while 2 * index + 1 < size:
child = 2 * index + 1
right = child + 1
if right < size and self.data[right] < self.data[child]:
child = right
if not self.data[child] < self.data[index]:
break
self.data[index], self.data[child] = self.data[child], self.data[index]
index = child
def pop(self):
if not self.data:
raise IndexError('heap empty')
answer = self.data[0]
last = self.data.pop()
if self.data:
self.data[0] = last
self._down(0)
return answer
heap = MinHeap([5, 1, 4, 1, 9])
heap.push(-2)
print(heap.peek(), len(heap))
print([heap.pop() for _ in range(len(heap))])
try:
heap.pop()
except IndexError:
print('empty rejected')
실행 결과
-2 6
[-2, 1, 1, 4, 5, 9]
empty rejectedbottom-up heapify는 왜 O(n)인가
배열을 힙으로 만들 때 모든 원소를 차례로 push하면 O(n log n)의 상한을 얻는다. bottom-up heapify는 마지막 내부 노드부터 루트까지 _down을 수행한다. 각 노드를 처리할 때 두 자식의 하위 트리는 이미 힙이다.
모든 노드가 log n만큼 내려가는 것은 아니다. 절반 정도는 잎이라 내려가지 않고, 그 위 층은 최대 한 칸, 더 위 층은 최대 두 칸을 내려간다. 내려갈 수 있는 높이가 큰 노드의 수는 빠르게 줄어든다.
높이 h를 가진 노드 수를 O(n/2^(h+1))로 잡으면 총 작업은 O(n·Σ h/2^(h+1))다. 이 급수는 상수로 수렴하므로 O(n)이다. 각 호출의 최악 O(log n)을 n번 곱하는 상한은 가능하지만 더 느슨한 분석이다.
아래 완전한 15노드 트리는 높이 3 노드 1개, 높이 2 노드 2개, 높이 1 노드 4개, 잎 8개다. 각 노드가 최대 높이만큼 내려간다고 세어도 합은 11이다. 실제 비교 횟수는 자식 선택 등으로 달라지지만 선형 합의 구조를 보여 준다.
예제 코드
counts = {0: 8, 1: 4, 2: 2, 3: 1}
print(sum(counts.values()))
print(sum(height * count for height, count in counts.items()))
실행 결과
15
11최소 힙·최대 힙과 비교 방향
최소 힙은 작은 값이 먼저 나오고 최대 힙은 큰 값이 먼저 나온다. “우선순위 숫자가 크다”가 더 급한 작업인지 덜 급한 작업인지는 응용 프로그램이 정하는 규칙이다. 자료구조 이름만 보고 업무 우선순위 의미를 추측하지 않는다.
Python의 기본 heapq는 최소 힙을 사용하며 정수 점수의 최대값을 얻기 위해 부호를 바꾸는 방법을 쓸 수 있다. 아래에서는 −점수를 넣고 꺼낼 때 다시 부호를 바꾼다. Python 정수에서는 최소 고정 폭 정수의 부호 반전 오버플로 문제가 없지만 다른 언어에 그대로 옮길 때는 범위를 확인한다.
문자열이나 복잡한 객체를 무조건 음수로 바꿀 수는 없다. 비교자 또는 별도의 정렬 키를 설계해야 한다. 부동소수점 NaN처럼 보통의 순서 관계를 만족하지 않는 값도 정책이 필요하다.
Rust BinaryHeap은 기본적으로 최대 힙이며 다른 언어의 우선순위 큐 기본 방향도 확인해야 한다. Python의 더 새로운 최대 힙 API는 실행 버전 지원 여부를 확인하고 사용한다. 여기서는 3.9.6에서 검증 가능한 기본 API로 설명한다.
예제 코드
import heapq
scores = [7, 2, 10, 7]
heap = [-score for score in scores]
heapq.heapify(heap)
print([-heapq.heappop(heap) for _ in range(len(heap))])
실행 결과
[10, 7, 7, 2]동률 처리와 비교할 수 없는 작업 객체
우선순위가 같을 때 먼저 들어온 작업을 먼저 처리하려면 삽입 순번을 별도 비교 키로 넣는다. (우선순위,순번,작업)의 튜플은 우선순위가 같으면 순번을 비교한다. 순번이 유일하면 마지막 작업 객체끼리 비교할 필요가 없다.
(우선순위,사전 객체)만 넣으면 동률에서 Python이 사전끼리 대소 비교하려 하여 TypeError가 날 수 있다. 서로 다른 작업 타입을 섞을 때도 마찬가지다. 우선순위 큐의 내부 비교 대상과 실제 작업 데이터를 분리한다.
아래 A와 B의 우선순위는 같지만 A가 먼저 들어왔으므로 A가 먼저 나온다. C는 더 작은 우선순위 값이라 먼저 나온다. 이 안정성은 이진 힙이 자동 제공하는 것이 아니라 순번이라는 추가 계약으로 만든 것이다.
우선순위를 갱신했을 때 원래 순번을 유지할지 새 순번을 줄지도 정책이다. 뒤의 갱신 큐는 매 갱신에 새 순번을 주므로 같은 우선순위라면 최근 갱신된 작업이 뒤로 간다. 두 정책은 서로 다르므로 명시한다.
예제 코드
import heapq
from itertools import count
sequence = count()
pending = []
for priority, name in [(2, 'A'), (2, 'B'), (1, 'C')]:
heapq.heappush(pending, (priority, next(sequence), {'name': name}))
print([heapq.heappop(pending)[2]['name'] for _ in range(len(pending))])
try:
heapq.heapify([(1, {'name': 'A'}), (1, {'name': 'B'})])
except TypeError:
print('payload comparison rejected')
실행 결과
['C', 'A', 'B']
payload comparison rejected힙 내부의 우선순위를 직접 바꾸면 안 된다
힙은 삽입 시 비교된 순서를 바탕으로 배치한다. 저장된 객체의 우선순위를 나중에 직접 바꿔도 그 객체가 자동으로 위나 아래로 이동하지 않는다. 루트보다 더 급한 항목이 아래에 남을 수 있다.
아래 [우선순위,이름] 항목 C의 값을 3에서 0으로 바꿨지만 루트는 여전히 A다. heapify로 다시 만들면 C가 앞으로 온다. heapify는 O(n)이므로 우선순위 갱신이 잦은 상황에서 매번 전체 재구성이 최선은 아닐 수 있다.
알려진 인덱스의 키를 바꾸고 적절한 방향으로 sift하는 구현, 키→현재 위치를 관리하는 인덱스 힙, 새 항목을 추가하고 옛 항목을 무시하는 지연 갱신 등 방법이 있다. 필요한 연산 빈도와 메모리 한계에 따라 선택한다.
비교 키가 외부 상태나 시간에 따라 저절로 변하는 객체도 같은 문제를 일으킨다. 비교할 때마다 결과가 달라지면 힙의 불변식 자체를 안정적으로 정의하기 어렵다. 저장 중의 순서 기준을 명확히 고정한다.
예제 코드
import heapq
pending = [[1, 'A'], [2, 'B'], [3, 'C']]
heapq.heapify(pending)
pending[2][0] = 0
print(pending[0][1])
heapq.heapify(pending)
print(pending[0][1])
실행 결과
A
C지연 갱신과 취소를 구현하기
UpdatingQueue는 키별 최신 (우선순위,버전)을 current 사전에 저장하고 힙에는 모든 갱신 항목을 추가한다. pop할 때 힙 항목이 current와 일치하면 유효하고, 일치하지 않으면 오래되었거나 취소된 항목이므로 건너뛴다.
버전은 전체에서 유일한 증가 순번이다. 같은 키가 취소된 뒤 같은 우선순위로 다시 등록되어도 이전 항목과 구별된다. 우선순위만 비교하면 오래된 항목이 새 작업으로 잘못 인정될 수 있으므로 버전을 함께 검사한다.
cancel은 논리 사전에서 제거하며 물리 힙을 즉시 검색하지 않는다. 없는 키 취소는 KeyError다. pop은 오래된 항목을 여러 개 버릴 수 있으므로 개별 호출 비용을 무조건 O(log 활성 항목 수)로 설명하면 안 된다. 물리 힙의 크기가 기준이다.
rebuild는 최신 항목만 모아 선형 heapify로 정리한다. 아래는 A를 5에서 1로 바꾸고 B를 취소한 뒤 A만 꺼낸다. 남은 옛 항목을 건너뛴 뒤 논리적으로 비었다는 오류를 반환한다.
예제 코드
class UpdatingQueue:
def __init__(self):
from itertools import count
self.heap = []
self.current = {}
self.sequence = count()
def set_priority(self, key, priority):
import heapq
version = next(self.sequence)
self.current[key] = (priority, version)
heapq.heappush(self.heap, (priority, version, key))
def cancel(self, key):
del self.current[key]
def pop(self):
import heapq
while self.heap:
priority, version, key = heapq.heappop(self.heap)
if self.current.get(key) == (priority, version):
del self.current[key]
return key, priority
raise KeyError('queue empty')
def rebuild(self):
import heapq
self.heap = [(priority, version, key)
for key, (priority, version) in self.current.items()]
heapq.heapify(self.heap)
queue = UpdatingQueue()
queue.set_priority('A', 5)
queue.set_priority('B', 2)
queue.set_priority('A', 1)
queue.cancel('B')
print(queue.pop())
try:
queue.pop()
except KeyError:
print('logically empty')
queue.set_priority('A', 3)
queue.set_priority('A', 3)
queue.rebuild()
print(len(queue.heap), queue.pop())
실행 결과
('A', 1)
logically empty
1 ('A', 3)물리 항목 수와 논리 작업 수를 구분하기
지연 갱신에서는 활성 작업이 1개여도 같은 키를 수천 번 갱신하면 힙에 수천 항목이 남을 수 있다. current의 길이는 논리 작업 수이고 heap의 길이는 물리 대기 항목 수다. 메모리와 pop 지연은 후자에 영향을 받는다.
오래된 항목은 나중에 한 번씩 제거되지만 한 번의 pop에서 많은 항목을 건너뛸 수도 있다. 처리량의 분할 분석과 개별 요청의 최대 지연은 다르다. 대기 시간을 엄격히 제한해야 한다면 지연 삭제가 적합한지 확인한다.
물리 길이가 활성 길이의 일정 배수를 넘을 때 재구성하는 정책을 둘 수 있다. 재구성 자체는 O(활성 항목 수) 시간과 새 목록 공간을 쓰며 특정 호출에 지연이 모일 수 있다. 임계값은 고정된 정답이 아니라 워크로드와 지연 예산에 따른 선택이다.
키→위치 사전을 갖는 인덱스 힙은 물리 항목을 직접 수정·삭제할 수 있지만 모든 교환에서 사전 위치를 정확히 갱신해야 한다. 구현 복잡성과 오류 가능성을 감수하고 더 강한 공간·갱신 계약을 얻는 선택이다.
heappushpop과 heapreplace는 계약이 다르다
heappushpop은 새 값을 넣은 뒤 최솟값을 꺼낸 효과를 한 번에 처리한다. 새 값이 현재 최솟값보다 작으면 그 새 값을 바로 반환하고 기존 힙을 유지할 수 있다. 빈 힙에서도 새 값을 반환하고 힙은 비어 있다.
heapreplace는 기존 최솟값을 반드시 제거하고 새 값을 넣는다. 빈 힙이면 제거할 기존 항목이 없으므로 오류다. 새 값이 더 작아도 기존 루트를 반환하고 작은 새 값을 힙에 남긴다.
아래 [4,7,9]에 2를 적용하면 heappushpop은 2를 반환하고 [4,7,9]를 유지하지만 heapreplace는 4를 반환하고 2를 남긴다. 상위 k개 선택에서 아무 조건 없이 heapreplace를 사용하면 더 나쁜 후보로 좋은 값을 바꾸는 오류가 생긴다.
두 함수가 모두 “넣고 빼기”처럼 보인다고 같은 것으로 취급하지 않는다. 무엇을 반드시 반환해야 하는지, 크기가 유지되는지, 빈 상태가 허용되는지에 따라 적합한 연산이 달라진다.
예제 코드
import heapq
first, second = [4, 7, 9], [4, 7, 9]
print(heapq.heappushpop(first, 2), sorted(first))
print(heapq.heapreplace(second, 2), sorted(second))
empty = []
print(heapq.heappushpop(empty, 3), empty)
try:
heapq.heapreplace([], 3)
except IndexError:
print('replace empty rejected')
실행 결과
2 [4, 7, 9]
4 [2, 7, 9]
3 []
replace empty rejected상위 k개 선택에는 왜 최소 힙을 쓰는가
가장 큰 k개 값을 유지하려면 현재 선택된 k개 중 가장 작은 값이 교체 기준이다. 따라서 선택 집합을 최소 힙으로 두면 루트가 현재 통과 기준이 된다. 새 값이 루트보다 크면 루트를 교체하고 아니면 버린다.
불변식은 지금까지 읽은 값 중 큰 min(k,읽은 수)개를 힙에 보관한다는 것이다. 처음 k개까지는 그냥 넣고, 그 뒤에는 새 값이 기존 경계보다 좋을 때만 교체한다. 중복 값은 값의 개수만큼 보존한다.
k≥1에서 입력 n개의 처리 시간은 O(n log(k+1))로 쓸 수 있고 힙 공간은 O(min(n,k))다. 이 함수는 최종 결과를 내림차순 정렬하므로 그 비용도 포함한다. 힙 내부 배열 그대로 반환하면 정렬된 결과가 아니다.
k=0은 빈 결과를 즉시 반환하며 이 구현에서는 입력 이터레이터를 소비하지 않는다. 음수 k는 오류다. k가 n보다 크면 모든 값을 정렬해 반환한다. 정확한 경계 계약을 정해 두면 작은 입력에서 생기는 예외를 줄일 수 있다.
예제 코드
def largest_k(values, k):
import heapq
if k < 0:
raise ValueError('negative k')
if k == 0:
return []
kept = []
for value in values:
if len(kept) < k:
heapq.heappush(kept, value)
elif kept[0] < value:
heapq.heapreplace(kept, value)
return sorted(kept, reverse=True)
print(largest_k([5, 1, 8, 3, 8, 2], 3))
print(largest_k([2, 1], 5))
print(largest_k([2, 1], 0))
실행 결과
[8, 8, 5]
[2, 1]
[]여러 정렬된 스트림 병합
각 입력이 오름차순으로 정렬되어 있다면 전체에서 다음으로 작은 값은 각 입력의 현재 맨 앞 값 중 하나다. 힙에 각 스트림의 앞 값 하나씩만 넣고 최솟값을 꺼낸 스트림에서 다음 값을 읽어 넣는다.
아래 merge_streams는 iterator와 값을 함께 보관하고 동률 비교용 순번을 둔다. iterator 객체 자체를 비교하지 않는다. 빈 스트림은 처음부터 건너뛴다. 생성기이므로 결과를 요청하는 동안 다음 원소를 하나씩 생산한다.
전체 원소 수 N, 스트림 수 k에서 초기 스트림 검사 O(k)와 각 출력의 힙 작업을 합쳐 O(k+N log(k+1))로 분석할 수 있다. 힙은 활성 스트림당 한 항목이므로 O(k) 보조 공간이다. 출력 전체를 list로 모으면 추가로 O(N) 출력 공간이 필요하다.
각 입력이 이미 정렬되어 있다는 전제가 필수다. 이 함수는 입력을 정렬하거나 완전히 검증하지 않는다. 동률 항목의 스트림 간 순서는 여기서 사용한 순번 정책에 따른 것으로 다른 라이브러리의 안정성 계약과 동일하다고 주장하지 않는다.
예제 코드
def merge_streams(streams):
import heapq
from itertools import count
ticket = count()
pending = []
for stream in streams:
iterator = iter(stream)
try:
value = next(iterator)
except StopIteration:
continue
heapq.heappush(pending, (value, next(ticket), iterator))
while pending:
value, _, iterator = heapq.heappop(pending)
yield value
try:
following = next(iterator)
except StopIteration:
continue
heapq.heappush(pending, (following, next(ticket), iterator))
streams = [[1, 4, 7], [], [1, 3, 8], [2, 6]]
print(list(merge_streams(streams)))
print(list(merge_streams([])))
실행 결과
[1, 1, 2, 3, 4, 6, 7, 8]
[]다익스트라에서 오래된 거리 항목을 건너뛰기
음이 아닌 가중치 그래프의 다익스트라는 현재 알려진 최소 거리 후보를 우선순위 큐에서 꺼낸다. 더 짧은 경로를 찾으면 새 거리 항목을 넣고 기존 항목은 남겨 둘 수 있다. 꺼낸 거리가 현재 사전의 거리와 다르면 오래된 항목으로 건너뛴다.
아래 예제에서는 A→B 비용 8을 먼저 알지만 A→C→B 비용 3으로 개선한다. 힙에는 B의 옛 거리 8과 새 거리 3이 함께 있을 수 있으며 옛 항목을 다시 처리하지 않는다. 정점 이름 비교를 피하려고 유일한 순번도 저장한다.
입력은 각 정점에서 (이웃,가중치) 목록으로 가는 사전이며 비음수 정수 가중치만 허용한다. 모든 저장된 간선을 먼저 검사하므로 도달하지 않는 부분의 음수 가중치도 거부한다. 최단 경로 자체의 깊은 증명과 경로 복원은 다익스트라 문서의 별도 주제다.
중복 항목 방식에서는 힙 크기가 V만으로 제한되지 않는다. 일반적으로 성공한 완화 수와 간선 수에 따른 저장 공간을 고려하며 시간 상한도 물리 힙 크기를 기준으로 설명해야 한다. decrease-key를 직접 구현한 인덱스 힙과 같은 공간 계약으로 설명하지 않는다.
예제 코드
def shortest_distances(graph, start):
import heapq
from itertools import count
for edges in graph.values():
for _, weight in edges:
if not isinstance(weight, int) or weight < 0:
raise ValueError('nonnegative integer weight required')
ticket = count()
distance = {start: 0}
pending = [(0, next(ticket), start)]
while pending:
saved, _, vertex = heapq.heappop(pending)
if saved != distance[vertex]:
continue
for neighbor, weight in graph.get(vertex, []):
candidate = saved + weight
if candidate < distance.get(neighbor, float('inf')):
distance[neighbor] = candidate
heapq.heappush(pending, (candidate, next(ticket), neighbor))
return distance
graph = {'A': [('B', 8), ('C', 2)], 'C': [('B', 1), ('D', 7)],
'B': [('D', 3)], 'D': []}
print(sorted(shortest_distances(graph, 'A').items()))
try:
shortest_distances({'A': [], 'Z': [('Y', -1)]}, 'A')
except ValueError:
print('negative weight rejected')
실행 결과
[('A', 0), ('B', 3), ('C', 2), ('D', 6)]
negative weight rejected힙 정렬과 공간 비용의 차이
최소 힙을 만들고 반복해서 최솟값을 꺼내면 오름차순 결과를 얻는다. 생성 O(n)과 n번 제거 O(n log n)을 합쳐 O(n log n)이다. 그러나 결과를 새 목록에 모으는 구현은 출력 공간 O(n)을 사용한다.
보통 제자리 힙 정렬은 최대 힙을 만들고 루트와 마지막 유효 원소를 교환한 뒤 힙 범위를 줄이는 방식이다. 배열 뒤쪽에 큰 값이 차례로 확정된다. 최소 힙에서 새 결과 목록으로 꺼내는 코드와 동일한 공간 분석을 붙이면 안 된다.
힙 정렬은 일반적으로 안정 정렬이 아니다. 같은 키의 원래 순서를 반드시 보존하려면 추가 키나 다른 정렬 방식을 선택해야 한다. 우선순위 큐에 순번을 넣는 방식도 그 추가 정보와 저장 비용을 포함해 분석한다.
전체 정렬만 필요하면 언어의 표준 정렬 API가 더 간단하고 실제 성능 특성도 다를 수 있다. 힙은 입력이 동적으로 바뀌면서 극값을 반복 선택하는 상황에서 장점이 분명하다. 하나의 구조가 모든 정렬·검색 요구에 최선이라는 뜻은 아니다.
예제 코드
values = [6, 1, 4, 1, 3]
heap = MinHeap(values)
result = []
while len(heap):
result.append(heap.pop())
print(values)
print(result)
실행 결과
[6, 1, 4, 1, 3]
[1, 1, 3, 4, 6]임의 원소 검색·삭제와 인덱스 힙
이진 힙은 극값 위치를 알기 쉽게 만들지만 임의 값이 어디 있는지는 알려 주지 않는다. 값으로 찾는 contains나 remove는 별도 인덱스가 없으면 O(n) 탐색을 요구할 수 있다. 힙의 모든 작업이 O(log n)이라는 설명은 틀리다.
특정 원소의 현재 인덱스를 이미 알고 있다면 마지막 값으로 메운 뒤 위나 아래로 이동시켜 복구할 수 있다. 하지만 삽입·제거가 다른 원소의 위치도 바꾸므로 외부에서 저장한 인덱스를 계속 유지하려면 모든 swap을 추적해야 한다.
인덱스 힙은 원소 식별자→위치 사전을 함께 관리한다. 키 변경과 삭제 시 위치를 찾고 sift하며, 각 교환마다 두 식별자의 위치를 갱신한다. 중복 값이 있더라도 식별자는 고유해야 한다. 값과 작업 정체성을 구분해야 하는 또 다른 사례다.
지연 갱신은 구현이 단순한 대신 오래된 항목을 보유하고, 인덱스 힙은 공간을 제한하기 쉬운 대신 부가 불변식이 많다. 사용할 연산, 최악 지연, 메모리 예산과 구현 검증 비용을 함께 비교한다.
비교 규칙과 NaN·가변 상태의 함정
힙의 정확성은 비교가 일관적이라는 전제에 의존한다. a가 b보다 앞서고 b가 c보다 앞서면 a와 c의 관계도 모순되지 않아야 한다. 비교 함수가 호출 횟수나 외부 시각에 따라 바뀌면 어느 배열이 올바른 힙인지 정의하기 어렵다.
부동소수점 NaN은 일반적인 수의 대소 관계를 따르지 않는다. NaN<3과 3<NaN이 모두 False일 수 있으므로 “부모보다 작은 자식이 없다”는 검사만으로 업무에서 기대하는 최소 숫자가 루트에 있다고 설명할 수 없다.
아래는 NaN을 힙에 넣어 임의의 결과를 정답처럼 제시하지 않고 비교 성질만 확인한다. 실무에서는 NaN을 거부하거나 맨 뒤로 보내는 명시적 키를 만들고 그 정책을 문서화할 수 있다. 무한대와 누락 값 역시 의미가 다른 상태다.
사용자 정의 비교가 예외를 던지면 연산 도중 배열이 일부 수정된 상태일 수 있다. 교육용 MinHeap은 예외 후 전체 롤백을 보장하지 않는다. 실제 라이브러리의 예외 안전성 계약과 자료의 유효 범위를 확인한다.
예제 코드
from math import isnan
value = float('nan')
print(value < 3, 3 < value, value == value)
print(isnan(value))
실행 결과
False False False
True동시 작업 큐와 스케줄링의 범위
heapq는 배열을 조작하는 자료구조 도구이며 작업 대기·스레드 간 동기화·완료 추적을 모두 제공하는 서비스 큐가 아니다. 여러 생산자와 소비자가 공유한다면 그 실행 모델에 맞는 동기화 도구가 필요하다.
Python queue.PriorityQueue는 잠금과 대기 동작을 포함한 우선순위 큐 인터페이스를 제공한다. 다만 우선순위 순서만으로 공정성, 최대 대기 시간, 작업 영속성이나 중복 실행 방지가 자동 해결되지는 않는다.
높은 우선순위 작업이 계속 들어오면 낮은 우선순위 작업이 오래 기다릴 수 있다. 기다린 시간에 따라 우선순위를 올리는 정책을 만들더라도 저장된 키를 어떻게 갱신할지 결정해야 한다. 시간이 흐르면 비교 값이 자동으로 변하는 객체를 넣는 것만으로 힙이 재정렬되지 않는다.
마감 시각 스케줄러는 다음 작업을 고르는 기능 외에 시계 종류, 취소, 깨어날 시점, 실행 실패와 종료 규칙이 필요하다. 이 문서의 자료구조 예제와 실제 작업 처리 시스템의 보장을 구분한다.
언어별 기본 방향과 API 계약
Python heapq는 목록 자체를 최소 힙으로 사용하며 함수를 호출해 유지한다. 사용자 코드가 그 목록을 임의 변경하면 계약이 깨질 수 있다. 동률에서는 튜플의 다음 필드까지 비교하므로 비교할 수 없는 페이로드를 분리한다.
Java PriorityQueue는 지정 비교자 또는 자연 순서의 가장 작은 항목을 앞에 둔다. 동률은 자동 FIFO 보장이 없고 반복자가 우선순위 순서로 순회하는 것도 아니다. contains와 값 지정 삭제가 선형이라는 점을 기본 삽입·제거의 비용과 구분한다.
Rust BinaryHeap은 기본 최대 힙이고 보관 중 상대 순서를 바꾸는 변경은 논리 오류다. 비교 키를 뒤집는 래퍼나 명시된 API를 사용한다. C++ priority_queue도 비교자의 방향과 strict weak ordering 계약을 확인해야 한다.
언어를 바꾸면서 “힙이니 제일 작은 것이 나온다”라고 가정하지 않는다. 기본 방향, 빈 상태, 동률, 임의 삭제, 반복 순서, 갱신과 스레드 안전성은 각각 따로 비교할 항목이다. 이 문서는 타 언어 구현을 실행했다고 주장하지 않는다.
작은 모든 입력의 heapify·정렬 대조
값 0,1,2의 길이 0부터 7까지 모든 배열은 3,280개다. 각 배열로 MinHeap을 만든 뒤 모든 부모·자식 관계를 검사하고 반복 pop 결과를 sorted와 비교한다. 같은 입력을 빈 힙에 차례로 push하는 방식도 대조한다.
이 검사는 이미 정렬된 입력뿐 아니라 역순·중복·빈 배열과 원소 하나를 포함한다. 같은 힙을 만드는 배열 배치는 여러 개일 수 있으므로 내부 배열이 heapq와 똑같은지 요구하지 않는다. 필요한 계약은 힙 조건과 제거 결과다.
두 생성 방식의 결과가 같은 정렬 순서를 내는지 확인하면 bottom-up 시작 인덱스, 오른쪽 자식 범위, 마지막 원소 제거 등의 오류를 찾을 수 있다. 테스트가 알고리즘의 O(n) heapify 증명을 대신하는 것은 아니다.
테스트를 확장할 때 비교가 불가능한 값의 예외, 실제 업무의 동률 정책과 값 표현 범위도 별도 대상으로 삼는다. 여기의 완전 열거는 작은 정수 값의 기본 최소 힙 계약에 한정된다.
예제 코드
from itertools import product
checks = 0
for length in range(8):
for values in product(range(3), repeat=length):
built = MinHeap(values)
pushed = MinHeap()
for value in values:
pushed.push(value)
for heap in [built, pushed]:
assert all(not heap.data[i] < heap.data[(i - 1) // 2]
for i in range(1, len(heap)))
assert [heap.pop() for _ in range(len(heap))] == sorted(values)
checks += 1
print(checks)
실행 결과
3280연산열·상위 k개·갱신 취소 대조
다음 검사는 MinHeap을 heapq와 4,000번의 혼합 삽입·제거로 비교한다. 매 단계 길이, 극값과 부모·자식 불변식을 검사한다. 단순히 한 번 만들고 모두 꺼내는 검사와 달리 제거 후 재삽입하는 경로도 포함된다.
상위 k개는 길이 0부터 6까지 모든 3진 값 배열에서 k=0부터 n+1까지 검사한다. 기준은 전체 정렬의 앞 k개다. 입력보다 큰 k와 중복 값도 포함한다.
UpdatingQueue는 별도 사전에 저장한 (우선순위,순번)을 직접 min으로 찾는 기준 모델과 3,000단계 대조한다. 갱신·취소·pop을 섞고 주기적으로 rebuild한다. 우선순위 동률과 같은 키 재등록의 세부 정책까지 결과를 비교한다.
기준 모델은 효율이 낮아도 단순하고 독립적인 방식으로 정답을 제공한다. 이 대조는 동시 실행이나 메모리 할당 실패를 검증하지 않으며 테스트의 성공을 운영 시스템의 내구성 보장으로 표현하지 않는다.
예제 코드
import heapq
from random import Random
from itertools import product
rng = Random(83)
actual, expected = MinHeap(), []
for _ in range(4000):
if not expected or rng.randrange(2):
value = rng.randrange(-10, 11)
actual.push(value); heapq.heappush(expected, value)
else:
assert actual.pop() == heapq.heappop(expected)
assert len(actual) == len(expected)
if expected:
assert actual.peek() == expected[0]
assert all(not actual.data[i] < actual.data[(i - 1) // 2]
for i in range(1, len(actual)))
top_checks = 0
for length in range(7):
for values in product(range(3), repeat=length):
for k in range(length + 2):
assert largest_k(iter(values), k) == sorted(values, reverse=True)[:k]
top_checks += 1
queue, model, serial = UpdatingQueue(), {}, 0
for step in range(3000):
operation = rng.randrange(3)
key = rng.randrange(12)
if operation == 0:
priority = rng.randrange(5)
queue.set_priority(key, priority)
model[key] = (priority, serial)
serial += 1
elif operation == 1 and key in model:
queue.cancel(key)
del model[key]
elif operation == 2:
if model:
chosen = min(model, key=lambda x: model[x])
assert queue.pop() == (chosen, model[chosen][0])
del model[chosen]
else:
try:
queue.pop()
except KeyError:
pass
else:
raise AssertionError('empty pop accepted')
if step % 31 == 0:
queue.rebuild()
assert queue.current == model
print(4000, top_checks, 3000)
실행 결과
4000 8201 3000다중 병합과 최단 거리의 독립 기준 검사
다중 병합은 세 스트림 각각의 길이 0부터 3, 값 0·1의 비감소 시퀀스를 모두 조합한다. 각 스트림 후보는 10개이므로 총 1,000개 조합을 전체 정렬 결과와 비교한다. 비어 있는 스트림과 여러 중복 값이 포함된다.
최단 거리는 정점 3개의 각 방향 간선을 없음·가중치0·가중치2로 선택한 729개 그래프에서 모든 시작점을 검사한다. 기준은 모든 간선을 두 번 반복 완화하는 방법으로, 힙의 선택 순서를 사용하지 않는다. 총 2,187개 거리 사전을 비교한다.
거리 값만 아니라 도달 가능한 정점 집합도 비교한다. 단절된 정점에 잘못된 거리를 넣거나 0 비용 순환에서 무한히 갱신하는 오류를 확인할 수 있다. 그래프가 작은 범위라는 한계는 유지한다.
이진 힙 자체의 정확성과 이를 이용한 알고리즘의 정확성은 따로 검증할 필요가 있다. 힙이 정상 동작해도 잘못된 완화 조건이나 정렬되지 않은 입력 스트림을 주면 응용 결과는 틀릴 수 있다.
예제 코드
from itertools import combinations_with_replacement, product
streams = [x for length in range(4)
for x in combinations_with_replacement(range(2), length)]
merge_checks = 0
for selected in product(streams, repeat=3):
assert list(merge_streams(selected)) == sorted(x for stream in selected for x in stream)
merge_checks += 1
edges = [(a, b) for a in range(3) for b in range(3) if a != b]
graph_checks = 0
for choices in product([-1, 0, 2], repeat=6):
graph = {a: [] for a in range(3)}
weighted = []
for (a, b), weight in zip(edges, choices):
if weight >= 0:
graph[a].append((b, weight))
weighted.append((a, b, weight))
for start in range(3):
distance = [99] * 3
distance[start] = 0
for _ in range(2):
for a, b, weight in weighted:
if distance[a] < 99:
distance[b] = min(distance[b], distance[a] + weight)
expected = {a: distance[a] for a in range(3) if distance[a] < 99}
assert shortest_distances(graph, start) == expected
graph_checks += 1
print(merge_checks, graph_checks)
실행 결과
1000 2187이진 힙을 선택하는 기준
다음 극값을 반복해서 읽고 제거하면서 중간에 새 항목을 넣는다면 이진 힙을 검토할 수 있다. 반면 전체 정렬을 한 번만 하거나 임의 값 검색·범위 검색·모든 순위 조회가 핵심이면 다른 자료구조가 더 직접적으로 맞을 수 있다.
우선순위의 방향과 동률 정책을 먼저 정하고, 갱신·취소가 필요한지 확인한다. 같은 값의 다른 작업을 구분할 식별자, 버전과 비교 가능한 키를 설계한다. 내부 배열을 직접 수정하면 불변식 복구 책임도 함께 생긴다.
성능 분석에서는 논리 항목 수와 물리 항목 수, 비교 비용, 기반 배열의 재할당과 지연 삭제를 구분한다. O(log n)이라는 표기 하나로 검색·갱신·최악 지연·메모리를 모두 설명할 수는 없다.
구현 학습은 배열 인덱스의 완전 트리 표현, 한 경로의 불변식 복구, bottom-up의 높이별 합으로 연결된다. 이를 이해하면 상위 k개 선택, 정렬된 입력 병합, 최단 경로의 후보 관리가 왜 같은 자료구조를 활용하는지 설명할 수 있다.