You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "kinolaev (via GitHub)" <gi...@apache.org> on 2026/03/22 23:55:50 UTC
[PR] Core, Spark: scan based remove dangling delete action [iceberg]
kinolaev opened a new pull request, #15727:
URL: https://github.com/apache/iceberg/pull/15727
The current implementation of RemoveDanglingDeletesSparkAction keeps equality delete files if their lower/upper bounds don't overlap with lower/upper bounds of data files with a lower sequence number.
These delete files are dangling because they are always skipped during scans. When they add up, a scan can slow down.
This PR collects all delete files from a full scan into a set and then removes all delete files that aren't in the set. This approach guarantees that all dangling delete files are deleted regardless of their type.
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5554548545
Thanks @szehon-ho for the review! I moved the scan to the executors, parallelized delete manifest reading, and implemented the anti-join using `JavaRDD.subtractByKey()`, so the driver now only collects the dangling deletes (d1558460e3a52ac06d75f0e4cd20b1de6c54e519).
My estimate is ~400 bytes per `DeleteFile` without stats, so a million delete files would need around 400MB of memory.
The day I first ran this in production, it removed ~250k of ~400k delete files, with peak driver memory consumption at 4.5GB. Over the last week, it removed ~200k of ~300k delete files every night, consuming 3-4GB on average. I'm still running the old version, where delete files are stored with stats. My driver has 16GB of memory, so I suspect the JVM just doesn't bother collecting garbage aggressively, which is probably why the actual numbers are higher than the estimates.
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r4014489369
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of(
+ "content",
+ "file_path",
+ "file_format",
+ "partition",
+ "record_count",
+ "file_size_in_bytes",
+ "key_metadata",
+ "split_offsets",
+ "equality_ids",
+ "sort_order_id",
+ "first_row_id",
+ "referenced_data_file",
+ "content_offset",
+ "content_size_in_bytes");
+ private static final Result EMPTY_RESULT =
+ ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(ImmutableList.of())
+ .build();
+
+ private final Table table;
+ private final BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ this.findDanglingDeletes = this::findDanglingDeletes;
+ }
+
+ public RemoveDanglingDeleteFilesAction(
+ Table table,
+ BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes) {
+ this.table = table;
+ this.findDanglingDeletes = findDanglingDeletes;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ protected Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Snapshot snapshot = table.snapshot(branch);
+ if (snapshot == null) {
+ Preconditions.checkArgument(
+ branch.equals(SnapshotRef.MAIN_BRANCH),
+ "Cannot remove dangling delete files from branch %s: branch does not exist",
+ branch);
+ return EMPTY_RESULT;
+ }
+
+ List<ManifestFile> deleteManifests = snapshot.deleteManifests(table.io());
+ if (deleteManifests.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ Collection<DeleteFile> danglingDeletes =
+ findDanglingDeletes.apply(snapshot.snapshotId(), deleteManifests);
+ if (danglingDeletes.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ RewriteFiles rewriteFiles = table.newRewrite().validateFromSnapshot(snapshot.snapshotId());
+ for (DeleteFile deleteFile : danglingDeletes) {
+ LOG.debug("Removing dangling delete file {}", deleteFile.location());
+ rewriteFiles.deleteFile(deleteFile);
+ }
+
+ commit(rewriteFiles.toBranch(branch));
+
+ return ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(danglingDeletes)
+ .build();
+ }
+
+ /**
+ * Dangling delete files can be identified with following steps
+ *
+ * <ol>
+ * <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(
+ long snapshotId, List<ManifestFile> deleteManifests) {
+ Set<DeleteFileKey> referencedKeys = Sets.newHashSet();
+ TableScan scan = table.newScan().useSnapshot(snapshotId);
+ try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
+ for (FileScanTask task : tasks) {
+ for (DeleteFile deleteFile : task.deletes()) {
+ referencedKeys.add(new DeleteFileKey(deleteFile));
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to scan: %s", scan);
+ }
+
+ List<DeleteFile> danglingDeletes = Lists.newArrayList();
+ for (ManifestFile manifest : deleteManifests) {
+ try (ManifestReader<DeleteFile> reader =
+ readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (!referencedKeys.contains(new DeleteFileKey(deleteFile))) {
+ danglingDeletes.add(deleteFile);
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to read manifest: %s", manifest);
+ }
+ }
+
+ return danglingDeletes;
+ }
+
+ public static ManifestReader<DeleteFile> readDeleteManifest(
+ ManifestFile manifest, FileIO io, Map<Integer, PartitionSpec> specsById) {
+ return ManifestFiles.readDeleteManifest(manifest, io, specsById).select(DELETE_COLUMNS);
+ }
+
+ public record DeleteFileKey(String location, Long contentOffset, Long contentSizeInBytes)
Review Comment:
Hi @manuzhang, thanks for catching this! Could you provide a few more details on how to reproduce the issue? Also, if you already have an idea of what Kryo doesn't support, that would be helpful - I haven't used Kryo before.
--
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: scan based remove dangling delete action [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3994174181
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java:
##########
@@ -18,189 +18,124 @@
*/
package org.apache.iceberg.spark.actions;
-import static org.apache.spark.sql.functions.col;
-import static org.apache.spark.sql.functions.min;
-
-import java.util.Collections;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Iterator;
import java.util.List;
-import java.util.stream.Collectors;
-import org.apache.iceberg.DataFile;
+import java.util.stream.StreamSupport;
import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.MetadataTableType;
-import org.apache.iceberg.Partitioning;
-import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
import org.apache.iceberg.Table;
-import org.apache.iceberg.actions.ImmutableRemoveDanglingDeleteFiles;
+import org.apache.iceberg.TableScan;
import org.apache.iceberg.actions.RemoveDanglingDeleteFiles;
-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.iceberg.actions.RemoveDanglingDeleteFilesAction;
+import org.apache.iceberg.actions.RemoveDanglingDeleteFilesAction.DeleteFileKey;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.io.ClosingIterator;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.source.SerializableTableWithSize;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.types.StructType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import scala.Tuple2;
/**
* 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>
implements RemoveDanglingDeleteFiles {
- private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeletesSparkAction.class);
private final Table table;
+ private final RemoveDanglingDeleteFilesAction action;
protected RemoveDanglingDeletesSparkAction(SparkSession spark, Table table) {
super(spark);
this.table = table;
+ this.action = new RemoveDanglingDeleteFilesAction(table, this::findDanglingDeletes);
}
@Override
protected RemoveDanglingDeletesSparkAction self() {
return this;
}
+ public RemoveDanglingDeletesSparkAction toBranch(String targetBranch) {
+ action.toBranch(targetBranch);
+ return this;
+ }
+
@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();
- }
-
+ commitSummary().forEach(action::set);
String desc = String.format("Removing dangling delete files in %s", table.name());
- JobGroupInfo info = newJobGroupInfo("REMOVE-DELETES", desc);
- return withJobGroupInfo(info, this::doExecute);
+ return withJobGroupInfo(newJobGroupInfo("REMOVE-DELETES", desc), action::execute);
}
- Result doExecute() {
- RewriteFiles rewriteFiles = table.newRewrite();
- DeleteFileSet danglingDeletes = DeleteFileSet.create();
- danglingDeletes.addAll(findDanglingDeletes());
- danglingDeletes.addAll(findDanglingDvs());
+ private List<DeleteFile> findDanglingDeletes(
+ long snapshotId, List<ManifestFile> deleteManifests) {
+ Broadcast<Table> tableBroadcast =
+ sparkContext().broadcast(SerializableTableWithSize.copyOf(table));
+
+ JavaPairRDD<DeleteFileKey, Void> referencedKeys =
+ sparkContext()
+ .parallelize(ImmutableList.of(snapshotId), 1)
+ .flatMap(
+ snapshot -> {
+ TableScan scan = tableBroadcast.value().newScan().useSnapshot(snapshot);
+ return new ClosingIterator<>(new DeleteFileKeyIterator(scan.planFiles()));
+ })
+ .mapToPair(key -> new Tuple2<>(key, (Void) null));
+
+ JavaPairRDD<DeleteFileKey, DeleteFile> allDeletes =
+ sparkContext()
+ .parallelize(
+ deleteManifests.stream().map(ManifestFileBean::fromManifest).toList(),
+ deleteManifests.size())
+ .flatMap(
+ manifest -> {
+ ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(
Review Comment:
Could we select all non-stat delete-file columns before iterating here? Without a projection, `ManifestReader.iterator()` materializes the column statistics before `copyWithoutStats()` discards them, adding avoidable manifest I/O and allocation.
--
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: scan based remove dangling delete action [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3974674670
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java:
##########
@@ -18,189 +18,123 @@
*/
package org.apache.iceberg.spark.actions;
-import static org.apache.spark.sql.functions.col;
-import static org.apache.spark.sql.functions.min;
-
-import java.util.Collections;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Iterator;
import java.util.List;
-import java.util.stream.Collectors;
-import org.apache.iceberg.DataFile;
+import java.util.stream.StreamSupport;
import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.MetadataTableType;
-import org.apache.iceberg.Partitioning;
-import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
-import org.apache.iceberg.actions.ImmutableRemoveDanglingDeleteFiles;
+import org.apache.iceberg.TableScan;
import org.apache.iceberg.actions.RemoveDanglingDeleteFiles;
-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.iceberg.actions.RemoveDanglingDeleteFilesAction;
+import org.apache.iceberg.actions.RemoveDanglingDeleteFilesAction.DeleteFileKey;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.io.ClosingIterator;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.source.SerializableTableWithSize;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.types.StructType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import scala.Tuple2;
/**
* 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>
implements RemoveDanglingDeleteFiles {
- private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeletesSparkAction.class);
private final Table table;
+ private final RemoveDanglingDeleteFilesAction action;
protected RemoveDanglingDeletesSparkAction(SparkSession spark, Table table) {
super(spark);
this.table = table;
+ this.action = new RemoveDanglingDeleteFilesAction(table, this::findDanglingDeletes);
}
@Override
protected RemoveDanglingDeletesSparkAction self() {
return this;
}
+ public RemoveDanglingDeletesSparkAction toBranch(String targetBranch) {
+ action.toBranch(targetBranch);
+ return this;
+ }
+
@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();
- }
-
+ commitSummary().forEach(action::set);
String desc = String.format("Removing dangling delete files in %s", table.name());
- JobGroupInfo info = newJobGroupInfo("REMOVE-DELETES", desc);
- return withJobGroupInfo(info, this::doExecute);
+ return withJobGroupInfo(newJobGroupInfo("REMOVE-DELETES", desc), action::execute);
}
- Result doExecute() {
- RewriteFiles rewriteFiles = table.newRewrite();
- DeleteFileSet danglingDeletes = DeleteFileSet.create();
- danglingDeletes.addAll(findDanglingDeletes());
- danglingDeletes.addAll(findDanglingDvs());
+ private List<DeleteFile> findDanglingDeletes(Snapshot snapshot) {
+ Broadcast<Table> tableBroadcast =
+ sparkContext().broadcast(SerializableTableWithSize.copyOf(table));
+
+ JavaPairRDD<DeleteFileKey, Void> referencedKeys =
+ sparkContext()
+ .parallelize(ImmutableList.of(snapshot.snapshotId()), 1)
+ .flatMap(
+ snapshotId -> {
+ TableScan scan = tableBroadcast.value().newScan().useSnapshot(snapshotId);
+ return new ClosingIterator<>(new DeleteFileKeyIterator(scan.planFiles()));
+ })
+ .mapToPair(key -> new Tuple2<>(key, (Void) null));
+
+ List<ManifestFileBean> deleteManifests =
+ snapshot.deleteManifests(table.io()).stream().map(ManifestFileBean::fromManifest).toList();
+ JavaPairRDD<DeleteFileKey, DeleteFile> allDeletes =
+ sparkContext()
+ .parallelize(deleteManifests, deleteManifests.size())
Review Comment:
Could we handle an empty `deleteManifests` list before creating this RDD, ideally before the broadcast and scan? Spark rejects `parallelize(..., 0)` with `IllegalArgumentException: Positive number of partitions required`, so this fails for a populated table with no delete manifests.
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java:
##########
@@ -18,189 +18,123 @@
*/
package org.apache.iceberg.spark.actions;
-import static org.apache.spark.sql.functions.col;
-import static org.apache.spark.sql.functions.min;
-
-import java.util.Collections;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Iterator;
import java.util.List;
-import java.util.stream.Collectors;
-import org.apache.iceberg.DataFile;
+import java.util.stream.StreamSupport;
import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.MetadataTableType;
-import org.apache.iceberg.Partitioning;
-import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
-import org.apache.iceberg.actions.ImmutableRemoveDanglingDeleteFiles;
+import org.apache.iceberg.TableScan;
import org.apache.iceberg.actions.RemoveDanglingDeleteFiles;
-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.iceberg.actions.RemoveDanglingDeleteFilesAction;
+import org.apache.iceberg.actions.RemoveDanglingDeleteFilesAction.DeleteFileKey;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.io.ClosingIterator;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.source.SerializableTableWithSize;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.types.StructType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import scala.Tuple2;
/**
* 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>
implements RemoveDanglingDeleteFiles {
- private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeletesSparkAction.class);
private final Table table;
+ private final RemoveDanglingDeleteFilesAction action;
protected RemoveDanglingDeletesSparkAction(SparkSession spark, Table table) {
super(spark);
this.table = table;
+ this.action = new RemoveDanglingDeleteFilesAction(table, this::findDanglingDeletes);
}
@Override
protected RemoveDanglingDeletesSparkAction self() {
return this;
}
+ public RemoveDanglingDeletesSparkAction toBranch(String targetBranch) {
+ action.toBranch(targetBranch);
+ return this;
+ }
+
@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();
- }
-
+ commitSummary().forEach(action::set);
String desc = String.format("Removing dangling delete files in %s", table.name());
- JobGroupInfo info = newJobGroupInfo("REMOVE-DELETES", desc);
- return withJobGroupInfo(info, this::doExecute);
+ return withJobGroupInfo(newJobGroupInfo("REMOVE-DELETES", desc), action::execute);
}
- Result doExecute() {
- RewriteFiles rewriteFiles = table.newRewrite();
- DeleteFileSet danglingDeletes = DeleteFileSet.create();
- danglingDeletes.addAll(findDanglingDeletes());
- danglingDeletes.addAll(findDanglingDvs());
+ private List<DeleteFile> findDanglingDeletes(Snapshot snapshot) {
+ Broadcast<Table> tableBroadcast =
+ sparkContext().broadcast(SerializableTableWithSize.copyOf(table));
+
+ JavaPairRDD<DeleteFileKey, Void> referencedKeys =
+ sparkContext()
+ .parallelize(ImmutableList.of(snapshot.snapshotId()), 1)
+ .flatMap(
+ snapshotId -> {
+ TableScan scan = tableBroadcast.value().newScan().useSnapshot(snapshotId);
+ return new ClosingIterator<>(new DeleteFileKeyIterator(scan.planFiles()));
+ })
+ .mapToPair(key -> new Tuple2<>(key, (Void) null));
+
+ List<ManifestFileBean> deleteManifests =
+ snapshot.deleteManifests(table.io()).stream().map(ManifestFileBean::fromManifest).toList();
+ JavaPairRDD<DeleteFileKey, DeleteFile> allDeletes =
+ sparkContext()
+ .parallelize(deleteManifests, deleteManifests.size())
+ .flatMap(
+ manifest -> {
+ ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(
+ manifest, tableBroadcast.value().io(), tableBroadcast.value().specs());
+ return new ClosingIterator<>(reader.iterator());
+ })
+ .mapToPair(file -> new Tuple2<>(new DeleteFileKey(file), file.copyWithoutStats()));
+
+ return allDeletes.subtractByKey(referencedKeys).values().collect();
+ }
- for (DeleteFile deleteFile : danglingDeletes) {
- LOG.debug("Removing dangling delete file {}", deleteFile.location());
- rewriteFiles.deleteFile(deleteFile);
+ private static class DeleteFileKeyIterator implements CloseableIterator<DeleteFileKey> {
+ private final Closeable closeable;
+ private final Iterator<DeleteFileKey> iterator;
+
+ DeleteFileKeyIterator(CloseableIterable<FileScanTask> tasks) {
+ this.closeable = tasks;
+ this.iterator =
+ StreamSupport.stream(tasks.spliterator(), false)
+ .flatMap(task -> task.deletes().stream())
+ .map(DeleteFileKey::new)
+ .distinct()
Review Comment:
Could we avoid performing `distinct()` inside this single scan task? It retains every referenced delete key in one executor’s heap, while removing it may significantly increase shuffle traffic when deletes apply to many file tasks. Is there a way to preserve deduplication without concentrating the full set in one executor?
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3851337932
##########
core/src/main/java/org/apache/iceberg/DeleteFileIndex.java:
##########
@@ -220,38 +222,26 @@ 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);
- validatePartitionMatch(dv, dataFile);
+ if (dv != null && (dv.dataSequenceNumber() < seq || !validatePartitionMatch(dv, dataFile))) {
+ return null;
}
return dv;
}
- private void validatePartitionMatch(DeleteFile deleteFile, DataFile dataFile) {
- ValidationException.check(
- deleteFile.specId() == dataFile.specId(),
- "Mismatched partition specs (%s, %s) for delete file %s and data file %s:"
- + " metadata is corrupted",
- deleteFile.specId(),
- dataFile.specId(),
- deleteFile.location(),
- dataFile.location());
+ private boolean validatePartitionMatch(DeleteFile deleteFile, DataFile dataFile) {
+ if (deleteFile.specId() != dataFile.specId()) {
+ // Mismatched partition specs for delete file and data file: metadata is corrupted
+ return false;
+ }
if (partitionComparatorsBySpecId != null) {
Comparator<StructLike> partitionComparator =
partitionComparatorsBySpecId.get(deleteFile.specId());
- ValidationException.check(
- partitionComparator.compare(deleteFile.partition(), dataFile.partition()) == 0,
- "Mismatched partition tuples (%s, %s) for delete file %s and data file %s:"
- + " metadata is corrupted",
- deleteFile.partition(),
- dataFile.partition(),
- deleteFile.location(),
- dataFile.location());
+ if (partitionComparator.compare(deleteFile.partition(), dataFile.partition()) != 0) {
+ // Mismatched partition tuples for delete file and data file: metadata is corrupted
+ return false;
Review Comment:
No, it doesn't, it just skips a delete file in case of partition mismatch. Check out the updated test - instead of throwing an exception, `task.deletes()` is empty. This matches the relaxation we discussed back in April.
The spec says nothing about failure, it says:
> A position delete file must be applied to a data file when all of the following are true:
> The data file's file_path is equal to the delete file's referenced_data_file if it is non-null
> The data file's partition (both spec and partition values) is equal to the delete file's partition
So, there is no spec violation here: a position delete file with a different partition won't be applied to a data file.
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3851533347
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java:
##########
@@ -113,94 +89,36 @@ 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());
- }
-
- 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);
+ TableScan scan = table.newScan();
+
+ 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);
+ }
- // 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());
- }
+ DeleteFileSet danglingDeletes = DeleteFileSet.create();
+ for (ManifestFile manifest : scan.snapshot().deleteManifests(table.io())) {
+ try (ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (!deletes.contains(deleteFile)) {
+ danglingDeletes.add(deleteFile);
Review Comment:
Maybe we can skip stats in the first place instead of making copies?
```java
ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs()).select(ImmutableList.of("file_path", "content_offset", "content_size_in_bytes"))
```
?
--
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: scan based remove dangling delete action [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5518696759
cc @szehon-ho @nastra Could you please also take a look? Thanks a lot!
--
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: scan based remove dangling delete action [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-4953431033
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5410387093
I've checked out https://github.com/Affirm/iceberg/pull/4 and found two interesting things:
- `scan.snapshot()` returns `table.currentSnapshot()` when snapshot isn't pinned, so the deleted files added after scan are deleted - fixed in 8888f7cccb03b4b9a58ee439c76f3788e91e19f2
- dangling deletes aren't removed when there is nothing to rewrite - fixed in 0d8075c6e2bd93d16574a382b1f33be86817aef4
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5620294990
Rebased and applied the changes to Spark 4.2
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3919145002
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.util.List;
+import org.apache.iceberg.DeleteFile;
+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.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.util.DeleteFileSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of("file_path", "content_offset", "content_size_in_bytes");
+
+ private final Table table;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+ private Snapshot snapshot;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ public Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ public RemoveDanglingDeleteFilesAction init() {
Review Comment:
Thank you for pointing this out - that was definitely a bad design. I just wanted to follow the same pattern as in `RewriteDataFilesSparkAction` (validate first, then wrap execution with `JobGroupInfo`), but the tradeoff clearly isn't worth it. I've moved snapshot resolution inside `execute()`.
--
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: scan based remove dangling delete action [iceberg]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r4014592728
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of(
+ "content",
+ "file_path",
+ "file_format",
+ "partition",
+ "record_count",
+ "file_size_in_bytes",
+ "key_metadata",
+ "split_offsets",
+ "equality_ids",
+ "sort_order_id",
+ "first_row_id",
+ "referenced_data_file",
+ "content_offset",
+ "content_size_in_bytes");
+ private static final Result EMPTY_RESULT =
+ ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(ImmutableList.of())
+ .build();
+
+ private final Table table;
+ private final BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ this.findDanglingDeletes = this::findDanglingDeletes;
+ }
+
+ public RemoveDanglingDeleteFilesAction(
+ Table table,
+ BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes) {
+ this.table = table;
+ this.findDanglingDeletes = findDanglingDeletes;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ protected Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Snapshot snapshot = table.snapshot(branch);
+ if (snapshot == null) {
+ Preconditions.checkArgument(
+ branch.equals(SnapshotRef.MAIN_BRANCH),
+ "Cannot remove dangling delete files from branch %s: branch does not exist",
+ branch);
+ return EMPTY_RESULT;
+ }
+
+ List<ManifestFile> deleteManifests = snapshot.deleteManifests(table.io());
+ if (deleteManifests.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ Collection<DeleteFile> danglingDeletes =
+ findDanglingDeletes.apply(snapshot.snapshotId(), deleteManifests);
+ if (danglingDeletes.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ RewriteFiles rewriteFiles = table.newRewrite().validateFromSnapshot(snapshot.snapshotId());
+ for (DeleteFile deleteFile : danglingDeletes) {
+ LOG.debug("Removing dangling delete file {}", deleteFile.location());
+ rewriteFiles.deleteFile(deleteFile);
+ }
+
+ commit(rewriteFiles.toBranch(branch));
+
+ return ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(danglingDeletes)
+ .build();
+ }
+
+ /**
+ * Dangling delete files can be identified with following steps
+ *
+ * <ol>
+ * <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(
+ long snapshotId, List<ManifestFile> deleteManifests) {
+ Set<DeleteFileKey> referencedKeys = Sets.newHashSet();
+ TableScan scan = table.newScan().useSnapshot(snapshotId);
+ try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
+ for (FileScanTask task : tasks) {
+ for (DeleteFile deleteFile : task.deletes()) {
+ referencedKeys.add(new DeleteFileKey(deleteFile));
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to scan: %s", scan);
+ }
+
+ List<DeleteFile> danglingDeletes = Lists.newArrayList();
+ for (ManifestFile manifest : deleteManifests) {
+ try (ManifestReader<DeleteFile> reader =
+ readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (!referencedKeys.contains(new DeleteFileKey(deleteFile))) {
+ danglingDeletes.add(deleteFile);
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to read manifest: %s", manifest);
+ }
+ }
+
+ return danglingDeletes;
+ }
+
+ public static ManifestReader<DeleteFile> readDeleteManifest(
+ ManifestFile manifest, FileIO io, Map<Integer, PartitionSpec> specsById) {
+ return ManifestFiles.readDeleteManifest(manifest, io, specsById).select(DELETE_COLUMNS);
+ }
+
+ public record DeleteFileKey(String location, Long contentOffset, Long contentSizeInBytes)
Review Comment:
With Kryo enabled, serialization fails with `UnsupportedOperationException: can't get field offset on a record class`. Kryo 4’s `FieldSerializer` calls `Unsafe.objectFieldOffset()`, which the JVM rejects for records.
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r2978224515
##########
core/src/main/java/org/apache/iceberg/DeleteFileIndex.java:
##########
@@ -205,12 +205,8 @@ 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);
+ if (dv != null && dv.dataSequenceNumber() < seq) {
Review Comment:
We can think about a DV file that reference a data file with a greater sequence number either 1) as a spec violation or 2) as a dangling delete file for a previously deleted data file with the same name.
In the first case, giving that different engines can work with the same table, I would prefer that spark ignore it instead of failing. Especially because there is no way in spark to fix the violation using only sql without java: remove dangling delete action can only be called as part of the rewrite_data_files procedure that will fail on this check during the scan before calling the action. And if you accept this PR, there will be no way to fix the violation with spark even using java.
The second case, I agree, is very unlikely but still possible, and scans should ignore dangling delete files. I'm sorry if I've missed the part of the spec that makes it a spec violation.
That is why I've proposed to relax the check. An alternative would be finding dangling equality deletes by copying DeleteFileIndex.canContainEqDeletesForFile method to the action (or making the class and the method public). I can do it, if it's a more appropriate solution.
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-4956016856
The PR is not stale
--
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: scan based remove dangling delete action [iceberg]
Posted by "manuzhang (via GitHub)" <gi...@apache.org>.
manuzhang commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r4014456162
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of(
+ "content",
+ "file_path",
+ "file_format",
+ "partition",
+ "record_count",
+ "file_size_in_bytes",
+ "key_metadata",
+ "split_offsets",
+ "equality_ids",
+ "sort_order_id",
+ "first_row_id",
+ "referenced_data_file",
+ "content_offset",
+ "content_size_in_bytes");
+ private static final Result EMPTY_RESULT =
+ ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(ImmutableList.of())
+ .build();
+
+ private final Table table;
+ private final BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ this.findDanglingDeletes = this::findDanglingDeletes;
+ }
+
+ public RemoveDanglingDeleteFilesAction(
+ Table table,
+ BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes) {
+ this.table = table;
+ this.findDanglingDeletes = findDanglingDeletes;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ protected Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Snapshot snapshot = table.snapshot(branch);
+ if (snapshot == null) {
+ Preconditions.checkArgument(
+ branch.equals(SnapshotRef.MAIN_BRANCH),
+ "Cannot remove dangling delete files from branch %s: branch does not exist",
+ branch);
+ return EMPTY_RESULT;
+ }
+
+ List<ManifestFile> deleteManifests = snapshot.deleteManifests(table.io());
+ if (deleteManifests.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ Collection<DeleteFile> danglingDeletes =
+ findDanglingDeletes.apply(snapshot.snapshotId(), deleteManifests);
+ if (danglingDeletes.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ RewriteFiles rewriteFiles = table.newRewrite().validateFromSnapshot(snapshot.snapshotId());
+ for (DeleteFile deleteFile : danglingDeletes) {
+ LOG.debug("Removing dangling delete file {}", deleteFile.location());
+ rewriteFiles.deleteFile(deleteFile);
+ }
+
+ commit(rewriteFiles.toBranch(branch));
+
+ return ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(danglingDeletes)
+ .build();
+ }
+
+ /**
+ * Dangling delete files can be identified with following steps
+ *
+ * <ol>
+ * <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(
+ long snapshotId, List<ManifestFile> deleteManifests) {
+ Set<DeleteFileKey> referencedKeys = Sets.newHashSet();
+ TableScan scan = table.newScan().useSnapshot(snapshotId);
+ try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
+ for (FileScanTask task : tasks) {
+ for (DeleteFile deleteFile : task.deletes()) {
+ referencedKeys.add(new DeleteFileKey(deleteFile));
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to scan: %s", scan);
+ }
+
+ List<DeleteFile> danglingDeletes = Lists.newArrayList();
+ for (ManifestFile manifest : deleteManifests) {
+ try (ManifestReader<DeleteFile> reader =
+ readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (!referencedKeys.contains(new DeleteFileKey(deleteFile))) {
+ danglingDeletes.add(deleteFile);
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to read manifest: %s", manifest);
+ }
+ }
+
+ return danglingDeletes;
+ }
+
+ public static ManifestReader<DeleteFile> readDeleteManifest(
+ ManifestFile manifest, FileIO io, Map<Integer, PartitionSpec> specsById) {
+ return ManifestFiles.readDeleteManifest(manifest, io, specsById).select(DELETE_COLUMNS);
+ }
+
+ public record DeleteFileKey(String location, Long contentOffset, Long contentSizeInBytes)
Review Comment:
@kinolaev This breaks kryo-enabled Spark jobs. Can you fix it in a follow-up PR?
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r4014489369
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of(
+ "content",
+ "file_path",
+ "file_format",
+ "partition",
+ "record_count",
+ "file_size_in_bytes",
+ "key_metadata",
+ "split_offsets",
+ "equality_ids",
+ "sort_order_id",
+ "first_row_id",
+ "referenced_data_file",
+ "content_offset",
+ "content_size_in_bytes");
+ private static final Result EMPTY_RESULT =
+ ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(ImmutableList.of())
+ .build();
+
+ private final Table table;
+ private final BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ this.findDanglingDeletes = this::findDanglingDeletes;
+ }
+
+ public RemoveDanglingDeleteFilesAction(
+ Table table,
+ BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes) {
+ this.table = table;
+ this.findDanglingDeletes = findDanglingDeletes;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ protected Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Snapshot snapshot = table.snapshot(branch);
+ if (snapshot == null) {
+ Preconditions.checkArgument(
+ branch.equals(SnapshotRef.MAIN_BRANCH),
+ "Cannot remove dangling delete files from branch %s: branch does not exist",
+ branch);
+ return EMPTY_RESULT;
+ }
+
+ List<ManifestFile> deleteManifests = snapshot.deleteManifests(table.io());
+ if (deleteManifests.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ Collection<DeleteFile> danglingDeletes =
+ findDanglingDeletes.apply(snapshot.snapshotId(), deleteManifests);
+ if (danglingDeletes.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ RewriteFiles rewriteFiles = table.newRewrite().validateFromSnapshot(snapshot.snapshotId());
+ for (DeleteFile deleteFile : danglingDeletes) {
+ LOG.debug("Removing dangling delete file {}", deleteFile.location());
+ rewriteFiles.deleteFile(deleteFile);
+ }
+
+ commit(rewriteFiles.toBranch(branch));
+
+ return ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(danglingDeletes)
+ .build();
+ }
+
+ /**
+ * Dangling delete files can be identified with following steps
+ *
+ * <ol>
+ * <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(
+ long snapshotId, List<ManifestFile> deleteManifests) {
+ Set<DeleteFileKey> referencedKeys = Sets.newHashSet();
+ TableScan scan = table.newScan().useSnapshot(snapshotId);
+ try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
+ for (FileScanTask task : tasks) {
+ for (DeleteFile deleteFile : task.deletes()) {
+ referencedKeys.add(new DeleteFileKey(deleteFile));
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to scan: %s", scan);
+ }
+
+ List<DeleteFile> danglingDeletes = Lists.newArrayList();
+ for (ManifestFile manifest : deleteManifests) {
+ try (ManifestReader<DeleteFile> reader =
+ readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (!referencedKeys.contains(new DeleteFileKey(deleteFile))) {
+ danglingDeletes.add(deleteFile);
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to read manifest: %s", manifest);
+ }
+ }
+
+ return danglingDeletes;
+ }
+
+ public static ManifestReader<DeleteFile> readDeleteManifest(
+ ManifestFile manifest, FileIO io, Map<Integer, PartitionSpec> specsById) {
+ return ManifestFiles.readDeleteManifest(manifest, io, specsById).select(DELETE_COLUMNS);
+ }
+
+ public record DeleteFileKey(String location, Long contentOffset, Long contentSizeInBytes)
Review Comment:
Hi @manuzhang, thanks for catching this! Could you provide a few more details on how to reproduce the issue? Also, if you already have an idea of what Kryo doesn't support, that would be helpful - I haven't used Kryo before.
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of(
+ "content",
+ "file_path",
+ "file_format",
+ "partition",
+ "record_count",
+ "file_size_in_bytes",
+ "key_metadata",
+ "split_offsets",
+ "equality_ids",
+ "sort_order_id",
+ "first_row_id",
+ "referenced_data_file",
+ "content_offset",
+ "content_size_in_bytes");
+ private static final Result EMPTY_RESULT =
+ ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(ImmutableList.of())
+ .build();
+
+ private final Table table;
+ private final BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ this.findDanglingDeletes = this::findDanglingDeletes;
+ }
+
+ public RemoveDanglingDeleteFilesAction(
+ Table table,
+ BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes) {
+ this.table = table;
+ this.findDanglingDeletes = findDanglingDeletes;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ protected Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Snapshot snapshot = table.snapshot(branch);
+ if (snapshot == null) {
+ Preconditions.checkArgument(
+ branch.equals(SnapshotRef.MAIN_BRANCH),
+ "Cannot remove dangling delete files from branch %s: branch does not exist",
+ branch);
+ return EMPTY_RESULT;
+ }
+
+ List<ManifestFile> deleteManifests = snapshot.deleteManifests(table.io());
+ if (deleteManifests.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ Collection<DeleteFile> danglingDeletes =
+ findDanglingDeletes.apply(snapshot.snapshotId(), deleteManifests);
+ if (danglingDeletes.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ RewriteFiles rewriteFiles = table.newRewrite().validateFromSnapshot(snapshot.snapshotId());
+ for (DeleteFile deleteFile : danglingDeletes) {
+ LOG.debug("Removing dangling delete file {}", deleteFile.location());
+ rewriteFiles.deleteFile(deleteFile);
+ }
+
+ commit(rewriteFiles.toBranch(branch));
+
+ return ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(danglingDeletes)
+ .build();
+ }
+
+ /**
+ * Dangling delete files can be identified with following steps
+ *
+ * <ol>
+ * <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(
+ long snapshotId, List<ManifestFile> deleteManifests) {
+ Set<DeleteFileKey> referencedKeys = Sets.newHashSet();
+ TableScan scan = table.newScan().useSnapshot(snapshotId);
+ try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
+ for (FileScanTask task : tasks) {
+ for (DeleteFile deleteFile : task.deletes()) {
+ referencedKeys.add(new DeleteFileKey(deleteFile));
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to scan: %s", scan);
+ }
+
+ List<DeleteFile> danglingDeletes = Lists.newArrayList();
+ for (ManifestFile manifest : deleteManifests) {
+ try (ManifestReader<DeleteFile> reader =
+ readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (!referencedKeys.contains(new DeleteFileKey(deleteFile))) {
+ danglingDeletes.add(deleteFile);
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to read manifest: %s", manifest);
+ }
+ }
+
+ return danglingDeletes;
+ }
+
+ public static ManifestReader<DeleteFile> readDeleteManifest(
+ ManifestFile manifest, FileIO io, Map<Integer, PartitionSpec> specsById) {
+ return ManifestFiles.readDeleteManifest(manifest, io, specsById).select(DELETE_COLUMNS);
+ }
+
+ public record DeleteFileKey(String location, Long contentOffset, Long contentSizeInBytes)
Review Comment:
@manuzhang I found out that Kryo 4 that is used in Spark doesn't support records. Is that what's causing the problem?
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5505674392
I'm actually thinking about moving the whole action in core, including tests. I plan to try it later today.
Let me know if you think it's a bad idea.
--
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: scan based remove dangling delete action [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3918743884
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.util.List;
+import org.apache.iceberg.DeleteFile;
+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.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.util.DeleteFileSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of("file_path", "content_offset", "content_size_in_bytes");
+
+ private final Table table;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+ private Snapshot snapshot;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ public Table table() {
Review Comment:
`table()` is `protected` in `BaseAction` and this widens it to `public`. It looks like that's only so the Spark wrapper can build its job description via `action.table().name()`. Could the wrapper keep its own `table` field instead and leave this `protected`?
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3074640305
##########
core/src/main/java/org/apache/iceberg/DeleteFileIndex.java:
##########
@@ -205,12 +205,8 @@ 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);
+ if (dv != null && dv.dataSequenceNumber() < seq) {
Review Comment:
Hello @huaxingao , I'd be happy to know, do you prefer keeping the check and reimplement delete-to-data file matching in the action class or it's acceptable to relax the check?
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3863569753
##########
core/src/main/java/org/apache/iceberg/DeleteFileIndex.java:
##########
@@ -220,38 +222,26 @@ 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);
- validatePartitionMatch(dv, dataFile);
+ if (dv != null && (dv.dataSequenceNumber() < seq || !validatePartitionMatch(dv, dataFile))) {
+ return null;
}
return dv;
}
- private void validatePartitionMatch(DeleteFile deleteFile, DataFile dataFile) {
- ValidationException.check(
- deleteFile.specId() == dataFile.specId(),
- "Mismatched partition specs (%s, %s) for delete file %s and data file %s:"
- + " metadata is corrupted",
- deleteFile.specId(),
- dataFile.specId(),
- deleteFile.location(),
- dataFile.location());
+ private boolean validatePartitionMatch(DeleteFile deleteFile, DataFile dataFile) {
+ if (deleteFile.specId() != dataFile.specId()) {
+ // Mismatched partition specs for delete file and data file: metadata is corrupted
+ return false;
+ }
if (partitionComparatorsBySpecId != null) {
Comparator<StructLike> partitionComparator =
partitionComparatorsBySpecId.get(deleteFile.specId());
- ValidationException.check(
- partitionComparator.compare(deleteFile.partition(), dataFile.partition()) == 0,
- "Mismatched partition tuples (%s, %s) for delete file %s and data file %s:"
- + " metadata is corrupted",
- deleteFile.partition(),
- dataFile.partition(),
- deleteFile.location(),
- dataFile.location());
+ if (partitionComparator.compare(deleteFile.partition(), dataFile.partition()) != 0) {
+ // Mismatched partition tuples for delete file and data file: metadata is corrupted
+ return false;
Review Comment:
@huaxingao, thank you for your persistence, and I apologize - I should have read the full discussion instead of just the diff before leaving my previous comment. I see the distinction now: a position delete file with a mismatched partition indicates corruption, it is not just a dangling delete file.
Also, according to the spec, there is no such thing as a dangling DV because:
> When removing a data file, writers must also remove any deletion vector that applies to that data file from delete manifests.
I checked, and this requirement is enforced by `ManifestFilterManager`.
Taking this into account, I think `RemoveDanglingDeletesSparkAction` should only worry about spec-compliant dangling delete files. That is why I reverted both validations in the core library and restricted the failing tests in `TestRemoveDanglingDeleteAction` to format versions lower than 3.
Does it sound right to you?
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3852374430
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java:
##########
@@ -113,94 +89,36 @@ 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());
- }
-
- 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);
+ TableScan scan = table.newScan();
+
+ 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);
+ }
- // 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());
- }
+ DeleteFileSet danglingDeletes = DeleteFileSet.create();
+ for (ManifestFile manifest : scan.snapshot().deleteManifests(table.io())) {
+ try (ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (!deletes.contains(deleteFile)) {
+ danglingDeletes.add(deleteFile);
Review Comment:
It looks like it works as expected (2e1d1311d10b1871c9ffcb8d5090f5b06789547b). And I've checked, [ManifestReader.iterator()](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/ManifestReader.java#L344-L347) makes copies without stats when the stats columns are not included.
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3919146837
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.util.List;
+import org.apache.iceberg.DeleteFile;
+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.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.util.DeleteFileSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of("file_path", "content_offset", "content_size_in_bytes");
+
+ private final Table table;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+ private Snapshot snapshot;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ public Table table() {
Review Comment:
Done
--
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: scan based remove dangling delete action [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho merged PR #15727:
URL: https://github.com/apache/iceberg/pull/15727
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r4014489369
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of(
+ "content",
+ "file_path",
+ "file_format",
+ "partition",
+ "record_count",
+ "file_size_in_bytes",
+ "key_metadata",
+ "split_offsets",
+ "equality_ids",
+ "sort_order_id",
+ "first_row_id",
+ "referenced_data_file",
+ "content_offset",
+ "content_size_in_bytes");
+ private static final Result EMPTY_RESULT =
+ ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(ImmutableList.of())
+ .build();
+
+ private final Table table;
+ private final BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ this.findDanglingDeletes = this::findDanglingDeletes;
+ }
+
+ public RemoveDanglingDeleteFilesAction(
+ Table table,
+ BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes) {
+ this.table = table;
+ this.findDanglingDeletes = findDanglingDeletes;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ protected Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Snapshot snapshot = table.snapshot(branch);
+ if (snapshot == null) {
+ Preconditions.checkArgument(
+ branch.equals(SnapshotRef.MAIN_BRANCH),
+ "Cannot remove dangling delete files from branch %s: branch does not exist",
+ branch);
+ return EMPTY_RESULT;
+ }
+
+ List<ManifestFile> deleteManifests = snapshot.deleteManifests(table.io());
+ if (deleteManifests.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ Collection<DeleteFile> danglingDeletes =
+ findDanglingDeletes.apply(snapshot.snapshotId(), deleteManifests);
+ if (danglingDeletes.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ RewriteFiles rewriteFiles = table.newRewrite().validateFromSnapshot(snapshot.snapshotId());
+ for (DeleteFile deleteFile : danglingDeletes) {
+ LOG.debug("Removing dangling delete file {}", deleteFile.location());
+ rewriteFiles.deleteFile(deleteFile);
+ }
+
+ commit(rewriteFiles.toBranch(branch));
+
+ return ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(danglingDeletes)
+ .build();
+ }
+
+ /**
+ * Dangling delete files can be identified with following steps
+ *
+ * <ol>
+ * <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(
+ long snapshotId, List<ManifestFile> deleteManifests) {
+ Set<DeleteFileKey> referencedKeys = Sets.newHashSet();
+ TableScan scan = table.newScan().useSnapshot(snapshotId);
+ try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
+ for (FileScanTask task : tasks) {
+ for (DeleteFile deleteFile : task.deletes()) {
+ referencedKeys.add(new DeleteFileKey(deleteFile));
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to scan: %s", scan);
+ }
+
+ List<DeleteFile> danglingDeletes = Lists.newArrayList();
+ for (ManifestFile manifest : deleteManifests) {
+ try (ManifestReader<DeleteFile> reader =
+ readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (!referencedKeys.contains(new DeleteFileKey(deleteFile))) {
+ danglingDeletes.add(deleteFile);
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to read manifest: %s", manifest);
+ }
+ }
+
+ return danglingDeletes;
+ }
+
+ public static ManifestReader<DeleteFile> readDeleteManifest(
+ ManifestFile manifest, FileIO io, Map<Integer, PartitionSpec> specsById) {
+ return ManifestFiles.readDeleteManifest(manifest, io, specsById).select(DELETE_COLUMNS);
+ }
+
+ public record DeleteFileKey(String location, Long contentOffset, Long contentSizeInBytes)
Review Comment:
Hi @manuzhang, thanks for catching this! Could you provide a few more details on how to reproduce the issue? Also, if you already have an idea of what Kryo doesn't support, that would be helpful - I haven't used Kryo before.
upd: I found out that Kryo 4 that is used in Spark doesn't support records. Is that what's causing the problem?
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3863569753
##########
core/src/main/java/org/apache/iceberg/DeleteFileIndex.java:
##########
@@ -220,38 +222,26 @@ 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);
- validatePartitionMatch(dv, dataFile);
+ if (dv != null && (dv.dataSequenceNumber() < seq || !validatePartitionMatch(dv, dataFile))) {
+ return null;
}
return dv;
}
- private void validatePartitionMatch(DeleteFile deleteFile, DataFile dataFile) {
- ValidationException.check(
- deleteFile.specId() == dataFile.specId(),
- "Mismatched partition specs (%s, %s) for delete file %s and data file %s:"
- + " metadata is corrupted",
- deleteFile.specId(),
- dataFile.specId(),
- deleteFile.location(),
- dataFile.location());
+ private boolean validatePartitionMatch(DeleteFile deleteFile, DataFile dataFile) {
+ if (deleteFile.specId() != dataFile.specId()) {
+ // Mismatched partition specs for delete file and data file: metadata is corrupted
+ return false;
+ }
if (partitionComparatorsBySpecId != null) {
Comparator<StructLike> partitionComparator =
partitionComparatorsBySpecId.get(deleteFile.specId());
- ValidationException.check(
- partitionComparator.compare(deleteFile.partition(), dataFile.partition()) == 0,
- "Mismatched partition tuples (%s, %s) for delete file %s and data file %s:"
- + " metadata is corrupted",
- deleteFile.partition(),
- dataFile.partition(),
- deleteFile.location(),
- dataFile.location());
+ if (partitionComparator.compare(deleteFile.partition(), dataFile.partition()) != 0) {
+ // Mismatched partition tuples for delete file and data file: metadata is corrupted
+ return false;
Review Comment:
@huaxingao, thank you for your persistence, and I apologize - I should have read the full discussion instead of just the diff before leaving my previous comment. I see the distinction now: a position delete file with a mismatched partition indicates corruption, it is not just a dangling delete file.
Also, according to the spec, there is no such thing as a dangling DV because:
> When removing a data file, writers must also remove any deletion vector that applies to that data file from delete manifests.
I checked, and this requirement is enforced by `ManifestFilterManager`.
Taking this into account, I think `RemoveDanglingDeletesSparkAction` should only worry about spec-compliant dangling delete files. That is why I reverted both validations in the core library and restricted the failing tests in `TestRemoveDanglingDeleteAction` to format versions lower than 3 (32f758a17be73ccd00ac5d3dde1a97ccab1ee036).
Does it sound right to you?
--
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: scan based remove dangling delete action [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3850365101
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java:
##########
@@ -113,94 +89,36 @@ 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());
- }
-
- 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);
+ TableScan scan = table.newScan();
+
+ 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);
+ }
- // 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());
- }
+ DeleteFileSet danglingDeletes = DeleteFileSet.create();
+ for (ManifestFile manifest : scan.snapshot().deleteManifests(table.io())) {
+ try (ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (!deletes.contains(deleteFile)) {
+ danglingDeletes.add(deleteFile);
Review Comment:
Can we drop the stats here? Every delete file we read keeps all of its column
statistics in driver memory. The set only needs the file
location and its offset and size, and copyWithoutStats() keeps all three.
```
danglingDeletes.add(deleteFile.copyWithoutStats());
```
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5291146352
Not stale, still awaiting review :)
--
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: scan based remove dangling delete action [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3918711955
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.util.List;
+import org.apache.iceberg.DeleteFile;
+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.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.util.DeleteFileSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of("file_path", "content_offset", "content_size_in_bytes");
+
+ private final Table table;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+ private Snapshot snapshot;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ public Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ public RemoveDanglingDeleteFilesAction init() {
Review Comment:
I'd drop `init()` and resolve the snapshot as a local in `execute()`. As a public method it makes the result depend on the order the caller makes its calls, and nothing guards against getting that order wrong.
Since `execute()` only calls `init()` when `snapshot` is null, `action.init().toBranch("x").execute()` resolves the snapshot while `branch` is still `main`, then commits that decision to `x`.
The only callers are the three Spark wrappers, and there `init()` only makes the "branch does not exist" check run slightly earlier, so this should be free to change.
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3941329979
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.util.List;
+import org.apache.iceberg.DeleteFile;
+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.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.util.DeleteFileSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of("file_path", "content_offset", "content_size_in_bytes");
+
+ private final Table table;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ protected Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Snapshot snapshot = table.snapshot(branch);
+ Preconditions.checkArgument(
+ snapshot != null,
+ "Cannot remove dangling delete files from branch %s: branch does not exist",
+ branch);
+
+ RewriteFiles rewriteFiles = table.newRewrite().validateFromSnapshot(snapshot.snapshotId());
+ DeleteFileSet danglingDeletes = findDanglingDeletes(table, snapshot);
+
+ for (DeleteFile deleteFile : danglingDeletes) {
+ LOG.debug("Removing dangling delete file {}", deleteFile.location());
+ rewriteFiles.deleteFile(deleteFile);
+ }
+
+ if (!danglingDeletes.isEmpty()) {
+ commit(rewriteFiles.toBranch(branch));
+ }
+
+ return ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(danglingDeletes)
+ .build();
+ }
+
+ /**
+ * Dangling delete files can be identified with following steps
+ *
+ * <ol>
+ * <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 static DeleteFileSet findDanglingDeletes(Table table, Snapshot snapshot) {
+ DeleteFileSet deletes = DeleteFileSet.create();
+ TableScan scan = table.newScan().useSnapshot(snapshot.snapshotId());
+ 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())
+ .select(DELETE_COLUMNS)) {
Review Comment:
> The absent partition can also interfere with commit retries after concurrent manifest rewriting.
Thanks for pointing this out! Fixed in b5b10468ab711045f8d1f99245791b7fefae8e4a.
--
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: scan based remove dangling delete action [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5671473520
Merged, thanks @kinolaev for pr and @huaxingao for reviews!
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-4680752373
The conflict is resolved.
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-4678733060
The PR is not stale
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-4424622695
Hello @huaxingao , I'm still hoping to get your opinion about relaxing the DV sequence number check and the PR in general.
In my production environment this PR significantly reduced the amount of equality deletes:
- from 6.3b to 3.2b records
- from 400k to 150k files
So the current `RemoveDanglingDeleteAction` implementation skips too many irrelevant equality deletes.
--
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: scan based remove dangling delete action [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5287912543
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: scan based remove dangling delete action [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5530260376
Can we consider a configurable Spark execution mode for very large tables? The previous implementation distributed metadata grouping and joins, collecting only dangling deletes to the driver. The new implementation is more accurate, but performs full scan planning and retains applicable deletes on the driver. Do we have benchmarks or memory estimates at million-delete-file scale?
--
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: scan based remove dangling delete action [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3849604134
##########
core/src/main/java/org/apache/iceberg/DeleteFileIndex.java:
##########
@@ -205,12 +205,8 @@ 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);
+ if (dv != null && dv.dataSequenceNumber() < seq) {
Review Comment:
Sorry for the long delay. Your explanation makes sense to me, let's keep the relaxation rather than reimplementing the matching in the action.
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5408227632
@huaxingao, thank you for the review! I've addressed your comments - please take another look when you have a chance.
--
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: scan based remove dangling delete action [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r2977846549
##########
core/src/main/java/org/apache/iceberg/DeleteFileIndex.java:
##########
@@ -205,12 +205,8 @@ 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);
+ if (dv != null && dv.dataSequenceNumber() < seq) {
Review Comment:
Why we need to relax this check?
--
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: scan based remove dangling delete action [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-4676153563
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: scan based remove dangling delete action [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5490452366
@szehon-ho @nastra would appreciate a second pair of eyes on this PR. Thanks!
--
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: scan based remove dangling delete action [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3927474993
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.util.List;
+import org.apache.iceberg.DeleteFile;
+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.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.util.DeleteFileSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of("file_path", "content_offset", "content_size_in_bytes");
+
+ private final Table table;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ protected Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Snapshot snapshot = table.snapshot(branch);
+ Preconditions.checkArgument(
+ snapshot != null,
+ "Cannot remove dangling delete files from branch %s: branch does not exist",
+ branch);
+
+ RewriteFiles rewriteFiles = table.newRewrite().validateFromSnapshot(snapshot.snapshotId());
+ DeleteFileSet danglingDeletes = findDanglingDeletes(table, snapshot);
+
+ for (DeleteFile deleteFile : danglingDeletes) {
+ LOG.debug("Removing dangling delete file {}", deleteFile.location());
+ rewriteFiles.deleteFile(deleteFile);
+ }
+
+ if (!danglingDeletes.isEmpty()) {
+ commit(rewriteFiles.toBranch(branch));
+ }
+
+ return ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(danglingDeletes)
+ .build();
+ }
+
+ /**
+ * Dangling delete files can be identified with following steps
+ *
+ * <ol>
+ * <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 static DeleteFileSet findDanglingDeletes(Table table, Snapshot snapshot) {
+ DeleteFileSet deletes = DeleteFileSet.create();
+ TableScan scan = table.newScan().useSnapshot(snapshot.snapshotId());
+ 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())
+ .select(DELETE_COLUMNS)) {
Review Comment:
Could we return complete stats-free `DeleteFile` objects here? This projection escapes through `removedDeleteFiles()`, so fields such as content, format, partition, and record count are invalid or missing. The absent partition can also interfere with commit retries after concurrent manifest rewriting.
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.util.List;
+import org.apache.iceberg.DeleteFile;
+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.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.util.DeleteFileSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of("file_path", "content_offset", "content_size_in_bytes");
+
+ private final Table table;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ protected Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Snapshot snapshot = table.snapshot(branch);
Review Comment:
Could an empty main branch return an empty result? A newly created unpartitioned table previously treated this action as a no-op, but now `snapshot` is null and the action reports that `main` does not exist. An explicitly missing named branch should still fail.
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3941329979
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.util.List;
+import org.apache.iceberg.DeleteFile;
+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.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.util.DeleteFileSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of("file_path", "content_offset", "content_size_in_bytes");
+
+ private final Table table;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ protected Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Snapshot snapshot = table.snapshot(branch);
+ Preconditions.checkArgument(
+ snapshot != null,
+ "Cannot remove dangling delete files from branch %s: branch does not exist",
+ branch);
+
+ RewriteFiles rewriteFiles = table.newRewrite().validateFromSnapshot(snapshot.snapshotId());
+ DeleteFileSet danglingDeletes = findDanglingDeletes(table, snapshot);
+
+ for (DeleteFile deleteFile : danglingDeletes) {
+ LOG.debug("Removing dangling delete file {}", deleteFile.location());
+ rewriteFiles.deleteFile(deleteFile);
+ }
+
+ if (!danglingDeletes.isEmpty()) {
+ commit(rewriteFiles.toBranch(branch));
+ }
+
+ return ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(danglingDeletes)
+ .build();
+ }
+
+ /**
+ * Dangling delete files can be identified with following steps
+ *
+ * <ol>
+ * <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 static DeleteFileSet findDanglingDeletes(Table table, Snapshot snapshot) {
+ DeleteFileSet deletes = DeleteFileSet.create();
+ TableScan scan = table.newScan().useSnapshot(snapshot.snapshotId());
+ 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())
+ .select(DELETE_COLUMNS)) {
Review Comment:
> The absent partition can also interfere with commit retries after concurrent manifest rewriting.
Thanks for pointing this out! Fixed in bb4563fc3c4349d8cc6072bcdfd23a8ef4f4fa5c.
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5443080565
@huaxingao I've addressed your feedback regarding partition validation above. I would love to get this landed in 1.12.0, so I'd really appreciate another look when you have a moment. Happy to make any further adjustments needed!
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5398767975
The conflict is resolved
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3979421721
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java:
##########
@@ -18,189 +18,123 @@
*/
package org.apache.iceberg.spark.actions;
-import static org.apache.spark.sql.functions.col;
-import static org.apache.spark.sql.functions.min;
-
-import java.util.Collections;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Iterator;
import java.util.List;
-import java.util.stream.Collectors;
-import org.apache.iceberg.DataFile;
+import java.util.stream.StreamSupport;
import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.MetadataTableType;
-import org.apache.iceberg.Partitioning;
-import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
-import org.apache.iceberg.actions.ImmutableRemoveDanglingDeleteFiles;
+import org.apache.iceberg.TableScan;
import org.apache.iceberg.actions.RemoveDanglingDeleteFiles;
-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.iceberg.actions.RemoveDanglingDeleteFilesAction;
+import org.apache.iceberg.actions.RemoveDanglingDeleteFilesAction.DeleteFileKey;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.io.ClosingIterator;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.source.SerializableTableWithSize;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.types.StructType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import scala.Tuple2;
/**
* 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>
implements RemoveDanglingDeleteFiles {
- private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeletesSparkAction.class);
private final Table table;
+ private final RemoveDanglingDeleteFilesAction action;
protected RemoveDanglingDeletesSparkAction(SparkSession spark, Table table) {
super(spark);
this.table = table;
+ this.action = new RemoveDanglingDeleteFilesAction(table, this::findDanglingDeletes);
}
@Override
protected RemoveDanglingDeletesSparkAction self() {
return this;
}
+ public RemoveDanglingDeletesSparkAction toBranch(String targetBranch) {
+ action.toBranch(targetBranch);
+ return this;
+ }
+
@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();
- }
-
+ commitSummary().forEach(action::set);
String desc = String.format("Removing dangling delete files in %s", table.name());
- JobGroupInfo info = newJobGroupInfo("REMOVE-DELETES", desc);
- return withJobGroupInfo(info, this::doExecute);
+ return withJobGroupInfo(newJobGroupInfo("REMOVE-DELETES", desc), action::execute);
}
- Result doExecute() {
- RewriteFiles rewriteFiles = table.newRewrite();
- DeleteFileSet danglingDeletes = DeleteFileSet.create();
- danglingDeletes.addAll(findDanglingDeletes());
- danglingDeletes.addAll(findDanglingDvs());
+ private List<DeleteFile> findDanglingDeletes(Snapshot snapshot) {
+ Broadcast<Table> tableBroadcast =
+ sparkContext().broadcast(SerializableTableWithSize.copyOf(table));
+
+ JavaPairRDD<DeleteFileKey, Void> referencedKeys =
+ sparkContext()
+ .parallelize(ImmutableList.of(snapshot.snapshotId()), 1)
+ .flatMap(
+ snapshotId -> {
+ TableScan scan = tableBroadcast.value().newScan().useSnapshot(snapshotId);
+ return new ClosingIterator<>(new DeleteFileKeyIterator(scan.planFiles()));
+ })
+ .mapToPair(key -> new Tuple2<>(key, (Void) null));
+
+ List<ManifestFileBean> deleteManifests =
+ snapshot.deleteManifests(table.io()).stream().map(ManifestFileBean::fromManifest).toList();
+ JavaPairRDD<DeleteFileKey, DeleteFile> allDeletes =
+ sparkContext()
+ .parallelize(deleteManifests, deleteManifests.size())
Review Comment:
Thanks for catching this! I moved the check to core, before calling `findDanglingDeletes` (1e536022da6464f7751e12e01262b9c8).
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3979807632
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java:
##########
@@ -18,189 +18,123 @@
*/
package org.apache.iceberg.spark.actions;
-import static org.apache.spark.sql.functions.col;
-import static org.apache.spark.sql.functions.min;
-
-import java.util.Collections;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Iterator;
import java.util.List;
-import java.util.stream.Collectors;
-import org.apache.iceberg.DataFile;
+import java.util.stream.StreamSupport;
import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.MetadataTableType;
-import org.apache.iceberg.Partitioning;
-import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
-import org.apache.iceberg.actions.ImmutableRemoveDanglingDeleteFiles;
+import org.apache.iceberg.TableScan;
import org.apache.iceberg.actions.RemoveDanglingDeleteFiles;
-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.iceberg.actions.RemoveDanglingDeleteFilesAction;
+import org.apache.iceberg.actions.RemoveDanglingDeleteFilesAction.DeleteFileKey;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.io.ClosingIterator;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.source.SerializableTableWithSize;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.types.StructType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import scala.Tuple2;
/**
* 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>
implements RemoveDanglingDeleteFiles {
- private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeletesSparkAction.class);
private final Table table;
+ private final RemoveDanglingDeleteFilesAction action;
protected RemoveDanglingDeletesSparkAction(SparkSession spark, Table table) {
super(spark);
this.table = table;
+ this.action = new RemoveDanglingDeleteFilesAction(table, this::findDanglingDeletes);
}
@Override
protected RemoveDanglingDeletesSparkAction self() {
return this;
}
+ public RemoveDanglingDeletesSparkAction toBranch(String targetBranch) {
+ action.toBranch(targetBranch);
+ return this;
+ }
+
@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();
- }
-
+ commitSummary().forEach(action::set);
String desc = String.format("Removing dangling delete files in %s", table.name());
- JobGroupInfo info = newJobGroupInfo("REMOVE-DELETES", desc);
- return withJobGroupInfo(info, this::doExecute);
+ return withJobGroupInfo(newJobGroupInfo("REMOVE-DELETES", desc), action::execute);
}
- Result doExecute() {
- RewriteFiles rewriteFiles = table.newRewrite();
- DeleteFileSet danglingDeletes = DeleteFileSet.create();
- danglingDeletes.addAll(findDanglingDeletes());
- danglingDeletes.addAll(findDanglingDvs());
+ private List<DeleteFile> findDanglingDeletes(Snapshot snapshot) {
+ Broadcast<Table> tableBroadcast =
+ sparkContext().broadcast(SerializableTableWithSize.copyOf(table));
+
+ JavaPairRDD<DeleteFileKey, Void> referencedKeys =
+ sparkContext()
+ .parallelize(ImmutableList.of(snapshot.snapshotId()), 1)
+ .flatMap(
+ snapshotId -> {
+ TableScan scan = tableBroadcast.value().newScan().useSnapshot(snapshotId);
+ return new ClosingIterator<>(new DeleteFileKeyIterator(scan.planFiles()));
+ })
+ .mapToPair(key -> new Tuple2<>(key, (Void) null));
+
+ List<ManifestFileBean> deleteManifests =
+ snapshot.deleteManifests(table.io()).stream().map(ManifestFileBean::fromManifest).toList();
+ JavaPairRDD<DeleteFileKey, DeleteFile> allDeletes =
+ sparkContext()
+ .parallelize(deleteManifests, deleteManifests.size())
+ .flatMap(
+ manifest -> {
+ ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(
+ manifest, tableBroadcast.value().io(), tableBroadcast.value().specs());
+ return new ClosingIterator<>(reader.iterator());
+ })
+ .mapToPair(file -> new Tuple2<>(new DeleteFileKey(file), file.copyWithoutStats()));
+
+ return allDeletes.subtractByKey(referencedKeys).values().collect();
+ }
- for (DeleteFile deleteFile : danglingDeletes) {
- LOG.debug("Removing dangling delete file {}", deleteFile.location());
- rewriteFiles.deleteFile(deleteFile);
+ private static class DeleteFileKeyIterator implements CloseableIterator<DeleteFileKey> {
+ private final Closeable closeable;
+ private final Iterator<DeleteFileKey> iterator;
+
+ DeleteFileKeyIterator(CloseableIterable<FileScanTask> tasks) {
+ this.closeable = tasks;
+ this.iterator =
+ StreamSupport.stream(tasks.spliterator(), false)
+ .flatMap(task -> task.deletes().stream())
+ .map(DeleteFileKey::new)
+ .distinct()
Review Comment:
The full set of referenced delete key is not a problem, it adds only a few dozens bytes per delete file. The real problem is `DeleteFileIndex`. `scan.planFiles()` loads all delete manifests eagerly to build `DeleteFileIndex` before planning any data files, and this index stays alive until the `tasks` iterable is closed. Consequently, the `String location`, `Long contentOffset`, and `Long contentSizeInBytes` for all delete files remain in the heap, even without `.distinct()`.
I haven't found a way to avoid keeping the full `DeleteFileIndex` on the driver or the executor. I checked [SparkDistributedDataScan](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/spark/v4.1/spark/src/main/java/org/apache/iceberg/SparkDistributedDataScan.java#L157-L175), but there's no magic there - it collects all delete files from executors to build the `DeleteFileIndex` on the driver.
@szehon-ho, do you have any ideas on how we could distribute building the referenced delete file key set?
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r4021598159
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of(
+ "content",
+ "file_path",
+ "file_format",
+ "partition",
+ "record_count",
+ "file_size_in_bytes",
+ "key_metadata",
+ "split_offsets",
+ "equality_ids",
+ "sort_order_id",
+ "first_row_id",
+ "referenced_data_file",
+ "content_offset",
+ "content_size_in_bytes");
+ private static final Result EMPTY_RESULT =
+ ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(ImmutableList.of())
+ .build();
+
+ private final Table table;
+ private final BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ this.findDanglingDeletes = this::findDanglingDeletes;
+ }
+
+ public RemoveDanglingDeleteFilesAction(
+ Table table,
+ BiFunction<Long, List<ManifestFile>, Collection<DeleteFile>> findDanglingDeletes) {
+ this.table = table;
+ this.findDanglingDeletes = findDanglingDeletes;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ protected Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Snapshot snapshot = table.snapshot(branch);
+ if (snapshot == null) {
+ Preconditions.checkArgument(
+ branch.equals(SnapshotRef.MAIN_BRANCH),
+ "Cannot remove dangling delete files from branch %s: branch does not exist",
+ branch);
+ return EMPTY_RESULT;
+ }
+
+ List<ManifestFile> deleteManifests = snapshot.deleteManifests(table.io());
+ if (deleteManifests.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ Collection<DeleteFile> danglingDeletes =
+ findDanglingDeletes.apply(snapshot.snapshotId(), deleteManifests);
+ if (danglingDeletes.isEmpty()) {
+ return EMPTY_RESULT;
+ }
+
+ RewriteFiles rewriteFiles = table.newRewrite().validateFromSnapshot(snapshot.snapshotId());
+ for (DeleteFile deleteFile : danglingDeletes) {
+ LOG.debug("Removing dangling delete file {}", deleteFile.location());
+ rewriteFiles.deleteFile(deleteFile);
+ }
+
+ commit(rewriteFiles.toBranch(branch));
+
+ return ImmutableRemoveDanglingDeleteFiles.Result.builder()
+ .removedDeleteFiles(danglingDeletes)
+ .build();
+ }
+
+ /**
+ * Dangling delete files can be identified with following steps
+ *
+ * <ol>
+ * <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(
+ long snapshotId, List<ManifestFile> deleteManifests) {
+ Set<DeleteFileKey> referencedKeys = Sets.newHashSet();
+ TableScan scan = table.newScan().useSnapshot(snapshotId);
+ try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
+ for (FileScanTask task : tasks) {
+ for (DeleteFile deleteFile : task.deletes()) {
+ referencedKeys.add(new DeleteFileKey(deleteFile));
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to scan: %s", scan);
+ }
+
+ List<DeleteFile> danglingDeletes = Lists.newArrayList();
+ for (ManifestFile manifest : deleteManifests) {
+ try (ManifestReader<DeleteFile> reader =
+ readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (!referencedKeys.contains(new DeleteFileKey(deleteFile))) {
+ danglingDeletes.add(deleteFile);
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to read manifest: %s", manifest);
+ }
+ }
+
+ return danglingDeletes;
+ }
+
+ public static ManifestReader<DeleteFile> readDeleteManifest(
+ ManifestFile manifest, FileIO io, Map<Integer, PartitionSpec> specsById) {
+ return ManifestFiles.readDeleteManifest(manifest, io, specsById).select(DELETE_COLUMNS);
+ }
+
+ public record DeleteFileKey(String location, Long contentOffset, Long contentSizeInBytes)
Review Comment:
Sorry for a delay, I opened a PR #18130 for Spark 4.2.
--
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: scan based remove dangling delete action [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5505619205
Maybe move `DanglingDeleteFiles` to `org.apache.iceberg.util` and rename it to `DanglingDeleteFileUtil`? Right now it sits in the actions package, so it looks like another action interface rather than a helper 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
Re: [PR] Core, Spark: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5491004314
> `findDanglingDeletes` is now pure core API (no Spark types). Worth lifting into a core helper rather than
keeping three copies across v3.5/v4.0/v4.1?
Thanks @huaxingao, I moved the method into a new `org.apache.iceberg.actions.DanglingDeleteFiles` class. Let me know if there's a better place for 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
Re: [PR] Core, Spark: scan based remove dangling delete action [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5490438806
@kinolaev `findDanglingDeletes` is now pure core API (no Spark types). Worth lifting into a core helper rather than
keeping three copies across v3.5/v4.0/v4.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
Re: [PR] Core, Spark: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#issuecomment-5511978361
Since this action no longer depends on Spark, I moved it and its tests to the core module and turned the Spark version into a thin adapter. How does this look to you, @huaxingao?
Prior to this, in commit 2285de716e3d05a206987b570c02f124fb043393, I renamed `org.apache.iceberg.actions.DanglingDeleteFiles` to `org.apache.iceberg.util.DanglingDeleteFileUtil`. If you'd prefer to keep only the utility in core rather than the entire action, let me know and I'll drop the latest commit.
[Diff before moving the action to core](https://github.com/apache/iceberg/compare/f4db4a56583d268a152a2e725ba4f798f86c4d74..2285de716e3d05a206987b570c02f124fb043393).
--
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: scan based remove dangling delete action [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3849657854
##########
core/src/main/java/org/apache/iceberg/DeleteFileIndex.java:
##########
@@ -220,38 +222,26 @@ 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);
- validatePartitionMatch(dv, dataFile);
+ if (dv != null && (dv.dataSequenceNumber() < seq || !validatePartitionMatch(dv, dataFile))) {
+ return null;
}
return dv;
}
- private void validatePartitionMatch(DeleteFile deleteFile, DataFile dataFile) {
- ValidationException.check(
- deleteFile.specId() == dataFile.specId(),
- "Mismatched partition specs (%s, %s) for delete file %s and data file %s:"
- + " metadata is corrupted",
- deleteFile.specId(),
- dataFile.specId(),
- deleteFile.location(),
- dataFile.location());
+ private boolean validatePartitionMatch(DeleteFile deleteFile, DataFile dataFile) {
+ if (deleteFile.specId() != dataFile.specId()) {
+ // Mismatched partition specs for delete file and data file: metadata is corrupted
+ return false;
+ }
if (partitionComparatorsBySpecId != null) {
Comparator<StructLike> partitionComparator =
partitionComparatorsBySpecId.get(deleteFile.specId());
- ValidationException.check(
- partitionComparator.compare(deleteFile.partition(), dataFile.partition()) == 0,
- "Mismatched partition tuples (%s, %s) for delete file %s and data file %s:"
- + " metadata is corrupted",
- deleteFile.partition(),
- dataFile.partition(),
- deleteFile.location(),
- dataFile.location());
+ if (partitionComparator.compare(deleteFile.partition(), dataFile.partition()) != 0) {
+ // Mismatched partition tuples for delete file and data file: metadata is corrupted
+ return false;
Review Comment:
This reverts the fix from #16957. Is that intentional? I think the scan
should still fail 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
Re: [PR] Core, Spark: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3851430709
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java:
##########
@@ -113,94 +89,36 @@ 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());
- }
-
- 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);
+ TableScan scan = table.newScan();
+
+ 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);
+ }
- // 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());
- }
+ DeleteFileSet danglingDeletes = DeleteFileSet.create();
+ for (ManifestFile manifest : scan.snapshot().deleteManifests(table.io())) {
+ try (ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (!deletes.contains(deleteFile)) {
+ danglingDeletes.add(deleteFile);
Review Comment:
Applied (43b9052222fca44a60adc468be62f0604dbde352), thanks!
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3852374430
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java:
##########
@@ -113,94 +89,36 @@ 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());
- }
-
- 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);
+ TableScan scan = table.newScan();
+
+ 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);
+ }
- // 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());
- }
+ DeleteFileSet danglingDeletes = DeleteFileSet.create();
+ for (ManifestFile manifest : scan.snapshot().deleteManifests(table.io())) {
+ try (ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (!deletes.contains(deleteFile)) {
+ danglingDeletes.add(deleteFile);
Review Comment:
It looks like it works as expected (2e1d1311d10b1871c9ffcb8d5090f5b06789547b). And I've checked, [ManifestReader.iterator()](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/ManifestReader.java#L344-L347) makes copies without stats when the stats columns are not included.
I also removed the danglingDeletes set to list conversion and used the set directly.
--
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: scan based remove dangling delete action [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3859301707
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java:
##########
@@ -113,94 +89,36 @@ 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());
- }
-
- 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);
+ TableScan scan = table.newScan();
+
+ 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);
+ }
- // 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());
- }
+ DeleteFileSet danglingDeletes = DeleteFileSet.create();
+ for (ManifestFile manifest : scan.snapshot().deleteManifests(table.io())) {
+ try (ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (!deletes.contains(deleteFile)) {
+ danglingDeletes.add(deleteFile);
Review Comment:
Even better, thanks. Skipping the stats on read beats copying without them.
--
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: scan based remove dangling delete action [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3859453331
##########
core/src/main/java/org/apache/iceberg/DeleteFileIndex.java:
##########
@@ -220,38 +222,26 @@ 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);
- validatePartitionMatch(dv, dataFile);
+ if (dv != null && (dv.dataSequenceNumber() < seq || !validatePartitionMatch(dv, dataFile))) {
+ return null;
}
return dv;
}
- private void validatePartitionMatch(DeleteFile deleteFile, DataFile dataFile) {
- ValidationException.check(
- deleteFile.specId() == dataFile.specId(),
- "Mismatched partition specs (%s, %s) for delete file %s and data file %s:"
- + " metadata is corrupted",
- deleteFile.specId(),
- dataFile.specId(),
- deleteFile.location(),
- dataFile.location());
+ private boolean validatePartitionMatch(DeleteFile deleteFile, DataFile dataFile) {
+ if (deleteFile.specId() != dataFile.specId()) {
+ // Mismatched partition specs for delete file and data file: metadata is corrupted
+ return false;
+ }
if (partitionComparatorsBySpecId != null) {
Comparator<StructLike> partitionComparator =
partitionComparatorsBySpecId.get(deleteFile.specId());
- ValidationException.check(
- partitionComparator.compare(deleteFile.partition(), dataFile.partition()) == 0,
- "Mismatched partition tuples (%s, %s) for delete file %s and data file %s:"
- + " metadata is corrupted",
- deleteFile.partition(),
- dataFile.partition(),
- deleteFile.location(),
- dataFile.location());
+ if (partitionComparator.compare(deleteFile.partition(), dataFile.partition()) != 0) {
+ // Mismatched partition tuples for delete file and data file: metadata is corrupted
+ return false;
Review Comment:
It does change #16957's behavior. that PR is titled "Fail scans when position deletes/DVs don't match data file partition," and this inverts all four of its tests. Could we ask its author and reviewers to weigh in?
This may also not be needed. `validatePartitionMatch` is only reached from `findPathDeletes` and `findDV`, since the partition keyed lookups already match on the data file's own partition. Equality deletes never reach it, so your 6.3b to 3.2b result should hold without this change.
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3941322156
##########
core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesAction.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import java.io.IOException;
+import java.util.List;
+import org.apache.iceberg.DeleteFile;
+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.SnapshotRef;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.util.DeleteFileSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoveDanglingDeleteFilesAction
+ extends BaseSnapshotUpdateAction<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result>
+ implements RemoveDanglingDeleteFiles {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeleteFilesAction.class);
+ private static final List<String> DELETE_COLUMNS =
+ ImmutableList.of("file_path", "content_offset", "content_size_in_bytes");
+
+ private final Table table;
+ private String branch = SnapshotRef.MAIN_BRANCH;
+
+ public RemoveDanglingDeleteFilesAction(Table table) {
+ this.table = table;
+ }
+
+ @Override
+ protected RemoveDanglingDeleteFilesAction self() {
+ return this;
+ }
+
+ @Override
+ protected Table table() {
+ return table;
+ }
+
+ public RemoveDanglingDeleteFilesAction toBranch(String targetBranch) {
+ Preconditions.checkArgument(targetBranch != null, "Invalid branch name: null");
+ this.branch = targetBranch;
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Snapshot snapshot = table.snapshot(branch);
Review Comment:
Done in cbe42cfeb660f5f4deec4597095b3a821eea3c08.
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3979421721
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java:
##########
@@ -18,189 +18,123 @@
*/
package org.apache.iceberg.spark.actions;
-import static org.apache.spark.sql.functions.col;
-import static org.apache.spark.sql.functions.min;
-
-import java.util.Collections;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Iterator;
import java.util.List;
-import java.util.stream.Collectors;
-import org.apache.iceberg.DataFile;
+import java.util.stream.StreamSupport;
import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.MetadataTableType;
-import org.apache.iceberg.Partitioning;
-import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
-import org.apache.iceberg.actions.ImmutableRemoveDanglingDeleteFiles;
+import org.apache.iceberg.TableScan;
import org.apache.iceberg.actions.RemoveDanglingDeleteFiles;
-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.iceberg.actions.RemoveDanglingDeleteFilesAction;
+import org.apache.iceberg.actions.RemoveDanglingDeleteFilesAction.DeleteFileKey;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.io.ClosingIterator;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.source.SerializableTableWithSize;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.types.StructType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import scala.Tuple2;
/**
* 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>
implements RemoveDanglingDeleteFiles {
- private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeletesSparkAction.class);
private final Table table;
+ private final RemoveDanglingDeleteFilesAction action;
protected RemoveDanglingDeletesSparkAction(SparkSession spark, Table table) {
super(spark);
this.table = table;
+ this.action = new RemoveDanglingDeleteFilesAction(table, this::findDanglingDeletes);
}
@Override
protected RemoveDanglingDeletesSparkAction self() {
return this;
}
+ public RemoveDanglingDeletesSparkAction toBranch(String targetBranch) {
+ action.toBranch(targetBranch);
+ return this;
+ }
+
@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();
- }
-
+ commitSummary().forEach(action::set);
String desc = String.format("Removing dangling delete files in %s", table.name());
- JobGroupInfo info = newJobGroupInfo("REMOVE-DELETES", desc);
- return withJobGroupInfo(info, this::doExecute);
+ return withJobGroupInfo(newJobGroupInfo("REMOVE-DELETES", desc), action::execute);
}
- Result doExecute() {
- RewriteFiles rewriteFiles = table.newRewrite();
- DeleteFileSet danglingDeletes = DeleteFileSet.create();
- danglingDeletes.addAll(findDanglingDeletes());
- danglingDeletes.addAll(findDanglingDvs());
+ private List<DeleteFile> findDanglingDeletes(Snapshot snapshot) {
+ Broadcast<Table> tableBroadcast =
+ sparkContext().broadcast(SerializableTableWithSize.copyOf(table));
+
+ JavaPairRDD<DeleteFileKey, Void> referencedKeys =
+ sparkContext()
+ .parallelize(ImmutableList.of(snapshot.snapshotId()), 1)
+ .flatMap(
+ snapshotId -> {
+ TableScan scan = tableBroadcast.value().newScan().useSnapshot(snapshotId);
+ return new ClosingIterator<>(new DeleteFileKeyIterator(scan.planFiles()));
+ })
+ .mapToPair(key -> new Tuple2<>(key, (Void) null));
+
+ List<ManifestFileBean> deleteManifests =
+ snapshot.deleteManifests(table.io()).stream().map(ManifestFileBean::fromManifest).toList();
+ JavaPairRDD<DeleteFileKey, DeleteFile> allDeletes =
+ sparkContext()
+ .parallelize(deleteManifests, deleteManifests.size())
Review Comment:
Thanks for catching this! I moved the check to the Core, before calling `findDanglingDeletes` (changes/1e536022da6464f7751e12e01262b9c8).
--
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: scan based remove dangling delete action [iceberg]
Posted by "kinolaev (via GitHub)" <gi...@apache.org>.
kinolaev commented on code in PR #15727:
URL: https://github.com/apache/iceberg/pull/15727#discussion_r3994554328
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java:
##########
@@ -18,189 +18,124 @@
*/
package org.apache.iceberg.spark.actions;
-import static org.apache.spark.sql.functions.col;
-import static org.apache.spark.sql.functions.min;
-
-import java.util.Collections;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Iterator;
import java.util.List;
-import java.util.stream.Collectors;
-import org.apache.iceberg.DataFile;
+import java.util.stream.StreamSupport;
import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.MetadataTableType;
-import org.apache.iceberg.Partitioning;
-import org.apache.iceberg.RewriteFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
import org.apache.iceberg.Table;
-import org.apache.iceberg.actions.ImmutableRemoveDanglingDeleteFiles;
+import org.apache.iceberg.TableScan;
import org.apache.iceberg.actions.RemoveDanglingDeleteFiles;
-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.iceberg.actions.RemoveDanglingDeleteFilesAction;
+import org.apache.iceberg.actions.RemoveDanglingDeleteFilesAction.DeleteFileKey;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.io.ClosingIterator;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.spark.source.SerializableTableWithSize;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.types.StructType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import scala.Tuple2;
/**
* 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>
implements RemoveDanglingDeleteFiles {
- private static final Logger LOG = LoggerFactory.getLogger(RemoveDanglingDeletesSparkAction.class);
private final Table table;
+ private final RemoveDanglingDeleteFilesAction action;
protected RemoveDanglingDeletesSparkAction(SparkSession spark, Table table) {
super(spark);
this.table = table;
+ this.action = new RemoveDanglingDeleteFilesAction(table, this::findDanglingDeletes);
}
@Override
protected RemoveDanglingDeletesSparkAction self() {
return this;
}
+ public RemoveDanglingDeletesSparkAction toBranch(String targetBranch) {
+ action.toBranch(targetBranch);
+ return this;
+ }
+
@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();
- }
-
+ commitSummary().forEach(action::set);
String desc = String.format("Removing dangling delete files in %s", table.name());
- JobGroupInfo info = newJobGroupInfo("REMOVE-DELETES", desc);
- return withJobGroupInfo(info, this::doExecute);
+ return withJobGroupInfo(newJobGroupInfo("REMOVE-DELETES", desc), action::execute);
}
- Result doExecute() {
- RewriteFiles rewriteFiles = table.newRewrite();
- DeleteFileSet danglingDeletes = DeleteFileSet.create();
- danglingDeletes.addAll(findDanglingDeletes());
- danglingDeletes.addAll(findDanglingDvs());
+ private List<DeleteFile> findDanglingDeletes(
+ long snapshotId, List<ManifestFile> deleteManifests) {
+ Broadcast<Table> tableBroadcast =
+ sparkContext().broadcast(SerializableTableWithSize.copyOf(table));
+
+ JavaPairRDD<DeleteFileKey, Void> referencedKeys =
+ sparkContext()
+ .parallelize(ImmutableList.of(snapshotId), 1)
+ .flatMap(
+ snapshot -> {
+ TableScan scan = tableBroadcast.value().newScan().useSnapshot(snapshot);
+ return new ClosingIterator<>(new DeleteFileKeyIterator(scan.planFiles()));
+ })
+ .mapToPair(key -> new Tuple2<>(key, (Void) null));
+
+ JavaPairRDD<DeleteFileKey, DeleteFile> allDeletes =
+ sparkContext()
+ .parallelize(
+ deleteManifests.stream().map(ManifestFileBean::fromManifest).toList(),
+ deleteManifests.size())
+ .flatMap(
+ manifest -> {
+ ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(
Review Comment:
I added the following columns:
```java
private static final List<String> DELETE_COLUMNS =
ImmutableList.of(
"content",
"file_path",
"file_format",
"partition",
"record_count",
"file_size_in_bytes",
"key_metadata",
"split_offsets",
"equality_ids",
"sort_order_id",
"first_row_id",
"referenced_data_file",
"content_offset",
"content_size_in_bytes");
```
The resulting `DeleteFile`s should match the ones from `ManifestFiles.readDeleteManifest` without `.select()`, but with `.copyWithoutStats()` (4747ec7f7121ce8c6491076e4ce902369945db9c).
--
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