Skip to content

Spark 3.5: scan-based dangling delete removal (backport of apache/iceberg#15727 / port of #4) #9

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
davkim wants to merge 1 commit into
base: 1.11.0-affirm-patch
Choose a base branch
from
Open

Spark 3.5: scan-based dangling delete removal (backport of apache/iceberg#15727 / port of #4) #9

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -205,12 +205,12 @@ private DeleteFile findDV(long seq, DataFile dataFile) {
}

DeleteFile dv = dvByPath.get(dataFile.location());
if (dv != null) {
ValidationException.check(
dv.dataSequenceNumber() >= seq,
"DV data sequence number (%s) must be greater than or equal to data file sequence number (%s)",
dv.dataSequenceNumber(),
seq);
// AFFIRM: from apache/iceberg#15727. A DV whose data sequence number is below the data file's
// simply does not apply, so return null instead of throwing. Required by the scan-based
// dangling-delete detection: it plans a full scan, and throwing here would abort that scan on
// exactly the stale DVs the cleanup exists to remove.
if (dv != null && dv.dataSequenceNumber() < seq) {
return null;
}
return dv;
}
Expand Down
4 changes: 1 addition & 3 deletions core/src/test/java/org/apache/iceberg/DeleteFileIndexTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -792,8 +792,6 @@ public void testInvalidDVSequenceNumber() {
DeleteFile dv = withDataSequenceNumber(1, newDV(FILE_A));
List<DeleteFile> dvs = Collections.singletonList(dv);
DeleteFileIndex index = DeleteFileIndex.builderFor(dvs).specsById(table.specs()).build();
assertThatThrownBy(() -> index.forDataFile(2, FILE_A))
.isInstanceOf(ValidationException.class)
.hasMessageContaining("must be greater than or equal to data file sequence number");
assertThat(index.forDataFile(2, FILE_A)).isEmpty();
}
}
180 changes: 65 additions & 115 deletions ...park/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,45 +18,32 @@
*/
package org.apache.iceberg.spark.actions;

import static org.apache.spark.sql.functions.col;
import static org.apache.spark.sql.functions.min;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.MetadataTableType;
import org.apache.iceberg.Partitioning;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.ManifestFiles;
import org.apache.iceberg.ManifestReader;
import org.apache.iceberg.RewriteFiles;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableScan;
import org.apache.iceberg.actions.ImmutableRemoveDanglingDeleteFiles;
import org.apache.iceberg.actions.RemoveDanglingDeleteFiles;
import org.apache.iceberg.exceptions.RuntimeIOException;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.spark.JobGroupInfo;
import org.apache.iceberg.spark.SparkDeleteFile;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.DeleteFileSet;
import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.StructType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
* An action that removes dangling delete files from the current snapshot. A delete file is dangling
* if its deletes no longer applies to any live data files.
*
* <p>The following dangling delete files are removed:
*
* <ul>
* <li>Position delete files with a data sequence number less than that of any data file in the
* same partition
* <li>Equality delete files with a data sequence number less than or equal to that of any data
* file in the same partition
* </ul>
*/
class RemoveDanglingDeletesSparkAction
extends BaseSnapshotUpdateSparkAction<RemoveDanglingDeletesSparkAction>
Expand All @@ -77,13 +64,12 @@ protected RemoveDanglingDeletesSparkAction self() {

@Override
public Result execute() {
if (table.specs().size() == 1 && table.spec().isUnpartitioned()) {
// ManifestFilterManager already performs this table-wide delete on each commit
return ImmutableRemoveDanglingDeleteFiles.Result.builder()
.removedDeleteFiles(Collections.emptyList())
.build();
}

// AFFIRM: dropped the 1.11-native "skip if single-spec unpartitioned" fast path here (assumed
// ManifestFilterManager already drops table-wide dangling deletes on every commit). It doesn't
// hold for deletes committed standalone via newRowDelta() -- exactly the CDC write pattern that
// produces dangling deletes in prod (see PR description) -- and TestRemoveDanglingDeleteAction
// #testUnpartitionedTable exercises precisely that gap. Skipping detection here would silently
// reintroduce a blind spot on unpartitioned tables.
String desc = String.format("Removing dangling delete files in %s", table.name());
JobGroupInfo info = newJobGroupInfo("REMOVE-DELETES", desc);
return withJobGroupInfo(info, this::doExecute);
Expand All @@ -93,7 +79,6 @@ Result doExecute() {
RewriteFiles rewriteFiles = table.newRewrite();
DeleteFileSet danglingDeletes = DeleteFileSet.create();
danglingDeletes.addAll(findDanglingDeletes());
danglingDeletes.addAll(findDanglingDvs());

for (DeleteFile deleteFile : danglingDeletes) {
LOG.debug("Removing dangling delete file {}", deleteFile.location());
Expand All @@ -113,94 +98,59 @@ Result doExecute() {
* Dangling delete files can be identified with following steps
*
* <ol>
* <li>Group data files by partition keys and find the minimum data sequence number in each
* group.
* <li>Left outer join delete files with partition-grouped data files on partition keys.
* <li>Find dangling deletes by comparing each delete file's sequence number to its partition's
* minimum data sequence number.
* <li>Collect results row to driver and use {@link SparkDeleteFile SparkDeleteFile} to wrap
* rows to valid delete files
* <li>Make a full scan and collect delete files from all file tasks.
* <li>Collect all delete file entries skipping files from the previous step.
* </ol>
*/
private List<DeleteFile> findDanglingDeletes() {
Dataset<Row> minSequenceNumberByPartition =
loadMetadataTable(table, MetadataTableType.ENTRIES)
// find live data files
.filter("data_file.content == 0 AND status < 2")
.selectExpr(
"data_file.partition as partition",
"data_file.spec_id as spec_id",
"sequence_number")
.groupBy("partition", "spec_id")
.agg(min("sequence_number"))
.toDF("grouped_partition", "grouped_spec_id", "min_data_sequence_number");

Dataset<Row> deleteEntries =
loadMetadataTable(table, MetadataTableType.ENTRIES)
// find live delete files
.filter("data_file.content != 0 AND status < 2");

Column joinOnPartition =
deleteEntries
.col("data_file.spec_id")
.equalTo(minSequenceNumberByPartition.col("grouped_spec_id"))
.and(
deleteEntries
.col("data_file.partition")
.equalTo(minSequenceNumberByPartition.col("grouped_partition")));

Column filterOnDanglingDeletes =
col("min_data_sequence_number")
// delete files without any data files in partition
.isNull()
// position delete files without any applicable data files in partition
.or(
col("data_file.content")
.equalTo("1")
.and(col("sequence_number").$less(col("min_data_sequence_number"))))
// equality delete files without any applicable data files in the partition
.or(
col("data_file.content")
.equalTo("2")
.and(col("sequence_number").$less$eq(col("min_data_sequence_number"))));

Dataset<Row> danglingDeletes =
deleteEntries
.join(minSequenceNumberByPartition, joinOnPartition, "left")
.filter(filterOnDanglingDeletes)
.select("data_file.*");
return danglingDeletes.collectAsList().stream()
// map on driver because SparkDeleteFile is not serializable
.map(row -> deleteFileWrapper(danglingDeletes.schema(), row))
.collect(Collectors.toList());
}
TableScan scan = table.newScan();

// AFFIRM: resolve the snapshot once and guard against null. Upstream calls scan.snapshot()
// inline below, which NPEs on a table with no current snapshot (reachable via a direct
// SparkActions.get().removeDanglingDeleteFiles(table) call).
Snapshot snapshot = scan.snapshot();
if (snapshot == null) {
return Collections.emptyList();
}

private List<DeleteFile> findDanglingDvs() {
Dataset<Row> dvs =
loadMetadataTable(table, MetadataTableType.DELETE_FILES)
.where(col("file_format").equalTo(FileFormat.PUFFIN.name()));
Dataset<Row> dataFiles = loadMetadataTable(table, MetadataTableType.DATA_FILES);

// a DV not pointing to a valid data file path is implicitly a dangling delete
List<Row> danglingDvs =
dvs.join(
dataFiles,
dvs.col("referenced_data_file").equalTo(dataFiles.col("file_path")),
"leftouter")
.filter(dataFiles.col("file_path").isNull())
.select(dvs.col("*"))
.collectAsList();
return danglingDvs.stream()
// map on driver because SparkDeleteFile is not serializable
.map(row -> deleteFileWrapper(dvs.schema(), row))
.collect(Collectors.toList());
}
// AFFIRM: pin the scan itself to the resolved snapshot ID rather than relying on
// TableScan#snapshot() being called before TableScan#planFiles() below. TableScan#snapshot()
// re-resolves table().currentSnapshot() on every call (SnapshotScan.java), so without this
// pin, a commit landing between the two calls could make planFiles() plan against a *newer*
// snapshot than the one whose delete manifests we read further down -- silently correct today
// only because of statement order, and one reordering away from misclassifying a still-live
// delete file as dangling.
scan = scan.useSnapshot(snapshot.snapshotId());

DeleteFileSet deletes = DeleteFileSet.create();
try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
for (FileScanTask task : tasks) {
deletes.addAll(task.deletes());
}
} catch (IOException e) {
throw new RuntimeIOException(e, "Failed to scan: %s", scan);
}

DeleteFileSet danglingDeletes = DeleteFileSet.create();
for (ManifestFile manifest : snapshot.deleteManifests(table.io())) {
try (ManifestReader<DeleteFile> reader =
ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) {
for (DeleteFile deleteFile : reader) {
if (!deletes.contains(deleteFile)) {
// AFFIRM: drop column stats before retaining. DeleteFileSet keys on location,
// contentOffset and contentSizeInBytes, all of which copyWithoutStats() preserves,
// and nothing on the rewrite path reads a metrics map -- so holding lower/upper
// bounds for every dangling delete is pure driver heap.
danglingDeletes.add(deleteFile.copyWithoutStats());
}
}
} catch (IOException e) {
throw new RuntimeIOException(e, "Failed to read manifest: %s", manifest);
}
}

private DeleteFile deleteFileWrapper(StructType sparkFileType, Row row) {
int specId = row.getInt(row.fieldIndex("spec_id"));
Types.StructType combinedFileType = DataFile.getType(Partitioning.partitionType(table));
// Set correct spec id
Types.StructType projection = DataFile.getType(table.specs().get(specId).partitionType());
return new SparkDeleteFile(combinedFileType, projection, sparkFileType).wrap(row);
// AFFIRM: upstream uses Stream.toList(), which is Java 16+. Iceberg's Flink/Spark modules
// compile with options.release = 11, so this is a hard compile error there.
return Lists.newArrayList(danglingDeletes);
}
}
34 changes: 29 additions & 5 deletions ...3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,6 +184,17 @@ public RewriteDataFiles.Result execute() {

if (plan.totalGroupCount() == 0) {
LOG.info("Nothing found to rewrite in {}", table.name());
// AFFIRM: still run dangling-delete removal on this path. `delete-file-threshold` normally
// guarantees a non-empty group set, EXCEPT in the exact stuck state this feature exists to
// escape: once the only remaining delete files are already dangling, no data file qualifies
// for rewrite, so totalGroupCount() == 0 and returning here would leave them in place
// forever. Not covered by apache/iceberg#15727, which does not touch this file; the hole is
// still present upstream on main.
if (removeDanglingDeletes) {
return removeDanglingDeletesAndBuild(
ImmutableRewriteDataFiles.Result.builder().rewriteResults(ImmutableList.of()).build());
}

return EMPTY_RESULT;
}

Expand All @@ -194,11 +205,7 @@ public RewriteDataFiles.Result execute() {
ImmutableRewriteDataFiles.Result result = resultBuilder.build();

if (removeDanglingDeletes) {
RemoveDanglingDeletesSparkAction action =
new RemoveDanglingDeletesSparkAction(spark(), table);
int removedDeleteFiles = Iterables.size(action.execute().removedDeleteFiles());
return result.withRemovedDeleteFilesCount(
result.removedDeleteFilesCount() + removedDeleteFiles);
return removeDanglingDeletesAndBuild(result);
}

return result;
Expand All @@ -218,6 +225,23 @@ private void init(long startingSnapshotId) {
validateAndInitOptions();
}

/**
* AFFIRM: extracted so the dangling-delete removal can also run on the "nothing to rewrite" early
* return path above, not just after a successful data rewrite. Takes the already-built Result,
* not a Builder: on 1.11, a rewrite can already report a non-zero removedDeleteFilesCount on its
* own (e.g. orphaned DVs dropped as part of the rewrite commit itself -- see
* TestRewriteDataFilesAction#testRemoveDangledPositionDeletesPartitionEvolution's v3 case), and
* result.removedDeleteFilesCount() is a sum over rewriteResults() computed lazily on first read.
* Setting a raw Builder field here would silently overwrite that baseline instead of adding to
* it.
*/
private ImmutableRewriteDataFiles.Result removeDanglingDeletesAndBuild(
ImmutableRewriteDataFiles.Result result) {
RemoveDanglingDeletesSparkAction action = new RemoveDanglingDeletesSparkAction(spark(), table);
int removedCount = Iterables.size(action.execute().removedDeleteFiles());
return result.withRemovedDeleteFilesCount(result.removedDeleteFilesCount() + removedCount);
}

@VisibleForTesting
RewriteFileGroup rewriteFiles(
FileRewritePlan<FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup> plan,
Expand Down
Loading
Loading