Spark 3.3: Add RemoveDanglingDeletes action #6581

szehon-ho · 2023-01-13T16:45:39Z

This adds an action to cleanup dangling (invalid) DeleteFiles that may otherwise keep getting carried over with the table's current snapshot and which may negatively impact read performance.

The problem and design doc is here: https://docs.google.com/document/d/11d-cIUR_89kRsMmWnEoxXGZCvp7L4TUmPJqUC60zB5M/edit#

In a nutshell, the current table-wide mechanism is crude and may miss many instances of aging off DeleteFiles, even after they become invalid after compaction. This implements a spark action to perform a partition-by-partition removal using the same rules.

szehon-ho · 2023-01-13T21:34:51Z

...3/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java

+            .toDF("partition", "spec_id", "min_data_sequence_number");
+
+    // Dangling position delete files
+    Column joinCond =


I can't reproduce it on my own machine, but on build machine I get the error if I try the cleaner: Dataset.join(ds, Seq columns)

error: no suitable method found for join(Dataset<Row>,Buffer<String>) see: https://github.com/apache/iceberg/actions/runs/3913210764/jobs/6688831726

So I make an explicit join condition, like RewriteManifestFileSparkAction

RussellSpitzer · 2023-01-17T20:49:01Z

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

+import org.apache.iceberg.DeleteFile;
+
+/**
+ * An action that removes dangling delete files from the current snapshot. A delete file is dangling


I think we need a note that this removes delete files only if they don't apply to any non-expired datafile. Just to make it clear that this isn't just about "live" delete files

RussellSpitzer · 2023-01-17T20:54:22Z

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

+ * <p>The following dangling delete files are removed:
+ *
+ * <ul>
+ *   <li>Position delete files with a sequence number less than that of any data file in the same


This is more of a technical detail right? Not sure we need it in the java doc

RussellSpitzer · 2023-01-17T21:14:44Z

core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesActionResult.java

+
+public class RemoveDanglingDeleteFilesActionResult implements RemoveDanglingDeleteFiles.Result {
+
+  private static final RemoveDanglingDeleteFilesActionResult EMPTY =


Do we really need this? Seems like in the code we can just add

new RemoveDanglingDeleteFileActionResult(Collections.emptyList())

at
https://github.com/apache/iceberg/pull/6581/files#diff-afa01360c6badf264da62497ccb1186cdd841cae290ab15b0a2b086cc100b9caR79

Doesn't seem like we really are making that many of these objects

RussellSpitzer · 2023-01-17T21:23:40Z

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java

+    if (useCaching) {
+      reusableDS = ds.cache();
+    } else {
+      int parallelism = SQLConf.get().numShufflePartitions();


This is a bit internal to spark, probably fine but i'm not a fan of touching this over spark.conf()

RussellSpitzer · 2023-01-17T21:24:42Z

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java

+    } else {
+      int parallelism = SQLConf.get().numShufflePartitions();
+      reusableDS =
+          ds.repartition(parallelism).map((MapFunction<T, T>) value -> value, ds.exprEnc());


why do we repartition here? (and encode)

szehon-ho · 2023-01-18T19:17:43Z

Chatting with @aokolnychyi , @RussellSpitzer , to do an analysis on when this can be useful.

There will be two types of operations that can remove delete files:

Operation	Cost	File Type	Description
RemoveDanglingDeletes (this one)	Metadata-Only, cost will be like querying files/partition table	Both	Removes position deletes with sequence number less than that of the min sequence number of all data files in each partition
RewritePositionDeletes (to be developed)	Data-operation, need to read/write all concerned delete files	Position only (Equality Deletes will need to be converted to PositionDeletes)	Read all position delete files satisfying given filter, write them back out , filtering out position delete entries that refer to data files that no longer exist

Analysis: RemoveDanglingDeleteFiles is cheaper and simpler, and can work across both types of files out of the box. However, to get it to exactly work, we need the following conditions: RewriteDataFiles being run with:

Filter that includes entire partition(s)
All data files in the partition with delete files gets rewritten, ie any of these:
- rewrite-all=true
- delete-file-threshold=1
- All data files happen to meet the criteria of rewrite without these flags.
'use-starting-sequence-number' needs to be false. This is to properly identify old delete files as invalid using sequence number rule. This is only needed for position-deletes, as equality-deletes are not applied to equivalent sequence number.

Note RemoveDanglingDeleteFiles can still remove some delete files if these conditions are not met, but just it may not do so for all dangling delete files, because an old data file (one with a low sequence number) not rewritten in a partition will prevent delete files from getting removed.

So Im open to whether there is a good use-case of this. One idea is to bundle this with RewriteDataFiles, and if trigger optimistically if these conditions are met, or trigger in any case in hopes it will remove delete files as, as its relatively cheap.

Otherwise, the complete solution (all to be developed) would be:
For position deletes, run RewritePositionDeletes across all partitions
For equality deletes, run ConvertToPosDeletes, then RewritePositionDeletes across all partitions.

amogh-jahagirdar · 2023-01-22T23:03:56Z

...3/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java

+                "data_file.spec_id as spec_id",
+                "data_file.file_path as file_path",
+                "data_file.content as content",
+                "data_file.file_size_in_bytes as file_size_in_bytes",


Sorry if it's a naive question, do we need to project file_size_in_bytes for this action?

Nvm, we need to serialize the result to a delete file so it's needed.

amogh-jahagirdar · 2023-01-22T23:06:38Z

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

+
+  /** The action result that contains a summary of the execution. */
+  interface Result {
+    /** Removes representation of removed delete files. */


Should this comment just be "List of removed delete files"?

amogh-jahagirdar · 2023-01-22T23:35:11Z

...3/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java

+      for (int i = 0; i < partitionRow.length(); i++) {
+        partition.set(i, partitionRow.get(i));
+      }


Style nit: newline after the loop

amogh-jahagirdar · 2023-01-22T23:50:52Z

@szehon-ho Thanks for the detailed analysis. On the surface it does make sense to combine with existing compaction mechanisms like RewriteDataFiles if the metadata only RemoveDanglingDeleteFiles is cheap and especially if many users are already running RewriteDataFiles periodically anyways. Is there a case we're missing where users would just want to run remove dangling delete files separately? I can't really think of a case.

singhpk234 · 2023-01-30T01:48:32Z

...3/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java

+    }
+
+    String desc = String.format("Remove dangling delete files for %s", table.name());
+    JobGroupInfo info = newJobGroupInfo("REWRITE-MANIFESTS", desc);


