XiaoHongbo-Hope commented on code in PR #9339: URL: https://github.com/apache/paimon/pull/9339#discussion_r3835566891
########## paimon-python/pypaimon/ray/row_id_conflict_rewriter.py: ########## @@ -0,0 +1,499 @@ +################################################################################ +# 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. +################################################################################ + +"""Rebase Ray self-merge updates after concurrent compaction.""" + +import logging +import random +import time +from dataclasses import dataclass, replace +from typing import Dict, List, Optional, Sequence, Tuple + +from pypaimon.common.options.core_options import CoreOptions +from pypaimon.manifest.schema.data_file_meta import DataFileMeta +from pypaimon.read.split import DataSplit +from pypaimon.table.row.generic_row import GenericRow +from pypaimon.table.special_fields import SpecialFields +from pypaimon.utils.range import Range +from pypaimon.write.commit.conflict_detection import RowIdExistenceConflict +from pypaimon.write.commit_message import CommitMessage + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class _StagedFile: + message_index: int + message: CommitMessage + file: DataFileMeta + + +@dataclass(frozen=True) +class _RewriteResult: + update_messages: List[CommitMessage] + superseded_messages: List[CommitMessage] + rewritten_file_count: int + + +def commit_self_merge_with_compaction_retry( + table, + update_messages: Sequence[CommitMessage], + other_messages: Sequence[CommitMessage], + *, + num_partitions: int, + ray_remote_args=None, +) -> None: + """Commit self-merge messages, rebasing stale updates when compaction wins.""" + current_updates = list(update_messages) + other_messages = list(other_messages) + superseded_messages = [] + retry_count = 0 + start_millis = int(time.time() * 1000) + + # Ray performs the rebase below without the local driver's size limit. + commit_table = table.copy_without_time_travel({ + CoreOptions.DATA_EVOLUTION_ROW_ID_CONFLICT_REWRITE_MAX_SIZE.key(): + "0 B", + }) + + base_snapshot_ids = _base_snapshot_ids(current_updates) + latest_snapshot = table.snapshot_manager().get_latest_snapshot() + if ( + len(base_snapshot_ids) == 1 + and latest_snapshot is not None + and latest_snapshot.id != next(iter(base_snapshot_ids)) + ): + result = _rewrite_updates( + table, + current_updates, + latest_snapshot, + num_partitions=num_partitions, + ray_remote_args=ray_remote_args, + ) + if result is not None: + current_updates = result.update_messages + superseded_messages.extend(result.superseded_messages) + logger.info( + "Rewrote %d stale self-merge file(s) against snapshot %d " + "before committing to table %s.", + result.rewritten_file_count, + latest_snapshot.id, + table.identifier, + ) + + while True: + commit = None + conflict = None + try: + commit = commit_table.new_batch_write_builder().new_commit() + # This layer owns stale row-id layout recovery. Do not enter the + # legacy compaction-rollback loop before the distributed rebase; + # ordinary FileStoreCommit retries remain enabled. + commit.file_store_commit.rollback = None + commit.commit(current_updates + other_messages) + except Exception as error: + conflict = _find_row_id_conflict(error) + if conflict is None: + raise + finally: + if commit is not None: + try: + commit.close() + except Exception as close_error: + logger.warning( + "Failed to close self-merge commit: %s", + close_error, + exc_info=close_error, + ) + + if conflict is None: + if superseded_messages: + from pypaimon.write.file_store_commit import ( + _abort_commit_messages, + ) + _abort_commit_messages(table, superseded_messages) Review Comment: > [P1] Preserve uncertain commit state before deleting superseded files > > This wrapper treats every surfaced `RowIdExistenceConflict` as a definite pre-commit failure. However, `FileStoreCommit` can surface that conflict after an earlier snapshot commit attempt became uncertain: the remote commit may have succeeded, its snapshot may be temporarily unreadable to duplicate detection, and a later compaction may change the row-ID layout. `_validate_no_logical_conflict` also skips the unreadable snapshot and the compaction snapshot, so a later successful rebase can reach this cleanup and delete a file that is still referenced by the earlier committed snapshot, tag, or branch. > > Please propagate the uncertain outcome to this coordinator and never rewrite or abort any generation that may have been committed unless its status is conclusively resolved. fixed now -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
