Algorithm Search⌘K

Roaring bitmap in Python

A compressed bitmap that stores 16-bit chunks as arrays, bitmaps or runs, depending on density.

File
roaring_bitmap.py, 269 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
roaring_bitmap.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/roaring-bitmap/. The authors are listed in prov/SA2AHY.jsonl.

from bisect import bisect_left, insort

ARRAY_MAX = 4096  # an array container holds at most this many values
WORDS = 1024  # a bitmap container is 1024 words of 64 bits = 2**16 bits
MASK64 = (1 << 64) - 1


class ArrayContainer:
    """A sorted list of distinct 16-bit values, at most ARRAY_MAX of them."""

    def __init__(self, values: list[int] | None = None) -> None:
        self.values = values if values is not None else []

    def cardinality(self) -> int:
        return len(self.values)

    def contains(self, low: int) -> bool:
        i = bisect_left(self.values, low)
        return i < len(self.values) and self.values[i] == low

    def members(self) -> list[int]:
        return list(self.values)


class BitmapContainer:
    """2**16 bits in 1024 words of 64 bits, with a cached count of ones."""

    def __init__(self) -> None:
        self.words = [0] * WORDS
        self.card = 0

    def cardinality(self) -> int:
        return self.card

    def contains(self, low: int) -> bool:
        return (self.words[low >> 6] >> (low & 63)) & 1 == 1

    def members(self) -> list[int]:
        out = []
        for j, word in enumerate(self.words):
            while word:
                t = word & -word
                out.append((j << 6) | (t.bit_length() - 1))
                word ^= t
        return out


Container = ArrayContainer | BitmapContainer


def to_bitmap(values: list[int]) -> BitmapContainer:
    b = BitmapContainer()
    for v in values:
        b.words[v >> 6] |= 1 << (v & 63)
    b.card = len(values)
    return b


def from_words(words: list[int]) -> Container | None:
    """Build the right container from 1024 words; None if all are zero."""
    card = sum(word.bit_count() for word in words)
    if card == 0:
        return None
    if card <= ARRAY_MAX:
        b = BitmapContainer()
        b.words = words
        return ArrayContainer(b.members())
    b = BitmapContainer()
    b.words = words
    b.card = card
    return b


