fix(expressions): bound the DNF expansion in rewrite_to_dnf (#3995)

Converting to DNF distributes every AND over the ORs below it, so a filter
with n two-branch OR groups expands to 2**n terms. Nothing bounded that, so a
small user-supplied filter could produce an expression no consumer can use:
20 groups reached 1,048,576 terms in ~17s and ~1 GiB.

Cap the expansion at MAX_DNF_TERMS and raise before the product is built, so
the same input now fails in 0.22s.

Author: Rodrigo-Palma

pyiceberg/expressions/visitors.py

@@ -974,6 +974,12 @@ def extract_field_ids(expr: BooleanExpression) -> set[int]:
     return visit(expr, _ExpressionFieldIDs())
 
 
+# Converting to DNF distributes every AND over the ORs below it, so the number of
+# terms is exponential in the worst case. The expansion is bounded to keep a small
+# filter from producing an expression that no consumer can use.
+MAX_DNF_TERMS = 1 << 14
+
+
 class _RewriteToDNF(BooleanExpressionVisitor[tuple[BooleanExpression, ...]]):
     def visit_true(self) -> tuple[BooleanExpression, ...]:
         return (AlwaysTrue(),)
@@ -991,6 +997,8 @@ def visit_and(
         # ((P OR Q) AND (R OR S)) AND (((P AND R) OR (P AND S)) OR ((Q AND R) OR ((Q AND S)))
         # A AND (B OR C) = (A AND B) OR (A AND C)
         # (A OR B) AND C = (A AND C) OR (B AND C)
+        if len(left_result) * len(right_result) > MAX_DNF_TERMS:
+            raise ValueError(f"Expression expands to more than the maximum of {MAX_DNF_TERMS} DNF terms")
         return tuple(And(le, re) for le in left_result for re in right_result)
 
     def visit_or(
@@ -1008,6 +1016,7 @@ def visit_bound_predicate(self, predicate: BoundPredicate) -> tuple[BooleanExpre
 def rewrite_to_dnf(expr: BooleanExpression) -> tuple[BooleanExpression, ...]:
     # Rewrites an arbitrary boolean expression to disjunctive normal form (DNF):
     # (A AND NOT(B) AND C) OR (NOT(D) AND E AND F) OR (G)
+    # Raises ValueError when the expansion exceeds MAX_DNF_TERMS terms.
     expr_without_not = rewrite_not(expr)
     return visit(expr_without_not, _RewriteToDNF())
 

tests/expressions/test_visitors.py

@@ -66,6 +66,7 @@
 )
 from pyiceberg.expressions.literals import Literal, literal
 from pyiceberg.expressions.visitors import (
+    MAX_DNF_TERMS,
     BindVisitor,
     BooleanExpressionVisitor,
     BoundBooleanExpressionVisitor,
@@ -1594,6 +1595,23 @@ def test_to_dnf_not_and() -> None:
     assert rewrite_to_dnf(expr) == (EqualTo("Q", "b"), NotEqualTo("R", "c"))
 
 
+def _chained_or_pairs(groups: int) -> BooleanExpression:
+    # (a_0 = x OR b_0 = y) AND ... AND (a_n = x OR b_n = y), which expands to 2**groups DNF terms
+    expr: BooleanExpression = Or(EqualTo("a0", "x"), EqualTo("b0", "y"))
+    for idx in range(1, groups):
+        expr = And(expr, Or(EqualTo(f"a{idx}", "x"), EqualTo(f"b{idx}", "y")))
+    return expr
+
+
+def test_to_dnf_at_expansion_limit() -> None:
+    assert len(rewrite_to_dnf(_chained_or_pairs(14))) == MAX_DNF_TERMS
+
+
+def test_to_dnf_rejects_unbounded_expansion() -> None:
+    with pytest.raises(ValueError, match=f"maximum of {MAX_DNF_TERMS} DNF terms"):
+        rewrite_to_dnf(_chained_or_pairs(20))
+
+
 def test_dnf_to_dask(table_schema_simple: Schema) -> None:
     expr = (
         BoundGreaterThan(