Skip to content

[core] Delete the output of a discarded compaction instead of leaking it #9345

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Stephen0421 wants to merge 1 commit into from
Closed

[core] Delete the output of a discarded compaction instead of leaking it #9345

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 37 additions & 23 deletions paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,8 +46,6 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;

import static java.util.Collections.emptyList;

/** Compact manager for {@link AppendOnlyFileStore}. */
public class BucketedAppendCompactManager extends CompactFutureManager {

Expand Down Expand Up @@ -115,15 +113,15 @@ private void triggerFullCompaction() {
LOG.debug("Submit full compaction with these files {}", toCompact);
}

taskFuture =
executor.submit(
new FullCompactTask(
dvMaintainer,
toCompact,
compactionFileSize,
forceRewriteAllFiles,
rewriter,
metricsReporter));
submitTask(
executor,
new FullCompactTask(
dvMaintainer,
toCompact,
compactionFileSize,
forceRewriteAllFiles,
rewriter,
metricsReporter));
recordCompactionsQueuedRequest();
compacting = new ArrayList<>(toCompact);
toCompact.clear();
Expand All @@ -147,10 +145,9 @@ private void triggerCompactionWithBestEffort() {
LOG.debug("Submit normal compaction with these files {}", compacting);
}

taskFuture =
executor.submit(
new AutoCompactTask(
dvMaintainer, compacting, rewriter, metricsReporter));
submitTask(
executor,
new AutoCompactTask(dvMaintainer, compacting, rewriter, metricsReporter));
recordCompactionsQueuedRequest();
}
}
Expand Down Expand Up @@ -281,7 +278,7 @@ protected CompactResult doCompact() throws Exception {
// do compaction
if (dvMaintainer != null) {
// if deletion vector enables, always trigger compaction.
return compact(dvMaintainer, toCompact, rewriter);
return compact(dvMaintainer, toCompact, rewriter, produced());
} else {
// compute small files
int big = 0;
Expand All @@ -295,13 +292,18 @@ protected CompactResult doCompact() throws Exception {
}
if (forceRewriteAllFiles
|| (small > big && toCompact.size() >= FULL_COMPACT_MIN_FILE)) {
return compact(null, toCompact, rewriter);
return compact(null, toCompact, rewriter, produced());
} else {
return result(emptyList(), emptyList());
return produced();
}
}
}

@Override
protected void deleteProduced(List<DataFileMeta> files) {
rewriter.delete(files);
}

private boolean hasDeletionFile(DataFileMeta file) {
return dvMaintainer != null
&& dvMaintainer.deletionVectorOf(file.fileName()).isPresent();
Expand Down Expand Up @@ -334,22 +336,28 @@ public AutoCompactTask(

@Override
protected CompactResult doCompact() throws Exception {
return compact(dvMaintainer, toCompact, rewriter);
return compact(dvMaintainer, toCompact, rewriter, produced());
}

@Override
protected void deleteProduced(List<DataFileMeta> files) {
rewriter.delete(files);
}
}

private static CompactResult compact(
@Nullable BucketedDvMaintainer dvMaintainer,
List<DataFileMeta> toCompact,
CompactRewriter rewriter)
CompactRewriter rewriter,
CompactResult toUpdate)
throws Exception {
List<DataFileMeta> rewrite = rewriter.rewrite(toCompact);
CompactResult result = result(toCompact, rewrite);
toUpdate.merge(result(toCompact, rewrite));
if (dvMaintainer != null) {
toCompact.forEach(f -> dvMaintainer.removeDeletionVectorOf(f.fileName()));
result.setDeletionFile(CompactDeletionFile.generateFiles(dvMaintainer));
toUpdate.setDeletionFile(CompactDeletionFile.generateFiles(dvMaintainer));
}
return result;
return toUpdate;
}