[minor] should we also mention it's rewriting manifest post removing danling delete files in the name ?

Good point.

singhpk234 · 2023-01-30T19:21:23Z

...3/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java

+    Dataset<Row> minDataSeqNumberPerPartition =
+        entries
+            .filter("content == 0") // data files
+            .groupBy("partition", "spec_id")
+            .agg(min("sequence_number"))
+            .toDF("partition", "spec_id", "min_data_sequence_number");


can we cache this df this as well, since we need this in two joins below ?

Yea I am not so sure, it seems cache is a bit controversial. I think the general approach in Iceberg spark action is not to use cache?

manuzhang · 2023-02-01T15:08:56Z

One idea is to bundle this with RewriteDataFiles, and if trigger optimistically if these conditions are met, or trigger in any case in hopes it will remove delete files as, as its relatively cheap.

@szehon-ho Have you considered partial-progress.enabled=true? Currently, delete files can't be removed when some file groups (partitions) are compacted and committed.

szehon-ho · 2023-02-01T17:31:29Z

Thanks all for reviews. After thinking, this will be good to have, but as the complete 100% position delete removal will be done via minor compaction of delete files (first part of the effort is here: #6365, more to come). Will work on that first.

@amogh-jahagirdar yea I initially thought it would be useful as a standalone action, but some chats with @aokolnychyi and maybe its a bit too tricky for user to know when to run it. Still open though.

@manuzhang do you mean integrating this commit with existing partial commits? Initially envisoning a separate commit altogether at the end of RewriteDataFiles.

manuzhang · 2023-02-02T01:25:22Z

@szehon-ho yes, integrating removing delete files per partition with partial commits. When new position delete files are generated during RewriteDataFiles with partial commits, all following commits will fail and no delete files will be removed at the end.

eric666666 · 2023-02-07T03:03:09Z

Is this action currently available? Is there any usage documentation.

eric666666 · 2023-02-07T03:07:18Z

Is this action currently available? Is there any usage documentation.

We have already applied the v2 table to production. It is too uncomfortable to delete the delete file when the snapshot expires. When will this Pr be merged into the master.

zinking · 2023-05-15T03:04:02Z

I am +1 for integrating these into existing Rewrite Action.

szehon-ho · 2023-05-15T18:18:15Z

Hi, can you guys please check #7389 , it is already merged. It is a way to do it manually for now (and also optimize position delete), we can come back later to integrate this action into rewriteDataFiles automatically if we need. There's also some ongoing post-pr improvements thats also in review like #7582 and #7572

zinking · 2024-02-03T12:04:34Z

...3/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java

+
+      return builder
+          .withPath(path)
+          .withPartition(partition)


the partition data here doesn't comply with the builder spec, so it will build out incorrect partition data.

is it a bit of overwork to rebuild the deleteFile, cant we just delete using paths? a lot of overhead could be avoided.

usually the copy needs to assert the specs are the same

public Builder copy(DeleteFile toCopy) { if (isPartitioned) { Preconditions.checkState( specId == toCopy.specId(), "Cannot copy a DeleteFile with a different spec"); this.partitionData = DataFiles.copyPartitionData(spec, toCopy.partition(), partitionData); }

it would be great if we can delete a deleteFile by filePaths only, but from mergingSnapshotProducer, looks like we never make it to work for path based deletion

iceberg/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java

Lines 193 to 202 in d32abe8

protected void delete(DeleteFile file) {

deleteFilterManager.delete(file);

}

/** Add a specific data path to be deleted in the new snapshot. */

protected void delete(CharSequence path) {

// this is an old call that never worked for delete files and can only be used to remove data

// files.

filterManager.delete(path);

}

Back-port of apache#6581

github-actions · 2024-08-24T00:13:25Z

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 [email protected] list. Thank you for your contributions.

github-actions · 2024-09-12T00:14:24Z

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.

Spark 3.3: Add RemoveDanglingDeletes action

5fc91bd

github-actions bot added API core spark labels Jan 13, 2023

szehon-ho force-pushed the remove_dangling_delete branch from 523408f to 8f59b01 Compare January 13, 2023 20:44

Try to solve no suitable method error

88a05db

szehon-ho force-pushed the remove_dangling_delete branch from 8f59b01 to 88a05db Compare January 13, 2023 21:03

szehon-ho commented Jan 13, 2023

View reviewed changes

RussellSpitzer reviewed Jan 17, 2023

View reviewed changes

amogh-jahagirdar reviewed Jan 22, 2023

View reviewed changes

singhpk234 reviewed Jan 30, 2023

View reviewed changes

wangtaohz mentioned this pull request Aug 9, 2023

[Feature]: Remove the Independent DeleteFiles for the Iceberg Format Table apache/amoro#1628

Closed

2 tasks

szehon-ho mentioned this pull request Jan 22, 2024

Core: rewrite should drop delete files by data sequence number partition wise #9454

Closed

zinking reviewed Feb 3, 2024

View reviewed changes

dramaticlly mentioned this pull request Feb 13, 2024

Core, Spark: Remove dangling deletes as part of RewriteDataFilesAction #9724

Merged

manuzhang added a commit to manuzhang/iceberg that referenced this pull request Jul 9, 2024

[HADP-52306] Spark 3.4: Add RemoveDanglingDeleteFilesProcedure

9eb8e13

Back-port of apache#6581

github-actions bot added the stale label Aug 24, 2024

github-actions bot closed this Sep 12, 2024


		public class RemoveDanglingDeleteFilesActionResult implements RemoveDanglingDeleteFiles.Result {

		private static final RemoveDanglingDeleteFilesActionResult EMPTY =

	protected void delete(DeleteFile file) {
	deleteFilterManager.delete(file);
	}

	/** Add a specific data path to be deleted in the new snapshot. */
	protected void delete(CharSequence path) {
	// this is an old call that never worked for delete files and can only be used to remove data
	// files.
	filterManager.delete(path);
	}

Spark 3.3: Add RemoveDanglingDeletes action #6581

Spark 3.3: Add RemoveDanglingDeletes action #6581

Uh oh!

Conversation

szehon-ho commented Jan 13, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RussellSpitzer Jan 17, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RussellSpitzer Jan 17, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szehon-ho commented Jan 18, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

amogh-jahagirdar commented Jan 22, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

manuzhang commented Feb 1, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

szehon-ho commented Feb 1, 2023

Uh oh!

manuzhang commented Feb 2, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

eric666666 commented Feb 7, 2023

Uh oh!

eric666666 commented Feb 7, 2023

Uh oh!

zinking commented May 15, 2023

Uh oh!

szehon-ho commented May 15, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions bot commented Aug 24, 2024

Uh oh!

github-actions bot commented Sep 12, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

8 participants

szehon-ho commented Jan 13, 2023 •

edited

Loading

RussellSpitzer Jan 17, 2023 •

edited

Loading

RussellSpitzer Jan 17, 2023 •

edited

Loading

szehon-ho commented Jan 18, 2023 •

edited

Loading

amogh-jahagirdar commented Jan 22, 2023 •

edited

Loading

manuzhang commented Feb 1, 2023 •

edited

Loading

manuzhang commented Feb 2, 2023 •

edited

Loading

szehon-ho commented May 15, 2023 •

edited

Loading