[core] Make bucketed append table write initial lighter

Author: JingsongLi

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

@@ -38,7 +38,6 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Comparator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Optional;
@@ -47,6 +46,7 @@
 import java.util.concurrent.ExecutorService;
 
 import static java.util.Collections.emptyList;
+import static org.apache.paimon.table.source.AppendOnlySplitGenerator.DATA_FILE_COMPARATOR;
 
 /** Compact manager for {@link AppendOnlyFileStore}. */
 public class BucketedAppendCompactManager extends CompactFutureManager {
@@ -78,7 +78,7 @@ public BucketedAppendCompactManager(
             @Nullable CompactionMetrics.Reporter metricsReporter) {
         this.executor = executor;
         this.dvMaintainer = dvMaintainer;
-        this.toCompact = new PriorityQueue<>(fileComparator(false));
+        this.toCompact = new PriorityQueue<>(DATA_FILE_COMPARATOR);
         this.toCompact.addAll(restored);
         this.minFileNum = minFileNum;
         this.targetFileSize = targetFileSize;
@@ -357,35 +357,4 @@ private static CompactResult result(List<DataFileMeta> before, List<DataFileMeta
     public interface CompactRewriter {
         List<DataFileMeta> rewrite(List<DataFileMeta> compactBefore) throws Exception;
     }
-
-    /**
-     * New files may be created during the compaction process, then the results of the compaction
-     * may be put after the new files, and this order will be disrupted. We need to ensure this
-     * order, so we force the order by sequence.
-     */
-    public static Comparator<DataFileMeta> fileComparator(boolean ignoreOverlap) {
-        return (o1, o2) -> {
-            if (o1 == o2) {
-                return 0;
-            }
-
-            if (!ignoreOverlap && isOverlap(o1, o2)) {
-                LOG.warn(
-                        String.format(
-                                "There should no overlap in append files, but Range1(%s, %s), Range2(%s, %s),"
-                                        + " check if you have multiple write jobs.",
-                                o1.minSequenceNumber(),
-                                o1.maxSequenceNumber(),
-                                o2.minSequenceNumber(),
-                                o2.maxSequenceNumber()));
-            }
-
-            return Long.compare(o1.minSequenceNumber(), o2.minSequenceNumber());
-        };
-    }
-
-    private static boolean isOverlap(DataFileMeta o1, DataFileMeta o2) {
-        return o2.minSequenceNumber() <= o1.maxSequenceNumber()
-                && o2.maxSequenceNumber() >= o1.minSequenceNumber();
-    }
 }

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

@@ -68,9 +68,16 @@ public BucketedAppendFileStoreWrite(
                 options,
                 dvMaintainerFactory,
                 tableName);
+        super.withIgnorePreviousFiles(options.writeOnly());
         this.commitUser = commitUser;
     }
 
+    @Override
+    public void withIgnorePreviousFiles(boolean ignorePrevious) {
+        // for append table, we need all writers to be empty if write only
+        super.withIgnorePreviousFiles(options.writeOnly());
+    }
+
     @Override
     protected CompactManager getCompactManager(
             BinaryRow partition,

paimon-core/src/main/java/org/apache/paimon/table/source/AppendOnlySplitGenerator.java

@@ -24,15 +24,17 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
-import static org.apache.paimon.append.BucketedAppendCompactManager.fileComparator;
-
 /** Append only implementation of {@link SplitGenerator}. */
 public class AppendOnlySplitGenerator implements SplitGenerator {
 
+    public static final Comparator<DataFileMeta> DATA_FILE_COMPARATOR =
+            Comparator.comparing(DataFileMeta::creationTime).thenComparing(DataFileMeta::fileName);
+
     private final long targetSplitSize;
     private final long openFileCost;
     private final BucketMode bucketMode;
@@ -52,7 +54,7 @@ public boolean alwaysRawConvertible() {
     @Override
     public List<SplitGroup> splitForBatch(List<DataFileMeta> input) {
         List<DataFileMeta> files = new ArrayList<>(input);
-        files.sort(fileComparator(bucketMode == BucketMode.BUCKET_UNAWARE));
+        files.sort(DATA_FILE_COMPARATOR);
         Function<DataFileMeta, Long> weightFunc = file -> Math.max(file.fileSize(), openFileCost);
         return BinPacking.packForOrdered(files, weightFunc, targetSplitSize).stream()
                 .map(SplitGroup::rawConvertibleGroup)