Algorithm Search⌘K

Gabow-Tarjan linear-time static union-find in Python

Linear-time algorithm for static tree set union using word-level parallelism.

File
gabow_tarjan_union_find.py, 177 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
gabow_tarjan_union_find.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/gabow-tarjan-union-find/. The authors are listed in prov/1DC4F6.jsonl.

from collections.abc import Sequence


class GabowTarjanUnionFind:
    """Static tree set union with microsets in words and a macro union-find forest.

    The tree is given by parent[v], with parent[root] == -1. link(v) unites the
    set of v with the set of its parent; find(v) returns the nearest ancestor
    of v, v included, that has not been linked. The tree is split into
    microsets of at most b nodes; inside a microset, the marks and the
    ancestor sets are bit masks, and a find inside it is one AND and one
    bit_length. A general union-find forest on the microset roots handles the
    jumps between microsets.
    """

    def __init__(self, parent: Sequence[int], b: int | None = None) -> None:
        n = len(parent)
        self.parent = list(parent)
        roots = [v for v in range(n) if parent[v] == -1]
        if n and len(roots) != 1:
            raise ValueError("parent must describe one rooted tree")
        if b is None:
            b = max(1, n.bit_length())  # @line L1
        self.b = b
        self.marked = [False] * n
        self.micro = [0] * n  # microset of each node
        self.local = [0] * n  # local index of each node in its microset
        self.anc = [0] * n  # mask of the ancestors of the node inside its piece
        self.members: list[list[int]] = []  # nodes of each microset, by local index
        self.mroot: list[int] = []  # root r(S) of each microset, or -1
        self.mask: list[int] = []  # marked nodes of each microset
        if n:
            self._partition(roots[0])  # @line L2
        for s, nodes in enumerate(self.members):  # @line L3
            for i, x in enumerate(nodes):
                self.local[x] = i
            for x in nodes:
                p = self.parent[x]
                inside = p != -1 and self.micro[p] == s
                self.anc[x] = (self.anc[p] if inside else 0) | (1 << self.local[x])
        # macro forest on the microset roots, with union by size and path compression
        self.mp: dict[int, int] = {}
        self.msize: dict[int, int] = {}
        self.mname: dict[int, int] = {}
        for r in self.mroot:  # @line L4
            if r != -1 and r not in self.mp:
                self.mp[r] = r
                self.msize[r] = 1
                self.mname[r] = r

    def _partition(self, root: int) -> None:
        """Split the tree into microsets of at most b nodes, bottom up."""
        n = len(self.parent)
        children: list[list[int]] = [[] for _ in range(n)]
        for v in range(n):
            if self.parent[v] != -1:
                children[self.parent[v]].append(v)
        order: list[int] = []
        stack = [root]
        while stack:
            v = stack.pop()
            order.append(v)
            stack.extend(children[v])
        order.reverse()  # children before parents
        half = (self.b + 1) // 2
        nxt = [-1] * n
        head = [-1] * n  # open list of each node, as a linked list
        tail = [-1] * n
        size = [0] * n

        def close(h: int, r: int) -> None:
            s = len(self.members)
            nodes = []
            while h != -1:
                nodes.append(h)
                self.micro[h] = s
                h = nxt[h]
            # the list is in preorder: each node comes after its parent in the piece
            self.members.append(nodes)
            self.mroot.append(r)
            self.mask.append(0)

        for v in order:
            gh, gt, gs = -1, -1, 0
            for c in children[v]:
                if gh == -1:
                    gh, gt = head[c], tail[c]
                else:
                    nxt[gt] = head[c]
                    gt = tail[c]
                gs += size[c]
                if gs >= half:
                    close(gh, v)
                    gh, gt, gs = -1, -1, 0
            nxt[v] = gh
            head[v], tail[v], size[v] = v, (gt if gh != -1 else v), gs + 1
        close(head[root], -1)

    def link(self, v: int) -> None:
        """Unite the set of v with the set of its parent. v must not be the root."""
        if self.parent[v] == -1:
            raise ValueError("the root cannot be linked")
        self.marked[v] = True
        self.mask[self.micro[v]] |= 1 << self.local[v]  # @line L5

    def _micro_find(self, x: int) -> int:
        s = self.micro[x]
        u = self.anc[x] & ~self.mask[s]  # @line L6
        if u == 0:  # @line L7
            return -1
        return self.members[s][u.bit_length() - 1]  # @line L8

    def _macro_name(self, y: int) -> int:
        r = y
        while self.mp[r] != r:
            r = self.mp[r]
        while self.mp[y] != r:
            nx = self.mp[y]
            self.mp[y] = r
            y = nx
        return self.mname[r]

    def _macro_union(self, z: int, y: int) -> None:
        name = self._macro_name(y)
        a, c = self._macro_root(z), self._macro_root(y)
        if self.msize[a] > self.msize[c]:
            a, c = c, a
        self.mp[a] = c
        self.msize[c] += self.msize[a]
        self.mname[c] = name

    def _macro_root(self, y: int) -> int:
        while self.mp[y] != y:
            y = self.mp[y]
        return y

    def find(self, v: int) -> int:
        """Return the nearest ancestor of v, v included, that has not been linked."""
        w = self._micro_find(v)  # @line L9
        if w != -1:
            return w
        z = self._macro_name(self.mroot[self.micro[v]])  # @line L10
        while True:
            w = self._micro_find(z)  # @line L11
            if w != -1:
                break
            self._macro_union(z, self.mroot[self.micro[z]])  # @line L12
            z = self._macro_name(z)  # @line L13
        return w  # @line L14

    def check_invariant(self) -> None:
        """Check the partition, the masks, and the names of the macro sets."""
        n = len(self.parent)
        seen = [0] * n
        for s, nodes in enumerate(self.members):
            assert 1 <= len(nodes) <= self.b
            m = 0
            for i, x in enumerate(nodes):
                seen[x] += 1
                assert self.micro[x] == s and self.local[x] == i
                p = self.parent[x]
                assert p == self.mroot[s] or (p != -1 and self.micro[p] == s)
                if p != -1 and self.micro[p] == s:
                    assert self.local[p] < i
                if self.marked[x]:
                    m |= 1 << i
            assert m == self.mask[s]
        assert all(c == 1 for c in seen)
        for y in self.mp:
            name = self._macro_name(y)
            a = y
            while a != name:  # the name is an ancestor of every member
                a = self.parent[a]
                assert a != -1
