Algorithm Search⌘K

Expression DAG in Python

Expression representation where shared subexpressions are stored once.

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

"""A hash-consed expression DAG: every distinct subexpression is stored once.

Nodes are small integers into the arrays of an ExprDAG. A node is a constant
(a Fraction), a symbol (a name), or an operator in + - * ^ with two child
nodes; the second child of ^ is a constant integer k >= 0. Because children
are created before their parents, every child id is smaller than its parent
id, so the arrays are already in topological order.

Cost model: make costs one hash-table lookup, O(1) expected. eval, subs and
diff visit each distinct node reachable from the root once, O(m) node visits
for m such nodes, whatever the size of the unfolded tree. equal is O(1).
"""

from fractions import Fraction
from typing import Dict, List, Optional, Tuple, Union

Number = Union[int, Fraction]
Key = Tuple[str, object, Tuple[int, ...]]


class ExprDAG:
    """A store of hash-consed expression nodes, with one method per operation."""

    def __init__(self) -> None:
        self.op: List[str] = []
        self.val: List[object] = []
        self.kids: List[Tuple[int, ...]] = []
        self.table: Dict[Key, int] = {}

    # construction -----------------------------------------------------------

    def make(self, op: str, val: object, kids: Tuple[int, ...]) -> int:
        """The unique node with this operator, value and children. O(1) expected."""
        key: Key = (op, val, kids)  # @line L1
        found = self.table.get(key)
        if found is not None:  # @line L2
            return found
        i = len(self.op)  # @line L3
        self.op.append(op)
        self.val.append(val)
        self.kids.append(kids)
        self.table[key] = i  # @line L4
        return i  # @line L5

    def const(self, c: Number) -> int:
        return self.make("const", Fraction(c), ())

    def sym(self, name: str) -> int:
        return self.make("sym", name, ())

    def add(self, a: int, b: int) -> int:
        return self.make("+", None, (a, b))

    def sub(self, a: int, b: int) -> int:
        return self.make("-", None, (a, b))

    def mul(self, a: int, b: int) -> int:
        return self.make("*", None, (a, b))

    def power(self, a: int, k: int) -> int:
        if k < 0:
            raise ValueError("exponent must be a nonnegative integer")
        return self.make("^", None, (a, self.const(k)))

    # smart constructors used by diff; they keep the value ---------------------

    def _is_const(self, u: int, c: Optional[int] = None) -> bool:
        return self.op[u] == "const" and (c is None or self.val[u] == c)

    def s_add(self, a: int, b: int) -> int:
        if self._is_const(a) and self._is_const(b):
            return self.const(self.val[a] + self.val[b])  # type: ignore[operator]
        if self._is_const(a, 0):
            return b
        if self._is_const(b, 0):
            return a
        return self.add(a, b)

    def s_sub(self, a: int, b: int) -> int:
        if self._is_const(a) and self._is_const(b):
            return self.const(self.val[a] - self.val[b])  # type: ignore[operator]
        if self._is_const(b, 0):
            return a
        return self.sub(a, b)

    def s_mul(self, a: int, b: int) -> int:
        if self._is_const(a) and self._is_const(b):
            return self.const(self.val[a] * self.val[b])  # type: ignore[operator]
        if self._is_const(a, 0) or self._is_const(b, 0):
            return self.const(0)
        if self._is_const(a, 1):
            return b
        if self._is_const(b, 1):
            return a
        return self.mul(a, b)

    # walks --------------------------------------------------------------------

    def reachable(self, root: int) -> List[int]:
        """Distinct nodes reachable from root, in increasing id order (children first)."""
        seen = {root}
        stack = [root]
        while stack:
            u = stack.pop()
            for k in self.kids[u]:
                if k not in seen:
                    seen.add(k)
                    stack.append(k)
        return sorted(seen)

    def eval(self, root: int, env: Dict[str, Number]) -> Fraction:
        """Value of root when each symbol s is env[s]. O(m) visits, m distinct nodes."""
        v: Dict[int, Fraction] = {}
        for u in self.reachable(root):
            op = self.op[u]
            if op == "const":
                v[u] = self.val[u]  # type: ignore[assignment]
            elif op == "sym":
                v[u] = Fraction(env[self.val[u]])  # type: ignore[index]
            else:
                a, b = v[self.kids[u][0]], v[self.kids[u][1]]
                if op == "+":
                    v[u] = a + b
                elif op == "-":
                    v[u] = a - b
                elif op == "*":
                    v[u] = a * b
                else:
                    v[u] = a ** int(b)
        return v[root]

    def subs(self, root: int, x: str, r: int) -> int:
        """root with the symbol x replaced by the node r. O(m) visits."""
        res: Dict[int, int] = {}
        for u in self.reachable(root):
            if self.op[u] == "sym" and self.val[u] == x:
                res[u] = r
            elif not self.kids[u]:
                res[u] = u
            else:
                res[u] = self.make(self.op[u], None, tuple(res[k] for k in self.kids[u]))
        return res[root]

    def diff(self, root: int, x: str) -> int:
        """Derivative of root with respect to x. O(m) visits and O(m) new nodes."""
        d: Dict[int, int] = {}
        for u in self.reachable(root):
            op = self.op[u]
            if op == "const":
                d[u] = self.const(0)
            elif op == "sym":
                d[u] = self.const(1 if self.val[u] == x else 0)
            else:
                a, b = self.kids[u]
                if op == "+":
                    d[u] = self.s_add(d[a], d[b])
                elif op == "-":
                    d[u] = self.s_sub(d[a], d[b])
                elif op == "*":
                    d[u] = self.s_add(self.s_mul(d[a], b), self.s_mul(a, d[b]))
                else:
                    k = int(self.val[b])  # type: ignore[arg-type]
                    if k == 0:
                        d[u] = self.const(0)
                    else:
                        pk = self.const(1) if k == 1 else (a if k == 2 else self.power(a, k - 1))
                        d[u] = self.s_mul(self.s_mul(self.const(k), pk), d[a])
        return d[root]

    def equal(self, a: int, b: int) -> bool:
        """Syntactic identity of the two expressions: one integer comparison."""
        return a == b

    def dag_size(self, root: int) -> int:
        """Number of distinct nodes reachable from root."""
        return len(self.reachable(root))

    def tree_size(self, root: int) -> int:
        """Number of nodes of the unfolded tree; may be exponential in dag_size."""
        t: Dict[int, int] = {}
        for u in self.reachable(root):
            t[u] = 1 + sum(t[k] for k in self.kids[u])
        return t[root]

    def to_string(self, root: int) -> str:
        """Fully parenthesized infix form of the unfolded tree (** for powers)."""
        s: Dict[int, str] = {}
        for u in self.reachable(root):
            op = self.op[u]
            if op == "const":
                c: Fraction = self.val[u]  # type: ignore[assignment]
                s[u] = "(%d)" % c.numerator if c.denominator == 1 else "Fraction(%d, %d)" % (c.numerator, c.denominator)
            elif op == "sym":
                s[u] = str(self.val[u])
            else:
                a, b = self.kids[u]
                s[u] = "(" + s[a] + " " + ("**" if op == "^" else op) + " " + s[b] + ")"
        return s[root]

    def check_invariant(self) -> None:
        """Arrays agree, the table is a bijection onto the nodes, children precede parents."""
        n = len(self.op)
        assert len(self.val) == n and len(self.kids) == n and len(self.table) == n
        for i in range(n):
            assert self.table[(self.op[i], self.val[i], self.kids[i])] == i
            op = self.op[i]
            if op == "const":
                assert isinstance(self.val[i], Fraction) and self.kids[i] == ()
            elif op == "sym":
                assert isinstance(self.val[i], str) and self.kids[i] == ()
            else:
                assert op in ("+", "-", "*", "^") and len(self.kids[i]) == 2
                assert all(0 <= k < i for k in self.kids[i])
                if op == "^":
                    k = self.kids[i][1]
                    assert self.op[k] == "const" and self.val[k].denominator == 1  # type: ignore[union-attr]
                    assert self.val[k] >= 0  # type: ignore[operator]
