Spark 3.2: Add `compareWithFileList` param to DeleteOrphanFiles action #4503

bijanhoule · 2022-04-05T18:06:55Z

This allows the user to specify an existing table of actual files for the DeleteOrphanFiles action, skipping the directory listing.

This can be used in cases where a cloud storage provider-specific implementation might be more efficient than a recursive FileSystem listing (e.g. leveraging S3 Inventory list for S3 to create a temp table).

…rocedure This allows the user to specify an existing table of actual files for the DeleteOrphanFiles action, skipping the directory listing.

rdblue · 2022-04-05T21:59:41Z

api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java

+  /**
+   * Passes a table which contains the list of actual files in the table. This skips the directory listing - any
+   * files in the actualFilesTable provided which are not found in table metadata will be deleted. Not compatible
+   * with `location` or `older_than` arguments - this assumes that the provided table of actual files has been


What if the table had a timestamp? For example, if the table has a modified_time then we could apply the older_than filter fairly easily rather than making the user handle it.

Adding modified_time sounds like a good idea as ignoring older_than is dangerous. Files just written but not yet committed could be cleaned up, corrupting the table.

I would even consider making it required. How many people will filter the locations manually before calling the action?

rdblue · 2022-04-05T22:06:17Z

api/src/main/java/org/apache/iceberg/actions/DeleteOrphanFiles.java

+   * @param tableName the table containing the actual files dataset.  Should have a single `file_path` string column
+   * @return this for method chaining
+   */
+  default DeleteOrphanFiles actualFilesTable(String tableName) {


I think it is a good idea to be able to use a temporary table for the stored procedure, but for the action API I would expect to be able to pass in a Dataset of some kind. Dataset<String> would work, and we could also expose a couple of Java classes to make Dataset<FileLocationAndModifiedTime> and Dataset<FileLocation> work.

I think that we can also make what's happening here a bit more clear in the API by naming the method something like compareToFileList(Dataset<String>):

Dataset<String> files = ... SparkActions.get() .deleteOrphanFiles(catalog.loadTable(identifier)) .compareToFileList(files) .execute();

I guess the problem is that the Action API is engine agnostic so we cannot refer to Dataset. We considered offering engine-specific actions at some point but that idea was discarded as it would overcomplicate everything.

I think we should move this into the Spark-specific API then.

@rdblue, you mean exposing this method in BaseDeleteOrphanFilesSparkAction similarly to expire in BaseExpireSnapshotsSparkAction or introduce a Spark-specific interface?

It is a bummer as this functionality can be useful for all engines.

We may be able to do this later, but I think the most natural way to do this for the Actions API is to use a Dataset. We can take another look at this later, too. Maybe we should parameterize this method so that you can specify what collection you want. Flink could use DataStream and Spark could use Dataset.

rdblue · 2022-04-05T22:10:53Z

...v3.2/spark/src/main/java/org/apache/iceberg/spark/procedures/RemoveOrphanFilesProcedure.java

@@ -49,7 +49,8 @@
      ProcedureParameter.optional("older_than", DataTypes.TimestampType),
      ProcedureParameter.optional("location", DataTypes.StringType),
      ProcedureParameter.optional("dry_run", DataTypes.BooleanType),
-      ProcedureParameter.optional("max_concurrent_deletes", DataTypes.IntegerType)
+      ProcedureParameter.optional("max_concurrent_deletes", DataTypes.IntegerType),
+      ProcedureParameter.optional("actual_files_table", DataTypes.StringType)


What about renaming this to file_list_view or listed_files_view? I think that we want some identifier that matches the name of the action method; that's why I'm thinking file_list or listed_files that are similar to compareToFileList. But we also want to make it clear that this is a table or temporary view. I think it's more likely that the DF is registered as a temp view rather than stored in a table, so I'd probably use the suffix _view.

rdblue · 2022-04-05T22:16:41Z

...k/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction.java

@@ -150,10 +150,20 @@ public void testDryRun() throws IOException, InterruptedException {
    Assert.assertEquals("Action should find 1 file", invalidFiles, result2.orphanFileLocations());
    Assert.assertTrue("Invalid file should be present", fs.exists(new Path(invalidFiles.get(0))));

+    String actualFilesTableName = "actualFilesTable";
+    spark.createDataset(allFiles, Encoders.STRING()).toDF("file_path").createOrReplaceTempView(actualFilesTableName);


I think that this test case should run after the actual orphan file delete. Otherwise, listing produces the same result and we're not actually verifying that we get a different result from the new option.

rdblue · 2022-04-05T22:18:06Z

Thanks, @bijanhoule! This looks close to being ready, but I think we should make a few minor modifications to make the actions API a bit more intuitive. I would expect that to accept a Dataset rather than requiring temp table registration.

aokolnychyi · 2022-04-06T22:30:11Z

I'd be curious to take a look as well tomorrow.

kbendick · 2022-04-07T15:33:07Z

...k/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction.java

+    Assert.assertEquals("Action should find 1 file", invalidFiles, result3.orphanFileLocations());
+    Assert.assertTrue("Invalid file should be present", fs.exists(new Path(invalidFiles.get(0))));
+
+    DeleteOrphanFiles.Result result4 = actions.deleteOrphanFiles(table)


If we add support for things like older_than, we should probably make a separate test case for the actual files table option so we can exercise various scenarios focused on that one area.

Agreed. We usually want to add separate test cases anyway.

kbendick · 2022-04-07T15:36:22Z

...v3.2/spark/src/main/java/org/apache/iceberg/spark/procedures/RemoveOrphanFilesProcedure.java

+            tableWithActualFilePaths == null || (location == null && olderThanMillis == null),
+            "actual_files_table cannot be used with `location` or `older_than`"


Nit: I think this is over-indented. Code continued on the next line should be indented 4 spaces over from its starting point.

That said, it does line up with the argument on the line above it.

+1, I think it uses 8 instead of 4. The line above is a mistake too.

aokolnychyi · 2022-04-08T23:57:05Z

...2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseDeleteOrphanFilesSparkAction.java

@@ -163,7 +194,7 @@ private String jobDesc() {
    Dataset<Row> validContentFileDF = buildValidContentFileDF(table);
    Dataset<Row> validMetadataFileDF = buildValidMetadataFileDF(table);
    Dataset<Row> validFileDF = validContentFileDF.union(validMetadataFileDF);
-    Dataset<Row> actualFileDF = buildActualFileDF();
+    Dataset<Row> actualFileDF = this.providedActualFilesDF == null ? buildActualFileDF() : providedActualFilesDF;


nit: redundant this

bijanhoule · 2022-04-12T18:51:30Z

Thanks for the feedback all! I'll take another pass and adjust a few things:

move to the spark-specific api and use a dataset input rather than temp table
add support for modified_time and enforce older_than. Originally the thought was to keep this out for simplicity, but I agree it's probably safer to just make it a requirement.
separate out test cases and add coverage for the above

- action: rename actualFilesTable -> compareWithFileList - procedure: rename actual_files_table -> file_list_view - interface change: remove from DeleteOrphanFiles, add to BaseDeleteOrphanFilesSparkAction - apply location and last_modified filtering to compareWithFileList - refactor test cases

bijanhoule · 2022-04-20T19:03:13Z

...2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseDeleteOrphanFilesSparkAction.java

+    if (location != null) {
+      files = files.filter(files.col(FILE_PATH).startsWith(location));
+    }


@aokolnychyi -- re: #4503 (comment) does this startsWith location filtering seem reasonable to you for a short-term solution? I know there's some discussion about making some larger changes here.

bijanhoule · 2022-04-20T19:04:47Z

FYI, updated this PR based on feedback above:

action: rename actualFilesTable -> compareWithFileList
procedure: rename actual_files_table -> file_list_view
interface change: remove from DeleteOrphanFiles, add to BaseDeleteOrphanFilesSparkAction
apply location and last_modified filtering to compareWithFileList
refactor test cases

rdblue · 2022-04-24T20:47:37Z

...k/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction.java

+    Assert.assertEquals("Should be 1 invalid file", 1, invalidFiles.size());
+
+    // sleep for 1 second to ensure files will be old enough
+    Thread.sleep(1000);


This should use waitUntilAfter rather than sleeping.

rdblue · 2022-04-24T20:49:44Z

This looks great. Thanks for adding this, @bijanhoule!

…e#4503)

…11) * Spark 3.2: Add compareWithFileList to DeleteOrphanFiles action (apache#4503) * Add regex replace support in remove orphan file action and procedure to allow for handling alternative filesystem schemes.

Spark 3.2: Add actualFilesTable param to DeleteOrphanFiles action/p…

88d210e

…rocedure This allows the user to specify an existing table of actual files for the DeleteOrphanFiles action, skipping the directory listing.

github-actions bot added API spark labels Apr 5, 2022

rdblue reviewed Apr 5, 2022

View reviewed changes

kbendick reviewed Apr 7, 2022

View reviewed changes

aokolnychyi reviewed Apr 8, 2022

View reviewed changes

bijanhoule commented Apr 20, 2022

View reviewed changes

bijanhoule changed the title ~~Spark 3.2: Add actualFilesTable param to DeleteOrphanFiles action~~ Spark 3.2: Add compareWithFileList param to DeleteOrphanFiles action Apr 21, 2022

rdblue reviewed Apr 24, 2022

View reviewed changes

rdblue approved these changes Apr 24, 2022

View reviewed changes

rdblue merged commit 17c0ff7 into apache:master Apr 24, 2022

SinghAsDev pushed a commit to SinghAsDev/iceberg that referenced this pull request Jun 23, 2022

Spark 3.2: Add compareWithFileList to DeleteOrphanFiles action (apach…

51cf5e1

…e#4503)

bitsondatadev mentioned this pull request Jul 26, 2023

Docs: document the compareWithFileList parameter #8155

Closed

anuragmantri mentioned this pull request Jun 5, 2024

Improve remove_orphan_files performance by using "inventory listing" #10426

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Spark 3.2: Add `compareWithFileList` param to DeleteOrphanFiles action #4503

Spark 3.2: Add `compareWithFileList` param to DeleteOrphanFiles action #4503

bijanhoule commented Apr 5, 2022 •

edited

Loading

rdblue Apr 5, 2022

aokolnychyi Apr 8, 2022

aokolnychyi Apr 8, 2022

rdblue Apr 5, 2022

aokolnychyi Apr 8, 2022 •

edited

Loading

rdblue Apr 10, 2022

aokolnychyi Apr 13, 2022 •

edited

Loading

rdblue Apr 13, 2022

rdblue Apr 5, 2022

rdblue Apr 5, 2022

rdblue commented Apr 5, 2022

aokolnychyi commented Apr 6, 2022

kbendick Apr 7, 2022

rdblue Apr 10, 2022

kbendick Apr 7, 2022

aokolnychyi Apr 8, 2022

aokolnychyi Apr 8, 2022

bijanhoule commented Apr 12, 2022

bijanhoule Apr 20, 2022

bijanhoule commented Apr 20, 2022

rdblue Apr 24, 2022

rdblue commented Apr 24, 2022

		tableWithActualFilePaths == null \|\| (location == null && olderThanMillis == null),
		"actual_files_table cannot be used with `location` or `older_than`"

Spark 3.2: Add compareWithFileList param to DeleteOrphanFiles action #4503

Spark 3.2: Add compareWithFileList param to DeleteOrphanFiles action #4503

Conversation

bijanhoule commented Apr 5, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi Apr 8, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi Apr 13, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rdblue commented Apr 5, 2022

aokolnychyi commented Apr 6, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bijanhoule commented Apr 12, 2022

Choose a reason for hiding this comment

bijanhoule commented Apr 20, 2022

Choose a reason for hiding this comment

rdblue commented Apr 24, 2022

Spark 3.2: Add `compareWithFileList` param to DeleteOrphanFiles action #4503

Spark 3.2: Add `compareWithFileList` param to DeleteOrphanFiles action #4503

bijanhoule commented Apr 5, 2022 •

edited

Loading

aokolnychyi Apr 8, 2022 •

edited

Loading

aokolnychyi Apr 13, 2022 •

edited

Loading