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 3145e0511b [python][daft] Push partition_filters into plan-time
pruning (#8427)
3145e0511b is described below
commit 3145e0511b915472a87a3056cf875b0efb09546e
Author: XiaoHongbo <[email protected]>
AuthorDate: Thu Jul 2 18:15:40 2026 +0800
[python][daft] Push partition_filters into plan-time pruning (#8427)
Daft delivers partition predicates on `pushdowns.partition_filters`,
separate from `pushdowns.filters`. Only `filters` were pushed to
Paimon's `plan()`; `partition_filters` were applied post-plan in Python,
so `plan()` enumerated every partition (full-table plan) and pruned
nothing at the manifest level.
Convert `partition_filters` to a Paimon predicate and push it into the
plan so partitions are pruned at the manifest level.
---
paimon-python/pypaimon/daft/daft_datasource.py | 32 ++++++
.../pypaimon/tests/daft/daft_data_test.py | 120 +++++++++++++++++++++
.../pypaimon/tests/daft/daft_explain_test.py | 4 +-
3 files changed, 155 insertions(+), 1 deletion(-)
diff --git a/paimon-python/pypaimon/daft/daft_datasource.py
b/paimon-python/pypaimon/daft/daft_datasource.py
index 68acae19dc..15cbb6f694 100644
--- a/paimon-python/pypaimon/daft/daft_datasource.py
+++ b/paimon-python/pypaimon/daft/daft_datasource.py
@@ -800,6 +800,12 @@ class PaimonDataSource(DataSource):
) -> _ReadPushdownState:
reader_predicate, filters_consumed =
self._pushdown_filter_state(pushdowns)
planning_predicate = self._planning_predicate(reader_predicate)
+ # Partition filters arrive on a separate Daft channel (pushdowns.
+ # partition_filters), not in pushdowns.filters. Convert and AND them in
+ # so plan() prunes partitions at the manifest level; otherwise plan()
+ # enumerates every split and we skip in Python -- a full-table plan.
+ planning_predicate = self._and_predicates(
+ planning_predicate, self._partition_planning_predicate(pushdowns))
requested_columns = self._valid_output_columns(pushdowns.columns)
task_columns = self._task_columns(table, requested_columns, pushdowns)
read_columns = self._fallback_read_columns(table, task_columns,
reader_predicate)
@@ -837,6 +843,32 @@ class PaimonDataSource(DataSource):
return pushdown_predicate
return None
+ def _partition_planning_predicate(self, pushdowns: Pushdowns) -> Predicate
| None:
+ """partition_filters -> Paimon predicate for plan-time pruning.
Excludes
+ isNull (_predicate_contains_is_null): pruning drops the whole null
+ partition, which the post-filter can't restore -- so exclude it even
for
+ PK tables (stricter than the row path's _can_plan_predicate)."""
+ partition_filters = getattr(pushdowns, "partition_filters", None)
+ if partition_filters is None:
+ return None
+ py_expr = getattr(partition_filters, "_expr", partition_filters)
+ _, remaining, paimon_predicate =
convert_filters_to_paimon(self._table, [py_expr])
+ if remaining:
+ # Unconverted parts still apply via _partition_filter_skips_split.
+ logger.debug("Partition filter not pushed to plan: %s", remaining)
+ if paimon_predicate is None or
self._predicate_contains_is_null(paimon_predicate):
+ return None # isNull -> post-filter only (see docstring)
+ return paimon_predicate
+
+ @staticmethod
+ def _and_predicates(left: Predicate | None, right: Predicate | None) ->
Predicate | None:
+ if left is None:
+ return right
+ if right is None:
+ return left
+ from pypaimon.common.predicate_builder import PredicateBuilder
+ return PredicateBuilder.and_predicates([left, right])
+
@staticmethod
def _source_limit(
pushdowns: Pushdowns,
diff --git a/paimon-python/pypaimon/tests/daft/daft_data_test.py
b/paimon-python/pypaimon/tests/daft/daft_data_test.py
index 5520b37100..a7a1842ba2 100644
--- a/paimon-python/pypaimon/tests/daft/daft_data_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_data_test.py
@@ -765,6 +765,126 @@ def test_read_paimon_partition_filter(append_only_table):
assert all(dt == "2024-01-01" for dt in result.column("dt").to_pylist())
+def test_partition_filter_prunes_at_plan_level(append_only_table):
+ """Regression: Daft routes partition predicates to
pushdowns.partition_filters
+ (a separate channel from pushdowns.filters). They must become a plan-time
+ predicate so plan() prunes partitions, instead of planning every split and
+ skipping in Python (a full-table plan)."""
+ from daft import context, runners
+ from daft.daft import StorageConfig
+ from daft.io.pushdowns import Pushdowns
+
+ from pypaimon.daft.daft_datasource import PaimonDataSource
+
+ table, _ = append_only_table
+ data = pa.table(
+ {
+ "id": pa.array([1, 2, 3], pa.int64()),
+ "name": pa.array(["a", "b", "c"], pa.string()),
+ "value": pa.array([1.0, 2.0, 3.0], pa.float64()),
+ "dt": pa.array(["2024-01-01", "2024-01-02", "2024-01-03"],
pa.string()),
+ }
+ )
+ _write_to_paimon(table, data)
+
+ 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={})
+
+ pushdowns = Pushdowns(partition_filters=(col("dt") == "2024-01-02"))
+
+ state = source._read_pushdown_state(table, pushdowns)
+ # partition filter must become a plan-time predicate
+ assert state.planning_predicate is not None
+
+ pruned = source._scan_read_builder(table, state).new_scan().plan().splits()
+ all_splits = table.new_read_builder().new_scan().plan().splits()
+
+ # pruning reduced the planned splits, and only the matching partition
remains
+ assert 0 < len(pruned) < len(all_splits)
+ assert all(s.partition.to_dict().get("dt") == "2024-01-02" for s in pruned)
+
+
+def
test_row_and_partition_filters_both_reach_planning_predicate(append_only_table):
+ """filters and partition_filters arrive on separate Daft channels; both
must
+ fold into the planning predicate (via _and_predicates)."""
+ from daft import context, runners
+ from daft.daft import StorageConfig
+ from daft.io.pushdowns import Pushdowns
+
+ from pypaimon.daft.daft_datasource import PaimonDataSource
+
+ table, _ = append_only_table
+ _write_to_paimon(table, pa.table(
+ {
+ "id": pa.array([1, 2, 3], pa.int64()),
+ "name": pa.array(["a", "b", "c"], pa.string()),
+ "value": pa.array([1.0, 2.0, 3.0], pa.float64()),
+ "dt": pa.array(["2024-01-01", "2024-01-02", "2024-01-02"],
pa.string()),
+ }
+ ))
+
+ 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"))
+
+ state = source._read_pushdown_state(table, pushdowns)
+ # both channels combined into one AND planning predicate
+ assert state.planning_predicate is not None
+ assert state.planning_predicate.method == "and"
+
+ pruned = source._scan_read_builder(table, state).new_scan().plan().splits()
+ assert pruned and all(s.partition.to_dict().get("dt") == "2024-01-02" for
s in pruned)
+
+
+def test_isnull_partition_filter_keeps_null_partition(append_only_table):
+ """isNull must NOT be pushed to plan pruning: plan pruning would drop the
+ null partition and the Python post-filter can't restore it. It must be left
+ to the post-filter so col(dt).is_null() still returns the null-partition
row."""
+ table, _ = append_only_table
+ _write_to_paimon(table, pa.table(
+ {
+ "id": pa.array([1, 2], pa.int64()),
+ "name": pa.array(["a", "b"], pa.string()),
+ "value": pa.array([1.0, 2.0], pa.float64()),
+ "dt": pa.array([None, "2024-01-02"], pa.string()),
+ }
+ ))
+
+ result = _read_table(table).where(col("dt").is_null()).to_arrow()
+
+ assert result.num_rows == 1
+ assert result.column("id").to_pylist() == [1]
+ assert result.column("dt").to_pylist() == [None]
+
+
+def test_isnull_partition_filter_not_pushed_even_for_pk_table(pk_table):
+ """isNull must be excluded from plan pushdown for ALL tables. The row gate
+ (_can_plan_predicate) would allow isNull on a PK table without deletion
+ vectors; the partition path must be stricter (_predicate_contains_is_null),
+ else a PK table would lose its null partition."""
+ from daft import context, runners
+ from daft.daft import StorageConfig
+ from daft.io.pushdowns import Pushdowns
+
+ from pypaimon.daft.daft_datasource import PaimonDataSource
+
+ table, _ = pk_table
+ 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={})
+
+ st_null = source._read_pushdown_state(
+ table, Pushdowns(partition_filters=col("dt").is_null()))
+ assert st_null.planning_predicate is None # would be non-None via
_can_plan_predicate
+ st_eq = source._read_pushdown_state(
+ table, Pushdowns(partition_filters=col("dt") == "2024-01-02"))
+ assert st_eq.planning_predicate is not None
+
+
def test_read_paimon_row_filter(append_only_table):
"""Row-level filter should be applied after reading data."""
table, _ = append_only_table
diff --git a/paimon-python/pypaimon/tests/daft/daft_explain_test.py
b/paimon-python/pypaimon/tests/daft/daft_explain_test.py
index 0846f59121..d0e6fbb387 100644
--- a/paimon-python/pypaimon/tests/daft/daft_explain_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_explain_test.py
@@ -278,7 +278,9 @@ def
test_explain_scan_applies_partition_filters_to_reader_counts(catalog_options
verbose=True,
)
- assert result.paimon_scan.split_count == 2
+ # partition_filters are pushed into the plan, so it prunes to the matching
+ # partition (1 split) instead of planning both and skipping in Python.
+ assert result.paimon_scan.split_count == 1
assert result.native_parquet_split_count == 1
assert result.pypaimon_fallback_split_count == 0
assert any("dt" in partition_filter for partition_filter in
result.partition_filters)
