[Python] Keep Native read planning in the selected execution backend

paimon-python/README.md

@@ -210,7 +210,7 @@ view or retained snapshot metadata. Current main-branch REST table sources
 with `query-auth.enabled=true` also use core when the catalog returns no row
 filter or column mask. Core requests authorization for every source column,
 even for a key-only DELETE, and refreshes it for each MERGE. A restricted or
-denied source fails before target actions, as with Python's `plan_all_files`.
+denied source fails before target actions, as with Python's `plan_all`.
 
 Python and native writers use Java's escaped partition directories for data,
 changelog, Blob and row sidecar files, including external locations. Both
@@ -561,7 +561,7 @@ live REST authorization and snapshot-file, deletion-vector and merge-group
 validation. Transported metadata never supplies a live grant. If restrictions
 changed incompatibly, the captured plan retains Python execution.
 Continuous streaming delegates planning and resume state to Rust core.
-`TableScan.plan_all_files()` includes L0 and every data-evolution column file,
+`TableScan.plan_all()` returns a split plan including L0 and every data-evolution column file,
 retaining scan filters and limits and requiring unrestricted query authorization.
 Writer restore, overwrite and index construction use this entrypoint and delegate
 planning to Rust core when Native planning is enabled.

paimon-python/conftest.py

@@ -107,7 +107,7 @@ def tracked_plan(self, *args, **kwargs):
 
         TableScan._try_native_plan = tracked_plan
 
-        from pypaimon.read.streaming_table_scan import StreamTableScan
+        from pypaimon.execution.native.read.streaming_table_scan import StreamTableScan
 
         original_stream_plan = StreamTableScan.plan
 

paimon-python/pypaimon/execution/README.md

@@ -98,8 +98,18 @@ Neither backend imports the other. Native adapters do not construct or retain
 Python executors. Python implementations do not load the optional Native SDK
 just because it is installed. Planning, reading, writing and committing retain
 independent options; shared plans and commit messages connect those stages.
-Explicit access to an existing Python scanner/router/committer property selects
-that implementation. Ordinary configuration and metadata inspection do not.
+The public `TableScan.file_scanner`, `AsyncStreamingTableScan` and
+`StreamTableScan` are backend-neutral facades, including direct construction.
+Accessing file configuration does not construct a Python scanner. File plans
+require unrestricted whole-table authorization because they expose raw file
+metadata. Internal writer metadata restoration uses the selected executor's
+private scanner and retains its write authorization boundary.
+
+Manifest-source and follow-up-scanner callbacks belong to Python execution
+internals. Public scanner configuration selects the existing snapshot, delta
+or changelog strategies without accepting custom scanner callbacks. Explicit
+Python router/committer properties retain their implementation-specific APIs.
+Ordinary configuration and metadata inspection do not select that backend.
 
 Deletion-vector codecs, file loading and row filtering belong to Python
 execution, corresponding to Rust Core's deletion-vector implementation.
@@ -200,7 +210,7 @@ Message conversion happens during preparation; publication uses the selected
 executor's prepared messages. Batch/stream binding details stay in the Native
 adapter, and overwrite details stay in each backend.
 
-Both batch scanners expose `plan(auth_result=None)`, `plan_all_files()` and
+Both batch scanners expose `plan(auth_result=None)`, `plan_all()` and
 `scan_with_stats(auth_result=None)`. Both readers expose `read_rows(splits)` and
 `read_batches(splits, schema, parallelism=None, blob_parallelism=1,
 ordered=False)`. Output materialization and Python parallel-read helpers are

paimon-python/pypaimon/execution/native/read/file_scan.py

