Use s3 batch deletes for iceberg expire_snapshots

[homar]

Description

Non-technical explanation

Release notes

( ) This is not user-visible or docs only and no release notes are required.
( ) Release notes are required, please propose a release note for me.
( ) Release notes are required, with the following suggested text:

# Section
* Fix some things. ({issue}`issuenumber`)

[alexjo2144]

There is a max number of keys allowed per request, if you have more than that you need to split them. There should be a constant defined somewhere but it's 1000 keys https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObjects.html

[homar]

Thanks. I am aware of this. For now I did this filtering in IcebergMetadata, I will move it here,

[alexjo2144]

There are a few permissions edge cases we need to think about. For example if you have two directories:

• /root/dir-a
• /root/dir-b

A user might have permission to delete files in dir-a but not dir-b. What should happen if that user tries to batch delete with [/root/dir-a/file-a, /root/dir-b/file-b]

[homar]

There are a few permissions edge cases we need to think about. For example if you have two directories:

• /root/dir-a
• /root/dir-b

A user might have permission to delete files in dir-a but not dir-b. What should happen if that user tries to batch delete with [/root/dir-a/file-a, /root/dir-b/file-b]

Is it possible for table metadata to have files in different locations like this ?
For such a case DeleteObjects would delete the files in dir-a but not in dir-b.
But the previous implementation would have same result. So I think this is ok ?

[findinpath]

I went through the hadoop-apache library and it seems that the components of the path which are used in the initialisation of the FileSystem instance are:

• scheme
• authority

See org.apache.hadoop.fs.Path#getFileSystem , org.apache.hadoop.fs.FileSystem#get(java.net.URI, org.apache.hadoop.conf.Configuration)

    String scheme = uri.getScheme();
    String authority = uri.getAuthority();

I don't know whether the grouping per directory is actually necessary.

[homar]

Probably it is not. @findepi as this was your idea WDYT ?

[findepi]

I went through the hadoop-apache library and it seems that the components of the path which are used in the initialisation of the FileSystem instance are:

• scheme
• authority

Are we talking about contract/interface or implementation?

Also, does https://trino.io/docs/current/connector/hive-s3.html#s3-security-mapping > "prefix" matter?

[findinpath]

I missed the fact that the path can be used to add, based on the Path dynamic configuration settings while creating the FileSystem implementation (see DynamicConfigurationProvider) .
So quite likely the whole path is needed.

See s3 predicate matching:

trino/plugin/trino-hive/src/main/java/io/trino/plugin/hive/s3/S3SecurityMapping.java

Lines 158 to 165 in fe608f2

	private static Predicate<URI> prefixPredicate(URI prefix)
	{
	checkArgument("s3".equals(prefix.getScheme()), "prefix URI scheme is not 's3': %s", prefix);
	checkArgument(prefix.getQuery() == null, "prefix URI must not contain query: %s", prefix);
	checkArgument(prefix.getFragment() == null, "prefix URI must not contain fragment: %s", prefix);
	return value -> extractBucketName(prefix).equals(extractBucketName(value)) &&
	value.getPath().startsWith(prefix.getPath());
	}

[findepi]

if we were to fully honor it, we would need to poke into dynamic configuration and see whether all paths in the group would yield the same configs. It's possible (Trino has all the information required), but hard (the information is well interfaced-out and inaccessible). I don't think there is any reasonable real-life use-case to have different configuration for two snapshot files of a table. That's why i deemed grouping by directory should work well in practice.

cc @losipiuk @arhimondr @electrum

[findinpath]

I spent some time thinking about this functionality and came up with a few ideas. Hopefully they will be useful for going further with the PR.

Both TrinoFileSystem and HdfsFileSystem do not have at the time of this writing a "batch" delete functionality.
This seems to be a particularity of TrinoS3FileSystem.

Grouping the files to remove after scheme could allow us to know which files are on a s3 file system.

With io.trino.plugin.hive.util.HiveWriteUtils#getRawFileSystem (it needs a path - take the first one from the group) we can know whether we are dealing with a TrinoS3FileSystem. In this specific case pass the files from the group directly to the raw file system to be deleted in batches. Otherwise, use the default TrinoFileSystem#delete method.

The wobbly part in the exposed algorithm is that getting the raw file system may eventually bypass unintentionally some FilterFileSystem which wraps the TrinoS3FileSystem. However, since we are using a method which does not belong to the FileSystem class, I think we are on the safe side.

[homar]

I spent some time thinking about this functionality and came up with a few ideas. Hopefully they will be useful for going further with the PR.

Both TrinoFileSystem and HdfsFileSystem do not have at the time of this writing a "batch" delete functionality. This seems to be a particularity of TrinoS3FileSystem.

Grouping the files to remove after scheme could allow us to know which files are on a s3 file system.

With io.trino.plugin.hive.util.HiveWriteUtils#getRawFileSystem (it needs a path - take the first one from the group) we can know whether we are dealing with a TrinoS3FileSystem. In this specific case pass the files from the group directly to the raw file system to be deleted in batches. Otherwise, use the default TrinoFileSystem#delete method.

