[python] Expand native planning across resolved tables and scan modes

paimon-python/README.md

@@ -50,6 +50,23 @@ 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.
 
+With Rust main's `Table.from_resolved_schema()` binding, filesystem and JDBC catalog
+tables preserve the Python table's resolved schema and complete effective
+options. Stale table objects, historical schemas, and `copy()` overrides or
+option removals no longer require catalog reloading or Python planning.
+Tables opened with `FileStoreTable.from_path(path, file_io_options=None)` use
+the same path with standard local, PyArrow or resolving FileIO. Storage options
+configure FileIO; use `copy()` for table read options.
+JDBC planning uses the resolved table location and storage properties without
+opening another database connection.
+REST tables use `Table.copy_with_resolved_schema()` to preserve the same schema
+and option semantics, including branches whose schemas are catalog-managed.
+The native table retains REST credentials, token refresh and catalog snapshot
+resolution. Database and table names containing dots are passed as separate
+identifier components. REST snapshot results (including empty results) take precedence over
+filesystem snapshots. REST errors, including HTTP 501, are propagated as in Java.
+Custom catalog/FileIO contexts still fall back when they cannot be reproduced.
+
 Explicit row ranges on data-evolution tables require `ReadBuilder.with_row_ranges()`.
 Watermark time travel requires Rust 0.4 or newer. Branch reads require the
 branch-aware binding exposing `Table.branch()`, and the resolved branch is
@@ -90,14 +107,28 @@ bucket sharding. Cross-partition key migration is maintained by the writer's ind
 Batch first-row scans follow Java and exclude un-compacted level-0 files; they can
 use native planning. With deletion vectors, batch scans exclude level 0 unless
 `deletion-vectors.merge-on-read=true`, in which case overlapping key ranges stay
-together for reader-side merging. Write scans and incremental scans retain level 0.
+together when they include L0 and require reader-side merging. Fully materialized
+DV files across levels use raw splits, including first-row clustering tables.
+First-row L0 runs can use native planning, including plans with materialized files
+in separate raw splits. Plans that require merging clustered materialized files
+still fall back to Python. Readers preserve physical row positions until deletion
+vectors are applied, then evaluate residual predicates after merging.
+Write scans and incremental scans retain level 0.
+
+Append and data-evolution chunk shuffle use Rust file and deletion-vector planning.
+Python retains live-row chunk sizing, seeded shuffle order and balanced worker
+assignment, so the same seed selects the same chunks with either planner.
+Projection does not remove aligned column files before chunk construction.
+Chunk shuffle supports partition predicates, deletion vectors and timestamp
+incremental scans; its existing restrictions on limits, slices, row ranges and
+global-index results still apply.
 
 Scored global-index results on data-evolution append tables use native row-range
 planning; Python attaches scores to the selected ranges and reads the data.
 Primary-key sorted indexes refine native batch splits through Python's existing
 index reader, preserving merge-required splits and the selected snapshot.
 
-Chunk shuffle, query authorization, batch first-row scans explicitly including L0,
+Query authorization, first-row plans mixing L0 with merge-required materialized files,
 and precomputed primary-key global-index results still use the Python planner.
 Continuous streaming and write planning also retain their Python entrypoints.
 Native planning remains optional and is disabled by default.

paimon-python/pypaimon/common/json_util.py

@@ -148,6 +148,8 @@ def __from_dict(data: Dict[str, Any], target_class: Type[T]) -> T:
                     kwargs[field_name] = (
                         None if value is None else decoder_mapping[json_name](value)
                     )
+                elif value is None:
+                    kwargs[field_name] = None
                 elif json_name in type_mapping:
                     tp = getattr(type_mapping[json_name], '__origin__', None)
                     if tp in (list, List):

paimon-python/pypaimon/deletionvectors/apply_deletion_vector_reader.py

