fix

Author: JingsongLi

docs/layouts/shortcodes/generated/core_configuration.html

@@ -166,7 +166,7 @@
         </tr>
         <tr>
             <td><h5>compaction.delete-ratio-threshold</h5></td>
-            <td style="word-wrap: break-word;">0.1</td>
+            <td style="word-wrap: break-word;">0.2</td>
             <td>Double</td>
             <td>Ratio of the deleted rows in a data file to be forced compacted for append-only table.</td>
         </tr>

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

@@ -639,7 +639,7 @@ public class CoreOptions implements Serializable {
     public static final ConfigOption<Double> COMPACTION_DELETE_RATIO_THRESHOLD =
             key("compaction.delete-ratio-threshold")
                     .doubleType()
-                    .defaultValue(0.1)
+                    .defaultValue(0.2)
                     .withDescription(
                             "Ratio of the deleted rows in a data file to be forced compacted for "
                                     + "append-only table.");

paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java

@@ -23,11 +23,11 @@
 import org.apache.paimon.format.FileFormatDiscover;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.manifest.ManifestCacheFilter;
+import org.apache.paimon.operation.AppendFileStoreWrite;
 import org.apache.paimon.operation.AppendOnlyFileStoreScan;
-import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
-import org.apache.paimon.operation.AppendOnlyFixedBucketFileStoreWrite;
-import org.apache.paimon.operation.AppendOnlyUnawareBucketFileStoreWrite;
+import org.apache.paimon.operation.BaseAppendFileStoreWrite;
 import org.apache.paimon.operation.BucketSelectConverter;
+import org.apache.paimon.operation.BucketedAppendFileStoreWrite;
 import org.apache.paimon.operation.RawFileSplitRead;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.schema.SchemaManager;
@@ -88,19 +88,19 @@ public RawFileSplitRead newRead() {
     }
 
     @Override
-    public AppendOnlyFileStoreWrite newWrite(String commitUser) {
+    public BaseAppendFileStoreWrite newWrite(String commitUser) {
         return newWrite(commitUser, null);
     }
 
     @Override
-    public AppendOnlyFileStoreWrite newWrite(
+    public BaseAppendFileStoreWrite newWrite(
             String commitUser, ManifestCacheFilter manifestFilter) {
         DeletionVectorsMaintainer.Factory dvMaintainerFactory =
                 options.deletionVectorsEnabled()
                         ? DeletionVectorsMaintainer.factory(newIndexFileHandler())
                         : null;
         if (bucketMode() == BucketMode.BUCKET_UNAWARE) {
-            return new AppendOnlyUnawareBucketFileStoreWrite(
+            return new AppendFileStoreWrite(
                     fileIO,
                     newRead(),
                     schema.id(),
@@ -113,7 +113,7 @@ public AppendOnlyFileStoreWrite newWrite(
                     dvMaintainerFactory,
                     tableName);
         } else {
-            return new AppendOnlyFixedBucketFileStoreWrite(
+            return new BucketedAppendFileStoreWrite(
                     fileIO,
                     newRead(),
                     schema.id(),

…areAppendTableCompactionCoordinator.java …mon/append/AppendCompactCoordinator.javapaimon-core/src/main/java/org/apache/paimon/append/UnawareAppendTableCompactionCoordinator.java renamed to paimon-core/src/main/java/org/apache/paimon/append/AppendCompactCoordinator.java paimon-core/src/main/java/org/apache/paimon/append/UnawareAppendTableCompactionCoordinator.java renamed to paimon-core/src/main/java/org/apache/paimon/append/AppendCompactCoordinator.java

@@ -22,8 +22,8 @@
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.data.BinaryRow;
-import org.apache.paimon.deletionvectors.append.AppendDeletionFileMaintainer;
-import org.apache.paimon.deletionvectors.append.UnawareAppendDeletionFileMaintainer;
+import org.apache.paimon.deletionvectors.append.AppendDeleteFileMaintainer;
+import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
 import org.apache.paimon.index.IndexFileHandler;
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.manifest.FileKind;
@@ -34,7 +34,6 @@
 import org.apache.paimon.table.source.EndOfScanException;
 import org.apache.paimon.table.source.ScanMode;
 import org.apache.paimon.table.source.snapshot.SnapshotReader;
-import org.apache.paimon.utils.Filter;
 import org.apache.paimon.utils.SnapshotManager;
 
 import javax.annotation.Nullable;
@@ -56,19 +55,19 @@
 /**
  * Compact coordinator for append only tables.
  *
- * <p>Note: {@link UnawareAppendTableCompactionCoordinator} scan files in snapshot, read APPEND and
- * COMPACT snapshot then load those new files. It will try it best to generate compaction task for
- * the restored files scanned in snapshot, but to reduce memory usage, it won't remain single file
- * for a long time. After ten times scan, single file with one partition will be ignored and removed
- * from memory, which means, it will not participate in compaction again until restart the
- * compaction job.
+ * <p>Note: {@link AppendCompactCoordinator} scan files in snapshot, read APPEND and COMPACT
+ * snapshot then load those new files. It will try it best to generate compaction task for the
+ * restored files scanned in snapshot, but to reduce memory usage, it won't remain single file for a
+ * long time. After ten times scan, single file with one partition will be ignored and removed from
+ * memory, which means, it will not participate in compaction again until restart the compaction
+ * job.
  *
  * <p>When a third task delete file in latest snapshot(including batch delete/update and overwrite),
  * the file in coordinator will still remain and participate in compaction task. When this happens,
  * compaction job will fail in commit stage, and fail-over to rescan the restored files in latest
  * snapshot.
  */
-public class UnawareAppendTableCompactionCoordinator {
+public class AppendCompactCoordinator {
 
     private static final int FILES_BATCH = 100_000;
 
@@ -87,11 +86,11 @@ public class UnawareAppendTableCompactionCoordinator {
     final Map<BinaryRow, PartitionCompactCoordinator> partitionCompactCoordinators =
             new HashMap<>();
 
-    public UnawareAppendTableCompactionCoordinator(FileStoreTable table, boolean isStreaming) {
+    public AppendCompactCoordinator(FileStoreTable table, boolean isStreaming) {
         this(table, isStreaming, null);
     }
 
-    public UnawareAppendTableCompactionCoordinator(
+    public AppendCompactCoordinator(
             FileStoreTable table, boolean isStreaming, @Nullable Predicate filter) {
         checkArgument(table.primaryKeys().isEmpty());
         this.snapshotManager = table.snapshotManager();
@@ -108,7 +107,7 @@ public UnawareAppendTableCompactionCoordinator(
         this.filesIterator = new FilesIterator(table, isStreaming, filter);
     }
 
-    public List<UnawareAppendCompactionTask> run() {
+    public List<AppendCompactTask> run() {
         // scan files in snapshot
         if (scan()) {
             // do plan compact tasks
@@ -154,29 +153,20 @@ FilesIterator filesIterator() {
 
     @VisibleForTesting
     void notifyNewFiles(BinaryRow partition, List<DataFileMeta> files) {
-        java.util.function.Predicate<DataFileMeta> filter =
-                file -> {
-                    if (dvMaintainerCache == null
-                            || dvMaintainerCache
-                                            .dvMaintainer(partition)
-                                            .getDeletionFile(file.fileName())
-                                    == null) {
-                        return file.fileSize() < compactionFileSize;
-                    }
-                    // if a data file has a deletion file, always be to compact.
-                    return true;
-                };
-        List<DataFileMeta> toCompact = files.stream().filter(filter).collect(Collectors.toList());
+        List<DataFileMeta> toCompact =
+                files.stream()
+                        .filter(file -> shouldCompact(partition, file))
+                        .collect(Collectors.toList());
         partitionCompactCoordinators
                 .computeIfAbsent(partition, pp -> new PartitionCompactCoordinator(partition))
                 .addFiles(toCompact);
     }
 
     @VisibleForTesting
     // generate compaction task to the next stage
-    List<UnawareAppendCompactionTask> compactPlan() {
+    List<AppendCompactTask> compactPlan() {
         // first loop to found compaction tasks
-        List<UnawareAppendCompactionTask> tasks =
+        List<AppendCompactTask> tasks =
                 partitionCompactCoordinators.values().stream()
                         .flatMap(s -> s.plan().stream())
                         .collect(Collectors.toList());
@@ -213,18 +203,18 @@ public PartitionCompactCoordinator(BinaryRow partition) {
             this.partition = partition;
         }
 
-        public List<UnawareAppendCompactionTask> plan() {
+        public List<AppendCompactTask> plan() {
             return pickCompact();
         }
 
         public BinaryRow partition() {
             return partition;
         }
 
-        private List<UnawareAppendCompactionTask> pickCompact() {
+        private List<AppendCompactTask> pickCompact() {
             List<List<DataFileMeta>> waitCompact = agePack();
             return waitCompact.stream()
-                    .map(files -> new UnawareAppendCompactionTask(partition, files))
+                    .map(files -> new AppendCompactTask(partition, files))
                     .collect(Collectors.toList());
         }
 
@@ -260,14 +250,11 @@ private List<List<DataFileMeta>> agePack() {
         }
 
         private List<List<DataFileMeta>> pack(Set<DataFileMeta> toCompact) {
-            // we compact smaller files first
-            // step 1, sort files by file size, pick the smaller first
+            // we don't know how many parallel compact works there should be, so in order to pack
+            // better, we will sort them first
             ArrayList<DataFileMeta> files = new ArrayList<>(toCompact);
             files.sort(Comparator.comparingLong(DataFileMeta::fileSize));
 
-            // step 2, when files picked size greater than targetFileSize(meanwhile file num greater
-            // than minFileNum) or file numbers bigger than maxFileNum, we pack it to a compaction
-            // task
             List<List<DataFileMeta>> result = new ArrayList<>();
             FileBin fileBin = new FileBin();
             for (DataFileMeta fileMeta : files) {
@@ -278,15 +265,19 @@ private List<List<DataFileMeta>> pack(Set<DataFileMeta> toCompact) {
                 }
             }
 
-            if (fileBin.enoughInputFiles() || fileBin.containsTooHighDeleteFile()) {
+            if (fileBin.enoughInputFiles()) {
                 result.add(new ArrayList<>(fileBin.bin));
-                fileBin.reset();
             }
+            // else skip these small files that are too few
+
             return result;
         }
 
         private List<List<DataFileMeta>> packInDeletionVectorVMode(Set<DataFileMeta> toCompact) {
             // we group the data files by their related index files.
+            // In the subsequent compact task, if any files with deletion vectors are compacted, we
+            // need to rewrite their corresponding deleted files. To avoid duplicate deleted files,
+            // we must group them according to the deleted files
             Map<String, List<DataFileMeta>> filesWithDV = new HashMap<>();
             Set<DataFileMeta> rest = new HashSet<>();
             for (DataFileMeta dataFile : toCompact) {
@@ -301,13 +292,35 @@ private List<List<DataFileMeta>> packInDeletionVectorVMode(Set<DataFileMeta> toC
                 }
             }
 
-            List<List<DataFileMeta>> result = new ArrayList<>(filesWithDV.values());
+            // To avoid too small a compact task, merge them
+            List<List<DataFileMeta>> dvGroups = new ArrayList<>(filesWithDV.values());
+            dvGroups.sort(Comparator.comparingLong(this::fileSizeOfList));
+
+            List<List<DataFileMeta>> result = new ArrayList<>();
+            FileBin fileBin = new FileBin();
+            for (List<DataFileMeta> dvGroup : dvGroups) {
+                fileBin.addFiles(dvGroup);
+                if (fileBin.enoughContent()) {
+                    result.add(new ArrayList<>(fileBin.bin));
+                    fileBin.reset();
+                }
+            }
+
+            // for file with deletion vectors, must do compaction
+            if (!fileBin.bin.isEmpty()) {
+                result.add(new ArrayList<>(fileBin.bin));
+            }
+
             if (rest.size() > 1) {
                 result.addAll(pack(rest));
             }
             return result;
         }
 
+        private long fileSizeOfList(List<DataFileMeta> list) {
+            return list.stream().mapToLong(DataFileMeta::fileSize).sum();
+        }
+
         /**
          * A file bin for {@link PartitionCompactCoordinator} determine whether ready to compact.
          */
@@ -321,6 +334,10 @@ public void reset() {
                 totalFileSize = 0;
             }
 
+            public void addFiles(List<DataFileMeta> files) {
+                files.forEach(this::addFile);
+            }
+
             public void addFile(DataFileMeta file) {
                 totalFileSize += file.fileSize() + openFileCost;
                 bin.add(file);
@@ -333,10 +350,6 @@ private boolean enoughContent() {
             private boolean enoughInputFiles() {
                 return bin.size() >= minFileNum;
             }
-
-            private boolean containsTooHighDeleteFile() {
-                return bin.stream().anyMatch(file -> tooHighDeleteRatio(partition, file));
-            }
         }
     }
 
@@ -345,8 +358,7 @@ private class DvMaintainerCache {
         private final IndexFileHandler indexFileHandler;
 
         /** Should be thread safe, ManifestEntryFilter will be invoked in many threads. */
-        private final Map<BinaryRow, UnawareAppendDeletionFileMaintainer> cache =
-                new ConcurrentHashMap<>();
+        private final Map<BinaryRow, AppendDeleteFileMaintainer> cache = new ConcurrentHashMap<>();
 
         private DvMaintainerCache(IndexFileHandler indexFileHandler) {
             this.indexFileHandler = indexFileHandler;
@@ -356,12 +368,12 @@ private void refresh() {
             this.cache.clear();
         }
 
-        private UnawareAppendDeletionFileMaintainer dvMaintainer(BinaryRow partition) {
-            UnawareAppendDeletionFileMaintainer maintainer = cache.get(partition);
+        private AppendDeleteFileMaintainer dvMaintainer(BinaryRow partition) {
+            AppendDeleteFileMaintainer maintainer = cache.get(partition);
             if (maintainer == null) {
                 synchronized (this) {
                     maintainer =
-                            AppendDeletionFileMaintainer.forUnawareAppend(
+                            BaseAppendDeleteFileMaintainer.forUnawareAppend(
                                     indexFileHandler, snapshotManager.latestSnapshot(), partition);
                 }
                 cache.put(partition, maintainer);
@@ -420,17 +432,10 @@ private void assignNewIterator() {
             if (dvMaintainerCache != null) {
                 dvMaintainerCache.refresh();
             }
-            Filter<ManifestEntry> entryFilter =
-                    entry -> {
-                        if (entry.file().fileSize() < compactionFileSize) {
-                            return true;
-                        }
-
-                        return tooHighDeleteRatio(entry.partition(), entry.file());
-                    };
             currentIterator =
                     snapshotReader
-                            .withManifestEntryFilter(entryFilter)
+                            .withManifestEntryFilter(
+                                    entry -> shouldCompact(entry.partition(), entry.file()))
                             .withSnapshot(snapshot)
                             .readFileIterator();
         }
@@ -458,6 +463,10 @@ public ManifestEntry next() {
         }
     }
 
+    private boolean shouldCompact(BinaryRow partition, DataFileMeta file) {
+        return file.fileSize() < compactionFileSize || tooHighDeleteRatio(partition, file);
+    }
+
     private boolean tooHighDeleteRatio(BinaryRow partition, DataFileMeta file) {
         if (dvMaintainerCache != null) {
             DeletionFile deletionFile =
@@ -472,7 +481,7 @@ private boolean tooHighDeleteRatio(BinaryRow partition, DataFileMeta file) {
     }
 
     @VisibleForTesting
-    UnawareAppendDeletionFileMaintainer dvMaintainer(BinaryRow partition) {
+    AppendDeleteFileMaintainer dvMaintainer(BinaryRow partition) {
         return dvMaintainerCache.dvMaintainer(partition);
     }
 }