[python] Expand native planning and align incremental reads with Java

.github/workflows/ci-python.yml

@@ -226,6 +226,8 @@ jobs:
                 'commit', source['vcs_info']['commit_id'])
           assert hasattr(PaimonCatalog, 'get_table'), 'Missing PaimonCatalog.get_table'
           assert hasattr(Split, 'serialize'), 'Missing Split.serialize'
+          assert hasattr(Split, 'is_streaming'), 'Missing stream-aware Split.is_streaming'
+          assert hasattr(ReadBuilder, 'new_incremental_scan'), 'Missing ReadBuilder.new_incremental_scan'
           assert hasattr(ReadBuilder, 'with_row_ranges'), 'Missing ReadBuilder.with_row_ranges'
           PY
 

paimon-python/README.md

@@ -72,11 +72,18 @@ 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.
+Timestamp incremental scans require `ReadBuilder.new_incremental_scan()` and
+stream-aware splits exposing `Split.is_streaming()`. Python resolves
+`(start_timestamp, end_timestamp]` to snapshot IDs; Rust packs the selected APPEND
+deltas into one plan. Like Java, readers retain physical change events, including
+repeated primary keys and retracts across commits. They do not merge the window
+into a final table state or apply endpoint deletion vectors or global indexes.
+Other commit kinds are excluded; the ending snapshot still supplies plan metadata.
+Rebuild development wheels from Rust main to obtain this contract.
+
+`scan.version` supports tags, snapshot IDs and `watermark-<value>`, resolving tags
+first and using the historical schema. Ordinary postpone-bucket batch scans can
+use native planning and exclude pending files in negative buckets.
 
 Dynamic and cross-partition primary-key buckets support native planning, including
 bucket sharding. Cross-partition key migration is maintained by the writer's index.
@@ -85,11 +92,15 @@ 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.
 
-Chunk shuffle, query authorization, first-row scans that include level 0, and scored
-or primary-key global-index results still use the Python planner. Native planning
-remains optional and is disabled by default. The combined incremental planner's
-window-end deletion-vector behavior described above differs from Java delta scans,
-which do not attach deletion vectors; full Java incremental parity remains pending.
+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,
+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.
 
 # Load LeRobot Dataset v3
 

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

@@ -557,6 +557,13 @@ class CoreOptions:
         .with_description("Optional tag name used in case of 'from-snapshot' scan mode.")
     )
 
+    SCAN_VERSION: ConfigOption[str] = (
+        ConfigOptions.key("scan.version")
+        .string_type()
+        .no_default_value()
+        .with_description("Time-travel version: tag name, watermark-<value>, or snapshot id; tags take precedence.")
+    )
+
     SCAN_SNAPSHOT_ID: ConfigOption[int] = (
         ConfigOptions.key("scan.snapshot-id")
         .long_type()
@@ -1409,7 +1416,8 @@ def startup_mode(self) -> 'StartupMode':
                 return StartupMode.FROM_TIMESTAMP
             elif (self.options.contains(CoreOptions.SCAN_SNAPSHOT_ID)
                   or self.options.contains(CoreOptions.SCAN_TAG_NAME)
-                  or self.options.contains(CoreOptions.SCAN_WATERMARK)):
+                  or self.options.contains(CoreOptions.SCAN_WATERMARK)
+                  or self.options.contains(CoreOptions.SCAN_VERSION)):
                 return StartupMode.FROM_SNAPSHOT
             elif self.options.contains(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP):
                 return StartupMode.INCREMENTAL

paimon-python/pypaimon/globalindex/indexed_split.py

@@ -24,6 +24,18 @@
 from pypaimon.read.split import Split
 
 
+def scores_for_ranges(score_getter, row_ranges):
+    """Scores follow row-id order, as in Java IndexedSplit, not relevance order."""
+    scores = []
+    for row_range in row_ranges:
+        for row_id in range(row_range.from_, row_range.to + 1):
+            score = score_getter(row_id)
+            if score is None:
+                raise ValueError("Missing score for selected row id %s" % row_id)
+            scores.append(score)
+    return scores
+
+
 class IndexedSplit(Split):
 
     def __init__(
@@ -94,6 +106,10 @@ def file_size(self):
         """Delegate to data_split."""
         return self._data_split.file_size
 
+    @property
+    def is_streaming(self):
+        return getattr(self._data_split, 'is_streaming', False)
+
     @property
     def raw_convertible(self):
         """Delegate to data_split."""

paimon-python/pypaimon/read/native_plan.py

@@ -147,6 +147,7 @@ def _read_options(table) -> dict:
     }
     table_options = table.options.options
     for option in (
+            CoreOptions.SCAN_VERSION,
             CoreOptions.SCAN_SNAPSHOT_ID,
             CoreOptions.SCAN_TAG_NAME,
             CoreOptions.SCAN_TIMESTAMP_MILLIS,
@@ -157,7 +158,7 @@ def _read_options(table) -> dict:
             CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE):
         if table_options.contains_key(option.key()):
             options[option.key()] = _option_value_to_string(
-                table_options.get(option))
+                table_options.to_map()[option.key()])
 
     # Rust takes epoch millis but PyPaimon also accepts a timestamp string.
     if table_options.contains_key(CoreOptions.SCAN_TIMESTAMP.key()):
