[Subtask]: Use a new configuration item to control whether master & slave mode is enabled. #3845

Author: wardli

amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/IcebergRewriteExecutor.java

@@ -101,12 +101,25 @@ protected TaskWriter<Record> dataWriter() {
   @Override
   protected long targetSize() {
     long targetSize = super.targetSize();
-    long inputSize =
-        Arrays.stream(input.rewrittenDataFiles()).mapToLong(DataFile::fileSizeInBytes).sum();
-    // When the input files’ total size is below targetSize, remove the output file size limit to
+    DataFile[] rewrittenDataFiles = input.rewrittenDataFiles();
+    if (rewrittenDataFiles == null || rewrittenDataFiles.length == 0) {
+      return targetSize;
+    }
+    long inputSize = Arrays.stream(rewrittenDataFiles).mapToLong(DataFile::fileSizeInBytes).sum();
+    // When the input files' total size is below targetSize, remove the output file size limit to
     // avoid outputting multiple files.
     // For more details, please refer to: https://github.com/apache/amoro/issues/3645
-    return inputSize < targetSize ? Long.MAX_VALUE : targetSize;
+    if (inputSize < targetSize) {
+      return Long.MAX_VALUE;
+    }
+    // Even if total size >= targetSize, if average file size is small (less than targetSize),
+    // we should still merge files to avoid outputting too many small files.
+    // This ensures that many small files can be merged effectively.
+    long averageFileSize = inputSize / rewrittenDataFiles.length;
+    if (averageFileSize < targetSize) {
+      return Long.MAX_VALUE;
+    }
+    return targetSize;
   }
 
   private PartitionSpec fileSpec() {

amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/plan/AbstractPartitionPlan.java

@@ -263,9 +263,7 @@ public Weight getWeight() {
   /**
    * When splitTask has only one undersized segment file, it needs to be triggered again to
    * determine whether to rewrite pos. If needed, add it to rewritePosDataFiles and bin-packing
-   * together. If it doesn't need to rewrite pos delete, add it to rewriteDataFiles so it can be
-   * merged with fragment files or other files during the second bin-packing, which prevents file
-   * loss and ensures proper file consolidation.
+   * together, else reserved delete files.
    */
   protected void disposeUndersizedSegmentFile(SplitTask splitTask) {
     Optional<DataFile> dataFile = splitTask.getRewriteDataFiles().stream().findFirst();
@@ -275,10 +273,7 @@ protected void disposeUndersizedSegmentFile(SplitTask splitTask) {
       if (evaluator().segmentShouldRewritePos(rewriteDataFile, deletes)) {
         rewritePosDataFiles.put(rewriteDataFile, deletes);
       } else {
-        // Add to rewriteDataFiles so it can be merged with other files (fragment files, etc.)
-        // during the second bin-packing. This prevents file loss and ensures files can be
-        // consolidated when possible.
-        rewriteDataFiles.put(rewriteDataFile, deletes);
+        reservedDeleteFiles(deletes);
       }
     }
   }