test_expression_dag.py 210 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/expression-dag/. The authors are listed in prov/46Z0DH.jsonl.

import random
from fractions import Fraction
from typing import Dict, List, Tuple, Union

import pytest

from expression_dag import ExprDAG

# The oracle: plain nested tuples, with no sharing and no hashing.
# ("c", Fraction) | ("s", name) | (op, left, right) | ("^", base, k)
T = tuple
Poly = Dict[Tuple[int, int], Fraction]


def t_eval(t: T, env: Dict[str, Fraction]) -> Fraction:
    if t[0] == "c":
        return t[1]
    if t[0] == "s":
        return env[t[1]]
    if t[0] == "^":
        return t_eval(t[1], env) ** t[2]
    a, b = t_eval(t[1], env), t_eval(t[2], env)
    return a + b if t[0] == "+" else a - b if t[0] == "-" else a * b


def t_size(t: T) -> int:
    if t[0] in "cs":
        return 1
    if t[0] == "^":
        return 2 + t_size(t[1])
    return 1 + t_size(t[1]) + t_size(t[2])


def pmul(a: Poly, b: Poly) -> Poly:
    r: Poly = {}
    for (i, j), c in a.items():
        for (k, l), d in b.items():
            r[(i + k, j + l)] = r.get((i + k, j + l), 0) + c * d
    return {m: c for m, c in r.items() if c != 0}


