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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
