[python][daft] Push filters into Paimon scan planning

paimon-python/pypaimon/daft/daft_datasource.py

@@ -559,9 +559,6 @@ def __init__(
         table_path = getattr(table, "table_path", None)
         self._table_path = str(table_path) if table_path is not None else None
         self._table_options = _extract_table_options(table)
-        self._pushed_filters: list[PyExpr] | None = None
-        self._paimon_predicate: Predicate | None = None
-        self._remaining_filters: list[PyExpr] | None = None
         self._init_table(table)
 
     def __getstate__(self) -> dict[str, Any]:
@@ -573,9 +570,6 @@ def __getstate__(self) -> dict[str, Any]:
             "_table_identifier": self._table_identifier,
             "_table_path": self._table_path,
             "_table_options": self._table_options,
-            "_pushed_filters": self._pushed_filters,
-            "_paimon_predicate": self._paimon_predicate,
-            "_remaining_filters": self._remaining_filters,
         }
 
     def __setstate__(self, state: dict[str, Any]) -> None:
@@ -585,9 +579,6 @@ def __setstate__(self, state: dict[str, Any]) -> None:
         self._table_identifier = state["_table_identifier"]
         self._table_path = state["_table_path"]
         self._table_options = state["_table_options"]
-        self._pushed_filters = state.get("_pushed_filters")
-        self._paimon_predicate = state["_paimon_predicate"]
-        self._remaining_filters = state["_remaining_filters"]
 
         table = _load_table(
             self._table_catalog_options,
@@ -673,27 +664,6 @@ def get_partition_fields(self) -> list[PartitionField]:
         partition_key_names = set(self._table.partition_keys)
         return [PartitionField.create(f) for f in self._schema if f.name in partition_key_names]
 
-    def push_filters(self, filters: list[PyExpr]) -> tuple[list[PyExpr], list[PyExpr]]:
-        """Push filters down to Paimon scan.
-
-        Converts Daft expressions to Paimon predicates where possible.
-        Returns (pushed_filters, remaining_filters).
-        """
-        pushed_filters, remaining_filters, paimon_predicate = convert_filters_to_paimon(self._table, filters)
-
-        self._pushed_filters = pushed_filters
-        self._paimon_predicate = paimon_predicate
-        self._remaining_filters = remaining_filters
-
-        if pushed_filters:
-            logger.debug(
-                "Paimon filter pushdown: %d filters pushed, %d remaining",
-                len(pushed_filters),
-                len(remaining_filters),
-            )
-
-        return pushed_filters, remaining_filters
-
     def _read_table_for_scan(self) -> FileStoreTable:
         if self._has_blob_columns:
             return self._table.copy({"blob-as-descriptor": "true"})
@@ -1081,12 +1051,6 @@ def _format_partition_filters(self, pushdowns: Pushdowns) -> list[str]:
         return self._format_pyexprs([getattr(pushdowns.partition_filters, "_expr", pushdowns.partition_filters)])
 
     def _filter_pushdown_explain(self, pushdowns: Pushdowns) -> tuple[list[str], list[str]]:
-        if self._remaining_filters is not None:
-            return (
-                self._format_pyexprs(self._pushed_filters or []),
-                self._format_pyexprs(self._remaining_filters),
-            )
-
         if pushdowns.filters is None:
             return [], []
 
@@ -1246,8 +1210,6 @@ def _read_pushdown_state(
         )
 
     def _pushdown_filter_state(self, pushdowns: Pushdowns) -> tuple[Predicate | None, bool]:
-        if self._remaining_filters is not None:
-            return self._paimon_predicate, not self._remaining_filters
         if pushdowns.filters is None:
             return None, True
 
@@ -1260,9 +1222,7 @@ def _planning_predicate(self, pushdown_predicate: Predicate | None) -> Predicate
             return None
         if not self._can_plan_predicate(pushdown_predicate):
             return None
-        if self._paimon_predicate is not None or self._requires_fallback_reader():
-            return pushdown_predicate
-        return None
+        return pushdown_predicate
 
     def _partition_planning_predicate(self, pushdowns: Pushdowns) -> Predicate | None:
         """partition_filters -> Paimon predicate for plan-time pruning. Excludes
@@ -1307,9 +1267,6 @@ def _source_limit(
             return None
         return pushdowns.limit
 
-    def _requires_fallback_reader(self) -> bool:
-        return not self._is_parquet or self._has_blob_columns or self._table.is_primary_key_table
-
     def _can_plan_predicate(self, predicate: Predicate) -> bool:
         # Missing value null-count stats make isNull unsafe for scan planning.
         if not self._predicate_contains_is_null(predicate):

paimon-python/pypaimon/daft/daft_paimon.py

@@ -178,22 +178,19 @@ def _read_table(
     return _source_for_table(table, catalog_options=catalog_options, io_config=io_config).read()
 
 
-def _normalize_explain_filters(filters: Any) -> tuple[Any, list[Any]]:
+def _normalize_explain_filters(filters: Any) -> Any:
     if filters is None:
-        return None, []
+        return None
 
     if isinstance(filters, (list, tuple)):
         if not filters:
-            return None, []
-        filter_exprs = list(filters)
-        combined = filter_exprs[0]
-        for filter_expr in filter_exprs[1:]:
+            return None
+        combined = filters[0]
+        for filter_expr in filters[1:]:
             combined = combined & filter_expr
-    else:
-        filter_exprs = [filters]
-        combined = filters
+        return combined
 
-    return combined, [getattr(filter_expr, "_expr", filter_expr) for filter_expr in filter_exprs]
+    return filters
 
 
 def _explain_table(
@@ -216,14 +213,10 @@ def _explain_table(
         table, snapshot_id=snapshot_id, tag_name=tag_name, timestamp=timestamp
     )
     source = _source_for_table(table, catalog_options=catalog_options, io_config=io_config)
-    filter_expr, filter_pyexprs = _normalize_explain_filters(filters)
-    partition_filter_expr, _ = _normalize_explain_filters(partition_filters)
-    if filter_pyexprs:
-        source.push_filters(filter_pyexprs)
     return source.explain_scan(
         Pushdowns(
-            filters=filter_expr,
-            partition_filters=partition_filter_expr,
+            filters=_normalize_explain_filters(filters),
+            partition_filters=_normalize_explain_filters(partition_filters),
             columns=columns,
             limit=limit,
         ),

paimon-python/pypaimon/tests/daft/daft_data_test.py

@@ -106,7 +106,6 @@ async def _read_paimon_source_batches(
     filter_expr=None,
     columns=None,
     limit=None,
-    call_push_filters=True,
 ):
     from daft import context, runners
     from daft.daft import StorageConfig
@@ -118,11 +117,6 @@ async def _read_paimon_source_batches(
     storage_config = StorageConfig(runners.get_or_create_runner().name != "ray", io_config)
     source = PaimonDataSource(table, storage_config=storage_config, catalog_options={})
 
-    if filter_expr is not None and call_push_filters:
-        pushed_filters, remaining_filters = source.push_filters([filter_expr._expr])
-        assert pushed_filters
-        assert not remaining_filters
-
     pushdowns = Pushdowns(filters=filter_expr, columns=columns, limit=limit)
     return await _collect_paimon_source_batches(source, pushdowns)
 
@@ -328,8 +322,8 @@ async def first_task():
         ) == explicit_values
 
 
-def test_read_paimon_source_serialization_preserves_pushed_filter_for_fallback(local_paimon_catalog):
-    """A serialized source must keep filters accepted by SupportsPushdownFilters."""
+def test_read_paimon_serialized_source_uses_current_filter_for_fallback(local_paimon_catalog):
+    """A serialized source must plan from the filters in the current request."""
     from daft import context, runners
     from daft.daft import StorageConfig
     from daft.io.pushdowns import Pushdowns
@@ -357,15 +351,12 @@ def test_read_paimon_source_serialization_preserves_pushed_filter_for_fallback(l
     io_config = context.get_context().daft_planning_config.default_io_config
     storage_config = StorageConfig(runners.get_or_create_runner().name != "ray", io_config)
     source = PaimonDataSource(table, storage_config=storage_config, catalog_options={})
-    pushed_filters, remaining_filters = source.push_filters([(col("id") == 999)._expr])
-    assert pushed_filters
-    assert not remaining_filters
 
     restored = loads(dumps(source))
     batches = asyncio.run(
         _collect_paimon_source_batches(
             restored,
-            Pushdowns(filters=None, limit=1),
+            Pushdowns(filters=col("id") == 999, limit=1),
         )
     )
 
@@ -664,8 +655,8 @@ def test_read_paimon_pk_fallback_filters_before_projection(pk_table):
     assert batches == [{"name": ["new_a"], "id": [1]}]
 
 
-def test_read_paimon_fallback_plans_pushdown_filter_without_push_filters(local_paimon_catalog):
-    """Fallback planning must use Pushdowns.filters even if push_filters was not called."""
+def test_read_paimon_fallback_plans_current_pushdown_filter(local_paimon_catalog):
+    """Fallback planning must use the filter from the current request."""
     catalog, _ = local_paimon_catalog
     schema = pypaimon.Schema.from_pyarrow_schema(
         pa.schema([
@@ -688,7 +679,6 @@ def test_read_paimon_fallback_plans_pushdown_filter_without_push_filters(local_p
             table,
             filter_expr=col("id") == 999,
             limit=1,
-            call_push_filters=False,
         )
     )
 
@@ -716,7 +706,6 @@ def test_read_paimon_fallback_not_in_filter_excludes_nulls_before_limit(local_pa
             table,
             filter_expr=~col("id").is_in([1, 2]),
             limit=1,
-            call_push_filters=False,
         )
     )
 
@@ -906,7 +895,6 @@ def test_row_and_partition_filters_both_reach_planning_predicate(append_only_tab
     io_config = context.get_context().daft_planning_config.default_io_config
     storage_config = StorageConfig(runners.get_or_create_runner().name != "ray", io_config)
     source = PaimonDataSource(table, storage_config=storage_config, catalog_options={})
-    source.push_filters([(col("id") > 1)._expr])
     pushdowns = Pushdowns(filters=(col("id") > 1),
                           partition_filters=(col("dt") == "2024-01-02"))
 

paimon-python/pypaimon/tests/daft/daft_integration_test.py

@@ -572,6 +572,52 @@ def test_read_paimon_filter(catalog_options):
     }
 
 
+def test_read_paimon_filter_pruning_matches_explain(catalog_options, monkeypatch):
+    from daft.io.source import DataSourceTask
+
+    pa_schema = pa.schema([
+        ("id", pa.int64()),
+        ("name", pa.string()),
+    ])
+    identifier, table = _create_table(
+        catalog_options,
+        "read_filter_pruning",
+        pa_schema,
+        options={
+            "bucket": "1",
+            "file.format": "parquet",
+            "metadata.stats-mode": "full",
+        },
+    )
+    _write_arrow(table, pa.table({"id": [1], "name": ["file-a"]}, schema=pa_schema))
+    _write_arrow(table, pa.table({"id": [999], "name": ["file-b"]}, schema=pa_schema))
+
+    planned_paths = []
+    original_parquet = DataSourceTask.parquet
+
+    def capture_parquet_task(**kwargs):
+        planned_paths.append(kwargs["path"])
+        return original_parquet(**kwargs)
+
+    monkeypatch.setattr(DataSourceTask, "parquet", capture_parquet_task)
+    predicate = col("id") == 999
+
+    result = read_paimon(identifier, catalog_options).where(predicate).to_pydict()
+    explain = explain_paimon_scan(
+        identifier,
+        catalog_options,
+        filters=predicate,
+        verbose=True,
+    )
+
+    assert result == {"id": [999], "name": ["file-b"]}
+    assert len(planned_paths) == 1
+    assert explain.total_file_count == len(planned_paths)
+    assert explain.paimon_scan.file_skipping is not None
+    assert explain.paimon_scan.file_skipping.before == 2
+    assert explain.paimon_scan.file_skipping.after == 1
+
+
 def test_read_paimon_limit(catalog_options):
     data = pa.table(
         {