def t_expand(t: T) -> Poly:
    if t[0] == "c":
        return {(0, 0): t[1]} if t[1] != 0 else {}
    if t[0] == "s":
        return {(1, 0) if t[1] == "x" else (0, 1): Fraction(1)}
    if t[0] == "^":
        r: Poly = {(0, 0): Fraction(1)}
        a = t_expand(t[1])
        for _ in range(t[2]):
            r = pmul(r, a)
        return r
    a, b = t_expand(t[1]), t_expand(t[2])
    if t[0] == "*":
        return pmul(a, b)
    r = dict(a)
    for m, c in b.items():
        r[m] = r.get(m, 0) + (c if t[0] == "+" else -c)
    return {m: c for m, c in r.items() if c != 0}


def d_expand(D: ExprDAG, u: int) -> Poly:
    """Expansion of a DAG node, through memoization over the distinct nodes."""
    memo: Dict[int, Poly] = {}
    for w in D.reachable(u):
        op = D.op[w]
        if op == "const":
            memo[w] = {(0, 0): D.val[w]} if D.val[w] != 0 else {}
        elif op == "sym":
            memo[w] = {(1, 0) if D.val[w] == "x" else (0, 1): Fraction(1)}
        else:
            a, b = D.kids[w]
            if op == "^":
                r: Poly = {(0, 0): Fraction(1)}
                for _ in range(int(D.val[b])):
                    r = pmul(r, memo[a])
                memo[w] = r
            elif op == "*":
                memo[w] = pmul(memo[a], memo[b])
            else:
                r = dict(memo[a])
                for m, c in memo[b].items():
                    r[m] = r.get(m, 0) + (c if op == "+" else -c)
                memo[w] = {m: c for m, c in r.items() if c != 0}
    return memo[u]


def build(D: ExprDAG, t: T) -> int:
    if t[0] == "c":
        return D.const(t[1])
    if t[0] == "s":
        return D.sym(t[1])
    if t[0] == "^":
        return D.power(build(D, t[1]), t[2])
    a, b = build(D, t[1]), build(D, t[2])
    return {"+": D.add, "-": D.sub, "*": D.mul}[t[0]](a, b)


def rand_tuple(rng: random.Random, depth: int, pool: List[Tuple[T, int]]) -> T:
    """Random tree of height at most depth that often reuses earlier subtrees,
    so the DAG has sharing; pool holds (subtree, height) pairs."""
    fits = [t for t, h in pool if h <= depth]
    if fits and rng.random() < 0.3:
        return rng.choice(fits)
    if depth <= 1 or rng.random() < 0.1:
        r = rng.random()
        t: T = ("c", Fraction(rng.randint(-4, 4), rng.randint(1, 3))) if r < 0.4 else ("s", "x" if r < 0.75 else "y")
        h = 0
    elif rng.random() < 0.15:
        t = ("^", rand_tuple(rng, depth - 2, pool), rng.randint(0, 3))
        h = depth
    else:
        t = (rng.choice("+-*"), rand_tuple(rng, depth - 1, pool), rand_tuple(rng, depth - 1, pool))
        h = depth
    pool.append((t, h))
    return t


