This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 4c5844c9de [python][daft] Split conjunctive filters for pushdown
(#8055)
4c5844c9de is described below
commit 4c5844c9decd597b8135e90710305a264b41a17e
Author: QuakeWang <[email protected]>
AuthorDate: Sun May 31 23:18:13 2026 +0800
[python][daft] Split conjunctive filters for pushdown (#8055)
Daft Paimon filter conversion previously treated each filter expression
as all-or-nothing. For an expression like `A AND unsupported(B)`, the
supported conjunct `A` was not pushed to Paimon, so it could not
participate in scan planning or file skipping.
This PR flattens only `AND` conjuncts in the Daft adapter layer and
converts each conjunct independently. Supported conjuncts are combined
into a Paimon `AND` predicate, while unsupported conjuncts remain as
Daft post-scan filters. `OR` expressions are still pushed only when the
whole `OR` is supported.
---
.../pypaimon/daft/daft_predicate_visitor.py | 57 ++++++++++++---
.../pypaimon/tests/daft/daft_data_test.py | 85 ++++++++++++++++++++++
.../pypaimon/tests/daft/daft_explain_test.py | 24 ++++++
3 files changed, 157 insertions(+), 9 deletions(-)
diff --git a/paimon-python/pypaimon/daft/daft_predicate_visitor.py
b/paimon-python/pypaimon/daft/daft_predicate_visitor.py
index 5dfe4bca3d..059efbec9d 100644
--- a/paimon-python/pypaimon/daft/daft_predicate_visitor.py
+++ b/paimon-python/pypaimon/daft/daft_predicate_visitor.py
@@ -28,6 +28,7 @@ import logging
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
+from daft.expressions import Expression
from daft.expressions.visitor import PredicateVisitor
if TYPE_CHECKING:
@@ -36,7 +37,6 @@ if TYPE_CHECKING:
from pypaimon.table.file_store_table import FileStoreTable
from daft.daft import PyExpr
- from daft.expressions import Expression
logger = logging.getLogger(__name__)
@@ -63,6 +63,43 @@ class _Unsupported:
_UNSUPPORTED = _Unsupported()
+class _AndSplitter(PredicateVisitor[tuple[Expression, Expression] | None]):
+ """Returns the direct children only when the expression root is AND."""
+
+ def visit_and(self, left: Expression, right: Expression) ->
tuple[Expression, Expression]:
+ return left, right
+
+ def _not_and(self, *args: Any) -> None:
+ return None
+
+ visit_or = _not_and
+ visit_not = _not_and
+ visit_equal = _not_and
+ visit_not_equal = _not_and
+ visit_less_than = _not_and
+ visit_less_than_or_equal = _not_and
+ visit_greater_than = _not_and
+ visit_greater_than_or_equal = _not_and
+ visit_between = _not_and
+ visit_is_in = _not_and
+ visit_is_null = _not_and
+ visit_not_null = _not_and
+ visit_col = _not_and
+ visit_lit = _not_and
+ visit_alias = _not_and
+ visit_cast = _not_and
+ visit_coalesce = _not_and
+ visit_function = _not_and
+
+
+def _split_conjuncts(expr: Expression) -> list[Expression]:
+ children = _AndSplitter().visit(expr)
+ if children is None:
+ return [expr]
+ left, right = children
+ return _split_conjuncts(left) + _split_conjuncts(right)
+
+
class PaimonPredicateVisitor(PredicateVisitor[Any]):
"""Tree fold visitor that converts Daft expressions to Paimon predicates.
@@ -272,14 +309,16 @@ def convert_filters_to_paimon(
for py_expr in py_filters:
expr = Expression._from_pyexpr(py_expr)
- predicate = converter.visit(expr)
-
- if isinstance(predicate, Predicate):
- pushed_filters.append(py_expr)
- predicates.append(predicate)
- else:
- remaining_filters.append(py_expr)
- logger.debug("Filter %s cannot be pushed down to Paimon", expr)
+
+ for conjunct in _split_conjuncts(expr):
+ predicate = converter.visit(conjunct)
+
+ if isinstance(predicate, Predicate):
+ pushed_filters.append(conjunct._expr)
+ predicates.append(predicate)
+ else:
+ remaining_filters.append(conjunct._expr)
+ logger.debug("Filter %s cannot be pushed down to Paimon",
conjunct)
combined_predicate: Predicate | None = None
if predicates:
diff --git a/paimon-python/pypaimon/tests/daft/daft_data_test.py
b/paimon-python/pypaimon/tests/daft/daft_data_test.py
index 9d7795cb97..e8dde610f5 100644
--- a/paimon-python/pypaimon/tests/daft/daft_data_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_data_test.py
@@ -39,6 +39,24 @@ from pypaimon.daft.daft_paimon import _read_table
from pypaimon.daft.daft_predicate_visitor import convert_filters_to_paimon
+def _contains_expr(py_expr):
+ from daft.expressions import Expression
+
+ expr_text = str(Expression._from_pyexpr(py_expr))
+ return "contains" in expr_text
+
+
+def _predicate_leaves(predicate):
+ if predicate is None:
+ return []
+ if predicate.method in ("and", "or"):
+ result = []
+ for child in predicate.literals:
+ result.extend(_predicate_leaves(child))
+ return result
+ return [(predicate.method, predicate.field, tuple(predicate.literals or
[]))]
+
+
# ---------------------------------------------------------------------------
# Helper
# ---------------------------------------------------------------------------
@@ -674,6 +692,47 @@ class TestFilterPushdown:
result = df.to_pydict()
assert result["id"] == [2, 3, 4]
+ def test_filter_pushdown_splits_supported_conjuncts(self, filter_table):
+ unsupported = col("value").contains(col("id"))
+ cases = [
+ ((col("id") == 1) & unsupported, [("equal", "id", (1,))]),
+ (unsupported & (col("id") == 1), [("equal", "id", (1,))]),
+ (
+ (col("id") == 1) & (unsupported & (col("value") == "a")),
+ [("equal", "id", (1,)), ("equal", "value", ("a",))],
+ ),
+ ]
+
+ for expr, expected_leaves in cases:
+ pushed_filters, remaining_filters, predicate =
convert_filters_to_paimon(filter_table, expr._expr)
+
+ assert len(pushed_filters) == len(expected_leaves)
+ assert len(remaining_filters) == 1
+ assert _contains_expr(remaining_filters[0])
+ assert _predicate_leaves(predicate) == expected_leaves
+
+ def test_filter_pushdown_does_not_split_or_with_unsupported_branch(self,
filter_table):
+ expr = (col("id") == 1) | col("value").contains(col("id"))
+
+ pushed_filters, remaining_filters, predicate =
convert_filters_to_paimon(filter_table, expr._expr)
+
+ assert pushed_filters == []
+ assert remaining_filters == [expr._expr]
+ assert predicate is None
+
+ def test_filter_pushdown_pushes_supported_or_conjunct(self, filter_table):
+ supported_or = (col("id") == 1) | (col("id") == 2)
+ expr = supported_or & col("value").contains(col("id"))
+
+ pushed_filters, remaining_filters, predicate =
convert_filters_to_paimon(filter_table, expr._expr)
+
+ assert len(pushed_filters) == 1
+ assert len(remaining_filters) == 1
+ assert _contains_expr(remaining_filters[0])
+ assert predicate is not None
+ assert predicate.method == "or"
+ assert _predicate_leaves(predicate) == [("equal", "id", (1,)),
("equal", "id", (2,))]
+
def test_unsupported_expression_remains_in_daft(self, filter_table):
expressions = [
col("id") == lit(1).cast("int64"),
@@ -718,6 +777,32 @@ class TestFilterPushdown:
assert result["id"] == [1, 3]
+ def test_mixed_conjunctive_expression_is_filtered_by_daft(self,
local_paimon_catalog):
+ catalog, tmp_path = local_paimon_catalog
+ pa_schema = pa.schema([
+ ("id", pa.int64()),
+ ("value", pa.string()),
+ ("pattern", pa.string()),
+ ])
+ paimon_schema = pypaimon.Schema.from_pyarrow_schema(pa_schema)
+ catalog.create_table("test_db.filter_mixed_conjunctive",
paimon_schema, ignore_if_exists=True)
+ table = catalog.get_table("test_db.filter_mixed_conjunctive")
+
+ data = pa.table(
+ {
+ "id": [1, 1, 2, 3],
+ "value": ["alpha", "bravo", "alps", "charlie"],
+ "pattern": ["lp", "zz", "lp", "lie"],
+ }
+ )
+ _write_to_paimon(table, data)
+
+ df = _read_table(table).where((col("id") == 1) &
col("value").contains(col("pattern")))
+ result = df.sort("value").to_pydict()
+
+ assert result["id"] == [1]
+ assert result["value"] == ["alpha"]
+
# ---------------------------------------------------------------------------
# Advanced data types
diff --git a/paimon-python/pypaimon/tests/daft/daft_explain_test.py
b/paimon-python/pypaimon/tests/daft/daft_explain_test.py
index 0df051adf9..ce1662a433 100644
--- a/paimon-python/pypaimon/tests/daft/daft_explain_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_explain_test.py
@@ -197,6 +197,30 @@ def
test_explain_scan_keeps_limit_above_remaining_filters(catalog_options):
assert result.paimon_scan.splits is None
+def
test_explain_scan_partially_pushes_conjuncts_and_keeps_limit(catalog_options):
+ pa_schema = pa.schema([
+ ("id", pa.int64()),
+ ("name", pa.string()),
+ ])
+ identifier, table = _create_table(
+ catalog_options,
+ "explain_partial_conjunct",
+ pa_schema,
+ options={"bucket": "-1", "file.format": "parquet"},
+ )
+ _write_arrow(table, pa.table({"id": [1, 2], "name": ["alpha", "bravo"]},
schema=pa_schema))
+
+ result = PaimonTable(table, catalog_options=catalog_options).explain_scan(
+ filters=(col("id") == 1) & col("name").contains(col("id")),
+ limit=1,
+ )
+
+ assert any("id" in pushed for pushed in result.pushed_filters)
+ assert any("contains" in remaining for remaining in
result.remaining_filters)
+ assert result.source_limit is None
+ assert result.limit_pushed is False
+
+
def
test_explain_scan_applies_partition_filters_to_reader_counts(catalog_options):
pa_schema = pa.schema([
("id", pa.int64()),
