You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@polaris.apache.org by "danielhumanmod (via GitHub)" <gi...@apache.org> on 2024/12/10 04:05:53 UTC
[PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
danielhumanmod opened a new pull request, #516:
URL: https://github.com/apache/polaris/pull/516
# Description
Please include a summary of the changes and the related issue. Please also include relevant motivation and context. List any dependencies that are required for this change.
Fixes #515
## Type of change
Please delete options that are not relevant.
- [X] Bug fix (non-breaking change which fixes an issue)
- [ ] Documentation update
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update
# How Has This Been Tested?
Please describe the tests that you ran to verify your changes. Provide instructions so we can reproduce. Please also list any relevant details for your test configuration
- [X] TableCleanupTaskHandlerTest.java
- [X] BatchFileCleanupTaskHandlerTest.java
- [X] ManifestFileCleanupTaskHandlerTest.java
**Test Configuration**:
* Hardware:
* Toolchain:
* SDK:
# Checklist:
Please delete options that are not relevant.
- [X] I have performed a self-review of my code
- [X] I have commented my code, particularly in hard-to-understand areas
- [X] I have made corresponding changes to the documentation
- [X] My changes generate no new warnings
- [X] If adding new functionality, I have discussed my implementation with the community using the linked GitHub issue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #516:
URL: https://github.com/apache/polaris/pull/516#discussion_r1879686170
##########
polaris-service/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.task;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.entity.AsyncTaskType;
+import org.apache.polaris.core.entity.TaskEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link BatchFileCleanupTaskHandler} responsible for batch file cleanup by processing multiple
+ * file deletions in a single task handler. Valid files are deleted asynchronously with retries for
+ * transient errors, while missing files are logged and skipped.
+ */
+public class BatchFileCleanupTaskHandler extends FileCleanupTaskHandler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(BatchFileCleanupTaskHandler.class);
+
+ public BatchFileCleanupTaskHandler(
+ Function<TaskEntity, FileIO> fileIOSupplier, ExecutorService executorService) {
+ super(fileIOSupplier, executorService);
+ }
+
+ @Override
+ public boolean canHandleTask(TaskEntity task) {
+ return task.getTaskType() == AsyncTaskType.BATCH_FILE_CLEANUP;
+ }
+
+ @Override
+ public boolean handleTask(TaskEntity task) {
+ BatchFileCleanupTask cleanupTask = task.readData(BatchFileCleanupTask.class);
+ TableIdentifier tableId = cleanupTask.getTableId();
+ List<String> batchFiles = cleanupTask.getBatchFiles();
+ try (FileIO authorizedFileIO = fileIOSupplier.apply(task)) {
+ List<String> validFiles =
+ batchFiles.stream().filter(file -> TaskUtils.exists(file, authorizedFileIO)).toList();
+ if (validFiles.isEmpty()) {
+ LOGGER
+ .atWarn()
+ .addKeyValue("batchFiles", batchFiles.toString())
+ .addKeyValue("tableId", tableId)
+ .log("File batch cleanup task scheduled, but the none of the file in batch exists");
+ return true;
+ }
+ if (validFiles.size() < batchFiles.size()) {
+ List<String> missingFiles =
+ batchFiles.stream().filter(file -> !TaskUtils.exists(file, authorizedFileIO)).toList();
+ LOGGER
+ .atWarn()
+ .addKeyValue("batchFiles", batchFiles.toString())
+ .addKeyValue("missingFiles", missingFiles.toString())
+ .addKeyValue("tableId", tableId)
+ .log(
+ "File batch cleanup task scheduled, but {} files in the batch are missing",
+ 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();
+
+ try {
+ // Wait for all delete operations to finish
+ CompletableFuture<Void> allDeletes =
+ CompletableFuture.allOf(deleteFutures.toArray(new CompletableFuture[0]));
+ allDeletes.join();
+ } catch (Exception e) {
+ LOGGER.error("Exception detected during batch files deletion", e);
+ return false;
Review Comment:
Imho this a bit too coarse-grained. The likelihood of having one failed delete will grow bigger with the number of files to delete. Shouldn't we return a more meaningful result, e.g. a statistics object reporting how many files were deleted, and how many failed? Not saying you should change this now though, because returning a boolean is imposed by `TaskHandler` API, so we need to look into improving the API itself.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #516:
URL: https://github.com/apache/polaris/pull/516#issuecomment-2586047340
This PR is stale because it has been open 30 days with no activity. Remove stale label or comment or this will be closed in 5 days.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "eric-maynard (via GitHub)" <gi...@apache.org>.
eric-maynard commented on code in PR #516:
URL: https://github.com/apache/polaris/pull/516#discussion_r1984470482
##########
service/common/src/main/java/org/apache/polaris/service/task/ManifestFileCleanupTaskHandler.java:
##########
@@ -42,50 +40,32 @@
import org.slf4j.LoggerFactory;
/**
- * {@link TaskHandler} responsible for deleting table files: 1. Manifest files: It contains all the
- * files in a manifest and the manifest itself. Since data files may be present in multiple
- * manifests across different snapshots, we assume a data file that doesn't exist is missing because
- * it was already deleted by another task. 2. Table metadata files: It contains previous metadata
- * and statistics files, which are grouped and deleted in batch
+ * {@link ManifestFileCleanupTaskHandler} responsible for deleting all the files in a manifest and
+ * the manifest itself. Since data files may be present in multiple manifests across different
+ * snapshots, we assume a data file that doesn't exist is missing because it was already deleted by
+ * another task.
*/
-// TODO: Rename this class since we introducing metadata cleanup here
-public class ManifestFileCleanupTaskHandler implements TaskHandler {
- public static final int MAX_ATTEMPTS = 3;
- public static final int FILE_DELETION_RETRY_MILLIS = 100;
+public class ManifestFileCleanupTaskHandler extends FileCleanupTaskHandler {
private static final Logger LOGGER =
LoggerFactory.getLogger(ManifestFileCleanupTaskHandler.class);
- private final TaskFileIOSupplier fileIOSupplier;
- private final ExecutorService executorService;
public ManifestFileCleanupTaskHandler(
TaskFileIOSupplier fileIOSupplier, ExecutorService executorService) {
- this.fileIOSupplier = fileIOSupplier;
- this.executorService = executorService;
+ super(fileIOSupplier, executorService);
}
@Override
public boolean canHandleTask(TaskEntity task) {
- return task.getTaskType() == AsyncTaskType.MANIFEST_FILE_CLEANUP
- || task.getTaskType() == AsyncTaskType.METADATA_FILE_BATCH_CLEANUP;
+ return task.getTaskType() == AsyncTaskType.MANIFEST_FILE_CLEANUP;
}
@Override
public boolean handleTask(TaskEntity task, CallContext callContext) {
ManifestCleanupTask cleanupTask = task.readData(ManifestCleanupTask.class);
- TableIdentifier tableId = cleanupTask.getTableId();
+ TableIdentifier tableId = cleanupTask.tableId();
Review Comment:
Just for my understanding, why did this change -- or why did we not just use `tableId` before?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "danielhumanmod (via GitHub)" <gi...@apache.org>.
danielhumanmod commented on code in PR #516:
URL: https://github.com/apache/polaris/pull/516#discussion_r1883188384
##########
polaris-service/src/main/java/org/apache/polaris/service/task/FileCleanupTaskHandler.java:
##########
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.task;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.entity.TaskEntity;
+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.
+ */
+public abstract class FileCleanupTaskHandler implements TaskHandler {
+
+ public static final int MAX_ATTEMPTS = 3;
+ public static final int FILE_DELETION_RETRY_MILLIS = 100;
+ public final Function<TaskEntity, FileIO> fileIOSupplier;
+ public final ExecutorService executorService;
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ManifestFileCleanupTaskHandler.class);
+
+ public FileCleanupTaskHandler(
+ Function<TaskEntity, FileIO> fileIOSupplier, ExecutorService executorService) {
+ this.fileIOSupplier = fileIOSupplier;
+ this.executorService = executorService;
+ }
+
+ @Override
+ public abstract boolean canHandleTask(TaskEntity task);
+
+ @Override
+ public abstract boolean handleTask(TaskEntity task);
+
+ public abstract Logger getLogger();
Review Comment:
> This is imho not necessary, using this class' `LOGGER` is enough.
Thanks for reminding, already fixed it!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on PR #516:
URL: https://github.com/apache/polaris/pull/516#issuecomment-2586211285
Do we still need this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "danielhumanmod (via GitHub)" <gi...@apache.org>.
danielhumanmod commented on PR #516:
URL: https://github.com/apache/polaris/pull/516#issuecomment-2586336134
> Do we still need this PR?
Apologies for the delay on my end. I’m currently on vacation with my family and will be back soon to continue this work. I believe there are still some comments worth considering in order to refine the code. Thanks for the patience here!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "danielhumanmod (via GitHub)" <gi...@apache.org>.
danielhumanmod commented on PR #516:
URL: https://github.com/apache/polaris/pull/516#issuecomment-2676480300
> Hi @eric-maynard, I’ve resolved the conflicts and refined the code based on the current settings after #1000 (which revert 589) . Would you mind taking a look to make sure I didn’t miss anything? Thanks!
The changes mainly in: https://github.com/apache/polaris/pull/516/commits/9270568172ef2da16df5f16347052b24726f9a82
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "danielhumanmod (via GitHub)" <gi...@apache.org>.
danielhumanmod commented on PR #516:
URL: https://github.com/apache/polaris/pull/516#issuecomment-2634688111
> @danielhumanmod , are you still working on it?
Hi @flyrain thanks for checking. Yes I just response to @snazy 's comment last week (Last pending comment of this PR) and is waiting for his feedback.
His suggestion about leveraging batch processing ability of object storage is great, but the main problem is our current deletion operation rely on Iceberg's FileIO interface, which does not support batch operation for now. Appreciate if the community could give some feedback on this as well!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #516:
URL: https://github.com/apache/polaris/pull/516#discussion_r1879682569
##########
polaris-service/src/main/java/org/apache/polaris/service/task/FileCleanupTaskHandler.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.task;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.entity.TaskEntity;
+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.
+ */
+public class FileCleanupTaskHandler implements TaskHandler {
+
+ public static final int MAX_ATTEMPTS = 3;
+ public static final int FILE_DELETION_RETRY_MILLIS = 100;
+ public final Function<TaskEntity, FileIO> fileIOSupplier;
+ public final ExecutorService executorService;
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ManifestFileCleanupTaskHandler.class);
+
+ public FileCleanupTaskHandler(
+ Function<TaskEntity, FileIO> fileIOSupplier, ExecutorService executorService) {
+ this.fileIOSupplier = fileIOSupplier;
+ this.executorService = executorService;
+ }
+
+ @Override
+ public boolean canHandleTask(TaskEntity task) {
+ throw new UnsupportedOperationException("This method must be implemented by subclasses.");
Review Comment:
Why not use an abstract class then?
##########
polaris-service/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.task;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.entity.AsyncTaskType;
+import org.apache.polaris.core.entity.TaskEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link BatchFileCleanupTaskHandler} responsible for batch file cleanup by processing multiple
+ * file deletions in a single task handler. Valid files are deleted asynchronously with retries for
+ * transient errors, while missing files are logged and skipped.
+ */
+public class BatchFileCleanupTaskHandler extends FileCleanupTaskHandler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(BatchFileCleanupTaskHandler.class);
+
+ public BatchFileCleanupTaskHandler(
+ Function<TaskEntity, FileIO> fileIOSupplier, ExecutorService executorService) {
+ super(fileIOSupplier, executorService);
+ }
+
+ @Override
+ public boolean canHandleTask(TaskEntity task) {
+ return task.getTaskType() == AsyncTaskType.BATCH_FILE_CLEANUP;
+ }
+
+ @Override
+ public boolean handleTask(TaskEntity task) {
+ BatchFileCleanupTask cleanupTask = task.readData(BatchFileCleanupTask.class);
+ TableIdentifier tableId = cleanupTask.getTableId();
+ List<String> batchFiles = cleanupTask.getBatchFiles();
+ try (FileIO authorizedFileIO = fileIOSupplier.apply(task)) {
+ List<String> validFiles =
+ batchFiles.stream().filter(file -> TaskUtils.exists(file, authorizedFileIO)).toList();
+ if (validFiles.isEmpty()) {
+ LOGGER
+ .atWarn()
+ .addKeyValue("batchFiles", batchFiles.toString())
+ .addKeyValue("tableId", tableId)
+ .log("File batch cleanup task scheduled, but the none of the file in batch exists");
Review Comment:
```suggestion
.log("File batch cleanup task scheduled, but none of the files in batch exists");
```
##########
polaris-service/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.task;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.entity.AsyncTaskType;
+import org.apache.polaris.core.entity.TaskEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link BatchFileCleanupTaskHandler} responsible for batch file cleanup by processing multiple
+ * file deletions in a single task handler. Valid files are deleted asynchronously with retries for
+ * transient errors, while missing files are logged and skipped.
+ */
+public class BatchFileCleanupTaskHandler extends FileCleanupTaskHandler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(BatchFileCleanupTaskHandler.class);
+
+ public BatchFileCleanupTaskHandler(
+ Function<TaskEntity, FileIO> fileIOSupplier, ExecutorService executorService) {
+ super(fileIOSupplier, executorService);
+ }
+
+ @Override
+ public boolean canHandleTask(TaskEntity task) {
+ return task.getTaskType() == AsyncTaskType.BATCH_FILE_CLEANUP;
+ }
+
+ @Override
+ public boolean handleTask(TaskEntity task) {
+ BatchFileCleanupTask cleanupTask = task.readData(BatchFileCleanupTask.class);
+ TableIdentifier tableId = cleanupTask.getTableId();
+ List<String> batchFiles = cleanupTask.getBatchFiles();
+ try (FileIO authorizedFileIO = fileIOSupplier.apply(task)) {
+ List<String> validFiles =
+ batchFiles.stream().filter(file -> TaskUtils.exists(file, authorizedFileIO)).toList();
+ if (validFiles.isEmpty()) {
+ LOGGER
+ .atWarn()
+ .addKeyValue("batchFiles", batchFiles.toString())
+ .addKeyValue("tableId", tableId)
+ .log("File batch cleanup task scheduled, but the none of the file in batch exists");
+ return true;
+ }
+ if (validFiles.size() < batchFiles.size()) {
+ List<String> missingFiles =
+ batchFiles.stream().filter(file -> !TaskUtils.exists(file, authorizedFileIO)).toList();
+ LOGGER
+ .atWarn()
+ .addKeyValue("batchFiles", batchFiles.toString())
+ .addKeyValue("missingFiles", missingFiles.toString())
+ .addKeyValue("tableId", tableId)
+ .log(
+ "File batch cleanup task scheduled, but {} files in the batch are missing",
+ 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();
+
+ try {
+ // Wait for all delete operations to finish
+ CompletableFuture<Void> allDeletes =
+ CompletableFuture.allOf(deleteFutures.toArray(new CompletableFuture[0]));
+ allDeletes.join();
+ } catch (Exception e) {
+ LOGGER.error("Exception detected during batch files deletion", e);
+ return false;
Review Comment:
Imho this a bit too coarse-grained. The likelihood of having one delete file will grow bigger with the number of files to delete. Shouldn't we return a more meaningful result, e.g. a statistics object reporting how many files were deleted, and how many failed? Not saying you should change this now though, because returning a boolean is imposed by `TaskHandler` API, so we need to look into improving the API itself.
##########
polaris-service/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.task;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.entity.AsyncTaskType;
+import org.apache.polaris.core.entity.TaskEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link BatchFileCleanupTaskHandler} responsible for batch file cleanup by processing multiple
+ * file deletions in a single task handler. Valid files are deleted asynchronously with retries for
+ * transient errors, while missing files are logged and skipped.
+ */
+public class BatchFileCleanupTaskHandler extends FileCleanupTaskHandler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(BatchFileCleanupTaskHandler.class);
+
+ public BatchFileCleanupTaskHandler(
+ Function<TaskEntity, FileIO> fileIOSupplier, ExecutorService executorService) {
+ super(fileIOSupplier, executorService);
+ }
+
+ @Override
+ public boolean canHandleTask(TaskEntity task) {
+ return task.getTaskType() == AsyncTaskType.BATCH_FILE_CLEANUP;
+ }
+
+ @Override
+ public boolean handleTask(TaskEntity task) {
+ BatchFileCleanupTask cleanupTask = task.readData(BatchFileCleanupTask.class);
+ TableIdentifier tableId = cleanupTask.getTableId();
+ List<String> batchFiles = cleanupTask.getBatchFiles();
+ try (FileIO authorizedFileIO = fileIOSupplier.apply(task)) {
+ List<String> validFiles =
+ batchFiles.stream().filter(file -> TaskUtils.exists(file, authorizedFileIO)).toList();
+ if (validFiles.isEmpty()) {
+ LOGGER
+ .atWarn()
+ .addKeyValue("batchFiles", batchFiles.toString())
+ .addKeyValue("tableId", tableId)
+ .log("File batch cleanup task scheduled, but the none of the file in batch exists");
+ return true;
+ }
+ if (validFiles.size() < batchFiles.size()) {
+ List<String> missingFiles =
+ batchFiles.stream().filter(file -> !TaskUtils.exists(file, authorizedFileIO)).toList();
+ LOGGER
+ .atWarn()
+ .addKeyValue("batchFiles", batchFiles.toString())
+ .addKeyValue("missingFiles", missingFiles.toString())
+ .addKeyValue("tableId", tableId)
+ .log(
+ "File batch cleanup task scheduled, but {} files in the batch are missing",
+ 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();
+
+ try {
+ // Wait for all delete operations to finish
+ CompletableFuture<Void> allDeletes =
+ CompletableFuture.allOf(deleteFutures.toArray(new CompletableFuture[0]));
+ allDeletes.join();
+ } catch (Exception e) {
+ LOGGER.error("Exception detected during batch files deletion", e);
+ return false;
+ }
+
+ return true;
+ }
+ }
+
+ @Override
+ public Logger getLogger() {
+ return LOGGER;
+ }
+
+ public static final class BatchFileCleanupTask {
Review Comment:
Can this be a `record`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "snazy (via GitHub)" <gi...@apache.org>.
snazy commented on code in PR #516:
URL: https://github.com/apache/polaris/pull/516#discussion_r1883787603
##########
polaris-service/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.task;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.entity.AsyncTaskType;
+import org.apache.polaris.core.entity.TaskEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link BatchFileCleanupTaskHandler} responsible for batch file cleanup by processing multiple
+ * file deletions in a single task handler. Valid files are deleted asynchronously with retries for
+ * transient errors, while missing files are logged and skipped.
+ */
+public class BatchFileCleanupTaskHandler extends FileCleanupTaskHandler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(BatchFileCleanupTaskHandler.class);
+
+ public BatchFileCleanupTaskHandler(
+ Function<TaskEntity, FileIO> fileIOSupplier, ExecutorService executorService) {
+ super(fileIOSupplier, executorService);
+ }
+
+ @Override
+ public boolean canHandleTask(TaskEntity task) {
+ return task.getTaskType() == AsyncTaskType.BATCH_FILE_CLEANUP;
+ }
+
+ @Override
+ public boolean handleTask(TaskEntity task) {
+ BatchFileCleanupTask cleanupTask = task.readData(BatchFileCleanupTask.class);
+ TableIdentifier tableId = cleanupTask.tableId();
+ List<String> batchFiles = cleanupTask.batchFiles();
+ try (FileIO authorizedFileIO = fileIOSupplier.apply(task)) {
+ List<String> validFiles =
+ batchFiles.stream().filter(file -> TaskUtils.exists(file, authorizedFileIO)).toList();
+ if (validFiles.isEmpty()) {
+ LOGGER
+ .atWarn()
+ .addKeyValue("batchFiles", batchFiles.toString())
+ .addKeyValue("tableId", tableId)
+ .log("File batch cleanup task scheduled, but none of the files in batch exists");
+ return true;
+ }
+ if (validFiles.size() < batchFiles.size()) {
+ List<String> missingFiles =
+ batchFiles.stream().filter(file -> !TaskUtils.exists(file, authorizedFileIO)).toList();
+ LOGGER
+ .atWarn()
+ .addKeyValue("batchFiles", batchFiles.toString())
+ .addKeyValue("missingFiles", missingFiles.toString())
+ .addKeyValue("tableId", tableId)
+ .log(
+ "File batch cleanup task scheduled, but {} files in the batch are missing",
+ 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))
Review Comment:
Hm, I'd have expected that this class actually leverages batched deletes of the underlying object store, but seems it does not.
##########
polaris-service/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.task;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.entity.AsyncTaskType;
+import org.apache.polaris.core.entity.TaskEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link BatchFileCleanupTaskHandler} responsible for batch file cleanup by processing multiple
+ * file deletions in a single task handler. Valid files are deleted asynchronously with retries for
+ * transient errors, while missing files are logged and skipped.
+ */
+public class BatchFileCleanupTaskHandler extends FileCleanupTaskHandler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(BatchFileCleanupTaskHandler.class);
+
+ public BatchFileCleanupTaskHandler(
+ Function<TaskEntity, FileIO> fileIOSupplier, ExecutorService executorService) {
+ super(fileIOSupplier, executorService);
+ }
+
+ @Override
+ public boolean canHandleTask(TaskEntity task) {
+ return task.getTaskType() == AsyncTaskType.BATCH_FILE_CLEANUP;
+ }
+
+ @Override
+ public boolean handleTask(TaskEntity task) {
+ BatchFileCleanupTask cleanupTask = task.readData(BatchFileCleanupTask.class);
+ TableIdentifier tableId = cleanupTask.tableId();
+ List<String> batchFiles = cleanupTask.batchFiles();
+ try (FileIO authorizedFileIO = fileIOSupplier.apply(task)) {
+ List<String> validFiles =
+ batchFiles.stream().filter(file -> TaskUtils.exists(file, authorizedFileIO)).toList();
Review Comment:
Here and in line 69 are exists-checks performed for each individual file, which is inefficient (slow) and also incurs unnecessary cost for cloud object storages.
I'd prefer an approach that
a) leverages object-storage batch requests
b) does not perform the same operation more than once
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "danielhumanmod (via GitHub)" <gi...@apache.org>.
danielhumanmod commented on code in PR #516:
URL: https://github.com/apache/polaris/pull/516#discussion_r1929657255
##########
polaris-service/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.task;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.entity.AsyncTaskType;
+import org.apache.polaris.core.entity.TaskEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link BatchFileCleanupTaskHandler} responsible for batch file cleanup by processing multiple
+ * file deletions in a single task handler. Valid files are deleted asynchronously with retries for
+ * transient errors, while missing files are logged and skipped.
+ */
+public class BatchFileCleanupTaskHandler extends FileCleanupTaskHandler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(BatchFileCleanupTaskHandler.class);
+
+ public BatchFileCleanupTaskHandler(
+ Function<TaskEntity, FileIO> fileIOSupplier, ExecutorService executorService) {
+ super(fileIOSupplier, executorService);
+ }
+
+ @Override
+ public boolean canHandleTask(TaskEntity task) {
+ return task.getTaskType() == AsyncTaskType.BATCH_FILE_CLEANUP;
+ }
+
+ @Override
+ public boolean handleTask(TaskEntity task) {
+ BatchFileCleanupTask cleanupTask = task.readData(BatchFileCleanupTask.class);
+ TableIdentifier tableId = cleanupTask.tableId();
+ List<String> batchFiles = cleanupTask.batchFiles();
+ try (FileIO authorizedFileIO = fileIOSupplier.apply(task)) {
+ List<String> validFiles =
+ batchFiles.stream().filter(file -> TaskUtils.exists(file, authorizedFileIO)).toList();
Review Comment:
> Here and in line 69 are exists-checks performed for each individual file, which is inefficient (slow) and also incurs unnecessary cost for cloud object storages.
>
> I'd prefer an approach that a) leverages object-storage batch requests b) does not perform the same operation more than once
Thanks for the suggestion @snazy , which is a good point! However, our current deletion operation rely on Iceberg's [`FileIO`](https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/io/FileIO.java) interface, which does not support batch operation for now. Would you be okay with keeping this implementation for now, and refine it once `FileIO` add batch support? (We could also consider contributing this feature to Iceberg as well)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "danielhumanmod (via GitHub)" <gi...@apache.org>.
danielhumanmod commented on PR #516:
URL: https://github.com/apache/polaris/pull/516#issuecomment-2683896208
Hi @dennishuo , I’ve refined the code about call context in task handler after #1057 , Would you mind taking a look to make sure I didn’t miss anything? Thanks!
The changes mainly in https://github.com/apache/polaris/pull/516/commits/13c70ecaafbf6336e2dbd5d7d46e0c6b630dd52d
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "danielhumanmod (via GitHub)" <gi...@apache.org>.
danielhumanmod commented on PR #516:
URL: https://github.com/apache/polaris/pull/516#issuecomment-2676473121
Hi @eric-maynard, I’ve resolved the conflicts and refined the code based on the current settings after #1000 . Would you mind taking a look to make sure I didn’t miss anything? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "adutra (via GitHub)" <gi...@apache.org>.
adutra commented on code in PR #516:
URL: https://github.com/apache/polaris/pull/516#discussion_r1881905988
##########
polaris-service/src/main/java/org/apache/polaris/service/task/FileCleanupTaskHandler.java:
##########
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.task;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.entity.TaskEntity;
+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.
+ */
+public abstract class FileCleanupTaskHandler implements TaskHandler {
+
+ public static final int MAX_ATTEMPTS = 3;
+ public static final int FILE_DELETION_RETRY_MILLIS = 100;
+ public final Function<TaskEntity, FileIO> fileIOSupplier;
+ public final ExecutorService executorService;
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ManifestFileCleanupTaskHandler.class);
+
+ public FileCleanupTaskHandler(
+ Function<TaskEntity, FileIO> fileIOSupplier, ExecutorService executorService) {
+ this.fileIOSupplier = fileIOSupplier;
+ this.executorService = executorService;
+ }
+
+ @Override
+ public abstract boolean canHandleTask(TaskEntity task);
+
+ @Override
+ public abstract boolean handleTask(TaskEntity task);
+
+ public abstract Logger getLogger();
Review Comment:
This is imho not necessary, using this class' `LOGGER` is enough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "danielhumanmod (via GitHub)" <gi...@apache.org>.
danielhumanmod commented on PR #516:
URL: https://github.com/apache/polaris/pull/516#issuecomment-2537901502
Thank you for the valuable feedback, @adutra ! Your comments were super helpful in improving the code quality. I’ve incorporated your suggestions and updated the code in the latest commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "danielhumanmod (via GitHub)" <gi...@apache.org>.
danielhumanmod commented on code in PR #516:
URL: https://github.com/apache/polaris/pull/516#discussion_r1881449620
##########
polaris-service/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.task;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.entity.AsyncTaskType;
+import org.apache.polaris.core.entity.TaskEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link BatchFileCleanupTaskHandler} responsible for batch file cleanup by processing multiple
+ * file deletions in a single task handler. Valid files are deleted asynchronously with retries for
+ * transient errors, while missing files are logged and skipped.
+ */
+public class BatchFileCleanupTaskHandler extends FileCleanupTaskHandler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(BatchFileCleanupTaskHandler.class);
+
+ public BatchFileCleanupTaskHandler(
+ Function<TaskEntity, FileIO> fileIOSupplier, ExecutorService executorService) {
+ super(fileIOSupplier, executorService);
+ }
+
+ @Override
+ public boolean canHandleTask(TaskEntity task) {
+ return task.getTaskType() == AsyncTaskType.BATCH_FILE_CLEANUP;
+ }
+
+ @Override
+ public boolean handleTask(TaskEntity task) {
+ BatchFileCleanupTask cleanupTask = task.readData(BatchFileCleanupTask.class);
+ TableIdentifier tableId = cleanupTask.getTableId();
+ List<String> batchFiles = cleanupTask.getBatchFiles();
+ try (FileIO authorizedFileIO = fileIOSupplier.apply(task)) {
+ List<String> validFiles =
+ batchFiles.stream().filter(file -> TaskUtils.exists(file, authorizedFileIO)).toList();
+ if (validFiles.isEmpty()) {
+ LOGGER
+ .atWarn()
+ .addKeyValue("batchFiles", batchFiles.toString())
+ .addKeyValue("tableId", tableId)
+ .log("File batch cleanup task scheduled, but the none of the file in batch exists");
+ return true;
+ }
+ if (validFiles.size() < batchFiles.size()) {
+ List<String> missingFiles =
+ batchFiles.stream().filter(file -> !TaskUtils.exists(file, authorizedFileIO)).toList();
+ LOGGER
+ .atWarn()
+ .addKeyValue("batchFiles", batchFiles.toString())
+ .addKeyValue("missingFiles", missingFiles.toString())
+ .addKeyValue("tableId", tableId)
+ .log(
+ "File batch cleanup task scheduled, but {} files in the batch are missing",
+ 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();
+
+ try {
+ // Wait for all delete operations to finish
+ CompletableFuture<Void> allDeletes =
+ CompletableFuture.allOf(deleteFutures.toArray(new CompletableFuture[0]));
+ allDeletes.join();
+ } catch (Exception e) {
+ LOGGER.error("Exception detected during batch files deletion", e);
+ return false;
Review Comment:
That’s a great point! Since this is the first time we’re introducing a batch files handler, it would be helpful to discuss how to refine the TaskHandler API for future scenarios. @collado-mike, do you have any suggestions or thoughts on this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "eric-maynard (via GitHub)" <gi...@apache.org>.
eric-maynard merged PR #516:
URL: https://github.com/apache/polaris/pull/516
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "danielhumanmod (via GitHub)" <gi...@apache.org>.
danielhumanmod commented on code in PR #516:
URL: https://github.com/apache/polaris/pull/516#discussion_r1984474418
##########
service/common/src/main/java/org/apache/polaris/service/task/ManifestFileCleanupTaskHandler.java:
##########
@@ -42,50 +40,32 @@
import org.slf4j.LoggerFactory;
/**
- * {@link TaskHandler} responsible for deleting table files: 1. Manifest files: It contains all the
- * files in a manifest and the manifest itself. Since data files may be present in multiple
- * manifests across different snapshots, we assume a data file that doesn't exist is missing because
- * it was already deleted by another task. 2. Table metadata files: It contains previous metadata
- * and statistics files, which are grouped and deleted in batch
+ * {@link ManifestFileCleanupTaskHandler} responsible for deleting all the files in a manifest and
+ * the manifest itself. Since data files may be present in multiple manifests across different
+ * snapshots, we assume a data file that doesn't exist is missing because it was already deleted by
+ * another task.
*/
-// TODO: Rename this class since we introducing metadata cleanup here
-public class ManifestFileCleanupTaskHandler implements TaskHandler {
- public static final int MAX_ATTEMPTS = 3;
- public static final int FILE_DELETION_RETRY_MILLIS = 100;
+public class ManifestFileCleanupTaskHandler extends FileCleanupTaskHandler {
private static final Logger LOGGER =
LoggerFactory.getLogger(ManifestFileCleanupTaskHandler.class);
- private final TaskFileIOSupplier fileIOSupplier;
- private final ExecutorService executorService;
public ManifestFileCleanupTaskHandler(
TaskFileIOSupplier fileIOSupplier, ExecutorService executorService) {
- this.fileIOSupplier = fileIOSupplier;
- this.executorService = executorService;
+ super(fileIOSupplier, executorService);
}
@Override
public boolean canHandleTask(TaskEntity task) {
- return task.getTaskType() == AsyncTaskType.MANIFEST_FILE_CLEANUP
- || task.getTaskType() == AsyncTaskType.METADATA_FILE_BATCH_CLEANUP;
+ return task.getTaskType() == AsyncTaskType.MANIFEST_FILE_CLEANUP;
}
@Override
public boolean handleTask(TaskEntity task, CallContext callContext) {
ManifestCleanupTask cleanupTask = task.readData(ManifestCleanupTask.class);
- TableIdentifier tableId = cleanupTask.getTableId();
+ TableIdentifier tableId = cleanupTask.tableId();
Review Comment:
Quote a prev comment:
> [adutra](https://github.com/adutra) [on Dec 11, 2024](https://github.com/apache/polaris/pull/516#discussion_r1879683280)
Can this be a record?
Here we take Alexandre's suggestion to make the `ManifestCleanupTask` a record as it can eliminate redundant code by automatically generating constructors, getters, equals(), hashCode(), and toString(), while enforcing immutability.
After making it a record, the getter method’s naming differs from the previous version.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "danielhumanmod (via GitHub)" <gi...@apache.org>.
danielhumanmod commented on code in PR #516:
URL: https://github.com/apache/polaris/pull/516#discussion_r1984474418
##########
service/common/src/main/java/org/apache/polaris/service/task/ManifestFileCleanupTaskHandler.java:
##########
@@ -42,50 +40,32 @@
import org.slf4j.LoggerFactory;
/**
- * {@link TaskHandler} responsible for deleting table files: 1. Manifest files: It contains all the
- * files in a manifest and the manifest itself. Since data files may be present in multiple
- * manifests across different snapshots, we assume a data file that doesn't exist is missing because
- * it was already deleted by another task. 2. Table metadata files: It contains previous metadata
- * and statistics files, which are grouped and deleted in batch
+ * {@link ManifestFileCleanupTaskHandler} responsible for deleting all the files in a manifest and
+ * the manifest itself. Since data files may be present in multiple manifests across different
+ * snapshots, we assume a data file that doesn't exist is missing because it was already deleted by
+ * another task.
*/
-// TODO: Rename this class since we introducing metadata cleanup here
-public class ManifestFileCleanupTaskHandler implements TaskHandler {
- public static final int MAX_ATTEMPTS = 3;
- public static final int FILE_DELETION_RETRY_MILLIS = 100;
+public class ManifestFileCleanupTaskHandler extends FileCleanupTaskHandler {
private static final Logger LOGGER =
LoggerFactory.getLogger(ManifestFileCleanupTaskHandler.class);
- private final TaskFileIOSupplier fileIOSupplier;
- private final ExecutorService executorService;
public ManifestFileCleanupTaskHandler(
TaskFileIOSupplier fileIOSupplier, ExecutorService executorService) {
- this.fileIOSupplier = fileIOSupplier;
- this.executorService = executorService;
+ super(fileIOSupplier, executorService);
}
@Override
public boolean canHandleTask(TaskEntity task) {
- return task.getTaskType() == AsyncTaskType.MANIFEST_FILE_CLEANUP
- || task.getTaskType() == AsyncTaskType.METADATA_FILE_BATCH_CLEANUP;
+ return task.getTaskType() == AsyncTaskType.MANIFEST_FILE_CLEANUP;
}
@Override
public boolean handleTask(TaskEntity task, CallContext callContext) {
ManifestCleanupTask cleanupTask = task.readData(ManifestCleanupTask.class);
- TableIdentifier tableId = cleanupTask.getTableId();
+ TableIdentifier tableId = cleanupTask.tableId();
Review Comment:
Quote a prev comment:
> [adutra](https://github.com/adutra) [on Dec 11, 2024](https://github.com/apache/polaris/pull/516#discussion_r1879683280)
Can this be a record?
Here we take Alexandre's suggestion to make the `ManifestCleanupTask` a record as it can eliminate redundant code by automatically generating constructors, getters, equals(), hashCode(), and toString(), while enforcing immutability.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on PR #516:
URL: https://github.com/apache/polaris/pull/516#issuecomment-2634643784
@danielhumanmod , are you still working on it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "flyrain (via GitHub)" <gi...@apache.org>.
flyrain commented on code in PR #516:
URL: https://github.com/apache/polaris/pull/516#discussion_r1941668187
##########
polaris-service/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.task;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.entity.AsyncTaskType;
+import org.apache.polaris.core.entity.TaskEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link BatchFileCleanupTaskHandler} responsible for batch file cleanup by processing multiple
+ * file deletions in a single task handler. Valid files are deleted asynchronously with retries for
+ * transient errors, while missing files are logged and skipped.
+ */
+public class BatchFileCleanupTaskHandler extends FileCleanupTaskHandler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(BatchFileCleanupTaskHandler.class);
+
+ public BatchFileCleanupTaskHandler(
+ Function<TaskEntity, FileIO> fileIOSupplier, ExecutorService executorService) {
+ super(fileIOSupplier, executorService);
+ }
+
+ @Override
+ public boolean canHandleTask(TaskEntity task) {
+ return task.getTaskType() == AsyncTaskType.BATCH_FILE_CLEANUP;
+ }
+
+ @Override
+ public boolean handleTask(TaskEntity task) {
+ BatchFileCleanupTask cleanupTask = task.readData(BatchFileCleanupTask.class);
+ TableIdentifier tableId = cleanupTask.tableId();
+ List<String> batchFiles = cleanupTask.batchFiles();
+ try (FileIO authorizedFileIO = fileIOSupplier.apply(task)) {
+ List<String> validFiles =
+ batchFiles.stream().filter(file -> TaskUtils.exists(file, authorizedFileIO)).toList();
Review Comment:
I agreed that leveraging object-storage batch requests would be a nice improvement. There are alternative ways as well. I'd consider a followup PR for that. With that, it isn't a blocker for this PR. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Enhancement] Refactor Cleanup Task Handler [polaris]
Posted by "danielhumanmod (via GitHub)" <gi...@apache.org>.
danielhumanmod commented on PR #516:
URL: https://github.com/apache/polaris/pull/516#issuecomment-2636113531
TODO: Solve conflict after #589 reverted
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org