class RoaringBitmap:
    """A set of 32-bit unsigned integers in the Roaring layout.

    Each value x is split into a key x >> 16 and a low part x & 0xFFFF. The
    keys are kept in a sorted list; each key has a container for its low
    parts: an array container when it holds at most 4096 values, otherwise
    a bitmap container. Empty containers are removed.
    """

    def __init__(self) -> None:
        self.keys: list[int] = []
        self.containers: list[Container] = []

    def _find(self, high: int) -> int:
        i = bisect_left(self.keys, high)
        if i < len(self.keys) and self.keys[i] == high:
            return i
        return -1

    def add(self, x: int) -> None:
        """Insert x, 0 <= x < 2**32. Cost O(log k + 4096) for k containers."""
        if not 0 <= x < (1 << 32):
            raise ValueError("x must be a 32-bit unsigned integer")
        high, low = x >> 16, x & 0xFFFF  # @line L1
        i = bisect_left(self.keys, high)  # @line L2
        if i == len(self.keys) or self.keys[i] != high:
            self.keys.insert(i, high)
            self.containers.insert(i, ArrayContainer())
        c = self.containers[i]  # @line L3
        if isinstance(c, ArrayContainer):  # @line L4
            if not c.contains(low):  # @line L5
                insort(c.values, low)
            if len(c.values) > ARRAY_MAX:  # @line L6
                self.containers[i] = to_bitmap(c.values)
        else:
            bit = 1 << (low & 63)  # @line L7
            if not c.words[low >> 6] & bit:
                c.words[low >> 6] |= bit
                c.card += 1

    def contains(self, x: int) -> bool:
        """Return True if x is a member. Cost O(log k + log 4096)."""
        if not 0 <= x < (1 << 32):
            return False
        i = self._find(x >> 16)  # @line L8
        if i < 0:
            return False
        return self.containers[i].contains(x & 0xFFFF)  # @line L9

    def remove(self, x: int) -> None:
        """Delete x if present. Cost O(log k + 4096)."""
        if not 0 <= x < (1 << 32):
            return
        high, low = x >> 16, x & 0xFFFF
        i = self._find(high)  # @line L10
        if i < 0:
            return
        c = self.containers[i]
        if isinstance(c, ArrayContainer):  # @line L11
            j = bisect_left(c.values, low)
            if j < len(c.values) and c.values[j] == low:
                c.values.pop(j)
        else:
            bit = 1 << (low & 63)
            if c.words[low >> 6] & bit:
                c.words[low >> 6] ^= bit
                c.card -= 1
            if c.card <= ARRAY_MAX:  # @line L12
                self.containers[i] = ArrayContainer(c.members())
        if self.containers[i].cardinality() == 0:  # @line L13
            self.keys.pop(i)
            self.containers.pop(i)

    def cardinality(self) -> int:
        """Return the number of members. Cost O(k)."""
        return sum(c.cardinality() for c in self.containers)

    def __iter__(self):
        for high, c in zip(self.keys, self.containers):
            for low in c.members():
                yield (high << 16) | low

    def to_list(self) -> list[int]:
        return list(self)

    @staticmethod
    def _combine(a: Container, b: Container, op: str) -> Container | None:
        if isinstance(a, ArrayContainer) and isinstance(b, ArrayContainer):
            if op == "and":
                out, i, j = [], 0, 0
                av, bv = a.values, b.values
                while i < len(av) and j < len(bv):
                    if av[i] < bv[j]:
                        i += 1
                    elif av[i] > bv[j]:
                        j += 1
                    else:
                        out.append(av[i])
                        i += 1
                        j += 1
                return ArrayContainer(out) if out else None
            out, i, j = [], 0, 0
            av, bv = a.values, b.values
            while i < len(av) or j < len(bv):
                if j == len(bv) or (i < len(av) and av[i] < bv[j]):
                    out.append(av[i])
                    i += 1
                elif i == len(av) or bv[j] < av[i]:
                    out.append(bv[j])
                    j += 1
                else:
                    out.append(av[i])
                    i += 1
                    j += 1
            if len(out) > ARRAY_MAX:
                return to_bitmap(out)
            return ArrayContainer(out)
        if op == "and" and isinstance(a, ArrayContainer) != isinstance(b, ArrayContainer):
            arr, bm = (a, b) if isinstance(a, ArrayContainer) else (b, a)
            out = [v for v in arr.values if bm.contains(v)]
            return ArrayContainer(out) if out else None
        wa = a.words if isinstance(a, BitmapContainer) else to_bitmap(a.values).words
        wb = b.words if isinstance(b, BitmapContainer) else to_bitmap(b.values).words
        if op == "and":
            words = [p & q for p, q in zip(wa, wb)]
        else:
            words = [p | q for p, q in zip(wa, wb)]
        return from_words(words)

    def _merge(self, other: "RoaringBitmap", op: str) -> "RoaringBitmap":
        res = RoaringBitmap()
        i = j = 0
        while i < len(self.keys) and j < len(other.keys):
            ki, kj = self.keys[i], other.keys[j]
            if ki < kj:
                if op == "or":
                    res.keys.append(ki)
                    res.containers.append(_copy(self.containers[i]))
                i += 1
            elif kj < ki:
                if op == "or":
                    res.keys.append(kj)
                    res.containers.append(_copy(other.containers[j]))
                j += 1
            else:
                c = self._combine(self.containers[i], other.containers[j], op)
                if c is not None:
                    res.keys.append(ki)
                    res.containers.append(c)
                i += 1
                j += 1
        if op == "or":
            for k in range(i, len(self.keys)):
                res.keys.append(self.keys[k])
                res.containers.append(_copy(self.containers[k]))
            for k in range(j, len(other.keys)):
                res.keys.append(other.keys[k])
                res.containers.append(_copy(other.containers[k]))
        return res

    def union(self, other: "RoaringBitmap") -> "RoaringBitmap":
        """Return a new bitmap with the members of either. Cost O(k + k') container merges."""
        return self._merge(other, "or")

    def intersection(self, other: "RoaringBitmap") -> "RoaringBitmap":
        """Return a new bitmap with the members of both."""
        return self._merge(other, "and")

    def check_invariant(self) -> None:
        """Raise AssertionError if the representation is not valid."""
        assert len(self.keys) == len(self.containers)
        assert all(0 <= k < (1 << 16) for k in self.keys)
        assert all(a < b for a, b in zip(self.keys, self.keys[1:]))
        for c in self.containers:
            if isinstance(c, ArrayContainer):
                assert 1 <= len(c.values) <= ARRAY_MAX
                assert all(0 <= v < (1 << 16) for v in c.values)
                assert all(a < b for a, b in zip(c.values, c.values[1:]))
            else:
                assert isinstance(c, BitmapContainer)
                assert len(c.words) == WORDS
                assert all(0 <= word <= MASK64 for word in c.words)
                assert c.card == sum(word.bit_count() for word in c.words)
                assert c.card > ARRAY_MAX


