This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 4def22901aa87cdbbd6b1e49f4efe0022a07526a 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].