@@ -257,7 +258,7 @@ def native_plan(
     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]
+    splits = [deserialize_split_v1(split.serialize(), pfields, kfields) for split in rust_splits]
     _restore_python_partition_paths(table, splits)
     snapshot_id = getattr(rust_plan, 'snapshot_id', None)
     if callable(snapshot_id):

paimon-python/pypaimon/read/query_auth_split.py

@@ -49,6 +49,10 @@ def partition(self):
     def bucket(self) -> int:
         return self._split.bucket
 
+    @property
+    def is_streaming(self):
+        return getattr(self._split, 'is_streaming', False)
+
     @property
     def raw_convertible(self) -> bool:
         return self._split.raw_convertible

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

@@ -18,7 +18,7 @@
 from collections import defaultdict
 from typing import List, Optional, Tuple
 
-from pypaimon.globalindex.indexed_split import IndexedSplit
+from pypaimon.globalindex.indexed_split import IndexedSplit, scores_for_ranges
 from pypaimon.utils.range import Range
 from pypaimon.utils.range_helper import RangeHelper
 from pypaimon.manifest.schema.data_file_meta import DataFileMeta
@@ -332,14 +332,8 @@ def _wrap_to_indexed_splits(self, splits: List[Split], row_ranges: List[Range])
                 # No intersection, skip this split
                 continue
 
-            # Create scores array if score_getter is provided
-            scores = None
-            if self.score_getter is not None:
-                scores = []
-                for r in expected:
-                    for row_id in range(r.from_, r.to + 1):
-                        score = self.score_getter(row_id)
-                        scores.append(score if score is not None else 0.0)
+            scores = (scores_for_ranges(self.score_getter, expected)
+                      if self.score_getter is not None else None)
 
             indexed_splits.append(IndexedSplit(split, expected, scores))
 

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

@@ -224,6 +224,7 @@ def __init__(
         limit: Optional[int] = None,
         partition_predicate: Optional[Predicate] = None,
         skip_level0: bool = False,
+        is_streaming: bool = False,
     ):
         from pypaimon.table.file_store_table import FileStoreTable
 
@@ -276,6 +277,7 @@ def __init__(
         self.data_evolution = options.data_evolution_enabled()
         self.deletion_vectors_enabled = options.deletion_vectors_enabled()
         self.skip_level0 = skip_level0
+        self.is_streaming = is_streaming
         self._global_index_result = None
         self._row_ranges = None
         self._scanned_snapshot = None
@@ -313,7 +315,7 @@ def _schema(self, schema_id: int):
         return self.table.schema_manager.get_schema(schema_id)
 
     def _deletion_files_map(self, entries: List[ManifestEntry]) -> Dict[tuple, Dict[str, DeletionFile]]:
-        if not self.deletion_vectors_enabled:
+        if self.is_streaming or not self.deletion_vectors_enabled:
             return {}
         # Extract unique partition-bucket pairs from file entries
         bucket_files = set()
@@ -393,14 +395,20 @@ def scan(self) -> Plan:
 
         # Generate splits
         splits = split_generator.create_splits(entries)
+        if self.is_streaming:
+            for split in splits:
+                while callable(getattr(split, 'data_split', None)):
+                    split = split.data_split()
+                split.is_streaming = True
+                split.snapshot_id = self._scanned_snapshot_id
 
         if self.data_evolution and self.scan_stats is not None:
             # Data-evolution stats pruning happens on complete row-id groups
             # inside the split generator, not in _filter_manifest_entry.
             self.scan_stats.entries_after_stats = sum(
                 len(split.files) for split in splits)
 
-        if self.table.is_primary_key_table:
+        if self.table.is_primary_key_table and not self.is_streaming:
             splits = self._apply_primary_key_sorted_indexes(splits)
 
         splits = self._apply_push_down_limit(splits)
@@ -411,10 +419,11 @@ def scan(self) -> Plan:
         )
         return Plan(splits, snapshot_id=self._scanned_snapshot_id)
 
-    def _apply_primary_key_sorted_indexes(self, splits):
+    def _apply_primary_key_sorted_indexes(self, splits, snapshot=None):
+        snapshot = snapshot if snapshot is not None else self._scanned_snapshot
         if (not self.table.options.global_index_enabled()
                 or self.predicate is None
-                or self._scanned_snapshot is None
+                or snapshot is None
                 or not splits):
             return splits
 
