Enable RemoveSnapshots#cleanExpiredSnapshots from ExpireSnapshots spark action and call procedure ?

[ajantha-bhat]

Currently table.expireSnapshots() will clean the data files and manifest files using RemoveSnapshots#cleanExpiredSnapshots.

But the Spark action ExpireSnapshots, always disable cleanExpiredSnapshots. we should provide a method to support it and also in the CALL procedure ? Because I think supporting cleaning then and there is more useful than making files orphan and cleaning it by calling DeleteOrphanFiles action.

If yes, please assign this issue to me.

[ajantha-bhat]

cc: @aokolnychyi , @RussellSpitzer

[ajantha-bhat]

I know that we support deleteWith method in spark action to collect expired data file, manifest file and manifest list.
But user cannot delete it as it might be referenced by other live snapshots.
So as RemoveSnapshots#cleanExpiredSnapshots is already having the logic to find out files that are not referenced by any live snapshots. I think we should support enabling it from actions also.

[RussellSpitzer]

The default behavior of the Action is to remove the files which are no longer live. The difference between the table api and the action, is that the Action determines the set of files to be removed by using a distributed job while the table api does this calculation locally.

The main reason behind writing the action was that the default api does not scale well for extremely large tables. To preserve the functionality of the original api we added the flag which causes the original api to just remove snapshots and not delete files. The Action then takes the difference in state between before and after running the api, and uses that information to delete the files.

This is explained in the Java Doc for the Class

iceberg/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseExpireSnapshotsSparkAction.java

Lines 54 to 67 in 9b285d0

	/**
	* An action that performs the same operation as {@link org.apache.iceberg.ExpireSnapshots} but uses Spark
	* to determine the delta in files between the pre and post-expiration table metadata. All of the same
	* restrictions of {@link org.apache.iceberg.ExpireSnapshots} also apply to this action.
	* <p>
	* This action first leverages {@link org.apache.iceberg.ExpireSnapshots} to expire snapshots and then
	* uses metadata tables to find files that can be safely deleted. This is done by anti-joining two Datasets
	* that contain all manifest and data files before and after the expiration. The snapshot expiration
	* will be fully committed before any deletes are issued.
	* <p>
	* This operation performs a shuffle so the parallelism can be controlled through 'spark.sql.shuffle.partitions'.
	* <p>
	* Deletes are still performed locally after retrieving the results from the Spark executors.
	*/

Here you can see the delete function is applied on the diff set result

iceberg/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseExpireSnapshotsSparkAction.java

Lines 214 to 221 in 9b285d0

	private ExpireSnapshots.Result doExecute() {
	boolean streamResults = PropertyUtil.propertyAsBoolean(options(), STREAM_RESULTS, false);
	if (streamResults) {
	return deleteFiles(expire().toLocalIterator());
	} else {
	return deleteFiles(expire().collectAsList().iterator());
	}
	}

The default delete action removes the files see

iceberg/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseExpireSnapshotsSparkAction.java

Lines 82 to 89 in 9b285d0

	private final Table table;
	private final TableOperations ops;
	private final Consumer<String> defaultDelete = new Consumer<String>() {
	@Override
	public void accept(String file) {
	ops.io().deleteFile(file);
	}
	};

---

So I do not believe any changes are needed here. If a user did want to use a separate delete facility I would actually suggest they use

iceberg/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseExpireSnapshotsSparkAction.java

Line 154 in 9b285d0

public Dataset<Row> expire() {

which was added explicitly for users who have some sort of distributed async delete solution

[ajantha-bhat]

Here you can see the delete function is applied on the diff set result

@RussellSpitzer : Yeah, that's my concern. we cannot just delete the files formed by diff set right? Because those files can still be referred by new snapshots which are not expired. I saw the logic to handle this problem in RemoveSnapshots#cleanExpiredSnapshots

But I didn't see that logic in Actions flow. So, how it is handled in actions flow ?

[RussellSpitzer]

It is handled by using the "ALL" versions of the respective metadata tables which check not just the set of live files, but also all files which are currently reachable by any known snapshot.

See

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

Line 119 in 9b285d0

Dataset<ManifestFileBean> allManifests = loadMetadataTable(table, ALL_MANIFESTS)

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

Line 119 in 9b285d0

Dataset<ManifestFileBean> allManifests = loadMetadataTable(table, ALL_MANIFESTS)

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

Lines 139 to 140 in 9b285d0

	otherMetadataFiles.addAll(ReachableFileUtil.metadataFileLocations(table, false));
	otherMetadataFiles.add(ReachableFileUtil.versionHintLocation(table));

[ajantha-bhat]

got it. Thanks for the clarifications.