You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "RussellSpitzer (via GitHub)" <gi...@apache.org> on 2023/01/27 23:27:28 UTC
[GitHub] [iceberg] RussellSpitzer opened a new pull request, #6682: Bulk delete
RussellSpitzer opened a new pull request, #6682:
URL: https://github.com/apache/iceberg/pull/6682
Changes all Deleting Spark Actions to use FileIO Bulk Operations, adds Bulk delete to HadoopIO
The basic idea here is all of our deletes should use the bulk api or have their parallelism controlled at the FileIO level primarily. All deletes should use some parallelism by default.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] amogh-jahagirdar commented on a diff in pull request #6682: Bulk delete
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1092454022
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -85,6 +88,7 @@
private static final Logger LOG = LoggerFactory.getLogger(BaseSparkAction.class);
private static final AtomicInteger JOB_COUNTER = new AtomicInteger();
private static final int DELETE_NUM_RETRIES = 3;
+ private static final int DELETE_GROUP_SIZE = 100000;
Review Comment:
yeah that makes sense, I was thinking that since the deletion is done on the driver people would want control over how much is brought into memory, but we don't need to overcomplicate it. 100000 seems like a good default
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1095009182
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -80,9 +84,16 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param executorService the service to use
* @return this for method chaining
+ * @deprecated All deletes should be performed using the bulk delete api if available. Use FileIO
+ * specific parallelism controls to adjust bulk delete concurrency within that api.
*/
+ @Deprecated
Review Comment:
I really don't think we should support external executor services. I think we are just pushing a ton of complexity into the user space which is only wildly used because our defaults are not good. If we don't allow passing an external executor to S3FileIO (or file io's in general) I don't think we should allow it here either.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1121109446
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +165,45 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
+ .suppressFailureWhenFinished()
+ .onFailure(
+ (f, e) -> {
+ LOG.error("Failure during bulk delete on file: {} ", f, e);
+ failureCount.incrementAndGet();
Review Comment:
This is going to increment the count on each failed attempt and won't be accurate. We could count the number of successfully deleted files instead and then use `Iterables.size(pathsToDelete)` to find how many we were supposed to delete.
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +165,45 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
Review Comment:
nit: Shall we define a constant for this too?
##########
api/src/main/java/org/apache/iceberg/actions/ExpireSnapshots.java:
##########
@@ -80,11 +80,14 @@ public interface ExpireSnapshots extends Action<ExpireSnapshots, ExpireSnapshots
ExpireSnapshots deleteWith(Consumer<String> deleteFunc);
/**
- * Passes an alternative executor service that will be used for manifests, data and delete files
- * deletion.
+ * Passes an alternative executor service that will be used for files removal. This service will
+ * only be used if a custom delete function is provided by {@link #deleteWith(Consumer)} or if the
+ * FileIO does not {@link org.apache.iceberg.io.SupportsBulkOperations support bulk deletes}.
Review Comment:
nit: Shall we import `SupportsBulkOperations` directly like in the two classes earlier?
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +261,38 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(SupportsBulkOperations io, Iterator<FileInfo> files) {
+ DeleteSummary summary = new DeleteSummary();
+ Iterator<List<FileInfo>> fileGroups = Iterators.partition(files, DELETE_GROUP_SIZE);
+
+ Tasks.foreach(fileGroups)
+ .suppressFailureWhenFinished()
+ .run(fileGroup -> deleteFileGroup(fileGroup, io, summary));
+
+ LOG.info("Deleted {} total files with bulk deletes", summary.totalFilesCount());
+
+ return summary;
+ }
+
+ private static void deleteFileGroup(
+ List<FileInfo> fileGroup, SupportsBulkOperations io, DeleteSummary summary) {
+ ImmutableListMultimap<String, FileInfo> filesByType =
+ Multimaps.index(fileGroup, FileInfo::getType);
+ ListMultimap<String, String> pathsByType =
+ Multimaps.transformValues(filesByType, FileInfo::getPath);
+ for (Map.Entry<String, Collection<String>> entry : pathsByType.asMap().entrySet()) {
+ String type = entry.getKey();
+ Collection<String> paths = entry.getValue();
+ int failures = 0;
+ try {
+ io.deleteFiles(paths);
+ } catch (BulkDeletionFailureException bulkDeletionFailureException) {
Review Comment:
nit: Is there value in such a long exception name compared to `e`?
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteReachableFilesSparkAction.java:
##########
@@ -132,7 +135,20 @@ private Dataset<FileInfo> reachableFileDS(TableMetadata metadata) {
}
private DeleteReachableFiles.Result deleteFiles(Iterator<FileInfo> files) {
- DeleteSummary summary = deleteFiles(deleteExecutorService, deleteFunc, files);
+ DeleteSummary summary;
Review Comment:
nit: What about an empty line before if/else blocks?
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteReachableFilesSparkAction.java:
##########
@@ -132,7 +135,20 @@ private Dataset<FileInfo> reachableFileDS(TableMetadata metadata) {
}
private DeleteReachableFiles.Result deleteFiles(Iterator<FileInfo> files) {
- DeleteSummary summary = deleteFiles(deleteExecutorService, deleteFunc, files);
+ DeleteSummary summary;
+ if (deleteFunc != null || !(io instanceof SupportsBulkOperations)) {
Review Comment:
Can we invert the condition as it is hard to interpret negation?
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/RemoveOrphanFilesProcedure.java:
##########
@@ -98,6 +101,13 @@ public InternalRow[] call(InternalRow args) {
String location = args.isNullAt(2) ? null : args.getString(2);
boolean dryRun = args.isNullAt(3) ? false : args.getBoolean(3);
Integer maxConcurrentDeletes = args.isNullAt(4) ? null : args.getInt(4);
+ if (maxConcurrentDeletes != null) {
Review Comment:
Same here.
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -246,12 +247,24 @@ private DeleteOrphanFiles.Result doExecute() {
List<String> orphanFiles =
findOrphanFiles(spark(), actualFileIdentDS, validFileIdentDS, prefixMismatchMode);
- Tasks.foreach(orphanFiles)
- .noRetry()
- .executeWith(deleteExecutorService)
- .suppressFailureWhenFinished()
- .onFailure((file, exc) -> LOG.warn("Failed to delete file: {}", file, exc))
- .run(deleteFunc::accept);
+ if (deleteFunc == null && table.io() instanceof SupportsBulkOperations) {
+ ((SupportsBulkOperations) table.io()).deleteFiles(orphanFiles);
+
+ } else {
+ if (deleteFunc == null) {
+ LOG.info("Bulk Deletes are not Supported by {}, using non-bulk deletes", table.io());
+ deleteFunc = defaultDelete;
Review Comment:
I don't think it is a good idea to init this variable here. I summarized my thoughts in other actions below. One way to avoid this:
```
if (deleteFunc == null) {
LOG.info("Table IO does not support bulk operations, using non-bulk deletes");
} else {
LOG.info("Custom delete function provided, using non-bulk deletes");
}
Tasks.foreach(orphanFiles)
.noRetry()
...
.run(deleteFunc != null ? deleteFunc::accept : table.io()::deleteFile);
```
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -125,7 +126,7 @@ public void accept(String file) {
private String location = null;
private long olderThanTimestamp = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(3);
private Dataset<Row> compareToFileList;
- private Consumer<String> deleteFunc = defaultDelete;
Review Comment:
I'd consider getting rid of `defaultDelete` and using `table.io()::deleteFile` explicitly.
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +261,38 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(SupportsBulkOperations io, Iterator<FileInfo> files) {
+ DeleteSummary summary = new DeleteSummary();
+ Iterator<List<FileInfo>> fileGroups = Iterators.partition(files, DELETE_GROUP_SIZE);
+
+ Tasks.foreach(fileGroups)
+ .suppressFailureWhenFinished()
+ .run(fileGroup -> deleteFileGroup(fileGroup, io, summary));
+
+ LOG.info("Deleted {} total files with bulk deletes", summary.totalFilesCount());
+
+ return summary;
+ }
+
+ private static void deleteFileGroup(
+ List<FileInfo> fileGroup, SupportsBulkOperations io, DeleteSummary summary) {
+ ImmutableListMultimap<String, FileInfo> filesByType =
Review Comment:
nit: Just `ListMultimap` instead of `ImmutableListMultimap`?
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +261,38 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(SupportsBulkOperations io, Iterator<FileInfo> files) {
+ DeleteSummary summary = new DeleteSummary();
+ Iterator<List<FileInfo>> fileGroups = Iterators.partition(files, DELETE_GROUP_SIZE);
+
+ Tasks.foreach(fileGroups)
+ .suppressFailureWhenFinished()
+ .run(fileGroup -> deleteFileGroup(fileGroup, io, summary));
+
+ LOG.info("Deleted {} total files with bulk deletes", summary.totalFilesCount());
+
+ return summary;
+ }
+
+ private static void deleteFileGroup(
+ List<FileInfo> fileGroup, SupportsBulkOperations io, DeleteSummary summary) {
+ ImmutableListMultimap<String, FileInfo> filesByType =
+ Multimaps.index(fileGroup, FileInfo::getType);
+ ListMultimap<String, String> pathsByType =
+ Multimaps.transformValues(filesByType, FileInfo::getPath);
+ for (Map.Entry<String, Collection<String>> entry : pathsByType.asMap().entrySet()) {
Review Comment:
These multi maps calls are pretty tricky and separating them out seems like a good idea.
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +261,38 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(SupportsBulkOperations io, Iterator<FileInfo> files) {
+ DeleteSummary summary = new DeleteSummary();
+ Iterator<List<FileInfo>> fileGroups = Iterators.partition(files, DELETE_GROUP_SIZE);
+
+ Tasks.foreach(fileGroups)
+ .suppressFailureWhenFinished()
+ .run(fileGroup -> deleteFileGroup(fileGroup, io, summary));
+
+ LOG.info("Deleted {} total files with bulk deletes", summary.totalFilesCount());
+
+ return summary;
+ }
+
+ private static void deleteFileGroup(
+ List<FileInfo> fileGroup, SupportsBulkOperations io, DeleteSummary summary) {
+ ImmutableListMultimap<String, FileInfo> filesByType =
+ Multimaps.index(fileGroup, FileInfo::getType);
+ ListMultimap<String, String> pathsByType =
+ Multimaps.transformValues(filesByType, FileInfo::getPath);
+ for (Map.Entry<String, Collection<String>> entry : pathsByType.asMap().entrySet()) {
Review Comment:
nit: What about adding empty lines after method args and prior to the for loop?
```
private static void deleteFileGroup(
List<FileInfo> fileGroup, SupportsBulkOperations io, DeleteSummary summary) {
ListMultimap<String, FileInfo> filesByType =
Multimaps.index(fileGroup, FileInfo::getType);
ListMultimap<String, String> pathsByType =
Multimaps.transformValues(filesByType, FileInfo::getPath);
for (Map.Entry<String, Collection<String>> entry : pathsByType.asMap().entrySet()) {
...
}
}
```
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +261,38 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(SupportsBulkOperations io, Iterator<FileInfo> files) {
+ DeleteSummary summary = new DeleteSummary();
+ Iterator<List<FileInfo>> fileGroups = Iterators.partition(files, DELETE_GROUP_SIZE);
+
+ Tasks.foreach(fileGroups)
+ .suppressFailureWhenFinished()
+ .run(fileGroup -> deleteFileGroup(fileGroup, io, summary));
+
+ LOG.info("Deleted {} total files with bulk deletes", summary.totalFilesCount());
+
+ return summary;
+ }
+
+ private static void deleteFileGroup(
+ List<FileInfo> fileGroup, SupportsBulkOperations io, DeleteSummary summary) {
+ ImmutableListMultimap<String, FileInfo> filesByType =
+ Multimaps.index(fileGroup, FileInfo::getType);
+ ListMultimap<String, String> pathsByType =
+ Multimaps.transformValues(filesByType, FileInfo::getPath);
+ for (Map.Entry<String, Collection<String>> entry : pathsByType.asMap().entrySet()) {
+ String type = entry.getKey();
+ Collection<String> paths = entry.getValue();
+ int failures = 0;
+ try {
+ io.deleteFiles(paths);
+ } catch (BulkDeletionFailureException bulkDeletionFailureException) {
+ failures = bulkDeletionFailureException.numberFailedObjects();
+ }
+ summary.deletedFiles(entry.getKey(), paths.size() - failures);
Review Comment:
Shall we use `type` variable defined above?
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteReachableFilesSparkAction.java:
##########
@@ -132,7 +135,20 @@ private Dataset<FileInfo> reachableFileDS(TableMetadata metadata) {
}
private DeleteReachableFiles.Result deleteFiles(Iterator<FileInfo> files) {
- DeleteSummary summary = deleteFiles(deleteExecutorService, deleteFunc, files);
+ DeleteSummary summary;
+ if (deleteFunc != null || !(io instanceof SupportsBulkOperations)) {
+ if (deleteFunc == null) {
+ LOG.info("Table IO does not support Bulk Operations. Using non-bulk deletes.");
+ deleteFunc = defaultDelete;
Review Comment:
If we follow the snippet above, we can remove `defaultDelete` completely.
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +261,38 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(SupportsBulkOperations io, Iterator<FileInfo> files) {
+ DeleteSummary summary = new DeleteSummary();
+ Iterator<List<FileInfo>> fileGroups = Iterators.partition(files, DELETE_GROUP_SIZE);
+
+ Tasks.foreach(fileGroups)
+ .suppressFailureWhenFinished()
+ .run(fileGroup -> deleteFileGroup(fileGroup, io, summary));
+
+ LOG.info("Deleted {} total files with bulk deletes", summary.totalFilesCount());
Review Comment:
I think this log message is redundant as each place that calls it logs the result afterwards. Each caller also logs whether it is going to use bulk deletes or not.
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteReachableFilesSparkAction.java:
##########
@@ -54,6 +55,8 @@
private static final Logger LOG = LoggerFactory.getLogger(DeleteReachableFilesSparkAction.class);
private final String metadataFileLocation;
+
+ @Deprecated
private final Consumer<String> defaultDelete =
Review Comment:
The code still uses this var even though it is marked as deprecated. Can we simply remove it?
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteReachableFilesSparkAction.java:
##########
@@ -132,7 +135,20 @@ private Dataset<FileInfo> reachableFileDS(TableMetadata metadata) {
}
private DeleteReachableFiles.Result deleteFiles(Iterator<FileInfo> files) {
- DeleteSummary summary = deleteFiles(deleteExecutorService, deleteFunc, files);
+ DeleteSummary summary;
+ if (deleteFunc != null || !(io instanceof SupportsBulkOperations)) {
+ if (deleteFunc == null) {
+ LOG.info("Table IO does not support Bulk Operations. Using non-bulk deletes.");
Review Comment:
nit: Why capital letter for some words?
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java:
##########
@@ -93,6 +97,12 @@ public InternalRow[] call(InternalRow args) {
Long olderThanMillis = args.isNullAt(1) ? null : DateTimeUtil.microsToMillis(args.getLong(1));
Integer retainLastNum = args.isNullAt(2) ? null : args.getInt(2);
Integer maxConcurrentDeletes = args.isNullAt(3) ? null : args.getInt(3);
+ if (maxConcurrentDeletes != null) {
+ LOG.warn(
Review Comment:
I don't think this description is accurate. I believe this property will have no impact now.
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/ExpireSnapshotsSparkAction.java:
##########
@@ -265,7 +266,20 @@ private Set<Long> findExpiredSnapshotIds(
}
private ExpireSnapshots.Result deleteFiles(Iterator<FileInfo> files) {
- DeleteSummary summary = deleteFiles(deleteExecutorService, deleteFunc, files);
+ DeleteSummary summary;
+ if (deleteFunc != null || !(table.io() instanceof SupportsBulkOperations)) {
Review Comment:
Same comments as in `DeleteReachableFilesSparkAction`.
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -246,12 +247,24 @@ private DeleteOrphanFiles.Result doExecute() {
List<String> orphanFiles =
findOrphanFiles(spark(), actualFileIdentDS, validFileIdentDS, prefixMismatchMode);
- Tasks.foreach(orphanFiles)
- .noRetry()
- .executeWith(deleteExecutorService)
- .suppressFailureWhenFinished()
- .onFailure((file, exc) -> LOG.warn("Failed to delete file: {}", file, exc))
- .run(deleteFunc::accept);
+ if (deleteFunc == null && table.io() instanceof SupportsBulkOperations) {
+ ((SupportsBulkOperations) table.io()).deleteFiles(orphanFiles);
Review Comment:
Unlike non-bulk deletes, this may fail the entire action if there is a bulk delete exception. I think we will have to add a helper method like this or similar?
```
private void deleteFiles(SupportsBulkOperations io, List<String> paths) {
try {
io.deleteFiles(paths);
LOG.info("Deleted {} files using bulk deletes", paths.size());
} catch (BulkDeletionFailureException e) {
int deletedFilesCount = paths.size() - e.numberFailedObjects();
LOG.warn(
"Deleted only {} of {} files using bulk deletes",
deletedFilesCount,
paths.size());
}
}
```
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -246,12 +247,24 @@ private DeleteOrphanFiles.Result doExecute() {
List<String> orphanFiles =
findOrphanFiles(spark(), actualFileIdentDS, validFileIdentDS, prefixMismatchMode);
- Tasks.foreach(orphanFiles)
- .noRetry()
- .executeWith(deleteExecutorService)
- .suppressFailureWhenFinished()
- .onFailure((file, exc) -> LOG.warn("Failed to delete file: {}", file, exc))
- .run(deleteFunc::accept);
+ if (deleteFunc == null && table.io() instanceof SupportsBulkOperations) {
+ ((SupportsBulkOperations) table.io()).deleteFiles(orphanFiles);
+
+ } else {
+ if (deleteFunc == null) {
+ LOG.info("Bulk Deletes are not Supported by {}, using non-bulk deletes", table.io());
Review Comment:
nit: Can we align log messages with other actions?
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +164,46 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
+ .suppressFailureWhenFinished()
+ .onFailure(
+ (f, e) -> {
+ LOG.error("Failure during bulk delete on file: {} ", f, e);
+ failureCount.incrementAndGet();
+ })
+ .run(this::deleteFile);
+
+ if (failureCount.get() != 0) {
+ throw new BulkDeletionFailureException(failureCount.get());
+ }
+ }
+
+ private int deleteThreads() {
+ return conf()
Review Comment:
I still think having an extra variable here would make it a bit more readable.
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteReachableFilesSparkAction.java:
##########
@@ -132,7 +135,20 @@ private Dataset<FileInfo> reachableFileDS(TableMetadata metadata) {
}
private DeleteReachableFiles.Result deleteFiles(Iterator<FileInfo> files) {
- DeleteSummary summary = deleteFiles(deleteExecutorService, deleteFunc, files);
+ DeleteSummary summary;
+ if (deleteFunc != null || !(io instanceof SupportsBulkOperations)) {
+ if (deleteFunc == null) {
+ LOG.info("Table IO does not support Bulk Operations. Using non-bulk deletes.");
+ deleteFunc = defaultDelete;
Review Comment:
Hm, it seems a bit odd to init this variable here. Also, setting this will change the behavior if the action is called multiple times. Can we pass delete func explicitly?
```
if (deleteFunc == null && io instanceof SupportsBulkOperations) {
LOG.info("Table IO supports bulk operations, using bulk deletes");
summary = deleteFiles((SupportsBulkOperations) io, files);
} else {
if (deleteFunc == null) {
LOG.info("Table IO does not support bulk operations, using non-bulk deletes");
summary = deleteFiles(deleteExecutorService, io::deleteFile, files);
} else {
LOG.info("Custom delete function provided, using non-bulk deletes");
summary = deleteFiles(deleteExecutorService, deleteFunc, files);
}
}
```
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -246,12 +247,24 @@ private DeleteOrphanFiles.Result doExecute() {
List<String> orphanFiles =
findOrphanFiles(spark(), actualFileIdentDS, validFileIdentDS, prefixMismatchMode);
- Tasks.foreach(orphanFiles)
- .noRetry()
- .executeWith(deleteExecutorService)
- .suppressFailureWhenFinished()
- .onFailure((file, exc) -> LOG.warn("Failed to delete file: {}", file, exc))
- .run(deleteFunc::accept);
+ if (deleteFunc == null && table.io() instanceof SupportsBulkOperations) {
+ ((SupportsBulkOperations) table.io()).deleteFiles(orphanFiles);
Review Comment:
Logging the number of removed files like above may help debugging.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] amogh-jahagirdar commented on pull request #6682: Bulk delete
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#issuecomment-1407464402
Thanks a ton for closing the loop on this @RussellSpitzer ! Left some comments
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#issuecomment-1410979830
I am getting to this today, hopefully.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1095234465
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -80,9 +84,16 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param executorService the service to use
* @return this for method chaining
+ * @deprecated All deletes should be performed using the bulk delete api if available. Use FileIO
+ * specific parallelism controls to adjust bulk delete concurrency within that api.
*/
+ @Deprecated
DeleteOrphanFiles executeDeleteWith(ExecutorService executorService);
+ default DeleteOrphanFiles deleteBulkWith(Consumer<Iterable<String>> deleteFunc) {
Review Comment:
My worry is that the current implementation breaks the current behavior in an incompatible way: it checks if the table IO supports bulk deletes and if yes, ignores `deleteWith`. It will break those dry-run scenarios if the table is configured with a non-bulk IO and the user passes a custom delete function.
Can you elaborate a bit on why this is needed? Why not just not go through the bulk delete API if a custom delete function is provided?
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -80,9 +84,16 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param executorService the service to use
* @return this for method chaining
+ * @deprecated All deletes should be performed using the bulk delete api if available. Use FileIO
+ * specific parallelism controls to adjust bulk delete concurrency within that api.
*/
+ @Deprecated
DeleteOrphanFiles executeDeleteWith(ExecutorService executorService);
+ default DeleteOrphanFiles deleteBulkWith(Consumer<Iterable<String>> deleteFunc) {
Review Comment:
My worry is that the current implementation breaks the behavior in an incompatible way: it checks if the table IO supports bulk deletes and if yes, ignores `deleteWith`. It will break those dry-run scenarios if the table is configured with a non-bulk IO and the user passes a custom delete function.
Can you elaborate a bit on why this is needed? Why not just not go through the bulk delete API if a custom delete function is provided?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] amogh-jahagirdar commented on a diff in pull request #6682: Bulk delete
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1093848968
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -85,6 +88,7 @@
private static final Logger LOG = LoggerFactory.getLogger(BaseSparkAction.class);
private static final AtomicInteger JOB_COUNTER = new AtomicInteger();
private static final int DELETE_NUM_RETRIES = 3;
+ private static final int DELETE_GROUP_SIZE = 100000;
Review Comment:
Yeah that makes sense, in that case we can just leave it as is.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1095006673
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -80,9 +84,16 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param executorService the service to use
* @return this for method chaining
+ * @deprecated All deletes should be performed using the bulk delete api if available. Use FileIO
+ * specific parallelism controls to adjust bulk delete concurrency within that api.
*/
+ @Deprecated
DeleteOrphanFiles executeDeleteWith(ExecutorService executorService);
+ default DeleteOrphanFiles deleteBulkWith(Consumer<Iterable<String>> deleteFunc) {
Review Comment:
I'd rather remove both, but this is required for testing the bulk delete method. The other option is I can rewrite the test suites which rely on "deleteWith" to instead just delete files for real.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1096394775
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -80,9 +84,16 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param executorService the service to use
* @return this for method chaining
+ * @deprecated All deletes should be performed using the bulk delete api if available. Use FileIO
+ * specific parallelism controls to adjust bulk delete concurrency within that api.
*/
+ @Deprecated
Review Comment:
The original proposal for `SupportsBulkOperations` included a way to pass a custom executor service, which we removed until we had a use case. Without having a default executor service in file IO, we would be pushing the complexity on the user. However, I was talking about having a default executor service in all file IO implementations but allowing to pass a custom one in `SupportsBulkOperations`. That way, our defaults are always good but if needed, one may pass a custom executor service.
```
void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException;
void deleteFiles(ExecutorService executorService, Iterable<String> pathsToDelete) throws BulkDeletionFailureException;
```
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122023544
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +261,38 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(SupportsBulkOperations io, Iterator<FileInfo> files) {
+ DeleteSummary summary = new DeleteSummary();
+ Iterator<List<FileInfo>> fileGroups = Iterators.partition(files, DELETE_GROUP_SIZE);
+
+ Tasks.foreach(fileGroups)
+ .suppressFailureWhenFinished()
+ .run(fileGroup -> deleteFileGroup(fileGroup, io, summary));
+
+ LOG.info("Deleted {} total files with bulk deletes", summary.totalFilesCount());
+
+ return summary;
+ }
+
+ private static void deleteFileGroup(
+ List<FileInfo> fileGroup, SupportsBulkOperations io, DeleteSummary summary) {
+ ImmutableListMultimap<String, FileInfo> filesByType =
+ Multimaps.index(fileGroup, FileInfo::getType);
+ ListMultimap<String, String> pathsByType =
+ Multimaps.transformValues(filesByType, FileInfo::getPath);
+ for (Map.Entry<String, Collection<String>> entry : pathsByType.asMap().entrySet()) {
Review Comment:
my kingdom for a flatmap
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1121996881
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +165,45 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
+ .suppressFailureWhenFinished()
+ .onFailure(
+ (f, e) -> {
+ LOG.error("Failure during bulk delete on file: {} ", f, e);
+ failureCount.incrementAndGet();
Review Comment:
I double checked this, it only fires off when all retries are exhausted so it is correct as is.
```scala
scala> def testFailure() = {
var failureCount =0
Tasks.foreach("value")
.retry(3)
.onFailure((y, x: Throwable) => failureCount += 1)
.suppressFailureWhenFinished()
.run(x => throw new Exception("ohNO"))
failureCount
}
scala> testFailure()
23/03/01 10:16:22 WARN Tasks: Retrying task after failure: ohNO
java.lang.Exception: ohNO
...
23/03/01 10:16:23 WARN Tasks: Retrying task after failure: ohNO
java.lang.Exception: ohNO
...
23/03/01 10:16:25 WARN Tasks: Retrying task after failure: ohNO
java.lang.Exception: ohNO
...
res21: Int = 1
````
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1120476371
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +164,46 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
Review Comment:
This is what used to be in the delete code, so I just copied 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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122682061
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java:
##########
@@ -114,7 +119,17 @@ public InternalRow[] call(InternalRow args) {
}
if (maxConcurrentDeletes != null) {
- action.executeDeleteWith(executorService(maxConcurrentDeletes, "expire-snapshots"));
+ if (table.io() instanceof SupportsBulkOperations) {
+ LOG.warn(
+ "max_concurrent_deletes only works with FileIOs that do not support bulk deletes. This"
+ + "table is currently using {} which supports bulk deletes so the parameter will be ignored. "
+ + "See that IO's documentation to learn how to adjust parallelism for that particular "
+ + "IO's bulk delete.",
+ table.io());
Review Comment:
Also `table.io().getClass().getName()`?
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/RemoveOrphanFilesProcedure.java:
##########
@@ -154,7 +158,17 @@ public InternalRow[] call(InternalRow args) {
}
if (maxConcurrentDeletes != null) {
- action.executeDeleteWith(executorService(maxConcurrentDeletes, "remove-orphans"));
+ if (table.io() instanceof SupportsBulkOperations) {
+ LOG.warn(
+ "max_concurrent_deletes only works with FileIOs that do not support bulk deletes. This"
+ + "table is currently using {} which supports bulk deletes so the parameter will be ignored. "
+ + "See that IO's documentation to learn how to adjust parallelism for that particular "
+ + "IO's bulk delete.",
+ table.io());
Review Comment:
Same 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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1095925633
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -67,7 +67,11 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param deleteFunc a function that will be called to delete files
* @return this for method chaining
+ * @deprecated Deletes are now performed in bulk see {@link #deleteBulkWith(Consumer)}. This
Review Comment:
I think adding that keeping the other apis makes things a bit more complicated. Let me try rewriting this a bit, but I think keeping deleteWith and Execute with means necessarily having two sets of tests one for each operation path. We could maybe get away with a single test for each of these ....
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] dramaticlly commented on a diff in pull request #6682: Bulk delete
Posted by "dramaticlly (via GitHub)" <gi...@apache.org>.
dramaticlly commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1093902105
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +257,39 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(
Review Comment:
I am wondering if we shall have some sort of tests to cover this, where the `bulkDeleteFunc` can add files to a separate set instead of actual delete, to see we are properly group by its type and verify the DeleteSummary results?
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -246,12 +268,22 @@ private DeleteOrphanFiles.Result doExecute() {
List<String> orphanFiles =
findOrphanFiles(spark(), actualFileIdentDS, validFileIdentDS, prefixMismatchMode);
- Tasks.foreach(orphanFiles)
- .noRetry()
- .executeWith(deleteExecutorService)
- .suppressFailureWhenFinished()
- .onFailure((file, exc) -> LOG.warn("Failed to delete file: {}", file, exc))
- .run(deleteFunc::accept);
+ if (table.io() instanceof SupportsBulkOperations) {
+ try {
+ bulkDeleteFunc.accept(orphanFiles);
+ } catch (BulkDeletionFailureException bulkDeletionFailureException) {
+ LOG.warn(
+ "Bulk delete failed to remove {} files",
+ bulkDeletionFailureException.numberFailedObjects());
+ }
+ } else {
Review Comment:
I was think would it help to leave a message stating that bulk deletion is not supported on current IO to encourage for further check? Similar to how we add `Falling back to non-bulk deletes because the given io {} does not support bulk deletes.` for expire snapshots, I feel orphan deletion might benefit from it 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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1095005897
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -67,7 +67,11 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param deleteFunc a function that will be called to delete files
* @return this for method chaining
+ * @deprecated Deletes are now performed in bulk see {@link #deleteBulkWith(Consumer)}. This
Review Comment:
One of my goals here is to remove those existing apis. I feel like custom delete function here is solely here for testing. Any user who actually wants to deal with their results would use the "toDataframe" option for Spark.
Currently we have too many options here and no real use cases. This really shouldn't be a set of configuration or options that users should be concerned with. If they are, the correct spot to be modifying this is the FileIO. For example, if I have multiple actions running at the same time, the coordination of requests to S3 or whatnot should happen at the FileIO, otherwise the user needs to manually be determining the correct values for each of their possible concurrent jobs.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1093452257
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -85,6 +88,7 @@
private static final Logger LOG = LoggerFactory.getLogger(BaseSparkAction.class);
private static final AtomicInteger JOB_COUNTER = new AtomicInteger();
private static final int DELETE_NUM_RETRIES = 3;
+ private static final int DELETE_GROUP_SIZE = 100000;
Review Comment:
Yeah if at all possible I want to just remove configuration options and instead just have things that always work
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1094075265
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -67,7 +67,11 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param deleteFunc a function that will be called to delete files
* @return this for method chaining
+ * @deprecated Deletes are now performed in bulk see {@link #deleteBulkWith(Consumer)}. This
Review Comment:
This will break the public API if a table is configured with a non-bulk IO and the user passes a custom delete function. In a lot of cases, a custom delete simply adds elements to a set. After this change, the custom delete function may be ignored and we will actually delete files.
Do we actually need to attempt bulk deletes in if a custom function is provided? I'd consider just going through the non-bulk API in that case. That can simplify the implementation.
```
private Consumer<String> deleteFunc = null;
...
@Override
public DeleteOrphanFilesSparkAction deleteWith(Consumer<String> newDeleteFunc) {
this.deleteFunc = newDeleteFunc;
return this;
}
...
private void deleteFiles(Iterable<String> paths) {
if (deleteFunc == null && table.io() instanceof SupportsBulkOperations) {
SupportsBulkOperations bulkIO = (SupportsBulkOperations) table.io();
if (deleteExecutorService != null) {
bulkIO.deleteFiles(deleteExecutorService, paths);
} else {
bulkIO.deleteFiles(paths);
}
} else {
Tasks.foreach(paths)
.noRetry()
.executeWith(deleteExecutorService)
.suppressFailureWhenFinished()
.onFailure((file, exc) -> LOG.warn("Failed to delete file: {}", file, exc))
.run(deleteFunc != null ? deleteFunc::accept : table.io()::deleteFile);
}
}
```
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -80,9 +84,16 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param executorService the service to use
* @return this for method chaining
+ * @deprecated All deletes should be performed using the bulk delete api if available. Use FileIO
+ * specific parallelism controls to adjust bulk delete concurrency within that api.
*/
+ @Deprecated
DeleteOrphanFiles executeDeleteWith(ExecutorService executorService);
+ default DeleteOrphanFiles deleteBulkWith(Consumer<Iterable<String>> deleteFunc) {
Review Comment:
Do we have a use case for this? If not, I'd skip it and just support the old `deleteWith` method.
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +257,39 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(
+ Consumer<Iterable<String>> bulkDeleteFunc, Iterator<FileInfo> files) {
+ DeleteSummary summary = new DeleteSummary();
+ Iterator<List<FileInfo>> groupedIterator = Iterators.partition(files, DELETE_GROUP_SIZE);
+
+ Tasks.foreach(groupedIterator)
+ .suppressFailureWhenFinished()
+ .run(
+ fileList -> {
+ Map<String, List<FileInfo>> filesByType =
+ fileList.stream().collect(Collectors.groupingBy(FileInfo::getType));
+ filesByType.entrySet().stream()
+ .forEach(
+ entry -> {
+ List<String> pathsToDelete =
Review Comment:
What about using `ListMultimap`?
```
ListMultimap<String, String> pathsByType =
Multimaps.newListMultimap(Maps.newHashMap(), Lists::newArrayList);
for (FileInfo fileInfo : fileGroup) {
pathsByType.put(fileInfo.getType(), fileInfo.getPath());
}
for (Map.Entry<String, Collection<String>> entry : filesByType.asMap().entrySet()) {
String type = entry.getKey();
Collection<String> pathsToDelete = entry.getValue();
...
}
```
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -85,6 +88,7 @@
private static final Logger LOG = LoggerFactory.getLogger(BaseSparkAction.class);
private static final AtomicInteger JOB_COUNTER = new AtomicInteger();
private static final int DELETE_NUM_RETRIES = 3;
+ private static final int DELETE_GROUP_SIZE = 100000;
Review Comment:
I don't think we would want to make it configurable but can 100K be too many? This will be used in the path that collects chunk by chunk to the driver to avoid OOM, which is supposed to help small drivers or scenarios with lots of files to delete. Do we think the records are tiny so 100K won't cause any issues?
I think S3FileIO would send 250 files in a single bulk delete by default.
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -261,6 +298,24 @@ static class DeleteSummary {
private final AtomicLong manifestListsCount = new AtomicLong(0L);
private final AtomicLong otherFilesCount = new AtomicLong(0L);
+ public void deletedFiles(String type, int numFiles) {
+ if (FileContent.DATA.name().equalsIgnoreCase(type)) {
+ dataFilesCount.addAndGet(numFiles);
Review Comment:
nit: What about empty lines after each if clause like in the method below? We do that quite frequently in large if-else statements.
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -18,27 +18,42 @@
*/
package org.apache.iceberg.hadoop;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.Iterator;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.BulkDeletionFailureException;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.FileInfo;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.SupportsBulkOperations;
import org.apache.iceberg.io.SupportsPrefixOperations;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Streams;
import org.apache.iceberg.util.SerializableMap;
import org.apache.iceberg.util.SerializableSupplier;
+import org.apache.iceberg.util.Tasks;
+import org.apache.iceberg.util.ThreadPools;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-public class HadoopFileIO implements FileIO, HadoopConfigurable, SupportsPrefixOperations {
+public class HadoopFileIO
+ implements FileIO, HadoopConfigurable, SupportsPrefixOperations, SupportsBulkOperations {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HadoopFileIO.class);
+ private static final String DELETE_FILE_PARALLELISM = "iceberg.hadoop.delete_file_parallelism";
Review Comment:
Do we have any other Hadoop configs? Which naming style did we use there? I thought we were using `-` to separate words.
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -80,9 +84,16 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param executorService the service to use
* @return this for method chaining
+ * @deprecated All deletes should be performed using the bulk delete api if available. Use FileIO
+ * specific parallelism controls to adjust bulk delete concurrency within that api.
*/
+ @Deprecated
Review Comment:
I am not sure the deprecation is worth it. Instead of deprecating a bunch of widely used methods in actions and arguments in procedures, can we consider supporting external executor services with bulk deletes so that these arguments can be still supported?
I like the idea of bulk deletes in `HadoopFileIO`, but I don't see anything fundamentally bad in an ability to override the delete parallelism in actions or procedures. Especially, because it can make the implementation simpler. We already discussed passing executor services in the original PR that added the bulk API.
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +257,39 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(
+ Consumer<Iterable<String>> bulkDeleteFunc, Iterator<FileInfo> files) {
+ DeleteSummary summary = new DeleteSummary();
+ Iterator<List<FileInfo>> groupedIterator = Iterators.partition(files, DELETE_GROUP_SIZE);
+
+ Tasks.foreach(groupedIterator)
+ .suppressFailureWhenFinished()
+ .run(
+ fileList -> {
+ Map<String, List<FileInfo>> filesByType =
Review Comment:
What about adding a private method to handle this? I feel it would help with formatting (if the call fits on a single line). Spotless formats these closures in a weird way.
```
Iterator<List<FileInfo>> fileGroups = Iterators.partition(files, DELETE_GROUP_SIZE);
Tasks.foreach(fileGroups)
.noRetry()
.suppressFailureWhenFinished()
.run(fileGroup -> deleteFileGroup(...));
LOG.info("Deleted {} total files with bulk deletes", summary.totalFilesCount());
return summary;
```
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +257,39 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(
+ Consumer<Iterable<String>> bulkDeleteFunc, Iterator<FileInfo> files) {
+ DeleteSummary summary = new DeleteSummary();
+ Iterator<List<FileInfo>> groupedIterator = Iterators.partition(files, DELETE_GROUP_SIZE);
Review Comment:
nit: Is there a more descriptive name? fileGroups, fileLists?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1095922475
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -80,9 +84,16 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param executorService the service to use
* @return this for method chaining
+ * @deprecated All deletes should be performed using the bulk delete api if available. Use FileIO
+ * specific parallelism controls to adjust bulk delete concurrency within that api.
*/
+ @Deprecated
DeleteOrphanFiles executeDeleteWith(ExecutorService executorService);
+ default DeleteOrphanFiles deleteBulkWith(Consumer<Iterable<String>> deleteFunc) {
Review Comment:
My goal is to remove the non-bulk delete api completely in the future. I really don't want to be in a situation where we have some apis that only work in conjunction with others.
I think if we allow `executeDeleteWith` and `deleteWith` we essentially have this permanently split api. Where sometimes these functions are able to do something but most of the time they can not. In the future I want to be in a situation where there is only `deleteBulkWith`.
If we don't add `deleteBulkWith` but keep an alternate pathway with `executeDeleteWith` and `deleteBulkWith` we will permanently need two sets of tests. One testing the bulk pathway and one testing the non-bulk pathway. Which pathway is used will also not be super clear as some procedures (if we keep dry run as is) will invoke the non bulk pathway.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1121972271
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +165,45 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
+ .suppressFailureWhenFinished()
+ .onFailure(
+ (f, e) -> {
+ LOG.error("Failure during bulk delete on file: {} ", f, e);
+ failureCount.incrementAndGet();
Review Comment:
I'm not sure we want to go over the iterable more than once ... let me think about 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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122025438
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +261,38 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(SupportsBulkOperations io, Iterator<FileInfo> files) {
+ DeleteSummary summary = new DeleteSummary();
+ Iterator<List<FileInfo>> fileGroups = Iterators.partition(files, DELETE_GROUP_SIZE);
+
+ Tasks.foreach(fileGroups)
+ .suppressFailureWhenFinished()
+ .run(fileGroup -> deleteFileGroup(fileGroup, io, summary));
+
+ LOG.info("Deleted {} total files with bulk deletes", summary.totalFilesCount());
+
+ return summary;
+ }
+
+ private static void deleteFileGroup(
+ List<FileInfo> fileGroup, SupportsBulkOperations io, DeleteSummary summary) {
+ ImmutableListMultimap<String, FileInfo> filesByType =
+ Multimaps.index(fileGroup, FileInfo::getType);
+ ListMultimap<String, String> pathsByType =
+ Multimaps.transformValues(filesByType, FileInfo::getPath);
+ for (Map.Entry<String, Collection<String>> entry : pathsByType.asMap().entrySet()) {
+ String type = entry.getKey();
+ Collection<String> paths = entry.getValue();
+ int failures = 0;
+ try {
+ io.deleteFiles(paths);
+ } catch (BulkDeletionFailureException bulkDeletionFailureException) {
+ failures = bulkDeletionFailureException.numberFailedObjects();
+ }
+ summary.deletedFiles(entry.getKey(), paths.size() - failures);
Review Comment:
Missed that, fixed
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122024981
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +261,38 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(SupportsBulkOperations io, Iterator<FileInfo> files) {
+ DeleteSummary summary = new DeleteSummary();
+ Iterator<List<FileInfo>> fileGroups = Iterators.partition(files, DELETE_GROUP_SIZE);
+
+ Tasks.foreach(fileGroups)
+ .suppressFailureWhenFinished()
+ .run(fileGroup -> deleteFileGroup(fileGroup, io, summary));
+
+ LOG.info("Deleted {} total files with bulk deletes", summary.totalFilesCount());
+
+ return summary;
+ }
+
+ private static void deleteFileGroup(
+ List<FileInfo> fileGroup, SupportsBulkOperations io, DeleteSummary summary) {
+ ImmutableListMultimap<String, FileInfo> filesByType =
+ Multimaps.index(fileGroup, FileInfo::getType);
+ ListMultimap<String, String> pathsByType =
+ Multimaps.transformValues(filesByType, FileInfo::getPath);
+ for (Map.Entry<String, Collection<String>> entry : pathsByType.asMap().entrySet()) {
+ String type = entry.getKey();
+ Collection<String> paths = entry.getValue();
+ int failures = 0;
+ try {
+ io.deleteFiles(paths);
+ } catch (BulkDeletionFailureException bulkDeletionFailureException) {
Review Comment:
just the default IntelliJ suggested, but i'll switch to e
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1094094981
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -67,7 +67,11 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param deleteFunc a function that will be called to delete files
* @return this for method chaining
+ * @deprecated Deletes are now performed in bulk see {@link #deleteBulkWith(Consumer)}. This
Review Comment:
I believe the above snippet would be backward compatible and would support the existing API?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] amogh-jahagirdar commented on a diff in pull request #6682: Bulk delete
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1089792632
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +164,45 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
Review Comment:
Perhaps this is out of scope for this PR since for other delete operations we only stop retry on FileNotFoundException, but it seems there are other cases where we should stop retrying like permission denied errors. https://github.com/apache/hadoop/blob/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java#L2848 or AccessControlExceptions in HDFS cases.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] amogh-jahagirdar commented on a diff in pull request #6682: Bulk delete
Posted by "amogh-jahagirdar (via GitHub)" <gi...@apache.org>.
amogh-jahagirdar commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1089790123
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -85,6 +88,7 @@
private static final Logger LOG = LoggerFactory.getLogger(BaseSparkAction.class);
private static final AtomicInteger JOB_COUNTER = new AtomicInteger();
private static final int DELETE_NUM_RETRIES = 3;
+ private static final int DELETE_GROUP_SIZE = 100000;
Review Comment:
Is it feasible to make this value configurable through the options passed in to the action?
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/RemoveOrphanFilesProcedure.java:
##########
@@ -98,6 +101,12 @@ public InternalRow[] call(InternalRow args) {
String location = args.isNullAt(2) ? null : args.getString(2);
boolean dryRun = args.isNullAt(3) ? false : args.getBoolean(3);
Integer maxConcurrentDeletes = args.isNullAt(4) ? null : args.getInt(4);
+ if (maxConcurrentDeletes != null) {
+ LOG.warn(
+ "{} is now deprecated, parallelism should now be configured in the FileIO bulk operations. Check the"
+ + "configured FileIO for more information",
+ PARAMETERS[4].name());
+ }
Review Comment:
Style nit, new line after the if block
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/ExpireSnapshotsSparkAction.java:
##########
@@ -265,7 +285,15 @@ private Set<Long> findExpiredSnapshotIds(
}
private ExpireSnapshots.Result deleteFiles(Iterator<FileInfo> files) {
- DeleteSummary summary = deleteFiles(deleteExecutorService, deleteFunc, files);
+ DeleteSummary summary;
+ if (ops.io() instanceof SupportsBulkOperations) {
+ LOG.info("Triggering Bulk Delete Operations");
+ summary = deleteFiles(bulkDeleteFunc, files);
+ } else {
+ LOG.warn("Warning falling back to non-bulk deletes");
Review Comment:
Nit on the warn log, I think we should make it more clear why it's a warning so users don't think anything bad is happening:
"Falling back to non-bulk deletes. Bulk deletes are recommended for better deletion throughput"
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +164,45 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
Review Comment:
Perhaps this is out of scope for this PR since for other delete operations we only stop retry on FileNotFoundException, but it seems there are other cases where we should stop retrying like permission denied errors. https://github.com/apache/hadoop/blob/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java#L2848 or AccessControlExceptions in Hadoop cases.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122003121
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +165,45 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
+ .suppressFailureWhenFinished()
+ .onFailure(
+ (f, e) -> {
+ LOG.error("Failure during bulk delete on file: {} ", f, e);
+ failureCount.incrementAndGet();
Review Comment:
https://github.com/apache/iceberg/blob/715c9b90f537a215f38da5a8ddad19977222860b/core/src/main/java/org/apache/iceberg/util/Tasks.java#L219-L225 Code in question (RunWithRetry) does all retries before hitting "onFailure"
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1121968696
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +165,45 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
+ .suppressFailureWhenFinished()
+ .onFailure(
+ (f, e) -> {
+ LOG.error("Failure during bulk delete on file: {} ", f, e);
+ failureCount.incrementAndGet();
Review Comment:
Ah I thought it was once per element
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122680575
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/ExpireSnapshotsSparkAction.java:
##########
@@ -265,7 +266,21 @@ private Set<Long> findExpiredSnapshotIds(
}
private ExpireSnapshots.Result deleteFiles(Iterator<FileInfo> files) {
- DeleteSummary summary = deleteFiles(deleteExecutorService, deleteFunc, files);
+ DeleteSummary summary;
+ if (deleteFunc == null && table.io() instanceof SupportsBulkOperations) {
+ summary = deleteFiles((SupportsBulkOperations) table.io(), files);
+ } else {
+
+ if (deleteFunc == null) {
+ LOG.info(
+ "Table IO {} does not support bulk operations. Using non-bulk deletes.", table.io());
Review Comment:
Same 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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122680273
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/ExpireSnapshotsSparkAction.java:
##########
@@ -85,7 +86,7 @@ public void accept(String file) {
private final Set<Long> expiredSnapshotIds = Sets.newHashSet();
private Long expireOlderThanValue = null;
private Integer retainLastValue = null;
- private Consumer<String> deleteFunc = defaultDelete;
Review Comment:
nit: Shall we remove `defaultDelete` as it is no longer used?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer merged pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer merged PR #6682:
URL: https://github.com/apache/iceberg/pull/6682
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122113255
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java:
##########
@@ -93,6 +97,12 @@ public InternalRow[] call(InternalRow args) {
Long olderThanMillis = args.isNullAt(1) ? null : DateTimeUtil.microsToMillis(args.getLong(1));
Integer retainLastNum = args.isNullAt(2) ? null : args.getInt(2);
Integer maxConcurrentDeletes = args.isNullAt(3) ? null : args.getInt(3);
+ if (maxConcurrentDeletes != null) {
+ LOG.warn(
Review Comment:
Moving this message to the application of the parameter and having it not applied if the underlying IO supports bulks. Message contents changing 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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1091268114
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -85,6 +88,7 @@
private static final Logger LOG = LoggerFactory.getLogger(BaseSparkAction.class);
private static final AtomicInteger JOB_COUNTER = new AtomicInteger();
private static final int DELETE_NUM_RETRIES = 3;
+ private static final int DELETE_GROUP_SIZE = 100000;
Review Comment:
Honestly I don't see a lot of use cases for changing this. It's really just to set a limit on how many records we pull into memory at a time.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1091266052
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +164,45 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
Review Comment:
For now I just wanted to copy the old logic
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1093958174
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +257,39 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(
Review Comment:
The ExpireSnapshots tests do this, which is why there are all of those modifications to the withBulkDeleteFunc. I think I missed a few though since there are still some failing tests. I'll check that tomorrow
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#issuecomment-1413222159
I agree with the overall direction but I'd try to support the existing API to avoid massive deprecation and simplify the implementation. It will be hard to test all possible scenarios.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1096378212
##########
core/src/main/java/org/apache/iceberg/BaseIncrementalScan.java:
##########
@@ -131,7 +131,8 @@ private Long fromSnapshotIdExclusive(long toSnapshotIdInclusive) {
} else {
// validate there is an ancestor of toSnapshotIdInclusive where parent is fromSnapshotId
Preconditions.checkArgument(
- SnapshotUtil.isParentAncestorOf(table(), toSnapshotIdInclusive, fromSnapshotId),
+ SnapshotUtil.isParentAncestorOf(table(), toSnapshotIdInclusive, fromSnapshotId)
+ || fromSnapshotId.equals(toSnapshotIdInclusive),
Review Comment:
Woops this needs to be removed
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1105204305
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -71,12 +71,15 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
DeleteOrphanFiles deleteWith(Consumer<String> deleteFunc);
/**
- * Passes an alternative executor service that will be used for removing orphaned files.
- *
- * <p>If this method is not called, orphaned manifests and data files will still be deleted in the
- * current thread.
- *
- * <p>
+ * Passes an alternative executor service that will be used for removing orphaned files. This
+ * service will only be used if a custom delete function is provided by {@link
+ * #deleteWith(Consumer)} or if the FileIO does not {@link
+ * org.apache.iceberg.io.SupportsBulkOperations support bulk deletes}. Otherwise, parallelism
Review Comment:
nit: Sometimes, we add qualified imports to shorten such references. Up to you, though.
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -71,12 +71,15 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
DeleteOrphanFiles deleteWith(Consumer<String> deleteFunc);
/**
- * Passes an alternative executor service that will be used for removing orphaned files.
- *
- * <p>If this method is not called, orphaned manifests and data files will still be deleted in the
- * current thread.
- *
- * <p>
+ * Passes an alternative executor service that will be used for removing orphaned files. This
Review Comment:
What about preserving the empty line after a short description of the method?
##########
api/src/main/java/org/apache/iceberg/actions/ExpireSnapshots.java:
##########
@@ -77,20 +77,25 @@ public interface ExpireSnapshots extends Action<ExpireSnapshots, ExpireSnapshots
* @param deleteFunc a function that will be called to delete manifests and data files
* @return this for method chaining
*/
+ @Deprecated
Review Comment:
Left over from the initial implementation?
##########
api/src/main/java/org/apache/iceberg/actions/DeleteReachableFiles.java:
##########
@@ -44,9 +44,11 @@
DeleteReachableFiles deleteWith(Consumer<String> deleteFunc);
/**
- * Passes an alternative executor service that will be used for files removal.
- *
- * <p>If this method is not called, files will be deleted in the current thread.
+ * Passes an alternative executor service that will be used for files removal. This service will
+ * only be used if a custom delete function is provided by {@link #deleteWith(Consumer)} or if the
+ * FileIO does not {@link org.apache.iceberg.io.SupportsBulkOperations support bulk deletes}.
+ * Otherwise, parallelism should be controlled by the IO specific {@link
+ * org.apache.iceberg.io.SupportsBulkOperations#deleteFiles(Iterable) deleteFiles} method.
Review Comment:
What about keeping the last sentence about what happens if not called?
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +164,46 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
+ .suppressFailureWhenFinished()
+ .onFailure(
+ (f, e) -> {
+ LOG.error("Failure during bulk delete on file: {} ", f, e);
+ failureCount.incrementAndGet();
+ })
+ .run(this::deleteFile);
+
+ if (failureCount.get() != 0) {
+ throw new BulkDeletionFailureException(failureCount.get());
+ }
+ }
+
+ private int deleteThreads() {
+ return conf()
Review Comment:
nit: What about an extra var to stay on one line?
```
int defaultValue = Runtime.getRuntime().availableProcessors() * DEFAULT_DELETE_CORE_MULTIPLE;
return conf().getInt(DELETE_FILE_PARALLELISM, defaultValue);
```
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -246,12 +247,23 @@ private DeleteOrphanFiles.Result doExecute() {
List<String> orphanFiles =
findOrphanFiles(spark(), actualFileIdentDS, validFileIdentDS, prefixMismatchMode);
- Tasks.foreach(orphanFiles)
- .noRetry()
- .executeWith(deleteExecutorService)
- .suppressFailureWhenFinished()
- .onFailure((file, exc) -> LOG.warn("Failed to delete file: {}", file, exc))
- .run(deleteFunc::accept);
+ if (deleteFunc != null || !(table.io() instanceof SupportsBulkOperations)) {
Review Comment:
nit: What about inverting this condition instead of using negation?
```
if (deleteFunc == null && table.io() instanceof SupportsBulkOperations) {
// bulk deletes
} else {
// non-bulk delete
}
```
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +164,46 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
+ .suppressFailureWhenFinished()
+ .onFailure(
+ (f, e) -> {
+ LOG.error("Failure during bulk delete on file: {} ", f, e);
+ failureCount.incrementAndGet();
+ })
+ .run(this::deleteFile);
+
+ if (failureCount.get() != 0) {
+ throw new BulkDeletionFailureException(failureCount.get());
+ }
+ }
+
+ private int deleteThreads() {
+ return conf()
+ .getInt(
+ DELETE_FILE_PARALLELISM,
+ Runtime.getRuntime().availableProcessors() * DEFAULT_DELETE_CORE_MULTIPLE);
+ }
+
+ private ExecutorService executorService() {
+ if (executorService == null) {
+ synchronized (HadoopFileIO.class) {
+ if (executorService == null) {
+ executorService =
+ ThreadPools.newWorkerPool("iceberg-hadoopfileio-delete", deleteThreads());
Review Comment:
nit: What about a constant to hold the pool name prefix? Maybe, it will then fit on one line too.
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -18,27 +18,42 @@
*/
package org.apache.iceberg.hadoop;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.Iterator;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.BulkDeletionFailureException;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.FileInfo;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.SupportsBulkOperations;
import org.apache.iceberg.io.SupportsPrefixOperations;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Streams;
import org.apache.iceberg.util.SerializableMap;
import org.apache.iceberg.util.SerializableSupplier;
+import org.apache.iceberg.util.Tasks;
+import org.apache.iceberg.util.ThreadPools;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-public class HadoopFileIO implements FileIO, HadoopConfigurable, SupportsPrefixOperations {
+public class HadoopFileIO
+ implements FileIO, HadoopConfigurable, SupportsPrefixOperations, SupportsBulkOperations {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HadoopFileIO.class);
+ private static final String DELETE_FILE_PARALLELISM = "iceberg.hadoop.delete_file_parallelism";
Review Comment:
nit: This var name sounds more like a value constant rather than a config name.
##########
api/src/main/java/org/apache/iceberg/actions/ExpireSnapshots.java:
##########
@@ -77,20 +77,25 @@ public interface ExpireSnapshots extends Action<ExpireSnapshots, ExpireSnapshots
* @param deleteFunc a function that will be called to delete manifests and data files
* @return this for method chaining
*/
+ @Deprecated
ExpireSnapshots deleteWith(Consumer<String> deleteFunc);
/**
- * Passes an alternative executor service that will be used for manifests, data and delete files
- * deletion.
+ * Passes an alternative executor service that will be used for files removal. This service will
+ * only be used if a custom delete function is provided by {@link #deleteWith(Consumer)} or if the
+ * FileIO does not {@link org.apache.iceberg.io.SupportsBulkOperations support bulk deletes}.
+ * Otherwise, parallelism should be controlled by the IO specific {@link
+ * org.apache.iceberg.io.SupportsBulkOperations#deleteFiles(Iterable) deleteFiles} method.
*
- * <p>If this method is not called, unnecessary manifests and content files will still be deleted
- * in the current thread.
+ * <p>If this method is not called and bulk deletes are not supported, unnecessary manifests and
+ * content files will still be deleted in the current thread.
*
* <p>Identical to {@link org.apache.iceberg.ExpireSnapshots#executeDeleteWith(ExecutorService)}
*
* @param executorService the service to use
* @return this for method chaining
*/
+ @Deprecated
Review Comment:
Also not needed anymore?
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +164,45 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
Review Comment:
+1 to just focus on bulk deletes in this PR.
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -85,6 +88,7 @@
private static final Logger LOG = LoggerFactory.getLogger(BaseSparkAction.class);
private static final AtomicInteger JOB_COUNTER = new AtomicInteger();
private static final int DELETE_NUM_RETRIES = 3;
+ private static final int DELETE_GROUP_SIZE = 100000;
Review Comment:
We would need to make sure this logic works with the streaming iterator in Spark.
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +164,46 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
Review Comment:
How did we come up with this number? What do we use in other places?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122108985
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/procedures/ExpireSnapshotsProcedure.java:
##########
@@ -93,6 +97,12 @@ public InternalRow[] call(InternalRow args) {
Long olderThanMillis = args.isNullAt(1) ? null : DateTimeUtil.microsToMillis(args.getLong(1));
Integer retainLastNum = args.isNullAt(2) ? null : args.getInt(2);
Integer maxConcurrentDeletes = args.isNullAt(3) ? null : args.getInt(3);
+ if (maxConcurrentDeletes != null) {
+ LOG.warn(
Review Comment:
It will still work but only if the IO does not support bulk
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122679902
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteReachableFilesSparkAction.java:
##########
@@ -132,7 +126,20 @@ private Dataset<FileInfo> reachableFileDS(TableMetadata metadata) {
}
private DeleteReachableFiles.Result deleteFiles(Iterator<FileInfo> files) {
- DeleteSummary summary = deleteFiles(deleteExecutorService, deleteFunc, files);
+ DeleteSummary summary;
+ if (deleteFunc == null && io instanceof SupportsBulkOperations) {
+ summary = deleteFiles((SupportsBulkOperations) io, files);
+ } else {
+
Review Comment:
I actually meant an empty line after `DeleteSummary` var but formatting here is up to you.
I like the new pattern.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122130931
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -239,19 +233,43 @@ private String jobDesc() {
return String.format("Deleting orphan files (%s) from %s", optionsAsString, table.name());
}
+ private void deleteFiles(SupportsBulkOperations io, List<String> paths) {
+ try {
+ io.deleteFiles(paths);
+ LOG.info("Deleted {} files using bulk deletes", paths.size());
+ } catch (BulkDeletionFailureException e) {
+ int deletedFilesCount = paths.size() - e.numberFailedObjects();
+ LOG.warn("Deleted only {} of {} files using bulk deletes", deletedFilesCount, paths.size());
+ }
+ }
+
private DeleteOrphanFiles.Result doExecute() {
Dataset<FileURI> actualFileIdentDS = actualFileIdentDS();
Dataset<FileURI> validFileIdentDS = validFileIdentDS();
List<String> orphanFiles =
findOrphanFiles(spark(), actualFileIdentDS, validFileIdentDS, prefixMismatchMode);
- Tasks.foreach(orphanFiles)
- .noRetry()
- .executeWith(deleteExecutorService)
- .suppressFailureWhenFinished()
- .onFailure((file, exc) -> LOG.warn("Failed to delete file: {}", file, exc))
- .run(deleteFunc::accept);
+ if (deleteFunc == null && table.io() instanceof SupportsBulkOperations) {
Review Comment:
This is the new pattern
```java
if (bulk)
Bulk
else {
if no custom delete
table.io:: delete
If custom delete
custom Delete
}
```
This logic is repeated in all of the actions
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122678306
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -239,19 +233,43 @@ private String jobDesc() {
return String.format("Deleting orphan files (%s) from %s", optionsAsString, table.name());
}
+ private void deleteFiles(SupportsBulkOperations io, List<String> paths) {
+ try {
+ io.deleteFiles(paths);
+ LOG.info("Deleted {} files using bulk deletes", paths.size());
+ } catch (BulkDeletionFailureException e) {
+ int deletedFilesCount = paths.size() - e.numberFailedObjects();
+ LOG.warn("Deleted only {} of {} files using bulk deletes", deletedFilesCount, paths.size());
+ }
+ }
+
private DeleteOrphanFiles.Result doExecute() {
Dataset<FileURI> actualFileIdentDS = actualFileIdentDS();
Dataset<FileURI> validFileIdentDS = validFileIdentDS();
List<String> orphanFiles =
findOrphanFiles(spark(), actualFileIdentDS, validFileIdentDS, prefixMismatchMode);
- Tasks.foreach(orphanFiles)
- .noRetry()
- .executeWith(deleteExecutorService)
- .suppressFailureWhenFinished()
- .onFailure((file, exc) -> LOG.warn("Failed to delete file: {}", file, exc))
- .run(deleteFunc::accept);
+ if (deleteFunc == null && table.io() instanceof SupportsBulkOperations) {
+ deleteFiles((SupportsBulkOperations) table.io(), orphanFiles);
+ } else {
+
+ Tasks.Builder<String> deleteTasks =
+ Tasks.foreach(orphanFiles)
+ .noRetry()
+ .executeWith(deleteExecutorService)
+ .suppressFailureWhenFinished()
+ .onFailure((file, exc) -> LOG.warn("Failed to delete file: {}", file, exc));
+
+ if (deleteFunc == null) {
+ LOG.info(
Review Comment:
We don't override `toString()` in `FileIO` classes. Did you mean `table.io().getClass().getName()`?
Applies to other two places in this PR. Also, what about using a shorter message to stay on one line?
```
LOG.info("Table IO {} only supports non-bulk deletes", table.io().getClass().getName());
```
Or something like that.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#issuecomment-1452169417
Thanks @amogh-jahagirdar , @dramaticlly and @aokolnychyi I'll merge when tests pass. I'll do the Backport Pr's after my subsurface talk.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1095010538
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -85,6 +88,7 @@
private static final Logger LOG = LoggerFactory.getLogger(BaseSparkAction.class);
private static final AtomicInteger JOB_COUNTER = new AtomicInteger();
private static final int DELETE_NUM_RETRIES = 3;
+ private static final int DELETE_GROUP_SIZE = 100000;
Review Comment:
This should just be paths, so 100K 1Kb Strings would still only be 100MB.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1095234465
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -80,9 +84,16 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param executorService the service to use
* @return this for method chaining
+ * @deprecated All deletes should be performed using the bulk delete api if available. Use FileIO
+ * specific parallelism controls to adjust bulk delete concurrency within that api.
*/
+ @Deprecated
DeleteOrphanFiles executeDeleteWith(ExecutorService executorService);
+ default DeleteOrphanFiles deleteBulkWith(Consumer<Iterable<String>> deleteFunc) {
Review Comment:
My worry is that the current implementation breaks the behavior in an incompatible way: it checks if the table IO supports bulk deletes and if yes, ignores `deleteWith`. It will break those dry-run scenarios if the table is configured with a non-bulk IO and the user passes a custom delete function.
Can you elaborate a bit on why this is needed? Why not just NOT go through the bulk delete API if a custom delete function is provided?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1095234465
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -80,9 +84,16 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param executorService the service to use
* @return this for method chaining
+ * @deprecated All deletes should be performed using the bulk delete api if available. Use FileIO
+ * specific parallelism controls to adjust bulk delete concurrency within that api.
*/
+ @Deprecated
DeleteOrphanFiles executeDeleteWith(ExecutorService executorService);
+ default DeleteOrphanFiles deleteBulkWith(Consumer<Iterable<String>> deleteFunc) {
Review Comment:
My worry is that the current implementation breaks the behavior in an incompatible way: it checks if the table IO supports bulk deletes and if yes, ignores `deleteWith`. It will break those dry-run scenarios if the table is configured with a non-bulk IO and the user passes a custom delete function.
Can you elaborate a bit on why this is needed? Why not just go through the non-bulk delete API if a custom delete function is provided?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1095013112
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -253,6 +257,39 @@ protected DeleteSummary deleteFiles(
return summary;
}
+ protected DeleteSummary deleteFiles(
+ Consumer<Iterable<String>> bulkDeleteFunc, Iterator<FileInfo> files) {
+ DeleteSummary summary = new DeleteSummary();
+ Iterator<List<FileInfo>> groupedIterator = Iterators.partition(files, DELETE_GROUP_SIZE);
+
+ Tasks.foreach(groupedIterator)
+ .suppressFailureWhenFinished()
+ .run(
+ fileList -> {
+ Map<String, List<FileInfo>> filesByType =
+ fileList.stream().collect(Collectors.groupingBy(FileInfo::getType));
+ filesByType.entrySet().stream()
+ .forEach(
+ entry -> {
+ List<String> pathsToDelete =
Review Comment:
Sure I wasn't aware of this class
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1095010538
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -85,6 +88,7 @@
private static final Logger LOG = LoggerFactory.getLogger(BaseSparkAction.class);
private static final AtomicInteger JOB_COUNTER = new AtomicInteger();
private static final int DELETE_NUM_RETRIES = 3;
+ private static final int DELETE_GROUP_SIZE = 100000;
Review Comment:
This should just be paths, so 100K 1Kb Strings would still only be 100MB. A 1 kb string could be between (in very extreme cases 250 characters to 1000 characters)
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1095231341
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -67,7 +67,11 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param deleteFunc a function that will be called to delete files
* @return this for method chaining
+ * @deprecated Deletes are now performed in bulk see {@link #deleteBulkWith(Consumer)}. This
Review Comment:
An empty delete function is also used for dry runs to analyze candidates prior to their physical removal. We can also collect go the data frame route but we offer `deleteWith` in a number of APIs, so I don't see a big problem.
> For example, if I have multiple actions running at the same time, the coordination of requests to S3 or whatnot should happen at the FileIO, otherwise the user needs to manually be determining the correct values for each of their possible concurrent jobs.
We should have a way to configure this global default in `FileIO` and it is a correct place to do it. My point is that I don't see harm allowing certain operations to override that default, especially if it makes the implementation simpler.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1095236241
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -67,7 +67,11 @@ public interface DeleteOrphanFiles extends Action<DeleteOrphanFiles, DeleteOrpha
*
* @param deleteFunc a function that will be called to delete files
* @return this for method chaining
+ * @deprecated Deletes are now performed in bulk see {@link #deleteBulkWith(Consumer)}. This
Review Comment:
My primary worry is the complexity in the code caused by the deprecation.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122055847
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteReachableFilesSparkAction.java:
##########
@@ -132,7 +135,20 @@ private Dataset<FileInfo> reachableFileDS(TableMetadata metadata) {
}
private DeleteReachableFiles.Result deleteFiles(Iterator<FileInfo> files) {
- DeleteSummary summary = deleteFiles(deleteExecutorService, deleteFunc, files);
+ DeleteSummary summary;
+ if (deleteFunc != null || !(io instanceof SupportsBulkOperations)) {
+ if (deleteFunc == null) {
+ LOG.info("Table IO does not support Bulk Operations. Using non-bulk deletes.");
Review Comment:
changed this and made it so the same messages appear in all three actions
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a diff in pull request #6682: Bulk delete
Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.
aokolnychyi commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1122666684
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +165,45 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
+ .suppressFailureWhenFinished()
+ .onFailure(
+ (f, e) -> {
+ LOG.error("Failure during bulk delete on file: {} ", f, e);
+ failureCount.incrementAndGet();
Review Comment:
You are right, we overlooked it while reviewing another PR. I like it more. I'll update `SparkCleanupUtil` to follow this patter 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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1120393853
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -18,27 +18,42 @@
*/
package org.apache.iceberg.hadoop;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.Iterator;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.BulkDeletionFailureException;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.FileInfo;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.SupportsBulkOperations;
import org.apache.iceberg.io.SupportsPrefixOperations;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Streams;
import org.apache.iceberg.util.SerializableMap;
import org.apache.iceberg.util.SerializableSupplier;
+import org.apache.iceberg.util.Tasks;
+import org.apache.iceberg.util.ThreadPools;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-public class HadoopFileIO implements FileIO, HadoopConfigurable, SupportsPrefixOperations {
+public class HadoopFileIO
+ implements FileIO, HadoopConfigurable, SupportsPrefixOperations, SupportsBulkOperations {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HadoopFileIO.class);
+ private static final String DELETE_FILE_PARALLELISM = "iceberg.hadoop.delete_file_parallelism";
Review Comment:
I don't think we have any others ... I looked for anything with "hadoop" in it and didn't find any, I can switch to -
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on a diff in pull request #6682: Bulk delete
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #6682:
URL: https://github.com/apache/iceberg/pull/6682#discussion_r1091266052
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java:
##########
@@ -149,6 +164,45 @@ public void deletePrefix(String prefix) {
}
}
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) throws BulkDeletionFailureException {
+ AtomicInteger failureCount = new AtomicInteger(0);
+ Tasks.foreach(pathsToDelete)
+ .executeWith(executorService())
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
Review Comment:
For now I just wanted to copy the old logic but I agree we can do much better than 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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org