TheR1sing3un commented on code in PR #7873:
URL: https://github.com/apache/paimon/pull/7873#discussion_r3523458894
##########
paimon-python/pypaimon/write/file_store_commit.py:
##########
@@ -124,33 +124,62 @@ def __init__(self, snapshot_commit: SnapshotCommit,
table, commit_user: str,
self.rollback = CommitRollback(table_rollback) if table_rollback is
not None else None
def commit(self, commit_messages: List[CommitMessage], commit_identifier:
int):
- """Commit the given commit messages in normal append mode."""
+ """Commit the given messages, splitting write and compaction results.
+
+ Each CommitMessage may carry a data_increment (normal write), a
+ compact_increment (compaction result), or both. Data increments feed an
+ APPEND (or OVERWRITE) snapshot; compact increments feed a separate
+ COMPACT snapshot, so a single commit() call produces up to two
+ snapshots, mirroring Java's FileStoreCommitImpl. Compaction never
+ assigns row-ids or runs conflict detection.
+ """
if not commit_messages:
return
- # Extract the minimum check_from_snapshot from commit messages
- valid_snapshots = [msg.check_from_snapshot for msg in commit_messages
- if msg.check_from_snapshot != -1]
- if valid_snapshots:
- self.conflict_detection._row_id_check_from_snapshot =
min(valid_snapshots)
+ self._reject_unwired_slots(commit_messages)
logger.info(
"Ready to commit to table %s, number of commit messages: %d",
self.table.identifier,
len(commit_messages),
)
- commit_entries = self._collect_manifest_entries(commit_messages)
+
+ # index_adds / index_deletes ride with the snapshot their payload
+ # belongs to: messages carrying a compact increment go to the COMPACT
+ # snapshot, everything else (writes and pure index updates) goes to the
+ # APPEND snapshot. Attributing them to exactly one side keeps the same
+ # index manifest change from being committed twice.
+ data_index_adds, data_index_deletes = [], []
+ compact_index_adds, compact_index_deletes = [], []
+ for msg in commit_messages:
+ if msg.compact_increment.is_empty():
+ data_index_adds.extend(msg.index_adds)
+ data_index_deletes.extend(msg.index_deletes)
+ else:
+ compact_index_adds.extend(msg.index_adds)
+ compact_index_deletes.extend(msg.index_deletes)
+
+ # APPEND/OVERWRITE snapshot first, COMPACT second, mirroring Java's
+ # FileStoreCommitImpl.commit ordering.
+ self._commit_data(commit_messages, commit_identifier,
+ data_index_adds, data_index_deletes)
+ self._commit_compact(commit_messages, commit_identifier,
+ compact_index_adds, compact_index_deletes)
+
+ def _commit_data(self, commit_messages: List[CommitMessage],
commit_identifier: int,
+ index_adds: List, index_deletes: List) -> None:
+ # Extract the minimum check_from_snapshot from commit messages
+ valid_snapshots = [msg.check_from_snapshot for msg in commit_messages
Review Comment:
fixed
--
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]