@@ -0,0 +1,39 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Lossless metadata adaptation for Core file scans."""
+
+from pypaimon.manifest.schema.manifest_entry import ManifestEntry
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.utils.file_store_path_factory import canonical_data_file_path
+from pypaimon.write.commit_message_serializer import deserialize_commit_message
+
+
+def from_native_file_plan(table, plan):
+    entries = []
+    for entry in plan.files():
+        # Reuse the existing v14 file-metadata bridge, including historical
+        # stats, external paths, BLOB descriptors and row-tracking metadata.
+        message = deserialize_commit_message(
+            entry._to_commit_message().serialize(), table.partition_keys_fields,
+            table.trimmed_primary_keys_fields)
+        file = message.new_files[0]
+        file.file_path = (file.physical_path() if file.external_path else canonical_data_file_path(
+            table, message.partition, entry.bucket, file.file_name))
+        entries.append(ManifestEntry(
+            entry.kind, GenericRow(list(message.partition), table.partition_keys_fields),
+            entry.bucket, entry.total_buckets, file))
+    return entries, plan.snapshot_id()

paimon-python/pypaimon/execution/native/read/plan.py

@@ -22,14 +22,13 @@
 from typing import List, Optional, Tuple
 
 from pypaimon.common.identifier import UNKNOWN_DATABASE
-from pypaimon.common.options.core_options import CoreOptions, ChangelogProducer
+from pypaimon.common.options.core_options import CoreOptions
 from pypaimon.common.predicate import Predicate
 from pypaimon.execution.native import runtime
 from pypaimon.execution.native import table as native_table
 from pypaimon.execution.native.read import builder as native_builder
 from pypaimon.read.plan import Plan
 from pypaimon.read.query_auth_split import QueryAuthSplit
-from pypaimon.read.scan_stats import ScanStats
 from pypaimon.read.scan_request import ScanRequest
 from pypaimon.read import split_serializer
 
@@ -56,6 +55,7 @@ def native_plan(
         shard: Optional[Tuple[int, int]] = None,
         partition_predicate: Optional[Predicate] = None,
         global_index_result=None,
+        file_filters: Optional[Tuple[bool, bool]] = None,
         all_files: bool = False) -> Plan:
     """Plan with pypaimon_rust, preserving snapshot metadata.
 
