You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "sririshindra (via GitHub)" <gi...@apache.org> on 2023/03/16 15:07:22 UTC
[GitHub] [iceberg] sririshindra opened a new pull request, #7127: [Core][Spark] Improve DeleteOrphanFiles action to return additional details of deleted orphan files
sririshindra opened a new pull request, #7127:
URL: https://github.com/apache/iceberg/pull/7127
### What changes are proposed in this PR?
---
This PR adds a new Interface OrphanFileStatus that indicates if an orphan file was deleted or not. In cases of failure during file deletion, It provides a reference to the encountered exception.
With this additional information, a user can choose to record this failure and retry the deletion process in a controlled fashion in the future.
### Why are the changes are needed?
During the execution of the DeleteOrphanFiles spark action or remove_orphan_files SQL procedure, a failure encountered during deletion of the orphan file is not bubbled up to the user nor is there any indication of the failure in the returned result.
The return value of the Spark action is Iterable and the SQL procedure simply displays a list of orphan files in a table. With this limited information, the only way for the user to know if the orphan file was deleted or not is to
- grep the logs for the warning
- for each of the locations returned in the iterable, query the cloud storage for its existence
One can re-run the expensive delete action to delete the files that failed during the previous delete action run, however, that is not very desirable since re-listing the dir contents to identify the orphan files would result in duplicate work plus wastage of precious API calls. One of the common causes of delete failure in the public cloud is hitting the API quotas and unnecessary re-runs of the delete action can lead to a temporary denial of access to other workloads accessing the same storage resource.
Providing information about the failed delete files gives users the ability to take appropriate action based on their use case.
## Output
### Before this change
#### DryRun => true/false
```
scala> sql("CALL spark_catalog.system.remove_orphan_files(table => 't1', older_than => TIMESTAMP '2022-05-21 00:00:00.000', dry_run => true)").show(false)
+--------------------------------------------------------------------------------------------------------------------------+
|orphan_file_location |
+--------------------------------------------------------------------------------------------------------------------------+
|file:/tmp/iceberg_warehouse/default/t1/non_table_files/part-00000-705370fd-ea3e-4ae7-8b40-c0362b0ba7df-c000.snappy.parquet|
|file:/tmp/iceberg_warehouse/default/t1/non_table_files/part-00001-705370fd-ea3e-4ae7-8b40-c0362b0ba7df-c000.snappy.parquet|
+--------------------------------------------------------------------------------------------------------------------------+
```
### After this change
#### DryRun => true
```
scala> sql("CALL spark_catalog.system.remove_orphan_files(table => 't1', older_than => TIMESTAMP '2022-05-21 00:00:00.000', dry_run => true)").show(false)
+--------------------------------------------------------------------------------------------------------------------------+-------+-------------+
|orphan_file_location |deleted|error_message|
+--------------------------------------------------------------------------------------------------------------------------+-------+-------------+
|file:/tmp/iceberg_warehouse/default/t1/non_table_files/part-00000-8251ec5f-dd9b-4754-8e68-2cdd01e66e56-c000.snappy.parquet|false |null |
|file:/tmp/iceberg_warehouse/default/t1/non_table_files/part-00001-8251ec5f-dd9b-4754-8e68-2cdd01e66e56-c000.snappy.parquet|false |null |
+--------------------------------------------------------------------------------------------------------------------------+-------+-------------+
```
#### DryRun => false
```
scala> sql("CALL spark_catalog.system.remove_orphan_files(table => 't1', older_than => TIMESTAMP '2022-05-21 00:00:00.000', dry_run => false)").show(false)
+--------------------------------------------------------------------------------------------------------------------------+-------+-------------+
|orphan_file_location |deleted|error_message|
+--------------------------------------------------------------------------------------------------------------------------+-------+-------------+
|file:/tmp/iceberg_warehouse/default/t1/non_table_files/part-00000-8251ec5f-dd9b-4754-8e68-2cdd01e66e56-c000.snappy.parquet|true |null |
|file:/tmp/iceberg_warehouse/default/t1/non_table_files/part-00001-8251ec5f-dd9b-4754-8e68-2cdd01e66e56-c000.snappy.parquet|true |null |
+--------------------------------------------------------------------------------------------------------------------------+-------+-------------+
```
#### DryRun => false, simulating failure during delete
```
+--------------------------------------------------------------------------------------------------------------------------+-------+---------------------------------------+
|orphan_file_location |deleted|error_message |
+--------------------------------------------------------------------------------------------------------------------------+-------+---------------------------------------+
|file:/tmp/iceberg_warehouse/default/t1/non_table_files/part-00000-615125af-bfbb-4279-b22b-559dee4f0b13-c000.snappy.parquet|true |null |
|file:/tmp/iceberg_warehouse/default/t1/non_table_files/part-00001-705370fd-ea3e-4ae7-8b40-c0362b0ba7df-c000.snappy.parquet|false |simulating failure during file deletion|
+--------------------------------------------------------------------------------------------------------------------------+-------+---------------------------------------+
```
## Code Credit
This PR is a reimplementation of #4817 published by @sumeetgajjar for spark3.2. As @sumeetgajjar mentioned in #4817 he moved on to other projects and I am taking this up. I tried cherry-picking Sumeet's changes but I had a hard time resolving the conflicts, It was easier to reimplement the change on master directly. I am not sure how to give credit to Sumeet's work. @sumeetgajjar maybe you can push a dummy commit on top of this branch so that if and when this PR gets merged, it will show your name as well in the commit message.
In the original PR(#4817) @RussellSpitzer and @aokolnychyi said that they can take a look at it after [#4652](https://github.com/apache/iceberg/pull/4652) is merged. @RussellSpitzer @aokolnychyi @szehon-ho Could you please take a look at this PR as [#4652](https://github.com/apache/iceberg/pull/4652) is now merged. Apart from the changes @sumeetgajjar did in #4817 I implemented this feature for bulk deletes as well. I also implemented it for spark3.3 rather than spark3.2. If this changes goes in I can raise a separate PRs for spark3.2 and spark3.1 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] sririshindra commented on pull request #7127: [Core][Spark] Improve DeleteOrphanFiles action to return additional details of deleted orphan files
Posted by "sririshindra (via GitHub)" <gi...@apache.org>.
sririshindra commented on PR #7127:
URL: https://github.com/apache/iceberg/pull/7127#issuecomment-1492784798
> > One of the common causes of delete failure in the public cloud is hitting the API quotas and unnecessary re-runs of the delete action
>
> Not really. What size of pages are you issuing delete requests to S3? Each object, even in a bulk delete, is one Write IOP; if you send a full 1000 entry list then under certain conditions the sleep and retry of that request is it's own thundering herd. Since [HADOOP-16823](https://issues.apache.org/jira/browse/HADOOP-16823) we've had a default page size of 200 and nobody complains about 503-triggered deletion failures, even when partition rebalancing operations massively reduce IO capacity . If you are seeing problems the S3A connector then complain. If you are seeing it in your own code -why not fix it there rather than expose the failure to apps?
Hi Steve, I am partially summarizing our offline conversation here. Maybe the statement that "causes of delete failure in the public cloud is hitting the API quotas" is wrong, but I think this PR still adds value.
When a remove orphan files procedure is called, currently a list of files that were supposed to be deleted are displayed, but that procedure doesn't give nay information about whether those files were in fact deleted. This PR simply displays the status of the delete file along with the delete file itself. If a failure occurs for some reason, the exception message is also displayed.
--
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] steveloughran commented on pull request #7127: [Core][Spark] Improve DeleteOrphanFiles action to return additional details of deleted orphan files
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on PR #7127:
URL: https://github.com/apache/iceberg/pull/7127#issuecomment-1474837901
> One of the common causes of delete failure in the public cloud is hitting the API quotas and unnecessary re-runs of the delete action
Not really. What size of pages are you issuing delete requests to S3? Each object, even in a bulk delete, is one Write IOP; if you send a full 1000 entry list then under certain conditions the sleep and retry of that request is it's own thundering herd. Since [HADOOP-16823](https://issues.apache.org/jira/browse/HADOOP-16823) we've had a default page size of 200 and nobody complains about 503-triggered deletion failures, even when partition rebalancing operations massively reduce IO capacity . If you are seeing problems the S3A connector then complain. If you are seeing it in your own code -why not fix it there rather than expose the failure to apps?
--
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
Re: [PR] [Core][Spark] Improve DeleteOrphanFiles action to return additional details of deleted orphan files [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #7127:
URL: https://github.com/apache/iceberg/pull/7127#issuecomment-2313786007
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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
Re: [PR] [Core][Spark] Improve DeleteOrphanFiles action to return additional details of deleted orphan files [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #7127: [Core][Spark] Improve DeleteOrphanFiles action to return additional details of deleted orphan files
URL: https://github.com/apache/iceberg/pull/7127
--
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] manuzhang commented on a diff in pull request #7127: [Core][Spark] Improve DeleteOrphanFiles action to return additional details of deleted orphan files
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on code in PR #7127:
URL: https://github.com/apache/iceberg/pull/7127#discussion_r1222859059
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -144,7 +146,25 @@ default DeleteOrphanFiles equalAuthorities(Map<String, String> newEqualAuthoriti
@Value.Immutable
interface Result {
/** Returns locations of orphan files. */
- Iterable<String> orphanFileLocations();
+ default Iterable<String> orphanFileLocations() {
Review Comment:
Can we add a summary like number of removed orphan files?
--
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
Re: [PR] [Core][Spark] Improve DeleteOrphanFiles action to return additional details of deleted orphan files [iceberg]
Posted by "steveloughran (via GitHub)" <gi...@apache.org>.
steveloughran commented on code in PR #7127:
URL: https://github.com/apache/iceberg/pull/7127#discussion_r1734723004
##########
api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java:
##########
@@ -144,7 +146,25 @@ default DeleteOrphanFiles equalAuthorities(Map<String, String> newEqualAuthoriti
@Value.Immutable
interface Result {
/** Returns locations of orphan files. */
- Iterable<String> orphanFileLocations();
+ default Iterable<String> orphanFileLocations() {
+ return StreamSupport.stream(statuses().spliterator(), false)
+ .map(DeleteOrphanFiles.OrphanFileStatus::location)
+ .collect(Collectors.toList());
+ }
+
+ Iterable<OrphanFileStatus> statuses();
+ }
+
+ interface OrphanFileStatus {
Review Comment:
prefer `DeleteOrphanFileOutcome` as `FileStatus` is broadly used elsewhere, same for implementation
##########
spark/v2.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction.java:
##########
@@ -301,6 +301,69 @@ public void orphanedFileRemovedWithParallelTasks() throws InterruptedException,
"remove-orphan-0", "remove-orphan-1", "remove-orphan-2", "remove-orphan-3"));
Assert.assertEquals("Should delete 4 files", 4, deletedFiles.size());
+ assertDeletedStatusAndFailureCause(result.statuses());
+ }
+
+ @Test
+ public void testOrphanFileDeleteThrowsException() {
+ Table table = TABLES.create(SCHEMA, SPEC, Maps.newHashMap(), tableLocation);
+
+ List<ThreeColumnRecord> records1 =
+ Lists.newArrayList(new ThreeColumnRecord(1, "AAAAAAAAAA", "AAAA"));
+ Dataset<Row> df1 = spark.createDataFrame(records1, ThreeColumnRecord.class).coalesce(1);
+
+ // original append
+ df1.select("c1", "c2", "c3").write().format("iceberg").mode("append").save(tableLocation);
+
+ List<ThreeColumnRecord> records2 =
+ Lists.newArrayList(new ThreeColumnRecord(2, "AAAAAAAAAA", "AAAA"));
+ Dataset<Row> df2 = spark.createDataFrame(records2, ThreeColumnRecord.class).coalesce(1);
+
+ // dynamic partition overwrite
+ df2.select("c1", "c2", "c3").write().format("iceberg").mode("overwrite").save(tableLocation);
+
+ // second append
+ df2.select("c1", "c2", "c3").write().format("iceberg").mode("append").save(tableLocation);
+
+ df2.coalesce(1).write().mode("append").parquet(tableLocation + "/data");
+ df2.coalesce(1).write().mode("append").parquet(tableLocation + "/data/c2_trunc=AA");
+ df2.coalesce(1).write().mode("append").parquet(tableLocation + "/data/c2_trunc=AA/c3=AAAA");
+ df2.coalesce(1).write().mode("append").parquet(tableLocation + "/data/invalid/invalid");
+
+ waitUntilAfter(System.currentTimeMillis());
+
+ String locationSubstringForException = "invalid";
+ DeleteOrphanFiles.Result result =
+ SparkActions.get()
+ .deleteOrphanFiles(table)
+ .olderThan(System.currentTimeMillis())
+ .deleteWith(
+ file -> {
+ if (file.contains(locationSubstringForException)) {
+ throw new RuntimeException("simulating failure during file deletion");
+ }
+ table.io().deleteFile(file);
+ })
+ .execute();
+
+ Assert.assertEquals("Should delete 4 files", 4, Iterables.size(result.statuses()));
Review Comment:
use assertj assertions on collections here
##########
spark/v2.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction.java:
##########
@@ -301,6 +301,69 @@ public void orphanedFileRemovedWithParallelTasks() throws InterruptedException,
"remove-orphan-0", "remove-orphan-1", "remove-orphan-2", "remove-orphan-3"));
Assert.assertEquals("Should delete 4 files", 4, deletedFiles.size());
+ assertDeletedStatusAndFailureCause(result.statuses());
+ }
+
+ @Test
+ public void testOrphanFileDeleteThrowsException() {
+ Table table = TABLES.create(SCHEMA, SPEC, Maps.newHashMap(), tableLocation);
+
+ List<ThreeColumnRecord> records1 =
+ Lists.newArrayList(new ThreeColumnRecord(1, "AAAAAAAAAA", "AAAA"));
+ Dataset<Row> df1 = spark.createDataFrame(records1, ThreeColumnRecord.class).coalesce(1);
+
+ // original append
+ df1.select("c1", "c2", "c3").write().format("iceberg").mode("append").save(tableLocation);
+
+ List<ThreeColumnRecord> records2 =
+ Lists.newArrayList(new ThreeColumnRecord(2, "AAAAAAAAAA", "AAAA"));
+ Dataset<Row> df2 = spark.createDataFrame(records2, ThreeColumnRecord.class).coalesce(1);
+
+ // dynamic partition overwrite
+ df2.select("c1", "c2", "c3").write().format("iceberg").mode("overwrite").save(tableLocation);
+
+ // second append
+ df2.select("c1", "c2", "c3").write().format("iceberg").mode("append").save(tableLocation);
+
+ df2.coalesce(1).write().mode("append").parquet(tableLocation + "/data");
+ df2.coalesce(1).write().mode("append").parquet(tableLocation + "/data/c2_trunc=AA");
+ df2.coalesce(1).write().mode("append").parquet(tableLocation + "/data/c2_trunc=AA/c3=AAAA");
+ df2.coalesce(1).write().mode("append").parquet(tableLocation + "/data/invalid/invalid");
+
+ waitUntilAfter(System.currentTimeMillis());
+
+ String locationSubstringForException = "invalid";
+ DeleteOrphanFiles.Result result =
+ SparkActions.get()
+ .deleteOrphanFiles(table)
+ .olderThan(System.currentTimeMillis())
Review Comment:
I'd propose actually makin the older than value "now + 1000" to deal with filesystem clock resolution quirks. I know, nobody will be testing on FAT32 with 2s granularity, but its still good to avoid brittle tests
##########
spark/v2.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction.java:
##########
@@ -301,6 +301,69 @@ public void orphanedFileRemovedWithParallelTasks() throws InterruptedException,
"remove-orphan-0", "remove-orphan-1", "remove-orphan-2", "remove-orphan-3"));
Assert.assertEquals("Should delete 4 files", 4, deletedFiles.size());
+ assertDeletedStatusAndFailureCause(result.statuses());
+ }
+
+ @Test
+ public void testOrphanFileDeleteThrowsException() {
+ Table table = TABLES.create(SCHEMA, SPEC, Maps.newHashMap(), tableLocation);
+
+ List<ThreeColumnRecord> records1 =
+ Lists.newArrayList(new ThreeColumnRecord(1, "AAAAAAAAAA", "AAAA"));
+ Dataset<Row> df1 = spark.createDataFrame(records1, ThreeColumnRecord.class).coalesce(1);
+
+ // original append
+ df1.select("c1", "c2", "c3").write().format("iceberg").mode("append").save(tableLocation);
+
+ List<ThreeColumnRecord> records2 =
+ Lists.newArrayList(new ThreeColumnRecord(2, "AAAAAAAAAA", "AAAA"));
+ Dataset<Row> df2 = spark.createDataFrame(records2, ThreeColumnRecord.class).coalesce(1);
+
+ // dynamic partition overwrite
+ df2.select("c1", "c2", "c3").write().format("iceberg").mode("overwrite").save(tableLocation);
+
+ // second append
+ df2.select("c1", "c2", "c3").write().format("iceberg").mode("append").save(tableLocation);
+
+ df2.coalesce(1).write().mode("append").parquet(tableLocation + "/data");
+ df2.coalesce(1).write().mode("append").parquet(tableLocation + "/data/c2_trunc=AA");
+ df2.coalesce(1).write().mode("append").parquet(tableLocation + "/data/c2_trunc=AA/c3=AAAA");
+ df2.coalesce(1).write().mode("append").parquet(tableLocation + "/data/invalid/invalid");
+
+ waitUntilAfter(System.currentTimeMillis());
Review Comment:
pull out to a variable and reuse on L339
##########
spark/v2.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestDeleteReachableFilesAction.java:
##########
@@ -286,11 +286,11 @@ public void testIgnoreMetadataFilesNotFound() {
DeleteOrphanFiles.Result result =
sparkActions().deleteOrphanFiles(table).olderThan(System.currentTimeMillis()).execute();
- Assert.assertEquals("Should delete 1 file", 1, Iterables.size(result.orphanFileLocations()));
+ Assert.assertEquals("Should delete 1 file", 1, Iterables.size(result.statuses()));
Review Comment:
as iceberg uses assertj in tests, use it for the new/updated assertions for better error reporting
--
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
Re: [PR] [Core][Spark] Improve DeleteOrphanFiles action to return additional details of deleted orphan files [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #7127:
URL: https://github.com/apache/iceberg/pull/7127#issuecomment-2408257986
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any 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