[core] Add compactIndexFiles to ManifestEntryChanges.changedPartitions

Author: JingsongLi

paimon-common/src/main/java/org/apache/paimon/utils/ListUtils.java

@@ -46,4 +46,11 @@ public static <T> List<T> toList(Iterator<T> iterator) {
         }
         return result;
     }
+
+    public static <E> List<E> union(List<? extends E> list1, List<? extends E> list2) {
+        ArrayList<E> result = new ArrayList<>(list1.size() + list2.size());
+        result.addAll(list1);
+        result.addAll(list2);
+        return result;
+    }
 }

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

@@ -352,7 +352,8 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) {
                                     changedPartitions(
                                             changes.appendTableFiles,
                                             changes.compactTableFiles,
-                                            changes.appendIndexFiles)));
+                                            changes.appendIndexFiles,
+                                            changes.compactIndexFiles)));
                     if (discardDuplicate) {
                         Set<FileEntry.Identifier> baseIdentifiers =
                                 baseEntries.stream()
@@ -897,7 +898,7 @@ CommitResult tryCommitOnce(
             // latestSnapshotId is different from the snapshot id we've checked for conflicts,
             // so we have to check again
             List<BinaryRow> changedPartitions =
-                    changedPartitions(deltaFiles, Collections.emptyList(), indexFiles);
+                    changedPartitions(deltaFiles, emptyList(), indexFiles, emptyList());
             if (retryResult != null && retryResult.latestSnapshot != null) {
                 baseDataFiles = new ArrayList<>(retryResult.baseDataFiles);
                 List<SimpleFileEntry> incremental =

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

@@ -25,6 +25,7 @@
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.table.sink.CommitMessage;
 import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.ListUtils;
 
 import java.util.ArrayList;
 import java.util.HashSet;
@@ -165,17 +166,15 @@ public String toString() {
     public static List<BinaryRow> changedPartitions(
             List<ManifestEntry> appendTableFiles,
             List<ManifestEntry> compactTableFiles,
-            List<IndexManifestEntry> appendIndexFiles) {
+            List<IndexManifestEntry> appendIndexFiles,
+            List<IndexManifestEntry> compactIndexFiles) {
         Set<BinaryRow> changedPartitions = new HashSet<>();
-        for (ManifestEntry appendTableFile : appendTableFiles) {
-            changedPartitions.add(appendTableFile.partition());
+        for (ManifestEntry file : ListUtils.union(appendTableFiles, compactTableFiles)) {
+            changedPartitions.add(file.partition());
         }
-        for (ManifestEntry compactTableFile : compactTableFiles) {
-            changedPartitions.add(compactTableFile.partition());
-        }
-        for (IndexManifestEntry appendIndexFile : appendIndexFiles) {
-            if (appendIndexFile.indexFile().indexType().equals(DELETION_VECTORS_INDEX)) {
-                changedPartitions.add(appendIndexFile.partition());
+        for (IndexManifestEntry file : ListUtils.union(appendIndexFiles, compactIndexFiles)) {
+            if (file.indexFile().indexType().equals(DELETION_VECTORS_INDEX)) {
+                changedPartitions.add(file.partition());
             }
         }
         return new ArrayList<>(changedPartitions);