The wobbly part in the exposed algorithm is that getting the raw file system may eventually bypass unintentionally some FilterFileSystem which wraps the TrinoS3FileSystem. However, since we are using a method which does not belong to the FileSystem class, I think we are on the safe side.

Sorry but I am not sure what is the difference between this and the current implementation - apart from obvious thing like I use entire path until last '/' vs use only scheme ?

[homar]

I spent some time thinking about this functionality and came up with a few ideas. Hopefully they will be useful for going further with the PR.
Both TrinoFileSystem and HdfsFileSystem do not have at the time of this writing a "batch" delete functionality. This seems to be a particularity of TrinoS3FileSystem.
Grouping the files to remove after scheme could allow us to know which files are on a s3 file system.
With io.trino.plugin.hive.util.HiveWriteUtils#getRawFileSystem (it needs a path - take the first one from the group) we can know whether we are dealing with a TrinoS3FileSystem. In this specific case pass the files from the group directly to the raw file system to be deleted in batches. Otherwise, use the default TrinoFileSystem#delete method.
The wobbly part in the exposed algorithm is that getting the raw file system may eventually bypass unintentionally some FilterFileSystem which wraps the TrinoS3FileSystem. However, since we are using a method which does not belong to the FileSystem class, I think we are on the safe side.

Sorry but I am not sure what is the difference between this and the current implementation - apart from obvious thing like I use entire path until last '/' vs use only scheme ?

Oh wait a second I mixed changes from different PRs, yeap checking down with getRawFile system may be useful here. I will take a look

[homar]

I hate the fact that I am doing similar thing twice but I didn't see a way to avoid this repetition..

[findepi]

call it deleteFiles and put right after deleteFile method.

also, let's have a javadoc explaining whether the operation is supposed to be atomic (it's not)

[findepi]

the if (!paths.isEmpty()) condition is redundant.

[findepi]

what if the path does not contain a slash at all? lastIndexOf will return -1 and substring will fail.
i know in practice it cannot have no slashes, but the intent of this line is not to validate this -- throwing some stupid exception if the caller e.g. erroneously provided a relative path.

you can avoid this with path.replaceFirst("/[^/]*$", ""). I assume regex's overhead is a problem here.

[homar]

I actually thought about this and assumed such a situation is not possible. But sure I will change this

[findepi]

iterate over entry set

[findepi]

.stream().findFirst().get() -> .get(0)?

but you're pulling from a wrong collection (i.e. it should be this directory's first path, not the paths first path)

[findepi]

remove this since on next line you check it's precisely 1.
instead, assert (earlier) that initialSnapshots.size() is more than 1

btw use assertThat(collection).hasSize, so that in case of failure, the exception message carries the collection contents.

[findepi]

listagg looks cool, but what is this for here? looks like redundant complexity, especially that we have just 2 rows
let's test with simpler query

[homar]

Because I always struggle to write tests that asserts more than one row :D

[findepi]

why 2? why size > 0?

document.
also recognizing batch by content-length looks very indirect. is there anything that looks more direct? eg what's the content?

[findepi]

maybe instead of this -- have the table with 10 snapshots and verify there is only deletion request to MinIO, not 10.

[homar]

ehh I thought I explained that during offline discussion. I can create 100 snapshots and there will be 100 events registered by Minio. I didn't manage to see incoming requests I can only see events that are registered and based on that I was not able to find a better solution than this one based on content-length. I was not able to access content itself.

BUT i found a little better way now :D

[findepi]

testExpireSnapshotsBatchDeletes ?

[findepi]

move (as in other cases)

[findepi]

Should we honor requesterPaysEnabled?

[homar]

yes, definitely, according to aws/aws-sdk-java#1219 that is a problem for single delete it should work for batch delete so I will simply set it

[homar]

in a threaded context many things go wrong

ArrayList can get corrupted, we may loose files
pathsToDelete.clear(); can be invoked with some paths never sent to deletion
deleteFiles can be invoked twice on same/similar sets of paths and may result in "file missing so cannot delete" exception

As I explained above, currently I assume this is run with a single thread. I am aware of all these problems if it would be called by many threads

[findepi]

Suggested change

	// deleteFunction should be thread safe if .executeDeleteWith() is used for expireSnapshots
	// deleteFunction is not accessed from multiple threads unless .executeDeleteWith() is used

[homar]

CI: #14686

[findepi]

CI: #14686

This prevented tests from running, so you need to eg push empty commit to have CI re-run.

[findepi]

it would be nice to have @electrum 's eyes on filesystem changes, but I know he's busy.
Will merge sometime later this week.

[alexjo2144]

Does this work if the file is in the root directory of the file system?

[homar]

Then it will be grouped by Path("" ) - is it incorrect ?

[alexjo2144]

Probably something more like s3://bucket/ in the s3 case at least. I think it's fine

[homar]

yes it should be fine imho

[alexjo2144]

ICEBERG_FILESYSTEM_ERROR ?

[alexjo2144]

ICEBERG_FILESYSTEM_ERROR ?

[homar]

CI: #14787

[losipiuk]

Aren't we moving away from use of hadoop's FileSystem towards TrinoFileSystem. Should TrinoFileSystem support batch delete?

[homar]

Maybe it should, though currently only HdfsFileSystem implements it and it doesn't support batch delete.