You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@polaris.apache.org by di...@apache.org on 2026/06/25 15:54:08 UTC
(polaris) branch main updated: Support bulk deletion in batch file cleanup task (#4850)
This is an automated email from the ASF dual-hosted git repository.
dimas-b pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/polaris.git
The following commit(s) were added to refs/heads/main by this push:
new 1e53274d48 Support bulk deletion in batch file cleanup task (#4850)
1e53274d48 is described below
commit 1e53274d48bbfa9c64bdef535476fa2e0305056d
Author: Nándor Kollár <na...@users.noreply.github.com>
AuthorDate: Thu Jun 25 17:54:02 2026 +0200
Support bulk deletion in batch file cleanup task (#4850)
Co-authored-by: Daniel Tu <da...@gmail.com>
---
.../service/task/BatchFileCleanupTaskHandler.java | 35 +++++++---
.../service/task/FileCleanupTaskHandler.java | 74 ++++++++++++++++++++--
.../service/task/TableCleanupTaskHandler.java | 4 +-
.../task/BatchFileCleanupTaskHandlerTest.java | 57 +++++++++++------
.../service/task/TableCleanupTaskHandlerTest.java | 17 +++--
5 files changed, 146 insertions(+), 41 deletions(-)
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java b/runtime/service/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java
index 67dcacb9d0..0b91194e7f 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java
@@ -77,17 +77,12 @@ public class BatchFileCleanupTaskHandler extends FileCleanupTaskHandler {
missingFiles.size());
}
- // Schedule the deletion for each file asynchronously
- List<CompletableFuture<Void>> deleteFutures =
- validFiles.stream()
- .map(file -> super.tryDelete(tableId, authorizedFileIO, null, file, null, 1))
- .toList();
+ CompletableFuture<Void> deleteFutures =
+ tryDelete(
+ tableId, authorizedFileIO, validFiles, cleanupTask.type().getValue(), true, null, 1);
try {
- // Wait for all delete operations to finish
- CompletableFuture<Void> allDeletes =
- CompletableFuture.allOf(deleteFutures.toArray(new CompletableFuture[0]));
- allDeletes.join();
+ deleteFutures.join();
} catch (Exception e) {
LOGGER.error("Exception detected during batch files deletion", e);
return false;
@@ -97,5 +92,25 @@ public class BatchFileCleanupTaskHandler extends FileCleanupTaskHandler {
}
}
- public record BatchFileCleanupTask(TableIdentifier tableId, List<String> batchFiles) {}
+ public enum BatchFileType {
+ TABLE_METADATA("table_metadata");
+
+ private final String value;
+
+ BatchFileType(String value) {
+ this.value = value;
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ @Override
+ public String toString() {
+ return value;
+ }
+ }
+
+ public record BatchFileCleanupTask(
+ TableIdentifier tableId, List<String> batchFiles, BatchFileType type) {}
}
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/task/FileCleanupTaskHandler.java b/runtime/service/src/main/java/org/apache/polaris/service/task/FileCleanupTaskHandler.java
index aa0b0d9f1e..b149508ed8 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/task/FileCleanupTaskHandler.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/task/FileCleanupTaskHandler.java
@@ -21,6 +21,7 @@ package org.apache.polaris.service.task;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.CatalogUtil;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.io.FileIO;
import org.apache.polaris.core.context.CallContext;
@@ -29,9 +30,12 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * {@link FileCleanupTaskHandler} responsible for cleaning up files in table tasks. Handles retries
- * for file deletions and skips files that are already missing. Subclasses must implement
- * task-specific logic.
+ * Abstract base class for handling file cleanup tasks within Apache Polaris.
+ *
+ * <p>This class is for performing asynchronous file deletions with retry logic.
+ *
+ * <p>Subclasses must implement {@link #canHandleTask(TaskEntity)} and {@link
+ * #handleTask(TaskEntity, CallContext)} to define task-specific handling logic.
*/
public abstract class FileCleanupTaskHandler implements TaskHandler {
@@ -53,6 +57,19 @@ public abstract class FileCleanupTaskHandler implements TaskHandler {
@Override
public abstract boolean handleTask(TaskEntity task, CallContext callContext);
+ /**
+ * Attempts to delete a single file with retry logic. If the file does not exist, it logs a
+ * message and does not retry. If an error occurs, it retries up to {@link #MAX_ATTEMPTS} times
+ * before failing.
+ *
+ * @param tableId The identifier of the table associated with the file.
+ * @param fileIO The {@link FileIO} instance used for file operations.
+ * @param baseFile An optional base file associated with the file being deleted (can be null).
+ * @param file The path of the file to be deleted.
+ * @param e The exception from the previous attempt, if any.
+ * @param attempt The current retry attempt count.
+ * @return A {@link CompletableFuture} representing the asynchronous deletion operation.
+ */
public CompletableFuture<Void> tryDelete(
TableIdentifier tableId,
FileIO fileIO,
@@ -97,10 +114,59 @@ public abstract class FileCleanupTaskHandler implements TaskHandler {
.addKeyValue("file", file)
.addKeyValue("tableIdentifier", tableId)
.addKeyValue("baseFile", baseFile != null ? baseFile : "")
- .log("Exception caught deleting data file", newEx);
+ .log("Exception caught deleting data file {}", newEx);
return tryDelete(tableId, fileIO, baseFile, file, newEx, attempt + 1);
},
CompletableFuture.delayedExecutor(
FILE_DELETION_RETRY_MILLIS, TimeUnit.MILLISECONDS, executorService));
}
+
+ /**
+ * Attempts to delete multiple files in a batch operation with retry logic. If an error occurs, it
+ * retries up to {@link #MAX_ATTEMPTS} times before failing.
+ *
+ * @param tableId The identifier of the table associated with the files.
+ * @param fileIO The {@link FileIO} instance used for file operations.
+ * @param files The list of file paths to be deleted.
+ * @param type The type of files being deleted (e.g., data files, metadata files).
+ * @param isConcurrent Whether the deletion should be performed concurrently.
+ * @param e The exception from the previous attempt, if any.
+ * @param attempt The current retry attempt count.
+ * @return A {@link CompletableFuture} representing the asynchronous batch deletion operation.
+ */
+ public CompletableFuture<Void> tryDelete(
+ TableIdentifier tableId,
+ FileIO fileIO,
+ Iterable<String> files,
+ String type,
+ Boolean isConcurrent,
+ Throwable e,
+ int attempt) {
+ if (e != null && attempt <= MAX_ATTEMPTS) {
+ LOGGER
+ .atWarn()
+ .addKeyValue("files", files)
+ .addKeyValue("attempt", attempt)
+ .addKeyValue("error", e.getMessage())
+ .addKeyValue("type", type)
+ .log("Error encountered attempting to delete files");
+ }
+ if (attempt > MAX_ATTEMPTS && e != null) {
+ return CompletableFuture.failedFuture(e);
+ }
+ return CompletableFuture.runAsync(
+ () -> CatalogUtil.deleteFiles(fileIO, files, type, isConcurrent), executorService)
+ .exceptionallyComposeAsync(
+ newEx -> {
+ LOGGER
+ .atWarn()
+ .addKeyValue("files", files)
+ .addKeyValue("tableIdentifier", tableId)
+ .addKeyValue("type", type)
+ .log("Exception caught deleting data files {}", newEx);
+ return tryDelete(tableId, fileIO, files, type, isConcurrent, newEx, attempt + 1);
+ },
+ CompletableFuture.delayedExecutor(
+ FILE_DELETION_RETRY_MILLIS, TimeUnit.MILLISECONDS, executorService));
+ }
}
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/task/TableCleanupTaskHandler.java b/runtime/service/src/main/java/org/apache/polaris/service/task/TableCleanupTaskHandler.java
index 81ae93e32e..d1bb1fb80d 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/task/TableCleanupTaskHandler.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/task/TableCleanupTaskHandler.java
@@ -232,7 +232,9 @@ public class TableCleanupTaskHandler implements TaskHandler {
.withTaskType(AsyncTaskType.BATCH_FILE_CLEANUP)
.withData(
new BatchFileCleanupTaskHandler.BatchFileCleanupTask(
- tableEntity.getTableIdentifier(), metadataBatch))
+ tableEntity.getTableIdentifier(),
+ metadataBatch,
+ BatchFileCleanupTaskHandler.BatchFileType.TABLE_METADATA))
.setInternalProperties(cleanupTask.getInternalPropertiesAsMap())
.build();
});
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandlerTest.java b/runtime/service/src/test/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandlerTest.java
index 01479363d7..23ea0aaea5 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandlerTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandlerTest.java
@@ -27,9 +27,7 @@ import io.quarkus.test.junit.QuarkusMock;
import io.quarkus.test.junit.QuarkusTest;
import jakarta.inject.Inject;
import java.io.IOException;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
@@ -53,6 +51,8 @@ import org.apache.polaris.core.entity.TaskEntity;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mockito;
@QuarkusTest
@@ -174,7 +174,10 @@ public class BatchFileCleanupTaskHandlerTest {
new TaskEntity.Builder()
.withTaskType(AsyncTaskType.BATCH_FILE_CLEANUP)
.withData(
- new BatchFileCleanupTaskHandler.BatchFileCleanupTask(tableIdentifier, cleanupFiles))
+ new BatchFileCleanupTaskHandler.BatchFileCleanupTask(
+ tableIdentifier,
+ cleanupFiles,
+ BatchFileCleanupTaskHandler.BatchFileType.TABLE_METADATA))
.setName(UUID.randomUUID().toString())
.build();
@@ -215,7 +218,9 @@ public class BatchFileCleanupTaskHandlerTest {
.withTaskType(AsyncTaskType.BATCH_FILE_CLEANUP)
.withData(
new BatchFileCleanupTaskHandler.BatchFileCleanupTask(
- tableIdentifier, List.of(statisticsFile.path())))
+ tableIdentifier,
+ List.of(statisticsFile.path()),
+ BatchFileCleanupTaskHandler.BatchFileType.TABLE_METADATA))
.setName(UUID.randomUUID().toString())
.build();
@@ -224,29 +229,38 @@ public class BatchFileCleanupTaskHandlerTest {
assertThat(handler.handleTask(task, polarisCallContext)).isTrue();
}
- @Test
- public void testCleanupWithRetries() throws IOException {
- Map<String, AtomicInteger> retryCounter = new HashMap<>();
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2, 3})
+ public void testCleanupWithRetries(int maxRetries) throws IOException {
+ AtomicInteger batchRetryCounter = new AtomicInteger(0);
FileIO fileIO =
new InMemoryFileIO() {
@Override
public void close() {
// no-op
}
-
+ };
+ TableIdentifier tableIdentifier = TableIdentifier.of(Namespace.of("db1", "schema1"), "table1");
+ Mockito.when(taskFileIOSupplier.apply(Mockito.any(), Mockito.any())).thenReturn(fileIO);
+ BatchFileCleanupTaskHandler handler =
+ new BatchFileCleanupTaskHandler(taskFileIOSupplier, executor) {
@Override
- public void deleteFile(String location) {
- int attempts =
- retryCounter.computeIfAbsent(location, k -> new AtomicInteger(0)).incrementAndGet();
- if (attempts < 3) {
- throw new RuntimeException("Simulating failure to test retries");
+ public CompletableFuture<Void> tryDelete(
+ TableIdentifier tableId,
+ FileIO fileIO,
+ Iterable<String> files,
+ String type,
+ Boolean isConcurrent,
+ Throwable e,
+ int attempt) {
+ if (attempt <= maxRetries) {
+ batchRetryCounter.incrementAndGet();
+ return tryDelete(tableId, fileIO, files, type, isConcurrent, e, attempt + 1);
} else {
- super.deleteFile(location);
+ return super.tryDelete(tableId, fileIO, files, type, isConcurrent, e, attempt);
}
}
};
- TableIdentifier tableIdentifier = TableIdentifier.of(Namespace.of("db1", "schema1"), "table1");
- BatchFileCleanupTaskHandler handler = newBatchFileCleanupTaskHandler(fileIO);
long snapshotId = 100L;
ManifestFile manifestFile =
TaskTestUtils.manifestFile(
@@ -268,7 +282,9 @@ public class BatchFileCleanupTaskHandlerTest {
.withTaskType(AsyncTaskType.BATCH_FILE_CLEANUP)
.withData(
new BatchFileCleanupTaskHandler.BatchFileCleanupTask(
- tableIdentifier, List.of(statisticsFile.path())))
+ tableIdentifier,
+ List.of(statisticsFile.path()),
+ BatchFileCleanupTaskHandler.BatchFileType.TABLE_METADATA))
.setName(UUID.randomUUID().toString())
.build();
@@ -284,11 +300,10 @@ public class BatchFileCleanupTaskHandlerTest {
// Wait for all async tasks to finish
future.join();
+ // Ensure that retries happened as expected
+ assertThat(batchRetryCounter.get()).isEqualTo(maxRetries);
+
// Check if the file was successfully deleted after retries
assertThat(TaskUtils.exists(statisticsFile.path(), fileIO)).isFalse();
-
- // Ensure that retries happened as expected
- assertThat(retryCounter.containsKey(statisticsFile.path())).isTrue();
- assertThat(retryCounter.get(statisticsFile.path()).get()).isEqualTo(3);
}
}
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/task/TableCleanupTaskHandlerTest.java b/runtime/service/src/test/java/org/apache/polaris/service/task/TableCleanupTaskHandlerTest.java
index d48c4432fb..1d3cc913c2 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/task/TableCleanupTaskHandlerTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/task/TableCleanupTaskHandlerTest.java
@@ -138,7 +138,8 @@ class TableCleanupTaskHandlerTest {
.returns(
new BatchFileCleanupTaskHandler.BatchFileCleanupTask(
tableIdentifier,
- List.of(snapshot.manifestListLocation(), statisticsFile.path())),
+ List.of(snapshot.manifestListLocation(), statisticsFile.path()),
+ BatchFileCleanupTaskHandler.BatchFileType.TABLE_METADATA),
entity ->
entity.readData(
BatchFileCleanupTaskHandler.BatchFileCleanupTask.class)));
@@ -257,7 +258,9 @@ class TableCleanupTaskHandlerTest {
.returns(AsyncTaskType.BATCH_FILE_CLEANUP, TaskEntity::getTaskType)
.returns(
new BatchFileCleanupTaskHandler.BatchFileCleanupTask(
- tableIdentifier, List.of(snapshot.manifestListLocation())),
+ tableIdentifier,
+ List.of(snapshot.manifestListLocation()),
+ BatchFileCleanupTaskHandler.BatchFileType.TABLE_METADATA),
entity ->
entity.readData(
BatchFileCleanupTaskHandler.BatchFileCleanupTask.class)),
@@ -268,7 +271,9 @@ class TableCleanupTaskHandlerTest {
.returns(AsyncTaskType.BATCH_FILE_CLEANUP, TaskEntity::getTaskType)
.returns(
new BatchFileCleanupTaskHandler.BatchFileCleanupTask(
- tableIdentifier, List.of(snapshot.manifestListLocation())),
+ tableIdentifier,
+ List.of(snapshot.manifestListLocation()),
+ BatchFileCleanupTaskHandler.BatchFileType.TABLE_METADATA),
entity ->
entity.readData(
BatchFileCleanupTaskHandler.BatchFileCleanupTask.class)),
@@ -392,7 +397,8 @@ class TableCleanupTaskHandlerTest {
snapshot.manifestListLocation(),
snapshot2.manifestListLocation(),
statisticsFile1.path(),
- statisticsFile2.path())),
+ statisticsFile2.path()),
+ BatchFileCleanupTaskHandler.BatchFileType.TABLE_METADATA),
entity ->
entity.readData(
BatchFileCleanupTaskHandler.BatchFileCleanupTask.class)));
@@ -562,7 +568,8 @@ class TableCleanupTaskHandlerTest {
statisticsFile1.path(),
statisticsFile2.path(),
partitionStatisticsFile1.path(),
- partitionStatisticsFile2.path())),
+ partitionStatisticsFile2.path()),
+ BatchFileCleanupTaskHandler.BatchFileType.TABLE_METADATA),
entity ->
entity.readData(
BatchFileCleanupTaskHandler.BatchFileCleanupTask.class)));