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
# 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