| # 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. |
| |
| """ |
| Conflict detection for commit operations. |
| """ |
| |
| |
| from pypaimon.manifest.manifest_list_manager import ManifestListManager |
| from pypaimon.manifest.schema.data_file_meta import DataFileMeta |
| from pypaimon.manifest.schema.file_entry import FileEntry |
| from pypaimon.utils.range import Range |
| from pypaimon.utils.range_helper import RangeHelper |
| from pypaimon.write.commit.commit_scanner import CommitScanner |
| |
| |
| class ConflictDetection: |
| """Detects conflicts between base and delta files during commit. |
| |
| This class provides row ID range conflict checks and row ID from snapshot conflict checks |
| for Data Evolution tables. |
| """ |
| |
| def __init__(self, data_evolution_enabled, snapshot_manager, |
| manifest_list_manager: ManifestListManager, table, commit_scanner: CommitScanner): |
| """Initialize ConflictDetection. |
| |
| Args: |
| data_evolution_enabled: Whether data evolution feature is enabled. |
| snapshot_manager: Manager for reading snapshot metadata. |
| manifest_list_manager: Manager for reading manifest lists. |
| table: The FileStoreTable instance. |
| """ |
| self.data_evolution_enabled = data_evolution_enabled |
| self.snapshot_manager = snapshot_manager |
| self.manifest_list_manager = manifest_list_manager |
| self.table = table |
| self._row_id_check_from_snapshot = None |
| self.commit_scanner = commit_scanner |
| |
| def should_be_overwrite_commit(self): |
| """Check if the commit should be treated as an overwrite commit. |
| |
| returns True if rowIdCheckFromSnapshot is set. |
| |
| Returns: |
| True if the commit should be treated as OVERWRITE. |
| """ |
| return self._row_id_check_from_snapshot is not None |
| |
| def check_conflicts(self, latest_snapshot, base_entries, delta_entries, commit_kind): |
| """Run all conflict checks and return the first detected conflict. |
| |
| merges base_entries and delta_entries, then runs conflict checks |
| on the merged result. |
| |
| Args: |
| latest_snapshot: The latest snapshot at commit time. |
| base_entries: All entries read from the latest snapshot. |
| delta_entries: The delta entries being committed. |
| commit_kind: The kind of commit (e.g. "APPEND", "COMPACT", "OVERWRITE"). |
| |
| Returns: |
| A RuntimeError if a conflict is detected, otherwise None. |
| """ |
| all_entries = list(base_entries) + list(delta_entries) |
| |
| try: |
| merged_entries = FileEntry.merge_entries(all_entries) |
| except Exception as e: |
| return RuntimeError( |
| "File deletion conflicts detected! Give up committing. " + str(e)) |
| |
| conflict = self.check_row_id_range_conflicts(commit_kind, merged_entries) |
| if conflict is not None: |
| return conflict |
| |
| return self.check_row_id_from_snapshot(latest_snapshot, delta_entries) |
| |
| def check_row_id_range_conflicts(self, commit_kind, commit_entries): |
| """Check for row ID range conflicts among merged entries. |
| |
| only enabled when data evolution is active, and checks that |
| overlapping row ID ranges in non-blob data files are identical. |
| |
| Args: |
| commit_kind: The kind of commit (e.g. "APPEND", "COMPACT"). |
| commit_entries: The entries being committed. |
| |
| Returns: |
| A RuntimeError if conflict is detected, otherwise None. |
| """ |
| if not self.data_evolution_enabled: |
| return None |
| if self._row_id_check_from_snapshot is None and commit_kind != "COMPACT": |
| return None |
| |
| entries_with_row_id = [ |
| entry for entry in commit_entries |
| if entry.file.first_row_id is not None |
| ] |
| |
| if not entries_with_row_id: |
| return None |
| |
| range_helper = RangeHelper(lambda entry: entry.file.row_id_range()) |
| merged_groups = range_helper.merge_overlapping_ranges(entries_with_row_id) |
| |
| for group in merged_groups: |
| data_files = [ |
| entry for entry in group |
| if not DataFileMeta.is_blob_file(entry.file.file_name) |
| ] |
| if not range_helper.are_all_ranges_same(data_files): |
| file_descriptions = [ |
| "{name}(rowId={row_id}, count={count})".format( |
| name=entry.file.file_name, |
| row_id=entry.file.first_row_id, |
| count=entry.file.row_count, |
| ) |
| for entry in data_files |
| ] |
| return RuntimeError( |
| "For Data Evolution table, multiple 'MERGE INTO' and 'COMPACT' " |
| "operations have encountered conflicts, data files: " |
| + str(file_descriptions)) |
| |
| return None |
| |
| def check_row_id_from_snapshot(self, latest_snapshot, commit_entries): |
| """Check for row ID conflicts from a specific snapshot onwards. |
| |
| collects row ID ranges from delta entries, then checks if any |
| incremental changes between the check snapshot and latest snapshot |
| have overlapping row ID ranges. |
| |
| Args: |
| latest_snapshot: The latest snapshot at commit time. |
| commit_entries: The delta entries being committed. |
| |
| Returns: |
| A RuntimeError if conflict is detected, otherwise None. |
| """ |
| if not self.data_evolution_enabled: |
| return None |
| if self._row_id_check_from_snapshot is None: |
| return None |
| |
| history_id_ranges = [] |
| for entry in commit_entries: |
| first_row_id = entry.file.first_row_id |
| row_count = entry.file.row_count |
| if first_row_id is not None: |
| history_id_ranges.append( |
| Range(first_row_id, first_row_id + row_count - 1)) |
| |
| check_snapshot = self.snapshot_manager.get_snapshot_by_id( |
| self._row_id_check_from_snapshot) |
| if check_snapshot is None or check_snapshot.next_row_id is None: |
| raise RuntimeError( |
| "Next row id cannot be null for snapshot " |
| "{snapshot}.".format(snapshot=self._row_id_check_from_snapshot)) |
| check_next_row_id = check_snapshot.next_row_id |
| |
| for snapshot_id in range( |
| self._row_id_check_from_snapshot + 1, |
| latest_snapshot.id + 1): |
| snapshot = self.snapshot_manager.get_snapshot_by_id(snapshot_id) |
| if snapshot is None: |
| continue |
| if snapshot.commit_kind == "COMPACT": |
| continue |
| |
| incremental_entries = self.commit_scanner.read_incremental_entries_from_changed_partitions( |
| snapshot, commit_entries) |
| for entry in incremental_entries: |
| file_range = entry.file.row_id_range() |
| if file_range is None: |
| continue |
| if file_range.from_ < check_next_row_id: |
| for history_range in history_id_ranges: |
| if history_range.overlaps(file_range): |
| return RuntimeError( |
| "For Data Evolution table, multiple 'MERGE INTO' " |
| "operations have encountered conflicts, updating " |
| "the same file, which can render some updates " |
| "ineffective.") |
| |
| return None |
