[python] Use Native scan statistics in explain

paimon-python/README.md

@@ -362,7 +362,23 @@ print(explanation.native_planned)
 The adapter checks the installed binding's capabilities and falls back to the
 Python planner for unsupported scans. New bindings preserve `plan.snapshot_id`
 even when pruning removes every split. Native explain output includes snapshot
-and split metadata; native pruning counters are not exposed.
+and split metadata plus counters for actual Native planning work, available in
+`explanation.planning_counters`. Statistics collection retains early manifest,
+sidecar and LIMIT optimizations. The Python scanner's before/after pruning funnel
+is unavailable for Native plans because skipped entries are not decoded.
+Final split, file and byte totals describe the returned Plan after Python's
+post-processing and deferred LIMIT; decoding counters retain the Native work.
+Every Native plan includes lightweight counters. Use
+`plan = builder.new_scan().plan()` followed by `plan.stats()`; conversion to
+Python counters is deferred until this getter is called. The existing
+`plan, stats = builder.new_scan().scan_with_stats()` API remains available
+for both planners. Snapshot metadata is available separately as `plan.snapshot_id`.
+
+Fixed-bucket Native planning also handles ORs of equality/IN predicates,
+intersects repeated key constraints, and specializes predicates per partition.
+It follows Java's 1000-candidate combination limit and keeps scans conservative
+when a predicate cannot determine buckets. Masked fields do not drive raw-value
+bucket pruning.
 
 To run both split planning and data-file reading in Rust, enable the independent
 native-read option:

paimon-python/conftest.py

@@ -104,9 +104,9 @@ def pytest_configure(config):
 
         original_plan = TableScan._try_native_plan
 
-        def tracked_plan(self):
+        def tracked_plan(self, *args, **kwargs):
             global _native_plan_count
-            plan = original_plan(self)
+            plan = original_plan(self, *args, **kwargs)
             if plan is not None and _force_native_for_test:
                 _native_plan_count += 1
             return plan

paimon-python/pypaimon/catalog/table_query_auth.py

@@ -17,6 +17,7 @@
 ################################################################################
 
 import json
+from dataclasses import replace
 from typing import Callable, Dict, List, Optional
 
 import pyarrow as pa
@@ -26,7 +27,6 @@
     extract_referenced_fields,
     parse_predicate_to_batch_filter,
 )
-from pypaimon.read.plan import Plan
 from pypaimon.read.query_auth_split import QueryAuthSplit
 from pypaimon.schema.data_types import DataField
 from pypaimon.table.special_fields import SpecialFields
@@ -132,7 +132,7 @@ def convert_plan(self, plan):
         if not self.has_restrictions:
             return plan
         auth_splits = [QueryAuthSplit(split, self) for split in plan.splits()]
-        return Plan(auth_splits, snapshot_id=plan.snapshot_id)
+        return replace(plan, _splits=auth_splits)
 
     def extract_row_filter(self) -> Optional[Callable[[pa.RecordBatch], pa.Array]]:
         if not self.filter:

paimon-python/pypaimon/read/explain.py

@@ -130,9 +130,10 @@ class ExplainResult:
     # Verbose-only
     splits: Optional[List[ExplainSplitInfo]] = None
 
-    # Native (pypaimon_rust) plan; pruning funnel not tracked. Kept last to
-    # preserve positional-arg compatibility.
+    # Whether the same core Native pass produced the plan and pruning counters.
     native_planned: bool = False
+    # Actual work performed by Native planning, retaining early metadata pruning.
+    planning_counters: Optional[Dict[str, Any]] = None
 
     def __str__(self) -> str:
         return render_explain(self)
@@ -168,14 +169,28 @@ def render_explain(result: ExplainResult) -> str:
 
     out.write("\n")
     if result.native_planned:
-        _line(out, "Planner", "native (pypaimon_rust); pruning not tracked")
+        _line(out, "Planner", "native (pypaimon_rust)")
     _line(out, "Partition pruning",
           result.partition_pruning.format() if result.partition_pruning else "n/a")
     _line(out, "Bucket pruning",
           result.bucket_pruning.format() if result.bucket_pruning else "n/a")
     _line(out, "File skipping",
           result.file_skipping.format() if result.file_skipping else "n/a")
 