def t_d(t: T) -> Poly:
    p = t_expand(t)
    return {(i - 1, j): c * i for (i, j), c in p.items() if i > 0}


def test_worked_example() -> None:
    D = ExprDAG()
    x, y = D.sym("x"), D.sym("y")
    s = D.add(x, y)
    e = D.add(D.mul(s, s), s)  # (x + y)(x + y) + (x + y)
    D.check_invariant()
    assert D.dag_size(e) == 5 and D.tree_size(e) == 11
    assert D.add(D.sym("x"), D.sym("y")) == s  # hash consing finds the old node
    assert len(D.op) == 5
    assert D.eval(e, {"x": 1, "y": 2}) == 12
    de = D.diff(e, "x")  # 2(x + y) + 1
    assert D.eval(de, {"x": 1, "y": 2}) == 7
    assert d_expand(D, de) == {(1, 0): 2, (0, 1): 2, (0, 0): 1}


def test_repeated_squaring() -> None:
    D = ExprDAG()
    u = D.add(D.sym("x"), D.const(1))
    for _ in range(40):
        u = D.mul(u, u)
    D.check_invariant()
    assert D.dag_size(u) == 43
    assert D.tree_size(u) == 2 ** 42 - 1
    assert D.eval(u, {"x": 0}) == 1
    assert D.eval(u, {"x": Fraction(-2)}) == 1
    D2 = ExprDAG()
    w = D2.add(D2.sym("x"), D2.const(1))
    for _ in range(12):
        w = D2.mul(w, w)
    assert D2.eval(w, {"x": 1}) == 2 ** 4096
    assert D2.eval(D2.diff(w, "x"), {"x": 0}) == 4096
    assert D2.dag_size(D2.diff(w, "x")) <= 6 * 15


def test_edge_cases() -> None:
    D = ExprDAG()
    z = D.const(0)
    assert D.eval(z, {}) == 0 and D.diff(z, "x") == z
    assert D.const(Fraction(2, 4)) == D.const(Fraction(1, 2))
    x = D.sym("x")
    assert D.diff(x, "x") == D.const(1) and D.diff(x, "y") == z
    assert D.diff(D.power(x, 0), "x") == z
    assert D.diff(D.power(x, 1), "x") == D.const(1)
    assert D.subs(z, "x", x) == z
    assert D.subs(x, "x", x) == x
    with pytest.raises(ValueError):
        D.power(x, -2)
    D.check_invariant()


def test_long_chain() -> None:
    D = ExprDAG()
    u = D.sym("x")
    for i in range(30000):
        u = D.add(u, D.const(i % 3))
    assert D.eval(u, {"x": 0}) == 30000
    assert D.eval(D.diff(u, "x"), {"x": 5}) == 1
    D.check_invariant()


def test_random_against_tuple_oracle() -> None:
    rng = random.Random(46)
    for _ in range(600):
        D = ExprDAG()
        pool: List[Tuple[T, int]] = []
        ts = [rand_tuple(rng, rng.randint(2, 8), pool) for _ in range(3)]
        ids = [build(D, t) for t in ts]
        D.check_invariant()
        # hash consing: equal ids exactly for equal trees
        for i in range(3):
            assert build(D, ts[i]) == ids[i]
            for j in range(3):
                assert D.equal(ids[i], ids[j]) == (ts[i] == ts[j])
        e, r = ids[0], ids[1]
        assert D.tree_size(e) == t_size(ts[0])
        env = {"x": Fraction(rng.randint(-5, 5), rng.randint(1, 3)), "y": Fraction(rng.randint(-5, 5))}
        assert D.eval(e, env) == t_eval(ts[0], env)
        assert Fraction(eval(D.to_string(e), {"Fraction": Fraction}, dict(env))) == t_eval(ts[0], env)
        de = D.diff(e, "x")
        assert d_expand(D, de) == t_d(ts[0])
        se = D.subs(e, "x", r)
        assert D.eval(se, env) == t_eval(ts[0], {"x": t_eval(ts[1], env), "y": env["y"]})
        assert D.subs(e, "q", r) == e
        D.check_invariant()

What the next level needs

  • the tests did not run in this build

Read the full page about Expression DAG