def _copy(c: Container) -> Container:
    if isinstance(c, ArrayContainer):
        return ArrayContainer(list(c.values))
    b = BitmapContainer()
    b.words = list(c.words)
    b.card = c.card
    return b
test_roaring_bitmap.py 117 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/roaring-bitmap/. The authors are listed in prov/SA2AHY.jsonl.

import random

import pytest

from roaring_bitmap import ArrayContainer, BitmapContainer, RoaringBitmap


def build(values) -> RoaringBitmap:
    r = RoaringBitmap()
    for v in values:
        r.add(v)
    r.check_invariant()
    return r


def test_page_example() -> None:
    r = build([1, 5, 70000, 70001, 196608 + 9])
    assert r.keys == [0, 1, 3]
    assert [c.members() for c in r.containers] == [[1, 5], [4464, 4465], [9]]
    assert r.contains(70001) and not r.contains(70002)
    assert r.cardinality() == 5
    r.remove(196617)
    r.check_invariant()
    assert r.keys == [0, 1]


def test_conversion_at_threshold() -> None:
    r = build(range(0, 2 * 4096, 2))
    assert isinstance(r.containers[0], ArrayContainer)
    assert r.cardinality() == 4096
    r.add(1)
    r.check_invariant()
    assert isinstance(r.containers[0], BitmapContainer)
    assert r.cardinality() == 4097
    r.remove(0)
    r.check_invariant()
    assert isinstance(r.containers[0], ArrayContainer)
    assert r.cardinality() == 4096


def test_full_chunk() -> None:
    r = build(range(5 << 16, 6 << 16))
    assert r.keys == [5]
    assert r.cardinality() == 65536
    for v in range(5 << 16, 6 << 16):
        r.remove(v)
    r.check_invariant()
    assert r.keys == [] and r.cardinality() == 0


def test_bad_values() -> None:
    r = RoaringBitmap()
    with pytest.raises(ValueError):
        r.add(1 << 32)
    with pytest.raises(ValueError):
        r.add(-1)
    assert not r.contains(-1)
    r.remove(1 << 32)


def random_values(rng: random.Random, n: int) -> list[int]:
    """Values that fill a few chunks densely and others sparsely."""
    out = []
    for _ in range(n):
        kind = rng.random()
        if kind < 0.5:
            out.append((3 << 16) | rng.randrange(1 << 13))  # dense chunk
        elif kind < 0.8:
            out.append(rng.randrange(4) << 16 | rng.randrange(1 << 16))
        else:
            out.append(rng.getrandbits(32))
    return out


def test_random_against_set_model() -> None:
    rng = random.Random(20261104)
    r = RoaringBitmap()
    model: set[int] = set()
    for step in range(20000):
        x = random_values(rng, 1)[0]
        if rng.random() < 0.65:
            r.add(x)
            model.add(x)
        else:
            y = x
            if model and rng.random() < 0.7:
                y = (x & ~0xFFFF) | (rng.randrange(1 << 16))
                if y not in model:
                    y = next(iter(model))
            r.remove(y)
            model.discard(y)
        if step % 1000 == 0:
            r.check_invariant()
            assert r.to_list() == sorted(model)
        probe = rng.getrandbits(32) if rng.random() < 0.5 else x
        assert r.contains(probe) == (probe in model)
    r.check_invariant()
    assert r.to_list() == sorted(model)
    assert r.cardinality() == len(model)


def test_set_operations_against_model() -> None:
    rng = random.Random(77)
    for _ in range(12):
        a = random_values(rng, rng.randrange(0, 9000))
        b = random_values(rng, rng.randrange(0, 9000))
        ra, rb = build(a), build(b)
        u = ra.union(rb)
        n = ra.intersection(rb)
        u.check_invariant()
        n.check_invariant()
        assert u.to_list() == sorted(set(a) | set(b))
        assert n.to_list() == sorted(set(a) & set(b))
        assert ra.to_list() == sorted(set(a))

What the next level needs

  • the tests did not run in this build

Read the full page about Roaring bitmap