test_gabow_tarjan_union_find.py 72 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/gabow-tarjan-union-find/. The authors are listed in prov/1DC4F6.jsonl.

import random

import pytest

from gabow_tarjan_union_find import GabowTarjanUnionFind


def naive_find(parent: list[int], marked: list[bool], v: int) -> int:
    while marked[v]:
        v = parent[v]
    return v


def test_example() -> None:
    parent = [-1, 0, 0, 1, 1, 2, 5, 5]
    g = GabowTarjanUnionFind(parent, b=4)
    assert sorted(map(sorted, g.members)) == [[0], [1, 2, 5], [3, 4], [6, 7]]
    ops = [("f", 6, 6), ("l", 6, 0), ("f", 6, 5), ("l", 5, 0), ("f", 7, 7), ("f", 6, 2),
           ("l", 3, 0), ("l", 1, 0), ("f", 3, 0), ("f", 4, 4), ("l", 2, 0), ("f", 6, 0)]
    for op, v, want in ops:
        if op == "l":
            g.link(v)
        else:
            assert g.find(v) == want
        g.check_invariant()


def test_edges() -> None:
    g = GabowTarjanUnionFind([-1])
    assert g.find(0) == 0
    with pytest.raises(ValueError):
        g.link(0)
    with pytest.raises(ValueError):
        GabowTarjanUnionFind([-1, -1])
    GabowTarjanUnionFind([]).check_invariant()


def test_long_path() -> None:
    n = 100000
    parent = [i + 1 for i in range(n)]
    parent[-1] = -1
    g = GabowTarjanUnionFind(parent)
    g.check_invariant()
    for v in range(n - 1):
        g.link(v)
        if v % 997 == 0:
            assert g.find(0) == v + 1
    assert g.find(0) == n - 1


def test_random_against_marks() -> None:
    rng = random.Random(2024)
    for _ in range(600):
        n = rng.randrange(1, 40)
        parent = [-1] + [rng.randrange(v) for v in range(1, n)]
        g = GabowTarjanUnionFind(parent, b=rng.choice([None, 1, 2, 3, 5, 8]))
        g.check_invariant()
        marked = [False] * n
        free = list(range(1, n))
        rng.shuffle(free)
        for _ in range(3 * n):
            if free and rng.random() < 0.4:
                v = free.pop()
                g.link(v)
                marked[v] = True
            else:
                v = rng.randrange(n)
                assert g.find(v) == naive_find(parent, marked, v)
        g.check_invariant()

What the next level needs

  • the tests did not run in this build

Read the full page about Gabow-Tarjan linear-time static union-find