hbgstc123 commented on code in PR #9790: URL: https://github.com/apache/paimon/pull/9790#discussion_r4056575939
########## paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SortCompactCommitter.java: ########## @@ -0,0 +1,262 @@ +/* + * 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. + */ + +package org.apache.paimon.flink.sink; + +import org.apache.paimon.append.SortCompactCommitMessageRewriter; +import org.apache.paimon.manifest.ManifestCommittable; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.table.sink.TableCommit; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +/** + * A {@link StoreCommitter} for the sort compact topology. It rewrites the append {@link + * CommitMessage}s produced by the sort compact write stage into compact commit messages (via {@link + * SortCompactCommitMessageRewriter}) before delegating to the normal {@link StoreCommitter} commit, + * so that the resulting snapshot is a {@code COMPACT} snapshot instead of an {@code OVERWRITE} + * snapshot. + */ +public class SortCompactCommitter extends StoreCommitter { + + private final SortCompactCommitMessageRewriter rewriter; + + public SortCompactCommitter( + FileStoreTable table, + TableCommit commit, + Context context, + SortCompactCommitMessageRewriter rewriter) { + super(table, commit, context); + this.rewriter = rewriter; + } + + @Override + protected long additionalBytesOut(CommitMessageImpl impl) { + return calcTotalFileSize(impl.compactIncrement().compactAfter()); + } + + @Override + protected long additionalRecordsOut(CommitMessageImpl impl) { + return calcTotalFileRowCount(impl.compactIncrement().compactAfter()); + } + + @Override + public void commit(List<ManifestCommittable> committables) + throws IOException, InterruptedException { + List<CommitMessage> writtenMessages = collectWrittenMessages(committables); + List<ManifestCommittable> rewritten; + try { + rewritten = rewriteAll(committables); + } catch (RuntimeException e) { + abortWrittenQuietly(writtenMessages, e); + throw e; + } + long snapshotIdBeforeCommit = rewriter.latestSnapshotIdOrZero(); + try { + super.commit(rewritten); + } catch (IOException | InterruptedException e) { + maybeAbortAfterFailedCommit(writtenMessages, rewritten, snapshotIdBeforeCommit, e); + throw e; + } catch (RuntimeException e) { + maybeAbortAfterFailedCommit(writtenMessages, rewritten, snapshotIdBeforeCommit, e); + throw e; + } + } + + @Override + public int filterAndCommit( + List<ManifestCommittable> globalCommittables, + boolean checkAppendFiles, + boolean partitionMarkDoneRecoverFromState) { + List<ManifestCommittable> sortedCommittables = + globalCommittables.stream() + .sorted(Comparator.comparingLong(ManifestCommittable::identifier)) + .collect(Collectors.toList()); + List<ManifestCommittable> retryCommittables = commit.filterCommitted(sortedCommittables); + if (retryCommittables.isEmpty()) { + // Delete-only compact commits are only valid at job end (filterAndCommit with + // checkAppendFiles=false, e.g. CommitterOperator#endInput). Recovery paths call + // filterAndCommit with checkAppendFiles=true and an empty restored list; treating + // that as delete-only would remove all planned input files before writers run. + if (!checkAppendFiles + && sortedCommittables.isEmpty() + && rewriter.hasInput() + && !rewriter.isPlannedInputAlreadyCommitted()) { + return filterAndCommitDeleteOnly( + globalCommittables, checkAppendFiles, partitionMarkDoneRecoverFromState); + } + commitListeners.notifyCommittable( + globalCommittables, partitionMarkDoneRecoverFromState); + return 0; + } + + List<CommitMessage> writtenMessages = collectWrittenMessages(retryCommittables); + List<ManifestCommittable> rewritten; + try { + rewritten = rewriteAll(retryCommittables); + } catch (RuntimeException e) { + abortWrittenQuietly(writtenMessages, e); + throw e; + } + long snapshotIdBeforeCommit = rewriter.latestSnapshotIdOrZero(); + int committed; + try { + committed = commit.filterAndCommitMultiple(rewritten, checkAppendFiles); + } catch (RuntimeException e) { + maybeAbortAfterFailedCommit(writtenMessages, rewritten, snapshotIdBeforeCommit, e); + throw e; + } + calcNumBytesAndRecordsOut(rewritten); + commitListeners.notifyCommittable(globalCommittables, partitionMarkDoneRecoverFromState); + return committed; + } + + private int filterAndCommitDeleteOnly( + List<ManifestCommittable> globalCommittables, + boolean checkAppendFiles, + boolean partitionMarkDoneRecoverFromState) { + List<ManifestCommittable> rewritten; + try { + rewritten = rewriteAll(Collections.emptyList()); + } catch (RuntimeException e) { + abortWrittenQuietly(Collections.emptyList(), e); + throw e; + } + if (rewritten.isEmpty()) { + return 0; + } + long snapshotIdBeforeCommit = rewriter.latestSnapshotIdOrZero(); + int committed; + try { + committed = commit.filterAndCommitMultiple(rewritten, checkAppendFiles); + } catch (RuntimeException e) { + maybeAbortAfterFailedCommit( + Collections.emptyList(), rewritten, snapshotIdBeforeCommit, e); + throw e; + } + calcNumBytesAndRecordsOut(rewritten); + commitListeners.notifyCommittable(globalCommittables, partitionMarkDoneRecoverFromState); + return committed; + } + + /** + * Merge all partial write outputs from multiple committables and rewrite once. Each individual + * rewrite would attach the full planned {@code compactBefore}, so committing partial outputs + * separately would delete all old files too early. + * + * <p>This also collapses multiple committables into a single one with the maximum identifier, + * which changes the per-identifier deduplication semantics used during job recovery. Sort + * compact is a batch job and does not rely on that recovery path, so this is acceptable here. + */ + private List<ManifestCommittable> rewriteAll(List<ManifestCommittable> committables) { + if (committables.isEmpty()) { + if (!rewriter.hasInput()) { + return committables; + } + + // A sort compact is a batch job. Even when all input rows are filtered out (for + // example, by deletion vectors), its planned input files still need to be removed. + return Collections.singletonList( + new ManifestCommittable( + Long.MAX_VALUE, + null, + rewriter.rewrite(Collections.emptyList()), + Collections.emptyMap())); + } + + List<CommitMessage> allWrittenMessages = new ArrayList<>(); + long identifier = Long.MIN_VALUE; + Long watermark = null; + Map<String, String> properties = new HashMap<>(); + + for (ManifestCommittable committable : committables) { + allWrittenMessages.addAll(committable.fileCommittables()); + identifier = Math.max(identifier, committable.identifier()); + if (committable.watermark() != null) { + watermark = + watermark == null + ? committable.watermark() + : Math.max(watermark, committable.watermark()); + } + properties.putAll(committable.properties()); + } + + List<CommitMessage> compactMessages = rewriter.rewrite(allWrittenMessages); + return Collections.singletonList( + new ManifestCommittable(identifier, watermark, compactMessages, properties)); + } + + private static List<CommitMessage> collectWrittenMessages( + List<ManifestCommittable> committables) { + List<CommitMessage> writtenMessages = new ArrayList<>(); + for (ManifestCommittable committable : committables) { + writtenMessages.addAll(committable.fileCommittables()); + } + return writtenMessages; + } + + private static List<CommitMessage> compactMessagesFrom( + List<ManifestCommittable> rewrittenCommittables) { + return collectWrittenMessages(rewrittenCommittables); + } + + private void maybeAbortAfterFailedCommit( + List<CommitMessage> writtenMessages, + List<ManifestCommittable> rewrittenCommittables, + long snapshotIdBeforeCommit, + Exception cause) { + List<CommitMessage> compactMessages = compactMessagesFrom(rewrittenCommittables); + if (rewriter.isBatchCompactCommitSucceeded(snapshotIdBeforeCommit, compactMessages)) { + return; + } + // Abort both the original written messages and the rewritten compact messages. The + // rewritten compact messages carry the new deletion-vector index files produced by + // dvMaintainer.persist() during rewrite, which are not referenced by the original written + // messages; aborting only the written messages would orphan them. For delete-only compact, + // writtenMessages is empty but compactMessages still carries the new DV index files, so it + // must be aborted (the previous writtenMessages.isEmpty() early return skipped cleanup + // entirely in that case). + abortWrittenQuietly(writtenMessages, cause); Review Comment: Thanks, agreed. The pre-snapshot abort path was deleting writer output (`compactAfter` is the same files). Flink batch recovery can restart only the committer and replay the same committables (`CommitterOperator#commitUpToCheckpoint`), so one transient failure made every retry fail with `Cannot recover ... files ... have been deleted`. Fix in 30fa106c3: - Keep writer output / `compactAfter` after a failed commit so the same committable can be replayed. - Clean only the new DV index files produced by rewrite (`abortNewIndexFiles`); they are regenerated on the next rewrite. - Add `testCommitFailureThenCommitterOnlyReplaySucceeds`: fail `filterAndCommit` once, then replay the same `Long.MAX_VALUE` committable with the same commit user. Spark is unchanged: a Spark procedure re-runs writers, so aborting written files there is still correct. -- 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]
