[core] Fix in CompactProcedure

Author: juntaozhang

paimon-core/src/main/java/org/apache/paimon/append/AppendCompactTask.java

@@ -35,7 +35,6 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.Comparator;
 import java.util.List;
 import java.util.Objects;
 
@@ -50,7 +49,6 @@ public class AppendCompactTask {
 
     public AppendCompactTask(BinaryRow partition, List<DataFileMeta> files) {
         Preconditions.checkArgument(files != null);
-        files.sort(Comparator.comparingLong(DataFileMeta::minSequenceNumber));
         this.partition = partition;
         compactBefore = new ArrayList<>(files);
         compactAfter = new ArrayList<>();

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java

@@ -87,6 +87,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -459,6 +460,13 @@ private void compactUnAwareBucketTable(
                                                             ser.deserialize(
                                                                     ser.getVersion(),
                                                                     taskIterator.next());
+                                                    if (coreOptions.rowTrackingEnabled()) {
+                                                        task.compactBefore()
+                                                                .sort(
+                                                                        Comparator.comparingLong(
+                                                                                DataFileMeta
+                                                                                        ::minSequenceNumber));
+                                                    }
                                                     messages.add(
                                                             messageSer.serialize(
                                                                     task.doCompact(table, write)));