[python] Align native scan planning with Rust

.github/workflows/ci-python.yml

@@ -34,6 +34,9 @@ jobs:
   # Lint + test on selected versions only (3.6 / 3.7 low end, 3.10 / 3.11 / 3.12 / 3.13 current), not every Python version.
   test:
     name: Tests / Python ${{ matrix.python-version }}
+    env:
+      # Real native planning is covered by the Rust main job below.
+      PYTEST_ADDOPTS: "-m 'not native_plan'"
     timeout-minutes: 90
     runs-on: ubuntu-latest
     container: "python:${{ matrix.python-version }}-slim"
@@ -148,6 +151,7 @@ jobs:
               python -c "import datasets, lerobot; print('datasets', datasets.__version__, 'lerobot', lerobot.__version__)"
             fi
 
+            # Required by SQL tests; native planner tests run against main below.
             python -m pip install pypaimon-rust
             if [[ "${{ matrix.python-version }}" == "3.11" ]]; then
               # Run the RoboMIND pipeline tests with synthetic local HDF5 data.

paimon-python/README.md

@@ -31,6 +31,59 @@ pip3 install dist/*.tar.gz
 
 The command will install the package and core dependencies to your local Python environment.
 
+# Native scan planning
+
+PyPaimon can plan splits with the optional `pypaimon-rust` package while retaining
+the Python reader:
+
+```python
+native_table = table.copy({"scan.native-plan.enabled": "true"})
+builder = native_table.new_read_builder()
+plan = builder.new_scan().plan()
+rows = builder.new_read().to_arrow(plan.splits())
+explanation = builder.explain()
+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.
+
+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
+checked before planning. Deletion-vector scans require `pypaimon-rust>=0.4.0`,
+which includes schema-aware decoding of Python-written index manifests and
+legacy bucket-index path compatibility. The reader honors explicit paths, then
+bucket paths, and can read older Python files placed in `table/index`.
+Bucket paths use the partition field types and `partition.legacy-name` to match
+Java formatting, including timestamp precision and different JVM float spellings.
+New Python writes honor `index-file-in-data-file-dir` and retain explicit paths
+when Python and Java partition-directory formatting differs. Older releases
+and prereleases before 0.4.0 use the Python planner for deletion vectors.
+When using an unreleased 0.4.0 development wheel, rebuild it with these fixes;
+package version checks cannot distinguish local builds with identical versions.
+
+Append scans support `with_shard()` and `with_slice()` with Rust 0.4 or newer,
+which preserves the file order needed for positional selection; primary-key scans support
+bucket-based `with_shard()`. Data-evolution position selection requires the
+binding's `TableScan.with_row_position_slice()` and `with_row_position_shard()`.
+Selection occurs before reader filtering and deletion vectors, so surviving row
+counts can differ between shards. Limits are applied after shard/slice selection.
+
+Timestamp incremental scans require `ReadBuilder.new_incremental_scan()`.
+Python resolves `(start_timestamp, end_timestamp]` to snapshot IDs; Rust combines
+the selected APPEND deltas into one plan, including merging primary-key versions
+across commits. Other commit kinds are excluded, and the ending snapshot supplies
+snapshot metadata and deletion vectors even if it contributes no APPEND files.
+
+Chunk shuffle, query authorization, first-row merge, deletion-vector merge-on-read,
+dynamic or cross-partition primary-key buckets, and scored or primary-key
+global-index results still use the Python planner. Rust also rejects floating-point
+partition-directory formatting; these scans fall back to Python. Native planning
+remains optional and is disabled by default.
+
 # Load LeRobot Dataset v3
 
 Install the optional dependency, then import a local directory, FileIO URI, or

paimon-python/conftest.py

@@ -38,6 +38,8 @@ def _native_plan_enabled():
 def pytest_configure(config):
     config.addinivalue_line(
         "markers", "python_plan: keep Python planner assertions on the Python lane")
+    config.addinivalue_line(
+        "markers", "native_plan: exercise the real Rust planner in the Rust main CI job")
     if not _native_plan_enabled():
         return
 
@@ -61,7 +63,8 @@ def enable_native_plan(request, monkeypatch):
     if (not _native_plan_enabled()
             or request.node.get_closest_marker("python_plan") is not None
             or request.path.name in (
-                "native_plan_test.py", "native_plan_integration_test.py")):
+                "native_plan_test.py", "native_plan_integration_test.py",
+                "native_plan_capabilities_test.py")):
         yield
         return
 

paimon-python/pypaimon/common/options/core_options.py

@@ -638,6 +638,13 @@ class CoreOptions:
         .with_description("Whether to enable deletion vectors.")
     )
 
+    INDEX_FILE_IN_DATA_FILE_DIR: ConfigOption[bool] = (
+        ConfigOptions.key("index-file-in-data-file-dir")
+        .boolean_type()
+        .default_value(False)
+        .with_description("Whether to store bucket index files in the data file directory.")
+    )
+
     SCAN_NATIVE_PLAN_ENABLED: ConfigOption[bool] = (
         ConfigOptions.key("scan.native-plan.enabled")
         .boolean_type()
@@ -1126,6 +1133,15 @@ class CoreOptions:
         )
     )
 
+    PARTITION_GENERATE_LEGACY_NAME: ConfigOption[bool] = (
+        ConfigOptions.key("partition.legacy-name")
+        .boolean_type()
+        .default_value(True)
+        .with_description(
+            "Use legacy Java toString partition names; otherwise use casts to string."
+        )
+    )
+
     DYNAMIC_PARTITION_OVERWRITE: ConfigOption[bool] = (
         ConfigOptions.key("dynamic-partition-overwrite")
         .boolean_type()
@@ -1484,6 +1500,9 @@ def global_index_column_update_action(self, default=None):
     def deletion_vectors_enabled(self, default=None):
         return self.options.get(CoreOptions.DELETION_VECTORS_ENABLED, default)
 
+    def index_file_in_data_file_dir(self, default=None):
+        return self.options.get(CoreOptions.INDEX_FILE_IN_DATA_FILE_DIR, default)
+
     def native_plan_enabled(self, default=None):
         return self.options.get(CoreOptions.SCAN_NATIVE_PLAN_ENABLED, default)
 

paimon-python/pypaimon/read/interval_partition.py

@@ -16,6 +16,7 @@
 # under the License.
 
 import heapq
+import math
 from dataclasses import dataclass
 from functools import cmp_to_key
 from typing import Callable, List
@@ -116,6 +117,12 @@ def default_key_comparator(key1: GenericRow, key2: GenericRow) -> int:
             return -1
         if val2 is None:
             return 1
+        # Preserve Java's ordering of signed zeros in composite key bounds.
+        if (isinstance(val1, float) and isinstance(val2, float)
+                and val1 == 0.0 and val2 == 0.0):
+            sign1, sign2 = math.copysign(1.0, val1), math.copysign(1.0, val2)
+            if sign1 != sign2:
+                return -1 if sign1 < sign2 else 1
         if val1 < val2:
             return -1
         elif val1 > val2:

paimon-python/pypaimon/read/native_plan.py

@@ -22,13 +22,15 @@
 still applies them while reading, so pushdown remains an optimization.
 """
 
-import re
 from typing import List, Optional, Tuple
 
+from packaging.version import InvalidVersion, Version
+
 from pypaimon.common.options.config import CatalogOptions, OssOptions
 from pypaimon.common.options.core_options import CoreOptions
 from pypaimon.common.options.options_utils import OptionsUtils
 from pypaimon.common.predicate import Predicate
+from pypaimon.read.plan import Plan
 from pypaimon.read.split import Split
 from pypaimon.read.split_serializer import deserialize_split_v1
 
@@ -49,18 +51,30 @@ def native_runtime_available() -> bool:
 
 def native_family_search_modes_available() -> bool:
     """Whether Rust supports family-specific global-index search modes."""
+    return native_version_at_least(0, 4)
+
+
+def native_version_at_least(major: int, minor: int, patch: int = 0) -> bool:
+    """Compare complete package versions, including patch and pre-release ordering."""
     if not native_runtime_available():
         return False
     try:
         from importlib.metadata import PackageNotFoundError, version
     except ImportError:
         return False
     try:
-        rust_version = version('pypaimon-rust')
-    except PackageNotFoundError:
+        rust_version = Version(version('pypaimon-rust'))
+    except (PackageNotFoundError, InvalidVersion):
+        return False
+    return rust_version >= Version('%d.%d.%d' % (major, minor, patch))
+
+
+def native_method_available(type_name: str, method: str) -> bool:
+    try:
+        from pypaimon_rust import datafusion
+        return callable(getattr(getattr(datafusion, type_name, None), method, None))
+    except ImportError:
         return False
-    match = re.match(r'^(\d+)\.(\d+)', rust_version)
-    return match is not None and tuple(map(int, match.groups())) >= (0, 4)
 
 
 def _partition_fields(table):
@@ -134,6 +148,7 @@ def _read_options(table) -> dict:
             CoreOptions.SCAN_SNAPSHOT_ID,
             CoreOptions.SCAN_TAG_NAME,
             CoreOptions.SCAN_TIMESTAMP_MILLIS,
+            CoreOptions.SCAN_WATERMARK,
             CoreOptions.GLOBAL_INDEX_SEARCH_MODE,
             CoreOptions.SCALAR_INDEX_SEARCH_MODE,
             CoreOptions.VECTOR_INDEX_SEARCH_MODE,
@@ -201,8 +216,11 @@ def native_plan(
         predicate: Optional[Predicate] = None,
         limit: Optional[int] = None,
         projection: Optional[List[str]] = None,
-        row_ranges: Optional[List[Tuple[int, int]]] = None) -> List[Split]:
-    """Plan with pypaimon_rust and return the decoded pypaimon splits.
+        row_ranges: Optional[List[Tuple[int, int]]] = None,
+        incremental_range: Optional[Tuple[int, int]] = None,
+        row_position_slice: Optional[Tuple[int, int]] = None,
+        row_position_shard: Optional[Tuple[int, int]] = None) -> Plan:
+    """Plan with pypaimon_rust, preserving snapshot metadata.
 
     Native conversion or planning failures are handled by TableScan, which
     falls back to the Python planner.
@@ -213,6 +231,10 @@ def native_plan(
     from pypaimon_rust.datafusion import PaimonCatalog
 
     rt = PaimonCatalog(_catalog_options(table)).get_table(table.identifier.get_full_name())
+    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)
@@ -222,10 +244,26 @@ def native_plan(
         builder = builder.with_limit(limit)
     if row_ranges is not None:
         builder = builder.with_row_ranges(row_ranges)
-    rust_splits = builder.new_scan().plan().splits()
+    scan = (builder.new_scan() if incremental_range is None
+            else builder.new_incremental_scan(*incremental_range))
+    if row_position_slice is not None:
+        scan = scan.with_row_position_slice(*row_position_slice)
+    if row_position_shard is not None:
+        scan = scan.with_row_position_shard(*row_position_shard)
+    rust_plan = scan.plan()
+    rust_splits = rust_plan.splits()
     pfields = _partition_fields(table)
     # Trimmed primary keys decode per-file min/max keys (PK merge-on-read).
     kfields = table.trimmed_primary_keys_fields
     splits = [deserialize_split_v1(s.serialize(), pfields, kfields) for s in rust_splits]
     _restore_python_partition_paths(table, splits)
-    return splits
+    snapshot_id = getattr(rust_plan, 'snapshot_id', None)
+    if callable(snapshot_id):
+        snapshot_id = snapshot_id()
+    elif splits:
+        snapshot_id = splits[0].snapshot_id
+    else:
+        # 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)

paimon-python/pypaimon/read/read_builder.py

@@ -208,16 +208,15 @@ def _resolve_dotted_paths(self, names: List[str]) -> List[List[int]]:
         return paths
 
 
-def _build_explain_result(table, scan: TableScan, plan, stats: ScanStats,
+def _build_explain_result(table, scan: TableScan, plan, stats: Optional[ScanStats],
                           predicate, projection, limit, verbose: bool) -> ExplainResult:
     """Translate one (Plan, ScanStats) pair into an ExplainResult."""
     splits: List[Split] = plan.splits()
 
     table_schema = table.table_schema
     bucket_mode_str = _safe_bucket_mode(table)
 
-    # stats is None when planned natively (pypaimon_rust): no manifest pruning
-    # funnel is tracked, so the split-level signals below are all we can report.
+    # Native plans expose split metadata without Python pruning counters.
     native_planned = stats is None
     if native_planned:
         partition_pruning = bucket_pruning = file_skipping = None

paimon-python/pypaimon/read/scan_distribution.py

@@ -0,0 +1,72 @@
+# 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.
+
+"""Row-position selection shared by the Python and native planning adapters."""
+
+from typing import List, Tuple
+
+from pypaimon.read.sliced_split import SlicedSplit
+from pypaimon.read.split import Split
+
+
+def validate_shard(index: int, count: int) -> None:
+    if not isinstance(count, int) or count <= 0:
+        raise ValueError("number_of_para_subtasks must be a positive integer")
+    if not isinstance(index, int) or index < 0:
+        raise ValueError("idx_of_this_subtask must be a non-negative integer")
+    if index >= count:
+        raise ValueError("idx_of_this_subtask must be less than number_of_para_subtasks")
+
+
+def validate_slice(start: int, end: int) -> None:
+    if not isinstance(start, int) or not isinstance(end, int) or start < 0 or start >= end:
+        raise ValueError("start_pos must be non-negative and less than end_pos; both must be integers")
+
+
+def shard_range(total: int, index: int, count: int) -> Tuple[int, int]:
+    base, remainder = divmod(total, count)
+    start = index * base + min(index, remainder)
+    return start, start + base + int(index < remainder)
+
+
+def slice_append_splits(splits: List[Split], start: int, end: int) -> List[Split]:
+    """Select physical rows in [start, end), preserving file order and DVs.
+
+    File statistics have already pruned the input; residual row predicates and
+    deletion vectors are applied by the reader after selecting these positions.
+    An end beyond the table is naturally clamped by each file's row count.
+    """
+    if start >= end:
+        return []
+    selected = []
+    offset = 0
+    for split in splits:
+        keep_ids, ranges = set(), {}
+        for file in split.files:
+            begin, stop = max(0, start - offset), min(file.row_count, end - offset)
+            offset += file.row_count
+            if begin < stop:
+                keep_ids.add(id(file))
+                if begin != 0 or stop != file.row_count:
+                    ranges[file.file_name] = (begin, stop)
+        if not keep_ids:
+            continue
+        kept = split.filter_file(lambda file: id(file) in keep_ids)
+        selected.append(SlicedSplit(kept, ranges) if ranges else kept)
+        if offset >= end:
+            break
+    return selected