@@ -84,6 +84,8 @@ def native_plan(
         scan = builder.new_incremental_scan(*incremental_range)
     else:
         scan = builder.new_incremental_scan(*incremental_range, incremental_mode)
+    if file_filters is not None:
+        scan = scan.with_file_filters(*file_filters)
     if global_index_result is not None and not table.options.query_auth_enabled:
         # A PK result already selected a snapshot and physical files. Java's
         # PrimaryKeyBatchScan returns those selections before ordinary planning.
@@ -120,7 +122,7 @@ def native_plan(
         scan = scan.with_chunk_shuffle(str(seed), chunk_size)
     if shard is not None:
         scan = scan.with_shard(*shard)
-    rust_plan = scan.plan_all_files() if all_files else scan.plan()
+    rust_plan = scan.plan_all() if all_files else scan.plan()
     return _from_native_plan(table, rust_plan)
 
 
@@ -168,9 +170,28 @@ def plan(self, auth_result=None):
         # Core requests and captures query authorization with its own plan.
         return self._plan()
 
-    def plan_all_files(self):
+    def plan_all(self):
         return self._plan(all_files=True)
 
+    def plan_files(self, skip_level0, only_read_real_buckets):
+        """Plan complete file metadata before query result transformations."""
+        try:
+            request = self._scan._scan_request()
+            builder = native_builder._configure_native_read_builder(
+                native_builder._native_read_builder(self._scan.table),
+                request.predicate, None, None)
+            if request.partition_filter is not None:
+                builder = builder.with_partition_filter(native_builder._predicate_to_native(
+                    request.partition_key_predicate))
+            plan = builder.new_scan().with_file_filters(
+                skip_level0, only_read_real_buckets).plan_files()
+            from pypaimon.execution.native.read.file_scan import from_native_file_plan
+            return from_native_file_plan(self._scan.table, plan)
+        except Exception as error:
+            runtime._raise_if_native_fork_safety_error(error)
+            logger.warning("Native file planning failed, falling back to Python: %s", error)
+            return None
+
     def scan_with_stats(self, auth_result=None):
         # Preserve explicit facade hooks while Core records counters as part
         # of the ordinary planning pass.
@@ -272,12 +293,8 @@ def _plan(self, all_files=False) -> Optional[Plan]:
             has_distribution = (request.idx_of_this_subtask is not None
                                 or request.start_pos_of_this_subtask is not None)
             extra_options = {}
-            if scan.table.options.options.contains(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP):
-                if request.incremental_snapshot_range is None:
-                    return Plan([], _scan_stats=ScanStats.empty_native(request.read_limit))
-                extra_options['incremental_range'] = request.incremental_snapshot_range
-                if scan.table.options.changelog_producer() != ChangelogProducer.NONE:
-                    extra_options['incremental_mode'] = 'changelog'
+            # Core resolves timestamp options after authorization, including
+            # empty windows. Never preselect snapshots with Python here.
             chunk_shuffle = request.chunk_shuffle
             if chunk_shuffle is not None:
                 extra_options['chunk_shuffle'] = chunk_shuffle
@@ -296,6 +313,8 @@ def _plan(self, all_files=False) -> Optional[Plan]:
             row_ranges = list(request.row_ranges) if request.row_ranges is not None else None
             if request.global_index_result is not None:
                 extra_options['global_index_result'] = request.global_index_result
+            if request.file_filters is not None:
+                extra_options['file_filters'] = request.file_filters
             if request.partition_filter is not None:
                 # Explicit partition selectors are scan-only. Keep the ordinary
                 # row predicate separate so core can apply Java's override and

paimon-python/pypaimon/execution/native/read/streaming_table_scan.py

@@ -141,6 +141,9 @@ def _check_generation(self, generation):
         if generation != self._generation:
             raise RuntimeError("stream scan was restored during planning")
 
+    def cancel_prefetch(self):
+        """Native plans frames on demand, with no speculative frame to cancel."""
+
     def close(self):
         """Stop polling without acknowledging a plan still owned by the caller."""
         self._closed = True

paimon-python/pypaimon/execution/native/write/bucket_metadata.py

@@ -24,7 +24,7 @@ def dynamic_bucket_data_buckets(table, partition, snapshot):
     from pypaimon.index.dynamic_bucket import bucket_read_builder
 
     read_builder = bucket_read_builder(table, partition, snapshot)
-    # plan_all_files includes level zero and requires unrestricted query
+    # plan_all includes level zero and requires unrestricted query
     # authorization. A policy which cannot supply complete metadata must fail
     # preparation before input assignment, never omit buckets from validation.
     plan = native_plan(read_builder.table, predicate=read_builder._predicate, all_files=True)

paimon-python/pypaimon/execution/native/write/native_merge_into.py

@@ -81,7 +81,7 @@ def create_native_merge_into(table, source, on, matched, not_matched, commit_use
             if not _rest_catalog_supported(source) or not _native_query_auth_supported(source):
                 return None
             # A source is a read, not a write. Core obtains a fresh REST grant
-            # and requires it to be unrestricted, as Python plan_all_files does.
+            # and requires it to be unrestricted, as Python plan_all does.
             native_source = _native_table(source)
         else:
             native_source = create_native_write_table(source)

paimon-python/pypaimon/execution/python/globalindex/create_global_index.py

@@ -78,7 +78,7 @@ def build(self, partition_filter, *, execution="local", concurrency=None, ray_re
             read_builder = read_builder.with_partition_filter(partition_filter)
 
         scan = read_builder.new_scan()
-        plan = scan.plan_all_files()
+        plan = scan.plan_all()
         splits = plan.splits()
         if not splits:
             return []

paimon-python/pypaimon/execution/python/read/auth_support.py

@@ -29,7 +29,10 @@ def prune_scanner_by_auth(table, scanner, auth_result):
     predicate = scanner._user_predicate
     effective = exclude_predicate_with_fields(predicate, masked)
     if effective != scanner.predicate:
+        partition_key_predicate = scanner.partition_key_predicate
         scanner._set_predicate(effective)
+        if scanner._explicit_partition_filter:
+            scanner.partition_key_predicate = partition_key_predicate
     scanner.auth_partition_predicate = None
     scanner.auth_has_non_partition_filter = bool(
         predicate is not None and masked & predicate_field_names(predicate))

paimon-python/pypaimon/execution/python/read/merge_engine_support.py

@@ -81,6 +81,14 @@
 _DEFAULT_AGGREGATE_FUNCTION_KEY = "fields.default-aggregate-function"
 
 
+def validate_read(read) -> None:
+    """Check Python merge capabilities once, after this backend is selected."""
+    if getattr(read, '_python_merge_validated', False):
+        return
+    check_supported(read.table)
+    read._python_merge_validated = True
+
+
 def check_supported(table) -> None:
     """Raise ``NotImplementedError`` if the table's merge-engine
     configuration is outside what pypaimon's read path implements, or
@@ -206,6 +214,6 @@ def _option_is_truthy(raw) -> bool:
     return bool(raw)
 
 __all__ = [
-    "check_supported", "check_sequence_field_valid", "check_sequence_field_supported",
+    "validate_read", "check_supported", "check_sequence_field_valid", "check_sequence_field_supported",
     "partial_update_unsupported_options", "aggregation_unsupported_options",
 ]

paimon-python/pypaimon/execution/python/read/streaming_table_scan.py

@@ -294,6 +294,10 @@ def _discard_prefetch(self):
         if future is not None:
             future.cancel()
 
+    def cancel_prefetch(self):
+        """Discard speculative work when a public polling iterator stops."""
+        self._discard_prefetch()
+
     def close(self):
         """Discard speculative work without acknowledging pending rows.