You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/07/01 03:22:10 UTC
(paimon) branch master updated: [core] Fix dry run in OrphanFilesClean and optimize result string
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 8c4be22c7 [core] Fix dry run in OrphanFilesClean and optimize result string
8c4be22c7 is described below
commit 8c4be22c7617f6a7b57c8646a44b7fde0377277d
Author: Jingsong <ji...@gmail.com>
AuthorDate: Mon Jul 1 11:21:57 2024 +0800
[core] Fix dry run in OrphanFilesClean and optimize result string
---
.../apache/paimon/operation/OrphanFilesClean.java | 71 +++++++++++-----------
.../paimon/operation/OrphanFilesCleanTest.java | 9 ++-
.../flink/action/RemoveOrphanFilesAction.java | 13 ++--
.../action/RemoveOrphanFilesActionFactory.java | 4 +-
.../procedure/RemoveOrphanFilesProcedure.java | 9 +--
.../procedure/RemoveOrphanFilesProcedure.java | 16 ++---
6 files changed, 57 insertions(+), 65 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
index 06bebf155..bccfee93e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
@@ -21,7 +21,6 @@ package org.apache.paimon.operation;
import org.apache.paimon.Changelog;
import org.apache.paimon.FileStore;
import org.apache.paimon.Snapshot;
-import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
@@ -55,6 +54,7 @@ import java.util.Set;
import java.util.TimeZone;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -92,11 +92,9 @@ public class OrphanFilesClean {
private final ManifestFile manifestFile;
private final IndexFileHandler indexFileHandler;
- // an estimated value of how many files were deleted
- private int deletedFilesNum = 0;
private final List<Path> deleteFiles;
private long olderThanMillis = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(1);
- private boolean isDryRun = false;
+ private Consumer<Path> fileCleaner;
public OrphanFilesClean(FileStoreTable table) {
this.snapshotManager = table.snapshotManager();
@@ -110,6 +108,17 @@ public class OrphanFilesClean {
this.manifestFile = store.manifestFileFactory().create();
this.indexFileHandler = store.newIndexFileHandler();
this.deleteFiles = new ArrayList<>();
+ this.fileCleaner =
+ path -> {
+ try {
+ if (fileIO.isDir(path)) {
+ fileIO.deleteDirectoryQuietly(path);
+ } else {
+ fileIO.deleteQuietly(path);
+ }
+ } catch (IOException ignored) {
+ }
+ };
}
public OrphanFilesClean olderThan(String timestamp) {
@@ -120,8 +129,8 @@ public class OrphanFilesClean {
return this;
}
- public OrphanFilesClean dryRun(boolean dryRun) {
- this.isDryRun = dryRun;
+ public OrphanFilesClean fileCleaner(Consumer<Path> fileCleaner) {
+ this.fileCleaner = fileCleaner;
return this;
}
@@ -133,14 +142,12 @@ public class OrphanFilesClean {
// specially handle the snapshot directory
List<Path> nonSnapshotFiles = snapshotManager.tryGetNonSnapshotFiles(this::oldEnough);
- nonSnapshotFiles.forEach(this::deleteFileOrDirQuietly);
- deletedFilesNum += nonSnapshotFiles.size();
+ nonSnapshotFiles.forEach(fileCleaner);
deleteFiles.addAll(nonSnapshotFiles);
// specially handle the changelog directory
List<Path> nonChangelogFiles = snapshotManager.tryGetNonChangelogFiles(this::oldEnough);
- nonChangelogFiles.forEach(this::deleteFileOrDirQuietly);
- deletedFilesNum += nonChangelogFiles.size();
+ nonChangelogFiles.forEach(fileCleaner);
deleteFiles.addAll(nonChangelogFiles);
Map<String, Path> candidates = getCandidateDeletingFiles();
@@ -148,22 +155,9 @@ public class OrphanFilesClean {
Set<String> deleted = new HashSet<>(candidates.keySet());
deleted.removeAll(usedFiles);
+ deleted.stream().map(candidates::get).forEach(fileCleaner);
- if (!isDryRun) {
- for (String file : deleted) {
- Path path = candidates.get(file);
- deleteFileOrDirQuietly(path);
- }
- }
-
- deletedFilesNum += deleted.size();
deleteFiles.addAll(deleted.stream().map(candidates::get).collect(Collectors.toList()));
-
- return deleteFiles;
- }
-
- @VisibleForTesting
- List<Path> getDeleteFiles() {
return deleteFiles;
}
@@ -532,10 +526,9 @@ public class OrphanFilesClean {
.map(FileStatus::getPath)
.forEach(
p -> {
- deleteFileOrDirQuietly(p);
+ fileCleaner.accept(p);
synchronized (deleteFiles) {
deleteFiles.add(p);
- deletedFilesNum++;
}
});
}
@@ -543,21 +536,25 @@ public class OrphanFilesClean {
return filtered;
}
- private void deleteFileOrDirQuietly(Path path) {
- try {
- if (fileIO.isDir(path)) {
- fileIO.deleteDirectoryQuietly(path);
- } else {
- fileIO.deleteQuietly(path);
- }
- } catch (IOException ignored) {
- }
- }
-
/** A helper functional interface for method {@link #retryReadingFiles}. */
@FunctionalInterface
private interface ReaderWithIOException<T> {
T read() throws IOException;
}
+
+ public static List<String> showDeletedFiles(List<Path> deleteFiles, int showLimit) {
+ int showSize = Math.min(deleteFiles.size(), showLimit);
+ List<String> result = new ArrayList<>();
+ if (deleteFiles.size() > showSize) {
+ result.add(
+ String.format(
+ "Total %s files, only %s lines are displayed.",
+ deleteFiles.size(), showSize));
+ }
+ for (int i = 0; i < showSize; i++) {
+ result.add(deleteFiles.get(i).toUri().getPath());
+ }
+ return result;
+ }
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/OrphanFilesCleanTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/OrphanFilesCleanTest.java
index 0222af4fa..09291c43e 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/OrphanFilesCleanTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/OrphanFilesCleanTest.java
@@ -177,9 +177,9 @@ public class OrphanFilesCleanTest {
// second check
orphanFilesClean = new OrphanFilesClean(table);
setOlderThan(orphanFilesClean);
- orphanFilesClean.clean();
+ List<Path> deleted = orphanFilesClean.clean();
try {
- validate(orphanFilesClean.getDeleteFiles(), snapshotData, new HashMap<>());
+ validate(deleted, snapshotData, new HashMap<>());
} catch (Throwable t) {
String tableOptions = "Table options:\n" + table.options();
@@ -367,9 +367,8 @@ public class OrphanFilesCleanTest {
// second check
orphanFilesClean = new OrphanFilesClean(table);
setOlderThan(orphanFilesClean);
- orphanFilesClean.clean();
- List<Path> cleanFiles = orphanFilesClean.getDeleteFiles();
- validate(cleanFiles, snapshotData, changelogData);
+ List<Path> deleted = orphanFilesClean.clean();
+ validate(deleted, snapshotData, changelogData);
}
/** Manually make a FileNotFoundException to simulate snapshot expire while clean. */
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesAction.java
index 46ea8302c..035882603 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesAction.java
@@ -18,7 +18,6 @@
package org.apache.paimon.flink.action;
-import org.apache.paimon.fs.Path;
import org.apache.paimon.operation.OrphanFilesClean;
import org.apache.paimon.table.FileStoreTable;
@@ -27,7 +26,6 @@ import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Map;
-import java.util.stream.Collectors;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -56,18 +54,15 @@ public class RemoveOrphanFilesAction extends TableActionBase {
return this;
}
- public RemoveOrphanFilesAction dryRun(Boolean dryRun) {
- this.orphanFilesClean.dryRun(dryRun);
+ public RemoveOrphanFilesAction dryRun() {
+ this.orphanFilesClean.fileCleaner(path -> {});
return this;
}
@Override
public void run() throws Exception {
- List<Path> orphanFiles = orphanFilesClean.clean();
- String files =
- orphanFiles.stream()
- .map(filePath -> filePath.toUri().getPath())
- .collect(Collectors.joining(", "));
+ List<String> result = OrphanFilesClean.showDeletedFiles(orphanFilesClean.clean(), 200);
+ String files = String.join(", ", result);
LOG.info("orphan files: [{}]", files);
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionFactory.java
index 632bc90da..0a461d839 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionFactory.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveOrphanFilesActionFactory.java
@@ -49,8 +49,8 @@ public class RemoveOrphanFilesActionFactory implements ActionFactory {
action.olderThan(params.get(OLDER_THAN));
}
- if (params.has(DRY_RUN)) {
- action.dryRun(Boolean.parseBoolean(params.get(DRY_RUN)));
+ if (params.has(DRY_RUN) && Boolean.parseBoolean(params.get(DRY_RUN))) {
+ action.dryRun();
}
return Optional.of(action);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
index 08545a32a..26b637a35 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RemoveOrphanFilesProcedure.java
@@ -68,11 +68,12 @@ public class RemoveOrphanFilesProcedure extends ProcedureBase {
orphanFilesClean.olderThan(olderThan);
}
- orphanFilesClean.dryRun(dryRun);
+ if (dryRun) {
+ orphanFilesClean.fileCleaner(path -> {});
+ }
- return orphanFilesClean.clean().stream()
- .map(filePath -> filePath.toUri().getPath())
- .toArray(String[]::new);
+ return OrphanFilesClean.showDeletedFiles(orphanFilesClean.clean(), 200)
+ .toArray(new String[0]);
}
@Override
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanFilesProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanFilesProcedure.java
index 902d932f9..b88b2d169 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanFilesProcedure.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanFilesProcedure.java
@@ -18,7 +18,6 @@
package org.apache.paimon.spark.procedure;
-import org.apache.paimon.fs.Path;
import org.apache.paimon.operation.OrphanFilesClean;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.StringUtils;
@@ -88,15 +87,16 @@ public class RemoveOrphanFilesProcedure extends BaseProcedure {
if (!StringUtils.isBlank(olderThan)) {
orphanFilesClean.olderThan(olderThan);
}
- orphanFilesClean.dryRun(dryRun);
+ if (dryRun) {
+ orphanFilesClean.fileCleaner(path -> {});
+ }
try {
- List<Path> orphanFiles = orphanFilesClean.clean();
- InternalRow[] rows = new InternalRow[orphanFiles.size()];
+ List<String> result =
+ OrphanFilesClean.showDeletedFiles(orphanFilesClean.clean(), 200);
+ InternalRow[] rows = new InternalRow[result.size()];
int index = 0;
- for (Path filePath : orphanFiles) {
- rows[index] =
- newInternalRow(
- UTF8String.fromString(filePath.toUri().getPath()));
+ for (String line : result) {
+ rows[index] = newInternalRow(UTF8String.fromString(line));
index++;
}
return rows;