JingsongLi commented on code in PR #7873:
URL: https://github.com/apache/paimon/pull/7873#discussion_r3485047833
##########
paimon-python/pypaimon/write/file_store_commit.py:
##########
@@ -189,6 +191,118 @@ def commit(self, commit_messages: List[CommitMessage],
commit_identifier: int):
allow_rollback=allow_rollback,
index_deletes=index_deletes)
+ def commit_compact(self, commit_messages: List[CommitMessage],
commit_identifier: int):
+ """Commit compaction-only messages.
+
+ Each message must carry no data increment; compact_before → DELETE
+ entries, compact_after → ADD entries; snapshot kind = COMPACT. Skips
+ row-id assignment and conflict detection (a compaction never produces
+ new rows).
+ """
+ if not commit_messages:
+ return
+
+ for msg in commit_messages:
+ if not msg.data_increment.is_empty():
+ raise ValueError(
+ "commit_compact rejects messages carrying a data
increment; "
+ "use commit() instead."
+ )
+
+ logger.info(
+ "Ready to commit compact to table %s, number of commit messages:
%d",
+ self.table.identifier,
+ len(commit_messages),
+ )
+ commit_entries = self._build_commit_entries(commit_messages)
+
+ # Forward index manifest deletions (e.g. global-index updates) carried
by
+ # the compact messages, mirroring commit(). Unlike commit() we do not
run
+ # the new_files-driven apply_global_index_update_action derivation,
since a
+ # compaction produces no new_files; we only ship deletes the caller set
+ # explicitly. A message whose only content is index_deletes still
needs to
+ # commit, so the empty-entries short-circuit must account for them.
+ index_deletes = []
+ for msg in commit_messages:
+ index_deletes.extend(msg.index_deletes)
+
+ if not commit_entries and not index_deletes:
+ return
+
+ logger.info("Finished collecting compact changes: %d entries",
len(commit_entries))
+
+ self._try_commit(
+ commit_kind="COMPACT",
+ commit_identifier=commit_identifier,
+ commit_entries_plan=lambda snapshot: commit_entries,
+ detect_conflicts=False,
+ allow_rollback=False,
+ index_deletes=index_deletes,
+ )
+
+ def _build_commit_entries(self, commit_messages: List[CommitMessage]) ->
List[ManifestEntry]:
+ # new_files / compact_before / compact_after become delta manifest
entries
+ # here; data_increment.changelog_files is handled separately by
+ # _collect_changelog_entries (written to the changelog manifest, not
the
+ # delta). Reject the remaining unwired slots loudly so a future writer
+ # that starts filling them cannot silently lose data at commit time.
+ #
+ # Entry ordering for a message that carries both kinds is
+ # ADD(new_files) -> DELETE(compact_before) -> ADD(compact_after). Today
+ # commit()/commit_compact() enforce that a single message never mixes
+ # data and compact increments, so the mixed case is unreachable; a
future
+ # change that relaxes that validation must keep this ordering in mind.
+ for msg in commit_messages:
+ di = msg.data_increment
+ if (di.deleted_files
+ or di.new_index_files or di.deleted_index_files):
+ raise NotImplementedError(
+ "FileStoreCommit does not yet handle
DataIncrement.deleted_files / "
+ "new_index_files / deleted_index_files; these slots "
+ "will be wired in by the feature that first needs each
one."
+ )
+ ci = msg.compact_increment
+ if (ci.changelog_files or ci.new_index_files or
ci.deleted_index_files):
+ raise NotImplementedError(
+ "FileStoreCommit does not yet handle
CompactIncrement.changelog_files / "
+ "new_index_files / deleted_index_files; these slots will
be wired in by "
+ "the feature that first needs each one."
+ )
+
+ entries: List[ManifestEntry] = []
+ for msg in commit_messages:
+ partition = GenericRow(list(msg.partition),
self.table.partition_keys_fields)
+ # Prefer the message's total_buckets (captured when the plan was
+ # built) over the current table value, so a plan that survived a
+ # bucket rescale is not silently rewritten with the new count.
+ total_buckets = msg.total_buckets if msg.total_buckets is not None
\
Review Comment:
This fixes `total_buckets` for the delta manifest entries, but
`_collect_changelog_entries()` still writes changelog manifest entries with
`self.table.total_buckets`. For a write plan prepared before a bucket-count
change, the data entries now keep `msg.total_buckets` while the changelog
entries are silently rewritten with the current table count, so the same commit
can publish inconsistent bucket metadata. Please apply the same
`msg.total_buckets if msg.total_buckets is not None else
self.table.total_buckets` logic in `_collect_changelog_entries()` (and ideally
cover a message with both `new_files` and `changelog_files`).
--
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]