private static CompactResult result(List<DataFileMeta> before, List<DataFileMeta> after) {
Expand All @@ -359,5 +367,11 @@ private static CompactResult result(List<DataFileMeta> before, List<DataFileMeta
/** Compact rewriter for append-only table. */
public interface CompactRewriter {
List<DataFileMeta> rewrite(List<DataFileMeta> compactBefore) throws Exception;

/**
* Delete files produced by this rewriter, used when a compaction result is discarded and
* its files can never be committed.
*/
void delete(List<DataFileMeta> files);
}
}
17 changes: 15 additions & 2 deletions paimon-core/src/main/java/org/apache/paimon/append/cluster/BucketedAppendClusterManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,7 +151,7 @@ private void submitCompaction(CompactUnit unit) {
file.fileName(), file.level(), file.fileSize()))
.collect(Collectors.joining(", ")));
}
taskFuture = executor.submit(task);
submitTask(executor, task);
}

@Override
Expand Down Expand Up @@ -202,7 +202,14 @@ public BucketedAppendClusterTask(
@Override
protected CompactResult doCompact() throws Exception {
List<DataFileMeta> rewrite = rewriter.rewrite(toCluster);
return new CompactResult(toCluster, upgrade(rewrite));
CompactResult result = produced();
result.merge(new CompactResult(toCluster, upgrade(rewrite)));
return result;
}

@Override
protected void deleteProduced(List<DataFileMeta> files) {
rewriter.delete(files);
}

protected List<DataFileMeta> upgrade(List<DataFileMeta> files) {
Expand All @@ -215,5 +222,11 @@ protected List<DataFileMeta> upgrade(List<DataFileMeta> files) {
/** Compact rewriter for append-only table. */
public interface CompactRewriter {
List<DataFileMeta> rewrite(List<DataFileMeta> compactBefore) throws Exception;

/**
* Delete files produced by this rewriter, used when a cluster result is discarded and its
* files can never be committed.
*/
void delete(List<DataFileMeta> files);
}
}
33 changes: 27 additions & 6 deletions paimon-core/src/main/java/org/apache/paimon/compact/CompactFutureManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,20 +20,35 @@

import org.apache.paimon.annotation.VisibleForTesting;

import javax.annotation.Nullable;

import java.util.Optional;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

/** Base implementation of {@link CompactManager} which runs compaction in a separate thread. */
public abstract class CompactFutureManager implements CompactManager {

protected Future<CompactResult> taskFuture;

@Nullable private CompactTask task;

protected void submitTask(ExecutorService executor, CompactTask task) {
this.task = task;
this.taskFuture = executor.submit(task);
}

@Override
public void cancelCompaction() {
// TODO this method may leave behind orphan files if compaction is actually finished
// but some CPU work still needs to be done
if (task != null) {
// Tell the task that its output is not needed anymore before interrupting it, so that
// it deletes the files it produced no matter whether it observes the interruption.
// See CompactTask#cancel for the invariant that this must not be followed by
// prepareCommit on the same writer/maintainer.
task.cancel();
}
if (taskFuture != null && !taskFuture.isCancelled()) {
taskFuture.cancel(true);
}
Expand All @@ -48,15 +63,21 @@ protected final Optional<CompactResult> innerGetCompactionResult(boolean blockin
throws ExecutionException, InterruptedException {
if (taskFuture != null) {
if (blocking || taskFuture.isDone()) {
CompactResult result;
CompactTask currentTask = task;
try {
result = obtainCompactResult();
return Optional.of(obtainCompactResult());
} catch (CancellationException e) {
return Optional.empty();
// Cancellation may have won the race against the completion of the task, in
// which case the future has dropped a result whose files are already on disk.
// Report it so that the caller can account for them. If the task instead
// observed the cancellation, it has deleted its own output and there is
// nothing to report here.
return Optional.ofNullable(
currentTask == null ? null : currentTask.completedResult());
} finally {
taskFuture = null;
task = null;
}
return Optional.of(result);
}
}
return Optional.empty();
Expand Down
Loading
Loading