[core] Commit independent Data Evolution compaction tasks on conflict

docs/docs/multimodal-table/data-evolution-maintenance.mdx

@@ -78,6 +78,12 @@ nor include more than one file smaller than `vector.target-file-size`. See
 [Dedicated Compaction](../maintenance/dedicated-compaction) for scheduling
 compaction jobs.
 
+`data-evolution.compaction.discard-conflicting-tasks` (default `false`) lets a
+data-only compaction commit skip complete tasks whose row ranges conflict with
+concurrent updates. Their input files remain live for later compaction; other
+conflicts still fail the commit. Unreferenced outputs can be reclaimed by
+`remove_orphan_files` after its safety window.
+
 Ordinary compaction **does not physically remove rows hidden by deletion
 vectors**. Removing their positions would change the alignment with untouched
 column files.

docs/generated/core_configuration.html

@@ -524,6 +524,12 @@
             <td>Duration</td>
             <td>The TTL in local index for cross partition upsert (primary keys not contain all partition fields), this can avoid maintaining too many indexes and lead to worse and worse performance, but please note that this may also cause data duplication.</td>
         </tr>
+        <tr>
+            <td><h5>data-evolution.compaction.discard-conflicting-tasks</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Skip complete data-evolution compaction tasks with row-range conflicts. Skipped inputs remain live for later compaction. Applies only to data-only compaction messages without row-id validation baselines.</td>
+        </tr>
         <tr>
             <td><h5>data-evolution.compaction.large-file-ratio</h5></td>
             <td style="word-wrap: break-word;">2.0</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -2774,6 +2774,15 @@ public String toString() {
                                     + "Row IDs and logical deletions are preserved, and associated "
                                     + "BLOB and VECTOR files are not rewritten by this option.");
 
+    public static final ConfigOption<Boolean> DATA_EVOLUTION_COMPACTION_DISCARD_CONFLICTING_TASKS =
+            key("data-evolution.compaction.discard-conflicting-tasks")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Skip complete data-evolution compaction tasks with row-range conflicts. "
+                                    + "Skipped inputs remain live for later compaction. Applies only to "
+                                    + "data-only compaction messages without row-id validation baselines.");
+
     public static final ConfigOption<Double> DATA_EVOLUTION_COMPACTION_LARGE_FILE_RATIO =
             key("data-evolution.compaction.large-file-ratio")
                     .doubleType()

paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java

@@ -29,6 +29,7 @@
 
 import javax.annotation.Nullable;
 
