Algorithm Search⌘K

PPJoin in Python

Extends prefix filtering with positional filtering to prune candidate pairs in set similarity joins.

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

import math
from collections import Counter
from fractions import Fraction
from typing import Hashable, Iterable, Sequence


def _ceil(q: Fraction) -> int:
    return math.ceil(q)  # exact on Fractions


def ppjoin(sets: Sequence[Iterable[Hashable]], t: Fraction | float | str, stats: dict | None = None) -> list[tuple[int, int]]:
    """All pairs (i, j), i < j, of input sets with Jaccard similarity at least t, 0 < t <= 1.

    Empty sets are ignored. stats, if given, receives the counts 'candidates' and 'pruned'.
    """
    t = Fraction(t)
    if not 0 < t <= 1:
        raise ValueError("t must be in (0, 1]")
    data = [set(s) for s in sets]
    freq = Counter(w for s in data for w in s)
    key = lambda w: (freq[w], repr(w))
    recs = [sorted(s, key=key) for s in data]  # @line L1
    order = sorted((i for i in range(len(recs)) if recs[i]), key=lambda i: (len(recs[i]), i))
    index: dict[Hashable, list[tuple[int, int]]] = {}  # @line L2
    out: list[tuple[int, int]] = []
    ncand = npruned = 0
    for xi in order:  # @line L3
        x = recs[xi]
        lx = len(x)
        overlap: dict[int, int] = {}  # @line L4
        pruned: set[int] = set()
        probe = lx - _ceil(t * lx) + 1
        indexed = lx - _ceil(2 * t / (1 + t) * lx) + 1
        for i in range(1, probe + 1):  # @line L5
            w = x[i - 1]  # @line L6
            for yi, j in index.get(w, []):  # @line L7
                ly = len(recs[yi])
                if ly < t * lx or yi in pruned:
                    continue
                alpha = _ceil(t / (1 + t) * (lx + ly))  # @line L8
                if overlap.get(yi, 0) + 1 + min(lx - i, ly - j) >= alpha:  # @line L9
                    overlap[yi] = overlap.get(yi, 0) + 1
                else:
                    pruned.add(yi)  # @line L10
            if i <= indexed:  # @line L11
                index.setdefault(w, []).append((xi, i))
        for yi in overlap:  # @line L12
            if yi in pruned:
                continue
            ncand += 1
            y = recs[yi]
            if len(data[xi] & data[yi]) >= _ceil(t / (1 + t) * (lx + len(y))):  # @line L13
                out.append((min(xi, yi), max(xi, yi)))
        npruned += len(pruned)
    if stats is not None:
        stats["candidates"] = ncand
        stats["pruned"] = npruned
    return sorted(out)


def jaccard(a: Iterable[Hashable], b: Iterable[Hashable]) -> Fraction:
    sa, sb = set(a), set(b)
    if not sa and not sb:
        return Fraction(1)
    return Fraction(len(sa & sb), len(sa | sb))
test_ppjoin.py 48 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/ppjoin/. The authors are listed in prov/YE460Y.jsonl.

import itertools
import random
from fractions import Fraction

from ppjoin import jaccard, ppjoin

R = ["ABCDE", "ABCDF", "BCDEFG", "AGH", "CDEFH", "ABCDEG"]


def brute(sets, t):
    t = Fraction(t)
    return sorted(
        (i, j)
        for i, j in itertools.combinations(range(len(sets)), 2)
        if set(sets[i]) and set(sets[j]) and jaccard(sets[i], sets[j]) >= t
    )


def test_example() -> None:
    stats: dict = {}
    assert ppjoin(R, "0.6", stats) == [(0, 1), (0, 5), (2, 5)]
    assert stats == {"candidates": 6, "pruned": 3}


def test_edges() -> None:
    assert ppjoin([], "0.5") == []
    assert ppjoin(["", "", "a"], "0.5") == []
    assert ppjoin(["ab", "ba", "ab"], 1) == [(0, 1), (0, 2), (1, 2)]
    assert ppjoin(["abc", "abd"], "0.5") == [(0, 1)]
    assert ppjoin(["abc", "abd"], "0.51") == []
    try:
        ppjoin(["a"], 0)
        raise AssertionError("t = 0 accepted")
    except ValueError:
        pass


def test_random_against_brute_force() -> None:
    rng = random.Random(2008)
    for _ in range(600):
        n = rng.randint(0, 25)
        universe = rng.randint(1, 12)
        sets = [{rng.randrange(universe) for _ in range(rng.randint(0, 8))} for _ in range(n)]
        t = rng.choice(["0.3", "0.5", "0.6", "2/3", "0.75", "0.8", "0.9", "1"])
        assert ppjoin(sets, t) == brute(sets, t)

What the next level needs

  • the tests did not run in this build

Read the full page about PPJoin