This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 2a38691bd1 [python] Expand native planning and align incremental reads
with Java (#9846)
2a38691bd1 is described below
commit 2a38691bd1419bc55b7c5694299dccbb0e74eab3
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 15 21:41:57 2026 +0800
[python] Expand native planning and align incremental reads with Java
(#9846)
---
.github/workflows/ci-python.yml | 2 +
paimon-python/README.md | 31 ++-
.../pypaimon/common/options/core_options.py | 10 +-
.../pypaimon/globalindex/indexed_split.py | 16 ++
paimon-python/pypaimon/read/native_plan.py | 5 +-
paimon-python/pypaimon/read/query_auth_split.py | 4 +
.../read/scanner/data_evolution_split_generator.py | 12 +-
.../pypaimon/read/scanner/file_scanner.py | 27 +-
paimon-python/pypaimon/read/sliced_split.py | 4 +
paimon-python/pypaimon/read/split.py | 7 +-
paimon-python/pypaimon/read/split_read.py | 20 +-
paimon-python/pypaimon/read/split_serializer.py | 3 +-
.../pypaimon/read/streaming_table_scan.py | 24 +-
paimon-python/pypaimon/read/table_read.py | 3 +-
paimon-python/pypaimon/read/table_scan.py | 42 +++-
.../pypaimon/snapshot/time_travel_util.py | 19 ++
.../source/primary_key_sorted_index_result.py | 2 +
paimon-python/pypaimon/tests/global_index_test.py | 3 +
.../pypaimon/tests/native_plan_expanded_test.py | 279 +++++++++++++++++++++
.../pypaimon/tests/native_plan_incremental_test.py | 132 +++++++++-
paimon-python/pypaimon/tests/native_plan_test.py | 28 +--
.../pypaimon/tests/reader_primary_key_test.py | 17 +-
.../pypaimon/tests/split_serializer_test.py | 13 +
23 files changed, 612 insertions(+), 91 deletions(-)
diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index ca5938da1e..637bcad9dc 100644
--- a/.github/workflows/ci-python.yml
+++ b/.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
diff --git a/paimon-python/README.md b/paimon-python/README.md
index 76155304b2..98dc4fc082 100644
--- a/paimon-python/README.md
+++ b/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
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 43a07a36a7..6586e051a5 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/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 @@ class CoreOptions:
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
diff --git a/paimon-python/pypaimon/globalindex/indexed_split.py
b/paimon-python/pypaimon/globalindex/indexed_split.py
index bd4d98a1d1..e2f203578d 100644
--- a/paimon-python/pypaimon/globalindex/indexed_split.py
+++ b/paimon-python/pypaimon/globalindex/indexed_split.py
@@ -24,6 +24,18 @@ from typing import List, Optional
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 @@ class IndexedSplit(Split):
"""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."""
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index bac2dd3784..253eeabb8f 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/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):
diff --git a/paimon-python/pypaimon/read/query_auth_split.py
b/paimon-python/pypaimon/read/query_auth_split.py
index 8aec57e3f8..bfbb1a2799 100644
--- a/paimon-python/pypaimon/read/query_auth_split.py
+++ b/paimon-python/pypaimon/read/query_auth_split.py
@@ -49,6 +49,10 @@ class QueryAuthSplit(Split):
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
diff --git
a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
b/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
index 5f5a1dad52..9a8015347c 100644
--- a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
+++ b/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 @@ class DataEvolutionSplitGenerator(AbstractSplitGenerator):
# 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))
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index a02e745bd6..c07bd7e2d3 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -224,6 +224,7 @@ class FileScanner:
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 @@ class FileScanner:
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 @@ class FileScanner:
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,6 +395,12 @@ class FileScanner:
# 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
@@ -400,7 +408,7 @@ class FileScanner:
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 @@ class FileScanner:
)
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 @@ class FileScanner:
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 @@ class FileScanner:
),
)
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 @@ class FileScanner:
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 @@ class FileScanner:
)
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
diff --git a/paimon-python/pypaimon/read/sliced_split.py
b/paimon-python/pypaimon/read/sliced_split.py
index 137e874628..e166be0451 100644
--- a/paimon-python/pypaimon/read/sliced_split.py
+++ b/paimon-python/pypaimon/read/sliced_split.py
@@ -89,6 +89,10 @@ class SlicedSplit(Split):
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
diff --git a/paimon-python/pypaimon/read/split.py
b/paimon-python/pypaimon/read/split.py
index bdfc537af1..a7800da72e 100644
--- a/paimon-python/pypaimon/read/split.py
+++ b/paimon-python/pypaimon/read/split.py
@@ -79,7 +79,8 @@ class DataSplit(Split):
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 @@ class DataSplit(Split):
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 @@ class DataSplit(Split):
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
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index 2d35830ecc..ec1ee56689 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -1049,15 +1049,19 @@ class MergeFileSplitRead(SplitRead):
)
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:
diff --git a/paimon-python/pypaimon/read/split_serializer.py
b/paimon-python/pypaimon/read/split_serializer.py
index ff5397c7c2..9a4e3a4270 100644
--- a/paimon-python/pypaimon/read/split_serializer.py
+++ b/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,
)
diff --git a/paimon-python/pypaimon/read/streaming_table_scan.py
b/paimon-python/pypaimon/read/streaming_table_scan.py
index e625b275ea..ee0cb669ee 100644
--- a/paimon-python/pypaimon/read/streaming_table_scan.py
+++ b/paimon-python/pypaimon/read/streaming_table_scan.py
@@ -354,32 +354,37 @@ class AsyncStreamingTableScan:
def _create_delta_plan(self, snapshot: Snapshot) -> Plan:
"""Read new files from delta_manifest_list
(changelog-producer=none)."""
manifest_files = self._manifest_list_manager.read_delta(snapshot)
- return self._create_plan_from_manifests(manifest_files)
+ return self._create_plan_from_manifests(manifest_files, snapshot.id)
def _create_changelog_plan(self, snapshot: Snapshot) -> Plan:
"""Read from changelog_manifest_list
(changelog-producer=input/full-compaction/lookup)."""
manifest_files = self._manifest_list_manager.read_changelog(snapshot)
- return self._create_plan_from_manifests(manifest_files)
+ return self._create_plan_from_manifests(manifest_files, snapshot.id)
- def _create_plan_from_manifests(self, manifest_files: List) -> Plan:
+ def _create_plan_from_manifests(self, manifest_files: List,
snapshot_id=None) -> Plan:
"""Create splits from manifest files, applying shard filtering."""
if not manifest_files:
- return Plan([])
+ return Plan([], snapshot_id=snapshot_id)
# Use configurable parallelism from table options
max_workers = max(8,
self.table.options.scan_manifest_parallelism(os.cpu_count() or 8))
- # Read manifest entries from manifest files
+ def require_add(entry):
+ if entry.kind != 0:
+ raise ValueError("Incremental manifests must contain only ADD
entries")
+ return True
+
+ # Validate before the manifest reader reconciles ADD/DELETE entries.
entries = self._manifest_file_manager.read_entries_parallel(
manifest_files,
- manifest_entry_filter=None,
+ manifest_entry_filter=require_add,
max_workers=max_workers
)
# Apply shard/bucket filtering for parallel consumption
entries = self._filter_entries_for_shard(entries) if entries else []
if not entries:
- return Plan([])
+ return Plan([], snapshot_id=snapshot_id)
# Get split options from table
options = self.table.options
@@ -403,7 +408,10 @@ class AsyncStreamingTableScan:
)
splits = split_generator.create_splits(entries)
- return Plan(splits)
+ for split in splits:
+ split.is_streaming = True
+ split.snapshot_id = snapshot_id
+ return Plan(splits, snapshot_id=snapshot_id)
def _should_use_diff_catch_up(self) -> bool:
"""Check if diff-based catch-up should be used (large gap to
latest)."""
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index c5b7af32df..2b0d5a4ea3 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -868,7 +868,8 @@ class TableRead:
) if push_down_limit else None
effective_read_type = read_type if read_type is not None else
self.read_type
scan_read_type = self._with_predicate_extra_fields(read_type) if
read_type is not None else self._scan_read_type
- if self.table.is_primary_key_table and not split.raw_convertible:
+ if self.table.is_primary_key_table and (
+ getattr(split, 'is_streaming', False) or not
split.raw_convertible):
inner_read_type = scan_read_type
outer_extract_name_paths: Optional[List[List[str]]] = None
if self.nested_name_paths and any(
diff --git a/paimon-python/pypaimon/read/table_scan.py
b/paimon-python/pypaimon/read/table_scan.py
index d2ee59a4b6..9a07b5ab9d 100755
--- a/paimon-python/pypaimon/read/table_scan.py
+++ b/paimon-python/pypaimon/read/table_scan.py
@@ -47,6 +47,7 @@ _NATIVE_FORWARDED_OPTIONS = frozenset({
CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(),
CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(),
CoreOptions.DELETION_VECTORS_MERGE_ON_READ.key(),
+ CoreOptions.SCAN_VERSION.key(),
CoreOptions.SCAN_SNAPSHOT_ID.key(),
CoreOptions.SCAN_TAG_NAME.key(),
CoreOptions.SCAN_TIMESTAMP.key(),
@@ -60,6 +61,7 @@ _NATIVE_PLAN_INDEPENDENT_OPTIONS = frozenset({
CoreOptions.READ_PARALLELISM.key(),
})
_NATIVE_TIME_TRAVEL_OPTIONS = frozenset({
+ CoreOptions.SCAN_VERSION.key(),
CoreOptions.SCAN_SNAPSHOT_ID.key(),
CoreOptions.SCAN_TAG_NAME.key(),
CoreOptions.SCAN_TIMESTAMP.key(),
@@ -113,8 +115,7 @@ class TableScan:
def _native_plan_supported_impl(self) -> bool:
"""Fall back to the Python scanner for scans native can't carry:
- chunk-shuffle, scored or primary-key
- global-index results, first-row scans which include L0, postpone
bucket,
+ chunk-shuffle, primary-key global-index results, first-row scans which
include L0,
a primary-key table whose trimmed PK is empty (PK equals the partition
key; Rust rejects this schema), a stale
schema without time travel, removed copy() options which Rust cannot
@@ -128,9 +129,10 @@ class TableScan:
if not native_runtime_available():
return False
fs = self.file_scanner
+ if fs.is_streaming and not native_method_available('Split',
'is_streaming'):
+ return False
if (getattr(fs, 'chunk_shuffle', None) is not None
- or not self._native_global_index_result_supported()
- or getattr(fs, 'only_read_real_buckets', False)):
+ or not self._native_global_index_result_supported()):
return False
# Positional append distribution needs the stable partition/file order
# introduced in 0.4. Older bindings can assign different rows per call.
@@ -181,7 +183,7 @@ class TableScan:
# Java batch first-row reads skip L0. Scans including L0 still need
# first-row overlap packing that Rust does not currently provide.
if (self.table.options.merge_engine() == 'first-row'
- and not fs.skip_level0):
+ and not fs.skip_level0 and not fs.is_streaming):
return False
# Rust rejects schemas whose primary keys are all partition keys.
if getattr(self.table, 'is_primary_key_table', False) \
@@ -215,8 +217,7 @@ class TableScan:
return False
from pypaimon.snapshot.time_travel_util import SCAN_KEYS
unsupported_scan_keys = set(SCAN_KEYS) - _NATIVE_TIME_TRAVEL_OPTIONS
- if any(options.contains_key(k) for k in unsupported_scan_keys) \
- or options.contains_key('scan.version'):
+ if any(options.contains_key(k) for k in unsupported_scan_keys):
return False
return (not options.contains(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP)
or native_method_available('ReadBuilder',
'new_incremental_scan'))
@@ -229,9 +230,7 @@ class TableScan:
or not self.file_scanner.data_evolution):
return False
from pypaimon.globalindex.global_index_result import GlobalIndexResult
- from pypaimon.globalindex.vector_search_result import
ScoredGlobalIndexResult
- return (isinstance(result, GlobalIndexResult)
- and not isinstance(result, ScoredGlobalIndexResult))
+ return isinstance(result, GlobalIndexResult)
def _native_row_ranges(self) -> Optional[List[Tuple[int, int]]]:
row_ranges = getattr(self.file_scanner, '_row_ranges', None)
@@ -293,6 +292,11 @@ class TableScan:
splits = [s for s in splits
if getattr(s, 'partition', None) is None
or partition_predicate.test(s.partition)]
+ if (self.table.is_primary_key_table and not fs.is_streaming and
self.predicate is not None
+ and self.table.options.global_index_enabled()
+ and plan.snapshot_id is not None):
+ snapshot =
self.table.snapshot_manager().get_snapshot_by_id(plan.snapshot_id)
+ splits = fs._apply_primary_key_sorted_indexes(splits, snapshot)
if has_distribution:
if self.table.is_primary_key_table:
splits = [s for s in splits
@@ -307,6 +311,15 @@ class TableScan:
start, end = fs.start_pos_of_this_subtask,
fs.end_pos_of_this_subtask
splits = slice_append_splits(splits, start, end)
splits = fs._apply_push_down_limit(splits)
+ # Attach scores to the row ranges retained by native planning.
+ from pypaimon.globalindex.indexed_split import IndexedSplit,
scores_for_ranges
+ from pypaimon.globalindex.vector_search_result import
ScoredGlobalIndexResult
+ result = fs._global_index_result
+ if fs._row_ranges is None and isinstance(result,
ScoredGlobalIndexResult):
+ splits = [IndexedSplit(
+ split.data_split(), split.row_ranges(),
+ scores_for_ranges(result.score_getter(),
split.row_ranges()),
+ ) for split in splits]
return Plan(splits, snapshot_id=plan.snapshot_id)
except Exception as e:
# Any native construction/planning/pruning failure -> fall back.
@@ -429,6 +442,7 @@ class TableScan:
self.limit,
partition_predicate=self.partition_predicate,
skip_level0=False,
+ is_streaming=True,
)
if has_time_travel:
@@ -499,6 +513,7 @@ class TableScan:
has_snapshot_id = options.contains(CoreOptions.SCAN_SNAPSHOT_ID)
has_tag_name = options.contains(CoreOptions.SCAN_TAG_NAME)
+ has_version = options.contains(CoreOptions.SCAN_VERSION)
has_watermark = options.contains(CoreOptions.SCAN_WATERMARK)
has_timestamp_millis =
options.contains(CoreOptions.SCAN_TIMESTAMP_MILLIS)
has_timestamp = options.contains(CoreOptions.SCAN_TIMESTAMP)
@@ -507,6 +522,8 @@ class TableScan:
has_creation_time =
options.contains(CoreOptions.SCAN_CREATION_TIME_MILLIS)
present_keys = []
+ if has_version:
+ present_keys.append(CoreOptions.SCAN_VERSION.key())
if has_snapshot_id:
present_keys.append(CoreOptions.SCAN_SNAPSHOT_ID.key())
if has_tag_name:
@@ -552,11 +569,12 @@ class TableScan:
CoreOptions.SCAN_SNAPSHOT_ID.key(),
CoreOptions.SCAN_TAG_NAME.key(),
CoreOptions.SCAN_WATERMARK.key(),
+ CoreOptions.SCAN_VERSION.key(),
}
- if not (has_snapshot_id or has_tag_name or has_watermark):
+ if not (has_snapshot_id or has_tag_name or has_watermark or
has_version):
raise ValueError(
"scan.mode is 'from-snapshot' but none of "
- "scan.snapshot-id, scan.tag-name, or scan.watermark is
set."
+ "scan.version, scan.snapshot-id, scan.tag-name, or
scan.watermark is set."
)
elif mode == StartupMode.INCREMENTAL:
allowed = {CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key()}
diff --git a/paimon-python/pypaimon/snapshot/time_travel_util.py
b/paimon-python/pypaimon/snapshot/time_travel_util.py
index 8a8609b147..99bfd3a1bf 100644
--- a/paimon-python/pypaimon/snapshot/time_travel_util.py
+++ b/paimon-python/pypaimon/snapshot/time_travel_util.py
@@ -26,6 +26,7 @@ from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.tag.tag_manager import TagManager
SCAN_KEYS = [
+ CoreOptions.SCAN_VERSION.key(),
CoreOptions.SCAN_SNAPSHOT_ID.key(),
CoreOptions.SCAN_TAG_NAME.key(),
CoreOptions.SCAN_WATERMARK.key(),
@@ -75,6 +76,7 @@ class TimeTravelUtil:
Try to travel to a snapshot based on the options.
Supports the following time travel options:
+ - scan.version: Resolve an existing tag, watermark prefix, or snapshot
id
- scan.tag-name: Travel to a specific tag
- scan.snapshot-id: Travel to a specific snapshot id
- scan.timestamp-millis: Travel to the latest snapshot <= the given
timestamp (ms)
@@ -95,6 +97,23 @@ class TimeTravelUtil:
required manager is not provided
"""
+ # Java adaptScanVersion resolves tags before watermark prefixes and
ids.
+ # Work on a copy so table.copy() retains its original effective
options.
+ if options.contains_key(CoreOptions.SCAN_VERSION.key()):
+ values = dict(options.to_map())
+ version = options.get(CoreOptions.SCAN_VERSION)
+ values.pop(CoreOptions.SCAN_VERSION.key())
+ tag = tag_manager.get(version)
+ if tag is not None:
+ values[CoreOptions.SCAN_TAG_NAME.key()] = version
+ elif version.startswith('watermark-'):
+ values[CoreOptions.SCAN_WATERMARK.key()] =
int(version[len('watermark-'):])
+ elif version and all('0' <= char <= '9' for char in version):
+ values[CoreOptions.SCAN_SNAPSHOT_ID.key()] = version
+ else:
+ raise ValueError("Cannot find a time travel version for %s" %
version)
+ options = Options(values)
+
scan_handle_keys = [key for key in SCAN_KEYS if
options.contains_key(key)]
if not scan_handle_keys:
diff --git
a/paimon-python/pypaimon/table/source/primary_key_sorted_index_result.py
b/paimon-python/pypaimon/table/source/primary_key_sorted_index_result.py
index ae98c7aef9..a92dba764c 100644
--- a/paimon-python/pypaimon/table/source/primary_key_sorted_index_result.py
+++ b/paimon-python/pypaimon/table/source/primary_key_sorted_index_result.py
@@ -96,4 +96,6 @@ def _single_file_split(file_plan):
bucket=source.bucket,
raw_convertible=False,
data_deletion_files=deletion_files,
+ snapshot_id=getattr(source, "snapshot_id", None),
+ is_streaming=getattr(source, "is_streaming", False),
)
diff --git a/paimon-python/pypaimon/tests/global_index_test.py
b/paimon-python/pypaimon/tests/global_index_test.py
index a5c2094862..044d08ae6c 100644
--- a/paimon-python/pypaimon/tests/global_index_test.py
+++ b/paimon-python/pypaimon/tests/global_index_test.py
@@ -319,6 +319,7 @@ class GlobalIndexScalarFallbackTest(unittest.TestCase):
predicate = Predicate(method="equal", index=0, field="id",
literals=[1])
scanner = FileScanner.__new__(FileScanner)
+ scanner.is_streaming = False
scanner.predicate = predicate
scanner.partition_key_predicate = None
scanner.table = _Table()
@@ -395,6 +396,7 @@ class GlobalIndexScalarFallbackTest(unittest.TestCase):
options = _Options()
scanner = FileScanner.__new__(FileScanner)
+ scanner.is_streaming = False
scanner.predicate = Predicate(
method="equal", index=0, field="id", literals=[1])
scanner.partition_key_predicate = None
@@ -411,6 +413,7 @@ class GlobalIndexScalarFallbackTest(unittest.TestCase):
result = scanner._eval_global_index(snapshot=object())
self.assertIsNone(result)
+
fake_scanner.scan_with_coverage.assert_called_once_with(scanner.predicate)
class PlanSnapshotFetchRegressionTest(
diff --git a/paimon-python/pypaimon/tests/native_plan_expanded_test.py
b/paimon-python/pypaimon/tests/native_plan_expanded_test.py
new file mode 100644
index 0000000000..06422c1c6d
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_plan_expanded_test.py
@@ -0,0 +1,279 @@
+# 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.
+
+"""Native planning coverage for Java time travel, postpone and scored DE
reads."""
+
+import json
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.globalindex.vector_search_result import
DictBasedScoredIndexResult
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.utils.range import Range
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.write.file_store_write import FileStoreWrite
+
+
[email protected](params=[False, pytest.param(True, marks=[
+ pytest.mark.native_plan, pytest.mark.skipif(
+ not native_runtime_available(), reason='Rust main required')])],
ids=['python', 'native'])
+def native(request):
+ return request.param
+
+
[email protected]
+def catalog(tmp_path):
+ result = CatalogFactory.create({'warehouse': str(tmp_path)})
+ result.create_database('default', True)
+ return result
+
+
+def create(catalog, schema, **kwargs):
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
**kwargs), False)
+ return catalog.get_table('default.t')
+
+
+def write(table, rows, schema):
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pydict(
+ {name: [row[name] for row in rows] for name in schema.names},
schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+
+def read(table, native, predicate=None, shard=None, limit=None, result=None,
ranges=None):
+ builder = table.copy({'scan.native-plan.enabled':
str(native).lower()}).new_read_builder()
+ if predicate is not None:
+ builder.with_filter(predicate)
+ if limit is not None:
+ builder.with_limit(limit)
+ scan = builder.new_scan()
+ if shard is not None:
+ scan.with_shard(*shard)
+ if result is not None:
+ scan.with_global_index_result(result)
+ if ranges is not None:
+ scan.with_row_ranges(ranges)
+ if native:
+ with patch.object(scan.file_scanner, 'scan',
side_effect=AssertionError('native fallback')):
+ plan = scan.plan()
+ else:
+ plan = scan.plan()
+ return plan, builder.new_read().to_arrow(plan.splits()).to_pylist()
+
+
[email protected]('version,expected', [('1', 1), ('base', 1),
('watermark-150', 2)])
+def test_scan_version_resolves_java_selectors(catalog, native, version,
expected):
+ schema = pa.schema([('id', pa.int64())])
+ table = create(catalog, schema)
+ for snapshot, value in enumerate((1, 2), 1):
+ write(table, [{'id': value}], schema)
+ path = table.snapshot_manager().get_snapshot_path(snapshot)
+ metadata = json.loads(table.file_io.read_file_utf8(path))
+ metadata['watermark'] = snapshot * 100
+ table.file_io.write_file(path, json.dumps(metadata), overwrite=True)
+ table.create_tag('base', 1)
+ plan, rows = read(table.copy({'scan.version': version}), native)
+ assert plan.snapshot_id == expected
+ assert sorted(row['id'] for row in rows) == list(range(1, expected + 1))
+ # A numeric tag wins over the snapshot with the same name, as in Java.
+ table.create_tag('2', 1)
+ assert read(table.copy({'scan.version': '2'}), native)[0].snapshot_id == 1
+
+
[email protected]('options', [
+ {'scan.version': 'missing'}, {'scan.version': 'watermark-invalid'},
+ {'scan.version': '1', 'scan.timestamp-millis': '1'},
+ {'scan.version': '1', 'incremental-between-timestamp': '1,2'},
+])
+def test_scan_version_rejects_invalid_or_conflicting_selectors(catalog,
options):
+ table = create(catalog, pa.schema([('id', pa.int64())]))
+ with pytest.raises((ValueError, RuntimeError)):
+ read(table.copy(options), False)
+
+
+def test_postpone_reads_only_real_buckets(catalog, native):
+ schema = pa.schema([('id', pa.int64()), ('p', pa.string()), ('v',
pa.string())])
+ table = create(catalog, schema, primary_keys=['id', 'p'],
partition_keys=['p'], options={
+ 'bucket': '-2', 'source.split.target-size': '1 b',
'source.split.open-file-cost': '1 b'})
+ writer = FileStoreWrite(table, 'postpone')
+ rows = [{'id': 1, 'p': 'a', 'v': 'real'}, {'id': 2, 'p': 'a', 'v':
'pending'},
+ {'id': 3, 'p': 'b', 'v': 'real'}, {'id': 4, 'p': 'b', 'v':
'pending'}]
+ try:
+ for row, bucket in zip(rows, (0, -1, 1, -1)):
+ batch = pa.RecordBatch.from_pydict(
+ {name: [row[name]] for name in schema.names}, schema=schema)
+ writer.write((row['p'],), bucket, batch)
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ commit.commit(writer.prepare_commit(1))
+ finally:
+ commit.close()
+ finally:
+ writer.close()
+ pb = table.new_read_builder().new_predicate_builder()
+ for predicate in (None, pb.equal('p', 'a'), pb.equal('id', 2),
pb.equal('v', 'real')):
+ for shard in (None, (0, 2), (1, 2)):
+ for limit in (None, 1):
+ plan, actual = read(table, native, predicate, shard, limit)
+ expected = [row for row, bucket in zip(rows, (0, -1, 1, -1))
+ if bucket >= 0 and (shard is None or bucket %
shard[1] == shard[0])
+ and (predicate is None or
predicate.test(GenericRow(list(row.values()), table.fields)))]
+ assert all(split.bucket >= 0 for split in plan.splits())
+ assert plan.snapshot_id == 1
+ expected_count = min(len(expected), limit) if limit is not
None else len(expected)
+ assert len(actual) == expected_count
+ assert all(row in expected for row in actual)
+ assert read(table.copy({'scan.snapshot-id': '1'}), native)[1] == [rows[0],
rows[2]]
+
+
[email protected]('dv', [False, True])
+def test_scored_de_preserves_score_mapping_and_selection(catalog, native, dv):
+ schema = pa.schema([('id', pa.int64()), ('v', pa.string())])
+ table = create(catalog, schema, options={
+ 'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true',
+ 'deletion-vectors.enabled': str(dv).lower(),
+ 'source.split.target-size': '1 b', 'source.split.open-file-cost': '1
b'})
+ for start in (0, 3):
+ write(table, [{'id': i, 'v': str(i)} for i in range(start, start +
3)], schema)
+ if dv:
+ builder = table.new_batch_write_builder()
+ commit = builder.new_commit()
+ try:
+ commit.commit(builder.new_update().delete_by_row_id([1]))
+ finally:
+ commit.close()
+ scores = {5: 0.9, 1: 0.2, 3: 0.5}
+ result = DictBasedScoredIndexResult(scores)
+ pb = table.new_read_builder().new_predicate_builder()
+ for shard in (None, (0, 2), (1, 2)):
+ for predicate in (None, pb.greater_than('id', 1)):
+ plan, actual = read(table, native, predicate, shard, result=result)
+ for split in plan.splits():
+ ids = [i for r in split.row_ranges() for i in range(r.from_,
r.to + 1)]
+ assert split.scores() == [scores[i] for i in ids]
+ python_rows = read(table, False, predicate, shard,
result=result)[1]
+ assert actual == python_rows
+ assert all(row['id'] in scores and (not dv or row['id'] != 1) for
row in actual)
+ limited = read(table, native, predicate, shard, limit=1,
result=result)[1]
+ assert len(limited) == min(len(actual), 1)
+ assert all(row in actual for row in limited)
+ assert {row['id'] for row in read(table, native, result=result)[1]} ==
({3, 5} if dv else {1, 3, 5})
+ empty, rows = read(table, native, result=DictBasedScoredIndexResult({}))
+ assert rows == [] and empty.snapshot_id == (3 if dv else 2)
+ with pytest.raises(ValueError, match='mutually exclusive'):
+ read(table, native, result=result, ranges=[Range(0, 0)])
+
+
[email protected]('covered', [True, False])
+def test_pk_sorted_index_refines_native_splits_at_selected_snapshot(catalog,
native, covered):
+ from pypaimon.globalindex.global_index_meta import GlobalIndexMeta
+ from pypaimon.globalindex.indexed_split import IndexedSplit
+ from pypaimon.index.index_file_meta import IndexFileMeta
+ from pypaimon.index.pk.primary_key_index_source_file import
PrimaryKeyIndexSourceFile
+ from pypaimon.index.pk.primary_key_index_source_meta import
PrimaryKeyIndexSourceMeta
+ from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+ from pypaimon.tests.primary_key_sorted_index_scan_test import _Reader
+
+ schema = pa.schema([('id', pa.int64()), ('v', pa.int64())])
+ table = create(catalog, schema, primary_keys=['id'], options={
+ 'bucket': '1', 'pk-btree.index.columns': 'v'})
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pydict(
+ {'id': [1, 2, 3], 'v': [10, 20, 10]}, schema=schema))
+ messages = writer.prepare_commit()
+ file = messages[0].new_files[0]
+ # A compaction-produced L1 file is eligible for the sorted index.
+ file.level, file.file_source = 1, 1
+ commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+ source_meta = PrimaryKeyIndexSourceMeta(1, [
+ PrimaryKeyIndexSourceFile(file.file_name if covered else
'retired-file', 3)]).serialize()
+ payload = IndexFileMeta('btree', 'index', 1, 3,
+ global_index_meta=GlobalIndexMeta(0, 2, 1,
source_meta=source_meta))
+ entry = IndexManifestEntry(0, GenericRow([], []), 0, payload)
+ write(table, [{'id': 2, 'v': 99}], schema)
+ predicate = table.new_read_builder().new_predicate_builder().equal('v', 20)
+ with patch('pypaimon.index.index_file_handler.IndexFileHandler.scan',
return_value=[entry]) as scan_index, \
+
patch('pypaimon.table.source.primary_key_sorted_index_scan.reader_factory',
+ return_value=lambda *args: _Reader([Range(1, 1)])):
+ plan, rows = read(table.copy({'scan.version': '1'}), native, predicate)
+ assert rows == [{'id': 2, 'v': 20}]
+ assert plan.snapshot_id == 1
+ assert scan_index.call_args[0][0].id == 1
+ assert any(isinstance(split, IndexedSplit) for split in plan.splits()) ==
covered
+ if native:
+ assert all(split.snapshot_id == 1 for split in plan.splits())
+ # The latest snapshot includes a newer version. Index pruning must not
+ # resurrect the old matching version from a merge-required split.
+ with patch('pypaimon.index.index_file_handler.IndexFileHandler.scan',
return_value=[entry]), \
+
patch('pypaimon.table.source.primary_key_sorted_index_scan.reader_factory',
+ return_value=lambda *args: _Reader([Range(1, 1)])):
+ assert read(table, native, predicate)[1] == []
+
+
+def test_scored_result_requires_a_score_for_each_selected_row(catalog, native):
+ from pypaimon.globalindex.global_index_result import GlobalIndexResult
+ from pypaimon.globalindex.vector_search_result import
ScoredGlobalIndexResult
+ schema = pa.schema([('id', pa.int64())])
+ table = create(catalog, schema, options={
+ 'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true'})
+ write(table, [{'id': 1}], schema)
+ result = ScoredGlobalIndexResult.create(
+ GlobalIndexResult.from_range(Range(0, 0)).results(), lambda _: None)
+ builder = table.copy({'scan.native-plan.enabled':
str(native).lower()}).new_read_builder()
+ with pytest.raises(ValueError, match='score'):
+ builder.new_scan().with_global_index_result(result).plan()
+
+
[email protected]('version', ['1', 'before-drop'])
+def test_scan_version_restores_historical_schema(catalog, native, version):
+ from pypaimon.schema.schema_change import SchemaChange
+ schema = pa.schema([('id', pa.int64()), ('old_value', pa.string())])
+ table = create(catalog, schema)
+ write(table, [{'id': 1, 'old_value': 'historic'}], schema)
+ table.create_tag('before-drop', 1)
+ catalog.alter_table('default.t', [SchemaChange.drop_column('old_value')],
False)
+ table = catalog.get_table('default.t')
+ historical = table.copy({'scan.version': version})
+ assert historical.field_names == ['id', 'old_value']
+ plan, rows = read(historical, native)
+ assert plan.snapshot_id == 1
+ assert rows == [{'id': 1, 'old_value': 'historic'}]
+ assert table.field_names == ['id']
+
+
[email protected]('version,key', [('1', 'scan.snapshot-id'), ('before',
'scan.tag-name')])
+def test_scan_version_overwrites_same_selector(catalog, native, version, key):
+ schema = pa.schema([('id', pa.int64())])
+ table = create(catalog, schema)
+ write(table, [{'id': 1}], schema)
+ table.create_tag('before', 1)
+ selected = table.copy({'scan.version': version, key:
'invalid-overridden-value'})
+ assert read(selected, native)[1] == [{'id': 1}]
+ assert selected.options.options.to_map()[key] == 'invalid-overridden-value'
diff --git a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
index 7eaac1ad8c..9dde912e44 100644
--- a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-"""Compare committed timestamp windows, including cross-snapshot PK merges."""
+"""Compare committed timestamp windows, including cross-snapshot change
events."""
import json
from contextlib import ExitStack
@@ -32,7 +32,7 @@ from pypaimon.utils.range import Range
@pytest.fixture(params=[False, pytest.param(
True, marks=[pytest.mark.native_plan, pytest.mark.skipif(
- not native_method_available('ReadBuilder', 'new_incremental_scan'),
+ not native_method_available('Split', 'is_streaming'),
reason='pypaimon_rust combined incremental planning API required')])],
ids=['python', 'native'])
def native(request):
@@ -125,16 +125,16 @@ def history(catalog):
@pytest.mark.parametrize('window,snapshot_id,expected', [
- ((100, 200), 3, [(1, 'latest'), (3, 'third'), (4, 'fourth')]),
+ ((100, 200), 3, [(1, 'intermediate'), (1, 'latest'), (3, 'third'), (4,
'fourth')]),
((200, 300), 4, []),
- ((100, 300), 4, [(1, 'latest'), (3, 'third'), (4, 'fourth')]),
+ ((100, 300), 4, [(1, 'intermediate'), (1, 'latest'), (3, 'third'), (4,
'fourth')]),
((100, 150), 1, []),
((500, 600), None, []),
((0, 50), None, []),
((0, 100), 1, [(1, 'base'), (2, 'base')]),
((200, 400), 5, [(5, 'fifth')]),
])
-def test_timestamp_windows_merge_appends_and_preserve_end_snapshot(
+def test_timestamp_windows_preserve_append_events_and_end_snapshot(
native, history, window, snapshot_id, expected):
plan, rows = _read(history, native, window)
assert plan.snapshot_id == snapshot_id
@@ -143,11 +143,11 @@ def
test_timestamp_windows_merge_appends_and_preserve_end_snapshot(
assert all(split.snapshot_id == snapshot_id for split in plan.splits())
-def test_predicate_does_not_resurrect_an_earlier_version(native, history):
+def test_predicate_can_select_an_earlier_event(native, history):
predicate = history.new_read_builder().new_predicate_builder().equal('v',
'intermediate')
plan, rows = _read(history, native, (100, 200), predicate=predicate,
with_stats=True)
assert plan.snapshot_id == 3
- assert rows == []
+ assert rows == [{'k': 1, 'v': 'intermediate'}]
def test_append_distribution_precedes_limit_in_incremental_window(native,
catalog):
@@ -161,7 +161,7 @@ def
test_append_distribution_precedes_limit_in_incremental_window(native, catalo
assert actual == rows[6:8]
-def test_primary_key_shards_merge_all_selected_commits(native, catalog):
+def test_primary_key_shards_preserve_all_selected_commits(native, catalog):
table = _table(catalog, 'pk_shards', True, {'bucket': '4'})
rows = [{'k': key, 'v': 'base'} for key in range(20)]
_write(table, 100, rows)
@@ -178,7 +178,7 @@ def
test_primary_key_shards_merge_all_selected_commits(native, catalog):
assert len(limited) == 1
assert limited[0] in actual
assert sorted((row['k'], row['v']) for row in result) == [
- (key, 'new') for key in range(20)]
+ (key, version) for key in range(20) for version in ('new', 'old')]
def test_data_evolution_positions_intersect_incremental_row_ranges(native,
catalog):
@@ -216,7 +216,7 @@ def
test_incremental_branch_uses_its_own_snapshot_history(native, catalog):
assert actual == [{'k': 2, 'v': 'branch'}]
-def test_incremental_uses_deletion_vectors_from_window_end(native, catalog):
+def test_incremental_ignores_deletion_vectors_from_window_end(native, catalog):
table = _table(catalog, 'deletions', options={
'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true',
'deletion-vectors.enabled': 'true', 'index-file-in-data-file-dir':
'true',
@@ -238,9 +238,9 @@ def
test_incremental_uses_deletion_vectors_from_window_end(native, catalog):
assert actual == rows[3:]
current, actual = _read(table, native, (100, 300))
assert current.snapshot_id == 3
- assert actual == [rows[3], rows[5]]
+ assert actual == rows[3:]
_, actual = _read(table, native, (100, 300), slice_=(1, 3), limit=1)
- assert actual == [rows[5]]
+ assert actual == [rows[4]]
@pytest.mark.parametrize('window', ['100,100', '200,100', '100', 'one,200'])
@@ -248,3 +248,111 @@ def
test_invalid_timestamp_window_is_rejected_even_for_empty_tables(catalog, win
table = _table(catalog, 'invalid')
with pytest.raises(ValueError):
table.copy({'incremental-between-timestamp':
window}).new_read_builder().new_scan()
+
+
[email protected]('engine,dv', [
+ ('deduplicate', False), ('deduplicate', True),
+ ('partial-update', False), ('first-row', False)])
+def test_incremental_includes_l0_across_merge_engines(catalog, native, engine,
dv):
+ table = _table(catalog, 'engines', True, {
+ 'bucket': '1', 'merge-engine': engine,
+ 'deletion-vectors.enabled': str(dv).lower(),
'source.split.target-size': '1b'})
+ _write(table, 100, [{'k': 1, 'v': 'old'}])
+ _write(table, 200, [{'k': 1, 'v': 'new'}])
+ for value in (None, 'old', 'new'):
+ predicate =
table.new_read_builder().new_predicate_builder().equal('v', value) \
+ if value is not None else None
+ plan, rows = _read(table, native, (0, 200), predicate=predicate,
+ limit=1 if value else None)
+ assert all(s.is_streaming and not s.data_deletion_files for s in
plan.splits())
+ assert sorted(row['v'] for row in rows) == ([value] if value else
['new', 'old'])
+
+
+def test_incremental_does_not_evaluate_endpoint_global_indexes(catalog,
native):
+ table = _table(catalog, 'endpoint_index', options={
+ 'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true',
+ 'deletion-vectors.enabled': 'true'})
+ _write(table, 100, [{'k': 1, 'v': 'old'}])
+ _write(table, 200, [{'k': 2, 'v': 'new'}])
+ path = table.snapshot_manager().get_snapshot_path(2)
+ data = json.loads(table.file_io.read_file_utf8(path))
+ data['indexManifest'] = 'missing-endpoint-index'
+ table.file_io.write_file(path, json.dumps(data), overwrite=True)
+ predicate = table.new_read_builder().new_predicate_builder().equal('v',
'old')
+ _, rows = _read(table, native, (0, 200), predicate=predicate)
+ assert rows == [{'k': 1, 'v': 'old'}]
+
+
+def test_incremental_rejects_manifest_delete_before_reconciliation(catalog,
native):
+ from pypaimon.read.streaming_table_scan import AsyncStreamingTableScan
+ table = _table(catalog, 'invalid_delta', True)
+ _write(table, 100, [{'k': 1, 'v': 'old'}])
+ _write(table, 200, [{'k': 1, 'v': 'new'}], overwrite=True)
+ path = table.snapshot_manager().get_snapshot_path(2)
+ data = json.loads(table.file_io.read_file_utf8(path))
+ data['commitKind'] = 'APPEND'
+ table.file_io.write_file(path, json.dumps(data), overwrite=True)
+ # Allow fallback here: both implementations must reject the malformed
input.
+ table = table.copy({'scan.native-plan.enabled': str(native).lower(),
+ 'incremental-between-timestamp': '0,200'})
+ with pytest.raises(ValueError, match='only ADD'):
+ table.new_read_builder().new_scan().plan()
+ scan = AsyncStreamingTableScan(table, prefetch_enabled=False)
+ with pytest.raises(ValueError, match='only ADD'):
+ scan._create_delta_plan(table.snapshot_manager().get_snapshot_by_id(2))
+
+
+def test_incremental_reader_preserves_all_physical_row_kinds(catalog, native):
+ import pyarrow.parquet as pq
+ from pypaimon.read.streaming_table_scan import AsyncStreamingTableScan
+ table = _table(catalog, 'row_kinds', True, {'bucket': '1'})
+ scan = AsyncStreamingTableScan(table, prefetch_enabled=False)
+ expected = []
+ for kind in range(4):
+ snapshot_id = _write(table, (kind + 1) * 100, [{'k': 1, 'v':
str(kind)}])
+ plan =
scan._create_delta_plan(table.snapshot_manager().get_snapshot_by_id(snapshot_id))
+ file = plan.splits()[0].files[0]
+ physical = pq.read_table(file.file_path)
+ index = physical.schema.get_field_index('_VALUE_KIND')
+ physical = physical.set_column(index, physical.schema.field(index),
pa.array([kind], pa.int8()))
+ pq.write_table(physical, file.file_path)
+ rows =
list(table.new_read_builder().new_read().to_iterator(plan.splits()))
+ assert [(row.get_field(1), row.get_row_kind().value) for row in rows]
== [(str(kind), kind)]
+ assert plan.snapshot_id == snapshot_id
+ expected.append((str(kind), kind))
+ builder = table.copy({'scan.native-plan.enabled': str(native).lower(),
+ 'incremental-between-timestamp':
'0,400'}).new_read_builder()
+ batch_scan = builder.new_scan()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(batch_scan.file_scanner, 'scan',
+
side_effect=AssertionError('native fallback')))
+ plan = batch_scan.plan()
+ rows = list(builder.new_read().to_iterator(plan.splits()))
+ assert sorted((row.get_field(1), row.get_row_kind().value) for row in
rows) == expected
+ # Initial streaming bootstrap remains a merged snapshot, where the last -D
removes the key.
+ initial =
scan._create_initial_plan(table.snapshot_manager().get_snapshot_by_id(4))
+ assert all(not split.is_streaming for split in initial.splits())
+ assert
list(table.new_read_builder().new_read().to_iterator(initial.splits())) == []
+
+
+def test_streaming_reader_honors_explicit_split_deletion_vector(catalog,
native, tmp_path):
+ from pypaimon.deletionvectors.bitmap_deletion_vector import
BitmapDeletionVector
+ from pypaimon.table.source.deletion_file import DeletionFile
+ table = _table(catalog, 'explicit_dv', True, {'bucket': '1'})
+ _write(table, 100, [{'k': key, 'v': str(key)} for key in (1, 2, 3)])
+ plan, rows = _read(table, native, (0, 100))
+ assert len(rows) == 3
+ vector = BitmapDeletionVector()
+ vector.delete(1)
+ encoded = vector.serialize()
+ path = tmp_path / 'explicit-dv'
+ path.write_bytes(encoded)
+ assert len(plan.splits()) == 1
+ split = plan.splits()[0]
+ assert split.is_streaming
+ split.data_deletion_files = [DeletionFile(str(path), 0, len(encoded) - 8,
1)]
+ # Planners do not attach endpoint DVs, but an explicit split DV is part of
+ # the reader contract, including in Java streaming frames.
+ result =
table.new_read_builder().new_read().to_arrow(plan.splits()).to_pylist()
+ assert result == [{'k': 1, 'v': '1'}, {'k': 3, 'v': '3'}]
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index b61b0c8fc6..315e15b91e 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -72,6 +72,7 @@ def _scan(native_enabled, file_scanner):
file_scanner._row_ranges = None # no explicit row ranges
file_scanner.deletion_vectors_enabled = False # no deletion vectors
file_scanner.data_evolution = False # no data evolution
+ file_scanner.is_streaming = False
file_scanner.skip_level0 = False
file_scanner.only_read_real_buckets = False # not postpone bucket
scan.file_scanner = file_scanner
@@ -226,21 +227,22 @@ class NativePlanTest(unittest.TestCase):
fs.scan.assert_not_called()
self.assertEqual(plan.splits(), [])
- def test_scored_global_index_result_falls_back(self):
+ def test_scored_global_index_result_uses_native_ranges(self):
+ from pypaimon.globalindex.indexed_split import IndexedSplit
+ from pypaimon.read.split import DataSplit
+
fs = Mock(partition_key_predicate=None)
- sentinel = object()
- fs.scan.return_value = sentinel
scan = _scan(native_enabled=True, file_scanner=fs)
fs.data_evolution = True
bitmap = GlobalIndexResult.from_range(Range(1, 1)).results()
- fs._global_index_result = ScoredGlobalIndexResult.create(
- bitmap, lambda _: 1.0)
-
- with patch('pypaimon.read.native_plan.native_plan') as np:
- self.assertIs(scan.plan(), sentinel)
-
- np.assert_not_called()
- fs.scan.assert_called_once_with()
+ fs._global_index_result = ScoredGlobalIndexResult.create(bitmap,
lambda _: 0.75)
+ split = IndexedSplit(DataSplit([], None, 0), [Range(1, 1)])
+ with patch('pypaimon.read.native_plan.native_plan',
return_value=Plan([split], 9)) as np:
+ plan = scan.plan()
+ self.assertEqual(np.call_args[1]['row_ranges'], [(1, 1)])
+ self.assertEqual(plan.splits()[0].scores(), [0.75])
+ self.assertEqual(plan.snapshot_id, 9)
+ fs.scan.assert_not_called()
def test_global_index_row_ranges_require_data_evolution_append_table(self):
result = GlobalIndexResult.from_range(Range(1, 1))
@@ -283,7 +285,7 @@ class NativePlanTest(unittest.TestCase):
check(lambda s, fs: setattr(fs, '_global_index_result', object()))
check(lambda s, fs: setattr(fs, '_row_ranges', [object()]))
check(lambda s, fs: setattr(fs, 'deletion_vectors_enabled', True))
- check(lambda s, fs: setattr(fs, 'only_read_real_buckets', True))
+ check(lambda s, fs: setattr(fs, 'is_streaming', True))
check(lambda s, fs: (setattr(s.table, 'is_primary_key_table', True),
setattr(s.table, 'trimmed_primary_keys', [])))
check(lambda s, fs: setattr(
@@ -291,8 +293,6 @@ class NativePlanTest(unittest.TestCase):
check(lambda s, fs:
setattr(s.table.schema_manager.latest.return_value, 'id', 2))
check(lambda s, fs: s.table.schema_manager.latest.__setattr__(
'side_effect', RuntimeError('metadata read failed')))
- check(lambda s, fs: s.table.options.options.contains_key.__setattr__(
- 'side_effect', lambda k: k == 'scan.version'))
check(lambda s, fs: s.table.options.merge_engine.__setattr__(
'return_value', 'first-row'))
check(lambda s, fs: setattr(s.table.options, 'query_auth_enabled',
True))
diff --git a/paimon-python/pypaimon/tests/reader_primary_key_test.py
b/paimon-python/pypaimon/tests/reader_primary_key_test.py
index 19a4f1558c..4a1d59043b 100644
--- a/paimon-python/pypaimon/tests/reader_primary_key_test.py
+++ b/paimon-python/pypaimon/tests/reader_primary_key_test.py
@@ -459,6 +459,10 @@ class PkReaderTest(unittest.TestCase):
timestamp = int(time.time() * 1000)
self._write_test_table(table)
+ # Snapshot reads still merge the two versions of user_id=2.
+ snapshot_rows =
self._read_test_table(table.new_read_builder()).sort_by('user_id')
+ self.assertEqual(self.expected, snapshot_rows)
+
snapshot_manager = table.snapshot_manager()
t1 = snapshot_manager.get_snapshot_by_id(1).time_millis
t2 = snapshot_manager.get_snapshot_by_id(2).time_millis
@@ -467,11 +471,18 @@ class PkReaderTest(unittest.TestCase):
read_builder = table.new_read_builder()
actual = self._read_test_table(read_builder)
self.assertEqual(len(actual), 0)
- # test 2
+ # The full incremental window retains both committed versions of
user_id=2.
table = table.copy({CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key():
str(timestamp) + ',' + str(t2)})
read_builder = table.new_read_builder()
- actual = self._read_test_table(read_builder).sort_by('user_id')
- self.assertEqual(self.expected, actual)
+ actual = self._read_test_table(read_builder).sort_by([
+ ('user_id', 'ascending'), ('behavior', 'ascending')])
+ expected = pa.Table.from_pydict({
+ 'user_id': [1, 2, 2, 3, 4, 5, 7, 8],
+ 'item_id': [1001, 1002, 1002, 1003, 1004, 1005, 1007, 1008],
+ 'behavior': ['a', 'b', 'b-new', 'c', None, 'e', 'g', 'h'],
+ 'dt': ['p1', 'p1', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2'],
+ }, schema=self.pa_schema)
+ self.assertEqual(expected, actual)
# test 3
table = table.copy({CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key():
str(t1) + ',' + str(t2)})
read_builder = table.new_read_builder()
diff --git a/paimon-python/pypaimon/tests/split_serializer_test.py
b/paimon-python/pypaimon/tests/split_serializer_test.py
index 0c35e10640..50561905e2 100644
--- a/paimon-python/pypaimon/tests/split_serializer_test.py
+++ b/paimon-python/pypaimon/tests/split_serializer_test.py
@@ -127,6 +127,19 @@ class SplitSerializerTest(unittest.TestCase):
(dv.dv_index_path, dv.offset, dv.length, dv.cardinality),
('dv/file-b', 2, 10, 3))
+ def test_streaming_java_flag_survives_decoding_and_selection(self):
+ data = bytearray(_GOLDEN_DATA_SPLIT_V1)
+ data[-2] = 1
+ split = deserialize_split_v1(bytes(data), self._partition_fields())
+ self.assertTrue(split.is_streaming)
+ selected = split.filter_file(lambda file: file.file_name == 'file-b')
+ self.assertTrue(selected.is_streaming)
+ self.assertEqual(selected.snapshot_id, split.snapshot_id)
+ indexed = IndexedSplit(selected, [])
+ self.assertTrue(indexed.is_streaming)
+ self.assertFalse(deserialize_split_v1(
+ _GOLDEN_DATA_SPLIT_V1, self._partition_fields()).is_streaming)
+
def test_decodes_min_max_keys_with_key_fields(self):
# Trimmed primary keys -> per-file min/max keys are decoded for PK
# merge-on-read. The golden files carry keys [1..10] and [11..20].