@@ -75,6 +75,7 @@ def __init__(
         self._reader = reader
         self._deletion_vector = deletion_vector
         self._returned_position = 0
+        self._record_iterator = None
 
     def reader(self) -> RecordReader:
         return self._reader
@@ -108,12 +109,18 @@ def read_batch(self) -> Optional[RecordIterator]:
         Returns:
             A RecordIterator with deletion filtering, or None if no more data.
         """
+        if self._record_iterator is not None:
+            self._returned_position = self._record_iterator.returned_position() + 1
         batch = self._reader.read_batch()
 
         if batch is None:
             return None
 
-        return ApplyDeletionRecordIterator(batch, self._deletion_vector)
+        # Positions address the whole file, including rows deleted in earlier
+        # batches. Starting each iterator at zero would apply the DV repeatedly.
+        self._record_iterator = ApplyDeletionRecordIterator(
+            batch, self._deletion_vector, self._returned_position)
+        return self._record_iterator
 
     def close(self):
         self._reader.close()
@@ -129,17 +136,19 @@ def __init__(
         self,
         iterator: RecordIterator,
         deletion_vector,
+        start_position: int = 0,
     ):
         """
         Initialize an ApplyDeletionRecordIterator.
 
         Args:
             iterator: The underlying record iterator.
             deletion_vector: The deletion vector to apply for filtering.
+            start_position: Position of the first record in this batch.
         """
         self._iterator = iterator
         self._deletion_vector = deletion_vector
-        self._returned_position = -1
+        self._returned_position = start_position - 1
 
     def iterator(self) -> RecordIterator:
         return self._iterator

paimon-python/pypaimon/read/native_plan.py

@@ -114,16 +114,23 @@ def _catalog_options(table) -> dict:
     loader = getattr(getattr(table, 'catalog_environment', None), 'catalog_loader', None)
     if loader is None:
         raise ValueError("native_plan requires a catalog-backed table (no catalog loader)")
+    metastore = _catalog_metastore(loader)
+    if metastore is None:
+        raise ValueError("native_plan requires an exact built-in catalog loader")
+    normalized = _catalog_context_options(table)
+    normalized[CatalogOptions.METASTORE.key()] = metastore
+    return normalized
+
+
+def _catalog_context_options(table) -> dict:
+    """Normalize catalog storage properties without rebuilding the metastore."""
+    loader = table.catalog_environment.catalog_loader
     options = loader.context().options.to_map()
     normalized = {
         str(key): _option_value_to_string(value)
         for key, value in options.items()
         if value is not None
     }
-    metastore = _catalog_metastore(loader)
-    if metastore is None:
-        raise ValueError("native_plan requires an exact built-in catalog loader")
-    normalized[CatalogOptions.METASTORE.key()] = metastore
     if str(getattr(table, 'table_path', '')).startswith('oss://'):
         from pypaimon.filesystem.jindo_file_system_handler import (
             JINDO_AVAILABLE,
@@ -169,6 +176,47 @@ def _read_options(table) -> dict:
     return options
 
 
+def _resolved_schema_file_io_options(table) -> Optional[dict]:
+    """FileIO properties for tables whose metadata needs no catalog resolution."""
+    if not native_method_available('Table', 'from_resolved_schema'):
+        return None
+    environment = table.catalog_environment
+    loader = environment.catalog_loader
+    if loader is None:
+        from pypaimon.catalog.catalog_environment import CatalogEnvironment
+        from pypaimon.filesystem.local_file_io import LocalFileIO
+        from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO
+        from pypaimon.filesystem.resolving_file_io import ResolvingFileIO
+        # A custom environment or FileIO can supply metadata outside the path.
+        if (type(environment) is not CatalogEnvironment
+                or type(table.file_io) not in (LocalFileIO, PyArrowFileIO, ResolvingFileIO)):
+            return None
+        return {str(key): _option_value_to_string(value)
+                for key, value in table.file_io.properties.to_map().items()
+                if value is not None}
+    from pypaimon.catalog.jdbc_catalog_loader import JdbcCatalogLoader
+    if _catalog_metastore(loader) != 'filesystem' and type(loader) is not JdbcCatalogLoader:
+        # REST tables must retain catalog snapshot loading and token refresh.
+        return None
+    context = loader.context()
+    if context.options is None or any(getattr(context, attr, None) is not None for attr in (
+            'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader')):
+        return None
+    # JDBC, like filesystem catalogs, uses on-disk snapshots. Its already
+    # resolved table does not need another database connection during planning.
+    return _catalog_context_options(table)
+
+
+def _resolved_schema_json(table) -> str:
+    from pypaimon.common.json_util import JSON
+    options = {str(key): _option_value_to_string(value)
+               for key, value in table.table_schema.options.items() if value is not None}
+    options.update(_read_options(table))
+    # The timestamp string has already been converted to epoch millis.
+    options.pop(CoreOptions.SCAN_TIMESTAMP.key(), None)
+    return JSON.to_json(table.table_schema.copy(new_options=options))
+
+
 def _predicate_to_native(predicate: Predicate) -> dict:
     """Convert PyPaimon's predicate tree to pypaimon-rust's dict API."""
     if predicate.method in ('and', 'or'):
@@ -231,14 +279,34 @@ def native_plan(
     if not native_runtime_available():
         raise RuntimeError(
             "scan.native-plan.enabled needs pypaimon-rust>=0.3.0 (split planning API)")
-    from pypaimon_rust.datafusion import PaimonCatalog
-
-    rt = PaimonCatalog(_catalog_options(table)).get_table(table.identifier.get_full_name())
+    file_io_options = _resolved_schema_file_io_options(table)
+    if file_io_options is not None:
+        from pypaimon_rust.datafusion import Table
+        rt = Table.from_resolved_schema(
+            table.table_path, _resolved_schema_json(table),
+            database=table.identifier.get_database_name(),
+            table=table.identifier.get_table_name(),
+            branch=table.current_branch(), options=file_io_options)
+        builder = rt.new_read_builder()
+    else:
+        from pypaimon_rust.datafusion import PaimonCatalog
+        catalog = PaimonCatalog(_catalog_options(table))
+        if native_method_available('Table', 'copy_with_resolved_schema'):
+            # REST may keep branch schemas in the catalog only. Load the base
+            # environment, then attach the schema/branch already resolved here.
+            rt = catalog.get_table((
+                table.identifier.get_database_name(), table.identifier.get_table_name()))
+            if rt.location() != table.table_path:
+                raise RuntimeError('Native catalog resolved a different table location')
+            rt = rt.copy_with_resolved_schema(_resolved_schema_json(table), branch=table.current_branch())
+            builder = rt.new_read_builder()
+        else:
+            rt = catalog.get_table(table.identifier.get_full_name())
+            builder = rt.new_read_builder(_read_options(table))
     if table.current_branch() != 'main':
         branch = getattr(rt, 'branch', None)
         if not callable(branch) or branch() != table.current_branch():
             raise RuntimeError("Native table did not resolve the requested branch")
-    builder = rt.new_read_builder(_read_options(table))
     if projection is not None:
         builder = builder.with_projection(projection)
     if predicate is not None:

paimon-python/pypaimon/read/reader/format_pyarrow_reader.py

@@ -341,6 +341,9 @@ def __init__(self, file_io: FileIO, file_format: str, file_path: str,
                  row_indices: Optional[List[int]] = None,
                  row_ranges: Optional[List[Tuple[int, int]]] = None,
                  row_group_cache: Optional[_DecodedRowGroupCache] = None):
+        from pypaimon.filesystem.resolving_file_io import ResolvingFileIO
+        if isinstance(file_io, ResolvingFileIO):
+            file_io = file_io._get_fileio(file_path)
         self._predicate_field_names = predicate_field_names or set()
         file_path_for_pyarrow = file_io.to_filesystem_path(file_path)
         self._row_group_cache = row_group_cache

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

@@ -230,12 +230,13 @@ def create_splits(self, file_entries: List[ManifestEntry]) -> List[Split]:
                 if f.file_name in seen_paths:
                     continue
                 seen_paths.add(f.file_name)
-                f.set_file_path(
-                    self.table.table_path,
-                    partition_row,
-                    bucket,
-                    self.default_part_value,
-                )
+                if not f.file_path:
+                    f.set_file_path(
+                        self.table.table_path,
+                        partition_row,
+                        bucket,
+                        self.default_part_value,
+                    )
             for segments in self._slice_group_into_chunks(entries_in_group):
                 all_chunks.append(_Chunk(partition_row, bucket, segments))
 

paimon-python/pypaimon/read/split_read.py

@@ -240,6 +240,13 @@ def file_reader_supplier(self, file: DataFileMeta, for_merge_read: bool,
             read_arrow_predicate,
             read_paimon_predicate,
         ) = self._get_fields_and_predicate(file.schema_id, read_fields)
+        if (file.file_name in self.deletion_file_readers
+                or (for_merge_read and self.row_ranges is not None)):
+            # DVs and indexed PK ranges refer to physical file positions.
+            # Filtering or skipping row groups here would renumber those rows.
+            # Apply the residual predicate after position selection and merging.
+            read_arrow_predicate = None
+            read_paimon_predicate = None
 
         # Use external_path if available, otherwise use file_path
         file_path = file.external_path if file.external_path else file.file_path
@@ -583,7 +590,11 @@ def _is_reachable(name: str) -> bool:
                 read_field for read_field in read_fields
                 if _is_reachable(read_field)
             ]
-            read_predicate = trim_predicate_by_fields(self.push_down_predicate, read_file_fields)
+            # File readers filter physical column names before field-id schema
+            # normalization. A renamed or re-added name can identify a different
+            # column, so cross-schema filtering must run after normalization.
+            read_predicate = (trim_predicate_by_fields(self.push_down_predicate, read_file_fields)
+                              if schema_id == self.table.table_schema.id else None)
             read_arrow_predicate = (
                 read_predicate.to_arrow()
                 if read_predicate and self._arrow_filter_pushdown_enabled
@@ -899,7 +910,10 @@ def create_reader(self) -> RecordReader:
         reader = concat_reader
         if (self.predicate_for_reader
                 and (self.table.is_primary_key_table
-                     or not self._arrow_filter_pushdown_enabled)):
+                     or not self._arrow_filter_pushdown_enabled
+                     or self.deletion_file_readers
+                     or any(file.schema_id != self.table.table_schema.id
+                            for file in self.split.files))):
             reader = FilterRecordBatchReader(
                 reader,
                 self.predicate_for_reader,