Karger-Klein-Tarjan algorithm in Python
Randomized expected linear-time MST using sampling, Borůvka steps and verification.
- File
karger_klein_tarjan_mst.py, 147 lines- Level
- I0 not checked yet
- Toolchain
- Python 3.14, pytest 9
- Test framework
- pytest
- Run the tests
python3 -m pytest
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/karger-klein-tarjan-mst/. The authors are listed in prov/YYGGT8.jsonl.
import random
def karger_klein_tarjan_mst(
n: int, edges: list[tuple[int, int, float]], rng: random.Random | None = None
) -> list[int]:
"""Return the indices of a minimum spanning forest, by Karger, Klein and Tarjan.
The vertices are 0 .. n - 1 and edges[i] = (u, v, w) is an undirected
edge. Weights may tie (ties are broken by the edge index, so the answer
is the unique forest that is minimum for the order by (w, i)); the graph
may be disconnected, and parallel edges and self-loops are allowed.
The algorithm is random, but its answer is not: the coins only change
the running time. Each call does two Boruvka steps, samples half of the
remaining edges, solves the sample, drops the edges that the sample's
forest proves heavy, and solves the rest.
Cost: this version checks heaviness with binary lifting instead of a
linear-time verification algorithm, so it takes O(m log n) expected
time instead of O(m); O(n + m) expected space.
"""
rng = rng if rng is not None else random.Random()
work = [(u, v, (w, i)) for i, (u, v, w) in enumerate(edges) if u != v]
return sorted(_msf(n, work, rng))
def _msf(n, edges, rng):
"""Minimum spanning forest of a graph with distinct keys; original indices."""
if not edges: # @line L1
return []
chosen: list[int] = []
for _ in range(2): # @line L2
n, edges, picked = _boruvka_step(n, edges)
chosen += picked
if not edges:
return chosen
sample = [e for e in edges if rng.random() < 0.5] # @line L3
forest = _msf(n, sample, rng) # @line L4
light = _drop_heavy(n, edges, set(forest)) # @line L5
return chosen + _msf(n, light, rng) # @line L6
def _boruvka_step(n, edges):
"""Pick the lightest edge at each vertex, contract, drop loops and isolated vertices."""
best = [None] * n
for e in edges:
u, v, key = e
for x in (u, v):
if best[x] is None or key < best[x][2]:
best[x] = e
parent = list(range(n))
def find(x):
while parent[x] != x:
parent[x] = parent[parent[x]]
x = parent[x]
return x
picked = []
for e in best:
if e is None:
continue
a, b = find(e[0]), find(e[1])
if a != b:
parent[a] = b
picked.append(e[2][1])
label: dict[int, int] = {}
out = []
for u, v, key in edges:
a, b = find(u), find(v)
if a != b:
a = label.setdefault(a, len(label))
b = label.setdefault(b, len(label))
out.append((a, b, key))
return len(label), out, picked
def _drop_heavy(n, edges, forest_ids):
"""Keep the edges that are not F-heavy, where F is the forest given by its ids."""
adj: list[list[tuple[int, tuple]]] = [[] for _ in range(n)]
for u, v, key in edges:
if key[1] in forest_ids:
adj[u].append((v, key))
adj[v].append((u, key))
depth = [-1] * n
root = [-1] * n
up = [list(range(n))]
mx: list[list[tuple | None]] = [[None] * n]
for s in range(n):
if depth[s] >= 0:
continue
depth[s] = 0
root[s] = s
stack = [s]
while stack:
x = stack.pop()
for y, key in adj[x]:
if depth[y] < 0:
depth[y] = depth[x] + 1
root[y] = s
up[0][y] = x
mx[0][y] = key
stack.append(y)
j = 1
while (1 << j) < n:
prev_up, prev_mx = up[-1], mx[-1]
up.append([prev_up[prev_up[x]] for x in range(n)])
mx.append([_max(prev_mx[x], prev_mx[prev_up[x]]) for x in range(n)])
j += 1
def path_max(a, b):
best = None
if depth[a] < depth[b]:
a, b = b, a
d = depth[a] - depth[b]
k = 0
while d:
if d & 1:
best = _max(best, mx[k][a])
a = up[k][a]
d >>= 1
k += 1
if a == b:
return best
for k in range(len(up) - 1, -1, -1):
if up[k][a] != up[k][b]:
best = _max(best, _max(mx[k][a], mx[k][b]))
a, b = up[k][a], up[k][b]
return _max(best, _max(mx[0][a], mx[0][b]))
keep = []
for u, v, key in edges:
if root[u] != root[v] or key <= path_max(u, v):
keep.append((u, v, key))
return keep
def _max(a, b):
if a is None:
return b
if b is None:
return a
return a if a > b else b
test_karger_klein_tarjan_mst.py 75 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/karger-klein-tarjan-mst/. The authors are listed in prov/YYGGT8.jsonl.
import random
from karger_klein_tarjan_mst import _drop_heavy, karger_klein_tarjan_mst
EXAMPLE = [(0, 1, 4), (0, 2, 3), (1, 2, 1), (1, 3, 2), (2, 3, 4), (2, 4, 5), (3, 4, 7), (3, 5, 6), (4, 5, 2)]
def kruskal(n, edges):
"""The minimum spanning forest for the order by (weight, index)."""
parent = list(range(n))
def find(x):
while parent[x] != x:
parent[x] = parent[parent[x]]
x = parent[x]
return x
out = []
for i in sorted(range(len(edges)), key=lambda i: (edges[i][2], i)):
a, b = find(edges[i][0]), find(edges[i][1])
if a != b:
parent[a] = b
out.append(i)
return sorted(out)
def test_example():
for seed in range(20):
assert karger_klein_tarjan_mst(6, EXAMPLE, random.Random(seed)) == [1, 2, 3, 5, 8]
def test_trivial_graphs():
assert karger_klein_tarjan_mst(0, []) == []
assert karger_klein_tarjan_mst(1, [(0, 0, 1)]) == []
assert karger_klein_tarjan_mst(4, []) == []
assert karger_klein_tarjan_mst(2, [(0, 1, 5)]) == [0]
def test_ties_and_parallel_edges():
edges = [(0, 1, 1), (0, 1, 1), (1, 2, 1), (0, 2, 1), (2, 2, 0)]
for seed in range(10):
assert karger_klein_tarjan_mst(3, edges, random.Random(seed)) == [0, 2]
def test_disconnected():
edges = [(0, 1, 2), (2, 3, 1), (3, 4, 1), (2, 4, 1), (6, 7, 0)]
assert karger_klein_tarjan_mst(8, edges, random.Random(1)) == kruskal(8, edges) == [0, 1, 2, 4]
def test_heavy_edges_are_dropped():
# Forest F = path 0 - 1 - 2 with keys 1 and 2; the edge 0-2 of key 3 is F-heavy.
edges = [(0, 1, (1, 0)), (1, 2, (2, 1)), (0, 2, (3, 2)), (0, 2, (1.5, 3)), (2, 3, (9, 4))]
keep = _drop_heavy(4, edges, {0, 1})
assert [e[2][1] for e in keep] == [0, 1, 3, 4]
def test_random_against_kruskal():
rng = random.Random(20261105)
for trial in range(800):
n = rng.randint(1, 15)
m = rng.randint(0, 40)
edges = [(rng.randrange(n), rng.randrange(n), rng.randint(-3, 5)) for _ in range(m)]
assert karger_klein_tarjan_mst(n, edges, random.Random(trial)) == kruskal(n, edges)
def test_larger_graphs():
rng = random.Random(9)
for trial in range(30):
n = rng.randint(100, 400)
m = rng.randint(n, 5 * n)
edges = [(rng.randrange(n), rng.randrange(n), rng.random()) for _ in range(m)]
assert karger_klein_tarjan_mst(n, edges, random.Random(trial)) == kruskal(n, edges)
What the next level needs
- the tests did not run in this build