+    if result.planning_counters is not None:
+        out.write("\nNative planning counters\n")
+        for label, name in (
+            ('Manifest entries decoded', 'manifest_entries_read'),
+            ('Bucket rejects', 'manifest_entries_pruned_by_bucket'),
+            ('Partition rejects', 'manifest_entries_pruned_by_partition'),
+            ('File statistics rejects', 'manifest_entries_pruned_by_data_stats'),
+            ('Cross-schema rejects', 'manifest_entries_pruned_by_cross_schema_stats'),
+            ('Manifest row-range skips', 'manifest_files_pruned_by_row_ranges'),
+            ('Split candidates built', 'split_candidates_built'),
+            ('LIMIT stopped planning', 'limit_early_stopped'),
+        ):
+            _line(out, label, str(result.planning_counters[name]))
+
     out.write("\n")
     _line(out, "Splits", str(result.split_count))
     if result.split_count > 0:

paimon-python/pypaimon/read/native_plan.py

@@ -514,7 +514,8 @@ def native_plan(
         scan = scan.with_chunk_shuffle(str(seed), chunk_size)
     if shard is not None:
         scan = scan.with_shard(*shard)
-    return _from_native_plan(table, scan.plan())
+    rust_plan = scan.plan()
+    return _from_native_plan(table, rust_plan)
 
 
 def _native_stream_auth_supported(table):
@@ -576,4 +577,4 @@ def _from_native_plan(table, rust_plan) -> Plan:
         # Older bindings cannot distinguish an empty committed snapshot from a
         # table without snapshots. Let the Python scanner recover the metadata.
         raise RuntimeError("Native runtime cannot report an empty plan's snapshot")
-    return Plan(splits, snapshot_id=snapshot_id)
+    return Plan(splits, snapshot_id=snapshot_id, _native_stats=rust_plan.stats())

paimon-python/pypaimon/read/plan.py

@@ -16,17 +16,47 @@
 # under the License.
 
 from dataclasses import dataclass, field
-from typing import List, Optional
+from typing import Any, List, Optional
 
 
 from pypaimon.read.split import Split
+from pypaimon.read.scan_stats import ScanStats
 
 
 @dataclass
 class Plan:
     """Implementation of Plan for native Python reading."""
     _splits: List[Split]
     snapshot_id: Optional[int] = field(default=None)
+    _scan_stats: Optional[ScanStats] = field(default=None, repr=False, compare=False)
+    _native_stats: Any = field(default=None, repr=False, compare=False)
+    # A LIMIT hint deferred until after Native row-position selection.
+    _stats_limit: Optional[int] = field(default=None, repr=False, compare=False)
 
     def splits(self) -> List[Split]:
         return self._splits
+
+    def stats(self) -> Optional[ScanStats]:
+        """Planning counters, converted from Native only when requested."""
+        native_stats = self._native_stats
+        if native_stats is not None:
+            self._scan_stats = ScanStats.from_native(native_stats)
+            # Python can filter or limit splits after Native planning. Keep the
+            # actual planning work, but summarize the final returned selection.
+            files, data_bytes = 0, 0
+            for split in self._splits:
+                for file in split.files:
+                    files += 1
+                    data_bytes += max(0, file.file_size)
+            counters = self._scan_stats.planning_counters
+            counters.update(final_splits=len(self._splits), splits_after_limit=len(self._splits),
+                            final_files=files, planned_data_file_bytes=min(data_bytes, 2 ** 64 - 1))
+            if self._stats_limit is not None:
+                counters['limit'] = self._stats_limit
+            self._native_stats = None
+        return self._scan_stats
+
+    def __getstate__(self):
+        # Native handles are process-local. Serialize counters as Python scalars.
+        self.stats()
+        return self.__dict__.copy()

paimon-python/pypaimon/read/read_builder.py

@@ -226,8 +226,8 @@ def explain(self, verbose: bool = False) -> ExplainResult:
 
         Cost: ``explain()`` reads manifest list + manifests but never
         opens data files. To produce accurate before/after counters it
-        suppresses the manifest-reader's early bucket filter and forces
-        single-threaded manifest decoding, so it can be measurably
+        suppresses early bucket and sidecar pruning and counts candidates
+        before applying LIMIT, so it can be measurably
         heavier than a regular ``new_scan().plan()`` on tables where the
         early filter usually prunes aggressively (e.g. very wide
         HASH_FIXED tables with a tight predicate).
@@ -482,9 +482,8 @@ def _build_explain_result(table, scan: TableScan, plan, stats: Optional[ScanStat
     table_schema = table.table_schema
     bucket_mode_str = _safe_bucket_mode(table)
 
-    # Native plans expose split metadata without Python pruning counters.
-    native_planned = stats is None
-    if native_planned:
+    native_planned = stats is None or stats.native_planned
+    if stats is None or native_planned:
         partition_pruning = bucket_pruning = file_skipping = None
     else:
         partition_pruning = _partition_pruning(stats, scan)
@@ -586,6 +585,7 @@ def _build_explain_result(table, scan: TableScan, plan, stats: Optional[ScanStat
         split_size_p95=sz_p95,
         has_auth=plan_has_auth,
         native_planned=native_planned,
+        planning_counters=stats.planning_counters if stats is not None else None,
         splits=split_infos if verbose else None,
     )
 

paimon-python/pypaimon/read/scan_stats.py

@@ -16,7 +16,38 @@
 # under the License.
 
 from dataclasses import dataclass, field
-from typing import Set, Tuple
+from typing import Any, Dict, Optional, Set, Tuple
+
+
+_NATIVE_COUNTER_FIELDS = (
+    'base_manifest_files',
+    'delta_manifest_files',
+    'manifest_files_before_partition_pruning',
+    'manifest_files_after_partition_pruning',
+    'manifest_files_pruned_by_row_ranges',
+    'manifest_entries_read',
+    'manifest_entries_pruned_by_bucket',
+    'manifest_entries_pruned_by_partition',
+    'manifest_entries_after_entry_pruning',
+    'manifest_entries_pruned_by_level',
+    'manifest_entries_pruned_by_row_ranges',
+    'manifest_entries_pruned_by_data_stats',
+    'manifest_entries_after_manifest_filters',
+    'manifest_entries_after_merge',
+    'manifest_entries_pruned_by_cross_schema_stats',
+    'manifest_entries_after_cross_schema_stats',
+    'data_evolution_groups_before_stats',
+    'data_evolution_groups_pruned_by_stats',
+    'data_evolution_groups_pruned_by_row_ranges',
+    'split_candidates_built',
+    'limit_early_stopped',
+    'splits_before_limit',
+    'splits_after_limit',
+    'final_splits',
+    'final_files',
+    'planned_data_file_bytes',
+    'limit',
+)
 
 
 @dataclass
@@ -38,14 +69,42 @@ class ScanStats:
     # manifest-level partition skip. ``entries_total`` is what actually reached
     # ``_filter_manifest_entry``; the gap between the two captures
     # manifest-level pruning.
-    entries_potential_total: int = 0
+    entries_potential_total: Optional[int] = 0
     entries_total: int = 0
-    entries_after_partition: int = 0
-    entries_after_bucket: int = 0
-    entries_after_stats: int = 0
+    entries_after_partition: Optional[int] = 0
+    entries_after_bucket: Optional[int] = 0
+    entries_after_stats: Optional[int] = 0
+
+    # Identity sets are Python-scanner bookkeeping. Native returns counters
+    # without transporting partition values back across the language boundary.
+    partition_keys_before: Optional[Set[Tuple]] = field(default_factory=set)
+    partition_keys_after: Optional[Set[Tuple]] = field(default_factory=set)
+
+    buckets_seen: Optional[Set[Tuple[Tuple, int]]] = field(default_factory=set)
+    buckets_after_pruning: Optional[Set[Tuple[Tuple, int]]] = field(default_factory=set)
+    native_planned: bool = False
+    # Native counters describe actual planning work; skipped metadata is not decoded
+    # to reconstruct the Python scanner's before/after funnel.
+    planning_counters: Optional[Dict[str, Any]] = None
+
+    @classmethod
+    def from_native(cls, stats):
+        return cls._from_native_counters({name: getattr(stats, name) for name in _NATIVE_COUNTER_FIELDS})
 
-    partition_keys_before: Set[Tuple] = field(default_factory=set)
-    partition_keys_after: Set[Tuple] = field(default_factory=set)
+    @classmethod
+    def empty_native(cls, limit=None):
+        counters = dict.fromkeys(_NATIVE_COUNTER_FIELDS, 0)
+        counters.update(limit=limit, limit_early_stopped=False)
+        return cls._from_native_counters(counters)
 
-    buckets_seen: Set[Tuple[Tuple, int]] = field(default_factory=set)
-    buckets_after_pruning: Set[Tuple[Tuple, int]] = field(default_factory=set)
+    @classmethod
+    def _from_native_counters(cls, counters):
+        return cls(
+            manifest_files_total=counters['manifest_files_before_partition_pruning'],
+            manifest_files_after_partition=counters['manifest_files_after_partition_pruning'],
+            entries_total=counters['manifest_entries_read'],
+            entries_potential_total=None, entries_after_partition=None,
+            entries_after_bucket=None, entries_after_stats=None,
+            planning_counters=counters,
+            partition_keys_before=None, partition_keys_after=None,
+            buckets_seen=None, buckets_after_pruning=None, native_planned=True)

paimon-python/pypaimon/read/scanner/bucket_select_converter.py

@@ -392,7 +392,7 @@ def __call__(self, *args) -> bool:
         # ``total_buckets <= 0`` shows up for postpone / legacy / special
         # entries and must NOT be pruned: returning False here would drop
         # rows the writer hashed under a different convention. Fail open.
-        if total_buckets <= 0:
+        if bucket < 0 or total_buckets <= 0:
             return True
         try:
             return bucket in self._compute(partition, total_buckets)
@@ -430,8 +430,9 @@ def _compute(self, partition, total_buckets: int) -> FrozenSet[int]:
                 self._predicate, partition_field_names, partition_values)
 
         if effective_predicate is _AlwaysFalse:
-            # No row in this partition can match — empty bucket set.
-            frozen: FrozenSet[int] = frozenset()
+            # Java cannot extract Equal/In values from AlwaysFalse. Leave
+            # rejection to the partition predicate rather than the key hash.
+            frozen: FrozenSet[int] = frozenset(range(total_buckets))
             self._cache[cache_key] = frozen
             return frozen
 

paimon-python/pypaimon/read/scanner/file_scanner.py

@@ -699,6 +699,7 @@ def scan_with_stats(self) -> Tuple[Plan, ScanStats]:
         """
         self.scan_stats = ScanStats()
         plan = self.scan()
+        plan._scan_stats = self.scan_stats
         return plan, self.scan_stats
 
     def with_chunk_shuffle(self, seed: int, chunk_size: int) -> 'FileScanner':

paimon-python/pypaimon/read/table_scan.py

@@ -18,6 +18,7 @@
 import json as _json
 import logging
 import re
+from dataclasses import replace
 from typing import List, Optional, Tuple
 
 from pypaimon.catalog.catalog_exception import TableNoPermissionException
@@ -200,7 +201,7 @@ def _try_native_plan(self) -> Optional[Plan]:
             extra_options = {}
             if self.table.options.options.contains(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP):
                 if self._incremental_snapshot_range is None:
-                    return Plan([])
+                    return Plan([], _scan_stats=ScanStats.empty_native(self.limit))
                 extra_options['incremental_range'] = self._incremental_snapshot_range
                 if self.table.options.changelog_producer() != ChangelogProducer.NONE:
                     extra_options['incremental_mode'] = 'changelog'
@@ -267,7 +268,7 @@ def _try_native_plan(self) -> Optional[Plan]:
                 if not (getattr(fs, 'deletion_vectors_enabled', False)
                         and not self.table.is_primary_key_table):
                     splits = fs._apply_push_down_limit(splits)
-            return Plan(splits, snapshot_id=plan.snapshot_id)
+            return replace(plan, _splits=splits, _stats_limit=self.limit if defer_limit else None)
         except Exception as e:
             _raise_if_native_fork_safety_error(e)
             # Any native construction/planning/pruning failure -> fall back.
@@ -296,23 +297,23 @@ def scan_with_stats(self) -> Tuple[Plan, Optional[ScanStats]]:
         """Run :meth:`plan` while recording manifest / pruning counters.
 
         Only used by :meth:`ReadBuilder.explain`; the regular read path
-        keeps going through :meth:`plan`. Native plans return stats=None;
-        explain reports their split metadata without pruning counters.
+        keeps going through :meth:`plan`. Native scans always collect lightweight
+        counters in the same core planning pass, without opening data files.
         """
         native_auth = (self.table.options.query_auth_enabled
                        and self._native_requested() and self._native_plan_supported())
         if native_auth:
             native = self._try_native_plan()
             if native is not None:
-                return native, None
+                return native, native.stats()
         auth_result = self.__auth_query()
         if self._query_auth_fn is not None:
             prune_scanner_by_auth(self.table, self.file_scanner, auth_result)
         if (not native_auth and auth_result is None and self._native_requested()
                 and self._native_plan_supported()):
             native = self._try_native_plan()
             if native is not None:
-                return native, None
+                return native, native.stats()
         plan, stats = self.file_scanner.scan_with_stats()
         return wrap_plan_with_auth(auth_result, plan), stats