@@ -430,7 +439,7 @@ def _apply_primary_key_sorted_indexes(self, splits):
             return splits
         field_ids = {definition.field_id for definition in definitions}
         entries = IndexFileHandler(self.table).scan(
-            self._scanned_snapshot,
+            snapshot,
             lambda entry: (
                 entry.kind == 0
                 and entry.index_file.global_index_meta is not None
@@ -439,7 +448,7 @@ def _apply_primary_key_sorted_indexes(self, splits):
             ),
         )
         index_plan = primary_key_sorted_index_scan.plan(
-            self._scanned_snapshot_id, splits, definitions, entries)
+            snapshot.id, splits, definitions, entries)
         evaluated = primary_key_sorted_index_scan.evaluate(
             index_plan,
             self.table.fields,
@@ -536,8 +545,8 @@ def plan_files(self) -> List[ManifestEntry]:
         return self.read_manifest_entries(manifest_files)
 
     def _eval_global_index(self, snapshot=None):
-        # No filter - nothing to evaluate
-        if self.predicate is None:
+        # Snapshot indexes describe current state, not historical change events.
+        if self.is_streaming or self.predicate is None:
             return None
 
         # Check if global index is enabled
@@ -830,6 +839,8 @@ def _init_bucket_selector(self):
         )
 
     def _filter_manifest_entry(self, entry: ManifestEntry) -> bool:
+        if self.is_streaming and entry.kind != 0:
+            raise ValueError("Incremental delta manifests must contain only ADD entries")
         stats = self.scan_stats
         if stats is not None:
             stats.entries_total += 1

paimon-python/pypaimon/read/sliced_split.py

@@ -89,6 +89,10 @@ def file_paths(self):
     def file_size(self):
         return self._data_split.file_size
 
+    @property
+    def is_streaming(self):
+        return getattr(self._data_split, 'is_streaming', False)
+
     @property
     def raw_convertible(self):
         return self._data_split.raw_convertible

paimon-python/pypaimon/read/split.py

@@ -79,7 +79,8 @@ def __init__(
         bucket: int,
         raw_convertible: bool = False,
         data_deletion_files: Optional[List[DeletionFile]] = None,
-        snapshot_id: Optional[int] = None
+        snapshot_id: Optional[int] = None,
+        is_streaming: bool = False,
     ):
         self._files = files
         self._partition = partition
@@ -88,6 +89,7 @@ def __init__(
         self.data_deletion_files = data_deletion_files
         # Scanned snapshot; None unless populated (e.g. by the native planner).
         self.snapshot_id = snapshot_id
+        self.is_streaming = is_streaming
 
     @property
     def files(self) -> List[DataFileMeta]:
@@ -125,7 +127,8 @@ def filter_file(self, func: Callable[[DataFileMeta], bool]) -> Optional['DataSpl
             bucket=self._bucket,
             raw_convertible=self.raw_convertible,
             data_deletion_files=filtered_data_deletion_files,
-            snapshot_id=self.snapshot_id
+            snapshot_id=self.snapshot_id,
+            is_streaming=self.is_streaming,
         )
 
     @property

paimon-python/pypaimon/read/split_read.py

@@ -1049,15 +1049,19 @@ def _build_merge_function(self):
         )
 
     def create_reader(self) -> RecordReader:
-        # Create a dict mapping data file name to deletion file reader method
         self._genarate_deletion_file_readers()
-        section_readers = []
-        sections = IntervalPartition(self.split.files).partition()
-        for section in sections:
-            supplier = partial(self.section_reader_supplier, section)
-            section_readers.append(supplier)
-        concat_reader = ConcatRecordReader(section_readers)
-        kv_unwrap_reader = KeyValueUnwrapRecordReader(DropDeleteRecordReader(concat_reader))
+        if getattr(self.split, 'is_streaming', False):
+            # Java streaming PK reads concatenate physical changes, including
+            # retracts, without merging versions. Respect explicitly supplied DVs.
+            kv_reader = ConcatRecordReader([
+                partial(self.kv_reader_supplier, file,
+                        self.deletion_file_readers.get(file.file_name)) for file in self.split.files])
+        else:
+            sections = IntervalPartition(self.split.files).partition()
+            concat_reader = ConcatRecordReader([
+                partial(self.section_reader_supplier, section) for section in sections])
+            kv_reader = DropDeleteRecordReader(concat_reader)
+        kv_unwrap_reader = KeyValueUnwrapRecordReader(kv_reader)
         if self.predicate_for_reader:
             reader = FilterRecordReader(kv_unwrap_reader, self.predicate_for_reader)
         else:

paimon-python/pypaimon/read/split_serializer.py

@@ -231,13 +231,14 @@ def _read_datasplit_body(r: _Reader, partition_fields: List[DataField],
     files = [_datafilemeta_from_row(r.take(r.i32()), bucket_path, arity, key_fields)
              for _ in range(file_count)]
     data_deletion_files = _read_deletion_list(r)
-    r.u8()    # isStreaming
+    is_streaming = r.u8() != 0
     raw_convertible = r.u8() != 0
     return DataSplit(
         files=files,
         partition=partition,
         bucket=bucket,
         raw_convertible=raw_convertible,
+        is_streaming=is_streaming,
         data_deletion_files=data_deletion_files,
         snapshot_id=snapshot_id,
     )