+import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -82,7 +83,9 @@ public interface FileEntry {
      * The same {@link Identifier} indicates that the {@link ManifestEntry} refers to the same data
      * file.
      */
-    class Identifier {
+    class Identifier implements Serializable {
+
+        private static final long serialVersionUID = 1L;
 
         public final BinaryRow partition;
         public final int bucket;

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java

@@ -51,6 +51,8 @@
 import org.apache.paimon.operation.commit.CommitScanner;
 import org.apache.paimon.operation.commit.ConflictDetection;
 import org.apache.paimon.operation.commit.DataEvolutionChangesProvider;
+import org.apache.paimon.operation.commit.DataEvolutionRowRangeConflictException;
+import org.apache.paimon.operation.commit.DiscardingCompactChangesProvider;
 import org.apache.paimon.operation.commit.ManifestEntryChanges;
 import org.apache.paimon.operation.commit.RetryCommitResult;
 import org.apache.paimon.operation.commit.RetryCommitResult.CommitFailRetryResult;
@@ -347,6 +349,7 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) {
         List<CommitMessage> commitMessages = committable.fileCommittables();
         configureRowIdCheckFromMessages(commitMessages);
         ManifestEntryChanges changes = collectChanges(commitMessages);
+        List<ManifestEntry> committedCompactFiles = changes.compactTableFiles;
         Set<Pair<BinaryRow, Integer>> materializedBuckets = materializedBuckets(commitMessages);
         try {
             List<SimpleFileEntry> appendSimpleEntries =
@@ -401,22 +404,23 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) {
             if (!changes.compactTableFiles.isEmpty()
                     || !changes.compactChangelog.isEmpty()
                     || !changes.compactIndexFiles.isEmpty()) {
+                CommitChangesProvider compactProvider =
+                        compactChangesProvider(changes, materializedBuckets, commitMessages);
                 attempts +=
                         tryCommit(
-                                dataEvolutionChangesProvider(
-                                        new CommitChanges(
-                                                changes.compactTableFiles,
-                                                changes.compactChangelog,
-                                                changes.compactIndexFiles),
-                                        CommitKind.COMPACT,
-                                        materializedBuckets),
+                                compactProvider,
                                 committable.identifier(),
                                 committable.watermark(),
                                 committable.properties(),
                                 CommitKind.COMPACT,
                                 false,
                                 true,
                                 null);
+                if (compactProvider instanceof DiscardingCompactChangesProvider) {
+                    committedCompactFiles =
+                            ((DiscardingCompactChangesProvider) compactProvider)
+                                    .retainedOriginalFiles();
+                }
                 generatedSnapshot += 1;
             }
         } finally {
@@ -429,7 +433,7 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) {
                 reportCommit(
                         changes.appendTableFiles,
                         changes.appendChangelog,
-                        changes.compactTableFiles,
+                        committedCompactFiles,
                         changes.compactChangelog,
                         commitDuration,
                         generatedSnapshot,
@@ -862,6 +866,56 @@ private Set<Pair<BinaryRow, Integer>> materializedBuckets(List<CommitMessage> co
         return result;
     }
 
+    private CommitChangesProvider compactChangesProvider(
+            ManifestEntryChanges changes,
+            Set<Pair<BinaryRow, Integer>> materializedBuckets,
+            List<CommitMessage> messages) {
+        CommitChangesProvider provider =
+                dataEvolutionChangesProvider(
+                        new CommitChanges(
+                                changes.compactTableFiles,
+                                changes.compactChangelog,
+                                changes.compactIndexFiles),
+                        CommitKind.COMPACT,
+                        materializedBuckets);
+        if (!options.dataEvolutionEnabled()
+                || !options.toConfiguration()
+                        .get(CoreOptions.DATA_EVOLUTION_COMPACTION_DISCARD_CONFLICTING_TASKS)
+                || !materializedBuckets.isEmpty()
+                || conflictDetection.shouldCheckRowIdFromSnapshot(CommitKind.COMPACT)
+                || !changes.appendTableFiles.isEmpty()
+                || !changes.appendChangelog.isEmpty()
+                || !changes.appendIndexFiles.isEmpty()
+                || !changes.compactChangelog.isEmpty()
+                || !changes.compactIndexFiles.isEmpty()) {
+            return provider;
+        }
+        List<List<ManifestEntry>> tasks = new ArrayList<>();
+        Set<FileEntry.Identifier> taskFiles = new HashSet<>();
+        for (CommitMessage message : messages) {
+            ManifestEntryChanges task = new ManifestEntryChanges(options.bucket());
+            task.collect(message);
+            if (task.compactTableFiles.stream().noneMatch(entry -> entry.kind() == FileKind.ADD)
+                    || task.compactTableFiles.stream()
+                            .noneMatch(entry -> entry.kind() == FileKind.DELETE)
+                    || task.compactTableFiles.stream()
+                            .anyMatch(
+                                    entry ->
+                                            entry.firstRowId() == null
+                                                    || isBlobFile(entry.fileName())
+                                                    || isVectorStoreFile(entry.fileName()))) {
+                return provider;
+            }
+            for (ManifestEntry entry : task.compactTableFiles) {
+                if (!taskFiles.add(entry.identifier())) {
+                    return provider;
+                }
+            }
+            tasks.add(task.compactTableFiles);
+        }
+        return new DiscardingCompactChangesProvider(provider, tasks);
+    }
+
     @VisibleForTesting
     CommitChangesProvider dataEvolutionChangesProvider(
             CommitChanges changes,
@@ -1251,6 +1305,27 @@ private CommitResult tryCommitOnceWithoutLock(
                             indexFiles,
                             rowIdConflictChecker,
                             commitKind);
+            while (exception.isPresent()
+                    && exception.get() instanceof DataEvolutionRowRangeConflictException
+                    && changesProvider instanceof DiscardingCompactChangesProvider
+                    && ((DiscardingCompactChangesProvider) changesProvider)
+                            .discard(
+                                    ((DataEvolutionRowRangeConflictException) exception.get())
+                                            .conflictingFiles())) {
+                LOG.warn(
+                        "Discarding conflicting compaction tasks for table {}: {}",
+                        tableName,
+                        exception.get().getMessage());
+                deltaFiles = changesProvider.provide(latestSnapshot).tableFiles;
+                exception =
+                        conflictDetection.checkConflicts(
+                                latestSnapshot,
+                                baseDataFiles,
+                                SimpleFileEntry.from(deltaFiles),
+                                indexFiles,
+                                rowIdConflictChecker,
+                                commitKind);
+            }
             if (exception.isPresent()) {
                 if (allowRollback
                         && rollback != null

paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java

@@ -412,7 +412,10 @@ private Optional<RuntimeException> checkDataFileRowIdRangeConflicts(
                                         + "have encountered conflicts, data files: "
                                         + dataFileGroup.stream()
                                                 .map(this::fileEntryToString)
-                                                .collect(Collectors.toList())));
+                                                .collect(Collectors.toList()),
+                                dataFileGroup.stream()
+                                        .map(FileEntry::identifier)
+                                        .collect(Collectors.toList())));
             }
         }
         return Optional.empty();

paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionRowRangeConflictException.java

@@ -18,10 +18,30 @@
 
 package org.apache.paimon.operation.commit;
 
+import org.apache.paimon.manifest.FileEntry;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
 /** Conflict caused by incompatible row-range boundaries in a data-evolution table. */
 public final class DataEvolutionRowRangeConflictException extends RuntimeException {
 
+    private static final long serialVersionUID = 1L;
+
+    private final List<FileEntry.Identifier> conflictingFiles;
+
     public DataEvolutionRowRangeConflictException(String message) {
+        this(message, Collections.emptyList());
+    }
+
+    public DataEvolutionRowRangeConflictException(
+            String message, List<FileEntry.Identifier> conflictingFiles) {
         super(message);
+        this.conflictingFiles = Collections.unmodifiableList(new ArrayList<>(conflictingFiles));
+    }
+
+    public List<FileEntry.Identifier> conflictingFiles() {
+        return conflictingFiles;
     }
 }

paimon-core/src/main/java/org/apache/paimon/operation/commit/DiscardingCompactChangesProvider.java

@@ -0,0 +1,84 @@
+/*
+ * 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.operation.commit;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Drops complete compaction messages, preserving their inputs for a later attempt. */
+public class DiscardingCompactChangesProvider implements CommitChangesProvider {
+    private final CommitChangesProvider delegate;
+    private final List<List<ManifestEntry>> tasks;
+    private final Set<FileEntry.Identifier> discarded = new HashSet<>();
+
+    public DiscardingCompactChangesProvider(
+            CommitChangesProvider delegate, List<List<ManifestEntry>> tasks) {
+        this.delegate = delegate;
+        this.tasks = tasks;
+    }
+
+    public boolean discard(List<FileEntry.Identifier> conflicts) {
+        Set<FileEntry.Identifier> identifiers = new HashSet<>(conflicts);
+        int previous = discarded.size();
+        for (List<ManifestEntry> task : tasks) {
+            if (task.stream()
+                    .anyMatch(
+                            entry ->
+                                    entry.kind() == FileKind.ADD
+                                            && identifiers.contains(entry.identifier()))) {
+                task.forEach(entry -> discarded.add(entry.identifier()));
+            }
+        }
+        return discarded.size() > previous;
+    }
+
+    @Override
+    public CommitChanges provide(@Nullable Snapshot latest) {
+        CommitChanges changes = delegate.provide(latest);
+        List<ManifestEntry> retained = new ArrayList<>();
+        for (ManifestEntry entry : changes.tableFiles) {
+            if (!discarded.contains(entry.identifier())) {
+                retained.add(entry);
+            }
+        }
+        return new CommitChanges(retained, changes.changelogFiles, changes.indexFiles);
+    }
+
+    public List<ManifestEntry> retainedOriginalFiles() {
+        return tasks.stream()
+                .flatMap(List::stream)
+                .filter(entry -> !discarded.contains(entry.identifier()))
+                .collect(Collectors.toList());
+    }
+
+    @Override
+    public Set<Long> rebasedReassignments() {
+        return delegate.rebasedReassignments();
+    }
+}