HADOOP-18679. Add API for bulk/paged delete of files

[mukund-thakur]

HADOOP-18679

Adding tests on top of #6494

Description of PR

How was this patch tested?

For code changes:

• Does the title or this PR starts with the corresponding JIRA issue id (e.g. 'HADOOP-17799. Your PR title ...')?
• Object storage: have the integration tests been executed and the endpoint declared according to the connector-specific documentation?
• If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under ASF 2.0?
• If applicable, have you updated the LICENSE, LICENSE-binary, NOTICE-binary files?

[steveloughran]

commented. I've also done a PR #6738 which tunes the API to work with iceberg, having just written a PoC of the iceberg binding.

My PR

• moved the wrapper methods to a new wrappedio.WrappedIO class
• add a probe for the api being available
• I also added an availability probe in the interface. not sure about that as we really should make it available everywhere, always.

Can you cherrypick this PR onto your branch and then do the review comments.

After which, please do not do any rebasing of your PR. That way, it is easier for me too keep my own branch in sync with your changes. Thanks.

PoC of iceberg integration, based on their S3FileIO one.

https://github.com/steveloughran/iceberg/blob/s3/HADOOP-18679-bulk-delete-api/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java#L208

The iceberg api passes in a collection of paths, which may span multiple filesystems.

To handle this,

• the bulk delete API should take a Collection, not a list
• it needs to be implemented in every FS, because trying to distinguish case-by-case on support would be really complex.

[steveloughran]

commented

[steveloughran]

this is always 1, isn't it? so much can be simplified here

• no need for the field
• no need to pass it in the constructor
• pageSize() to return 1

[steveloughran]

there's only even going to be 1 entry here

[steveloughran]

good q. I'd say yes. or actually do a getFileStatus() cal and see what is there for

• file doesn't exist (not an error)
• path is a directory: add to result

key is that file not found isn't escalated

[steveloughran]

on a store where bulk delete page size == 1, use that as the range

[hadoop-yetus]

💔 -1 overall

Vote	Subsystem	Runtime	Logfile	Comment
			_ Prechecks _
+1 💚	dupname	0m 05s		No case conflicting files found.
+0 🆗	codespell	0m 01s		codespell was not available.
+0 🆗	detsecrets	0m 01s		detect-secrets was not available.
+0 🆗	xmllint	0m 01s		xmllint was not available.
+0 🆗	spotbugs	0m 01s		spotbugs executables are not available.
+0 🆗	markdownlint	0m 01s		markdownlint was not available.
+1 💚	@author	0m 00s		The patch does not contain any @author tags.
+1 💚	test4tests	0m 00s		The patch appears to include 6 new or modified test files.
			_ trunk Compile Tests _
+0 🆗	mvndep	2m 31s		Maven dependency ordering for branch
+1 💚	mvninstall	91m 09s		trunk passed
+1 💚	compile	40m 32s		trunk passed
+1 💚	checkstyle	6m 09s		trunk passed
-1 ❌	mvnsite	4m 42s	/branch-mvnsite-hadoop-common-project_hadoop-common.txt	hadoop-common in trunk failed.
+1 💚	javadoc	14m 12s		trunk passed
+1 💚	shadedclient	171m 45s		branch has no errors when building and testing our client artifacts.
			_ Patch Compile Tests _
+0 🆗	mvndep	2m 24s		Maven dependency ordering for patch
+1 💚	mvninstall	11m 00s		the patch passed
+1 💚	compile	38m 33s		the patch passed
+1 💚	javac	38m 33s		the patch passed
-1 ❌	blanks	0m 00s	/blanks-eol.txt	The patch has 5 line(s) that end in blanks. Use git apply --whitespace=fix <<patch_file>>. Refer https://git-scm.com/docs/git-apply
+1 💚	checkstyle	6m 31s		the patch passed
-1 ❌	mvnsite	4m 36s	/patch-mvnsite-hadoop-common-project_hadoop-common.txt	hadoop-common in the patch failed.
+1 💚	javadoc	14m 19s		the patch passed
+1 💚	shadedclient	185m 02s		patch has no errors when building and testing our client artifacts.
			_ Other Tests _
-1 ❌	asflicense	5m 46s	/results-asflicense.txt	The patch generated 1 ASF License warnings.
		555m 55s

Subsystem	Report/Notes
GITHUB PR	#6726
Optional Tests	dupname asflicense compile javac javadoc mvninstall mvnsite unit shadedclient codespell detsecrets xmllint spotbugs checkstyle markdownlint
uname	MINGW64_NT-10.0-17763 cfb6e8c364ad 3.4.10-87d57229.x86_64 2024-02-14 20:17 UTC x86_64 Msys
Build tool	maven
Personality	/c/hadoop/dev-support/bin/hadoop.sh
git revision	trunk / 7415427
Default Java	Azul Systems, Inc.-1.8.0_332-b09
Test Results	https://ci-hadoop.apache.org/job/hadoop-multibranch-windows-10/job/PR-6726/1/testReport/
modules	C: hadoop-common-project/hadoop-common hadoop-tools/hadoop-aws hadoop-tools/hadoop-azure U: .
Console output	https://ci-hadoop.apache.org/job/hadoop-multibranch-windows-10/job/PR-6726/1/console
versions	git=2.44.0.windows.1
Powered by	Apache Yetus 0.14.0 https://yetus.apache.org

This message was automatically generated.

[steveloughran]

commented.

• I'd prefer that BulkDeleteSource to be in FileSystem, with the base implementatoon returning a DefaultBulkDeleteOperation instance.
• and so we should have one of the contract tests doing it directly through the API on the getFileSystem() instance
• other than that though, looking really good.

[steveloughran]

Commented. Now, one more thing based on my changes in #6686.

it's adding many more methods to this class, so I'm giving them the name of the class/interface + "_" + method name.

can you do the same here? some style checker will complain but it will help us to separate the methods in the new class.

other than that, all good

[steveloughran]

This doesn't hold for the subclasses. better to say

Create a bulk delete operation.
The default implementation returns an instance of {@link DefaultBulkDeleteOperation}

[mukund-thakur]

can you do the same here? some style checker will complain but it will help us to separate the methods in the new class.

I don't understand what to do here.

[steveloughran]

@mukund-thakur tried to clarify what i mean: we have the class/interface split from the method/operation by a _ character

it makes a lot more sense when you look at my PR, which brings a lot more methods into the same class. Why all the same class? less reflection code

[steveloughran]

rename bulkDelete_pageSize

[steveloughran]

rename bulkDelete_delete

[hadoop-yetus]

💔 -1 overall

Vote	Subsystem	Runtime	Logfile	Comment
			_ Prechecks _
+1 💚	dupname	0m 07s		No case conflicting files found.
+0 🆗	codespell	0m 01s		codespell was not available.
+0 🆗	detsecrets	0m 01s		detect-secrets was not available.
+0 🆗	xmllint	0m 01s		xmllint was not available.
+0 🆗	spotbugs	0m 01s		spotbugs executables are not available.
+0 🆗	markdownlint	0m 01s		markdownlint was not available.
+1 💚	@author	0m 01s		The patch does not contain any @author tags.
+1 💚	test4tests	0m 00s		The patch appears to include 11 new or modified test files.
			_ trunk Compile Tests _
+0 🆗	mvndep	2m 42s		Maven dependency ordering for branch
+1 💚	mvninstall	107m 34s		trunk passed
+1 💚	compile	48m 30s		trunk passed
+1 💚	checkstyle	7m 10s		trunk passed
-1 ❌	mvnsite	5m 19s	/branch-mvnsite-hadoop-common-project_hadoop-common.txt	hadoop-common in trunk failed.
+1 💚	javadoc	23m 51s		trunk passed
+1 💚	shadedclient	225m 30s		branch has no errors when building and testing our client artifacts.
-0 ⚠️	patch	228m 42s		Used diff version of patch file. Binary files and potentially other changes not applied. Please rebase and squash commits if necessary.
			_ Patch Compile Tests _
+0 🆗	mvndep	2m 57s		Maven dependency ordering for patch
+1 💚	mvninstall	19m 17s		the patch passed
+1 💚	compile	45m 43s		the patch passed
+1 💚	javac	45m 43s		the patch passed
+1 💚	blanks	0m 01s		The patch has no blanks issues.
+1 💚	checkstyle	7m 25s		the patch passed
-1 ❌	mvnsite	5m 28s	/patch-mvnsite-hadoop-common-project_hadoop-common.txt	hadoop-common in the patch failed.
+1 💚	javadoc	24m 00s		the patch passed
+1 💚	shadedclient	232m 12s		patch has no errors when building and testing our client artifacts.
			_ Other Tests _
+1 💚	asflicense	7m 14s		The patch does not generate ASF License warnings.
		700m 51s

Subsystem	Report/Notes
GITHUB PR	#6726
Optional Tests	dupname asflicense compile javac javadoc mvninstall mvnsite unit shadedclient codespell detsecrets xmllint spotbugs checkstyle markdownlint
uname	MINGW64_NT-10.0-17763 296d6abd6fb2 3.4.10-87d57229.x86_64 2024-02-14 20:17 UTC x86_64 Msys
Build tool	maven
Personality	/c/hadoop/dev-support/bin/hadoop.sh
git revision	trunk / e37d88f
Default Java	Azul Systems, Inc.-1.8.0_332-b09
Test Results	https://ci-hadoop.apache.org/job/hadoop-multibranch-windows-10/job/PR-6726/5/testReport/
modules	C: hadoop-common-project/hadoop-common hadoop-hdfs-project/hadoop-hdfs hadoop-tools/hadoop-aws hadoop-tools/hadoop-azure U: .
Console output	https://ci-hadoop.apache.org/job/hadoop-multibranch-windows-10/job/PR-6726/5/console
versions	git=2.44.0.windows.1
Powered by	Apache Yetus 0.14.0 https://yetus.apache.org

This message was automatically generated.

[steveloughran]

mukund, if you can do those naming changes then I'm +1

[steveloughran]

+1

I was about to merge then I realised that yetus wasn't ready. Here is my draft commit message

1. as soon as this is in I will rebase HADOOP-19131. Assist reflection IO with WrappedOperations class #6686 onto it, which extends WrappedIO and adds the reflection utility classes from Parquet to assist in testing.
2. I'll leave you to do the cherrypick and merge onto 3.4.x
3. And I want to get a minimal version into 3.3.x, maybe with a page size of 1 even on S3A, but without the safety checks, so still saves on LIST/HEAD calls.

---

create a BulkDelete implementation from a
BulkDeleteSource; the BulkDelete interface provides
the pageSize(): the maximum number of entries which can be
deleted, and a bulkDelete(Collection paths)
method which can take a collection up to pageSize() long.

This is optimized for object stores with bulk delete APIs;
the S3A connector will offer the page size of
fs.s3a.bulk.delete.page.size unless bulk delete has
been disabled.

Even with a page size of 1, the S3A implementation is
more efficient than delete(path)
as there are no safety checks for the path being a directory
or probes for the need to recreate directories.

The interface BulkDeleteSource is implemented by
all FileSystem implementations, with a page size
of 1 and mapped to delete(pathToDelete, false).
This means that callers do not need to have special
case handling for object stores versus classic filesystems.

To aid use through reflection APIs, the class
org.apache.hadoop.io.wrappedio.WrappedIO
has been created with "reflection friendly" methods.

Contributed by Mukund Thakur and Steve Loughran

[steveloughran]

merged to trunk. mukund, can you do the backport to branch-3.4.1, while we can think about what to do for 3.3.9? speaking of which, we should think about that...

[steveloughran]

fyi #6686 adds dynamic load for the wrapped methods, and calls them through the tests.

[mukund-thakur]

While backporting this to branch-3.4 I see this failure. Will check if this is happening on trunk as well.
[INFO] [ERROR] Failures: [ERROR] TestStagingCommitter.testJobCommitFailure:662 [Committed objects compared to deleted paths org.apache.hadoop.fs.s3a.commit.staging.StagingTestBase$ClientResults@2de1acf4{ requests=12, uploads=12, parts=12, tagsByUpload=12, commits=5, aborts=7, deletes=0}] Expecting: <["s3a://bucket-name/output/path/r_0_0_c055250c-58c7-47ea-8b14-215cb5462e89", "s3a://bucket-name/output/path/r_1_1_9111aa65-96c2-465c-b278-696aff7707e3", "s3a://bucket-name/output/path/r_0_0_dec7f398-ee4e-4a53-a783-6b72cead569a", "s3a://bucket-name/output/path/r_1_1_39ad0eba-1053-4217-aa63-ddc8edfa7c64", "s3a://bucket-name/output/path/r_0_0_6c0518f6-7c1b-418f-a3e4-7db568880e6a"]> to contain exactly in any order: <[]> but the following elements were unexpected: <["s3a://bucket-name/output/path/r_0_0_c055250c-58c7-47ea-8b14-215cb5462e89", "s3a://bucket-name/output/path/r_1_1_9111aa65-96c2-465c-b278-696aff7707e3", "s3a://bucket-name/output/path/r_0_0_dec7f398-ee4e-4a53-a783-6b72cead569a", "s3a://bucket-name/output/path/r_1_1_39ad0eba-1053-4217-aa63-ddc8edfa7c64", "s3a://bucket-name/output/path/r_0_0_6c0518f6-7c1b-418f-a3e4-7db568880e6a"]>

[steveloughran]

yeah, just seen those too. the mocked s3 client isn't getting the delete one object calls via the store object.

got a small PR up to help debug it, which is not a fix

[steveloughran]

problem is store is null in innermost s3afs, triggers an NPE in deleteObject() before the aws client has its delete operation called, so list of deleted paths is not updated.

easiest immediate fix is to mock deleteObject(), longer term we should actually be stubbing the entire store, as that's what the interface/impl split is meant to assist

[mukund-thakur]

fix this with mockito magic somehow. https://github.com/apache/hadoop/pull/6843/files
this was not easy to debug.

[steveloughran]

I tried to do it yesterday too. I think my solution would have pulled out createStore() into a method and overrode it, or added S3AInternals.setStore() call.

[steveloughran]

update, #5081 does successfully show the staging UT failure (good!) but also some in hadoop-common due to the new interface...we need to modify the tests to indicate it is safe to not override the base implementation.

[alkis]

It is great that we call this out, but can we do better at the API level?

Had the API took Collection<FileStatus>, we could make the operation idempotent in most clouds by passing the version/generationid/etag in the request.

[steveloughran]

Aws sdk delete with version id actually requires more IAM permissions than unversioned delete, which always removes HEAD object, because granting that permission allows the caller to delete backups. Deployments where apps can delete HEAD but not versions are not unusual for this reason.

This is why S3A doesn't use it even in simple listing -> delete calls where the status is known.

you might also need to issue getFileStatus/list calls, which would massively increase the cost if the process didn't have those values already.

A bulk delete with a tuple of (path, version) for each entry could work, if the store could be configured to use that version ID/type. for S3A we would leave it off by default. the tuple would be Map.entry to be reflection friendly.

if you do thing version/etag support would be a blocker to use, well, things haven't shipped yet, though @mukund-thakur is preparing a 3.4.1 alpha release.

You (and it would be you, sorry) will need to modify the api with

• Collection<Map.Entry<Path, version>>[]
• S3A impl to not use version by default, option to turn it on, parameterized testing for this if a versioned bucket is the test bucket.
• WrappedIO changed to match

This isn't that useful for table compaction, as the engines tend to use randomness in their names to spread the s3 store load across shards. But it could have other uses.

for example. here's some work to do version printing, recovery and copy within the same bucket, lets you pull out the layers underneath a directory tree

https://github.com/steveloughran/cloudstore/tree/main/src/main/extra/org/apache/hadoop/fs/s3a/extra

question is: do we want to make something that complex part of a broader api with tests, specification, commitments to maintain etc, or do we just say call S3A.getInternals().getClient() and then sort it out yourself?

[steveloughran]

• some interface getVersionForDeleteOperations() -> String to put on s3/abfs file status, the way we do for getEtag(), so you can extract version info without having to play class cast games, built with hadoop-aws and hadoop-azure on CP etc.

#3633 shows what's required there.

[steveloughran]

BTW, the "is delete idempotent?" question is a recurrent one; search the mail lists for history on it, including even HDFS. In a world with parallel writers, it clearly isn't. But if you are going from paged list -> delete you are already getting non-atomic listings where race conditions can do odd things (rename file zzzzzzz to aaaaa and not have the listing find it at all).

You can rebuild S3AFS with InternalConstants.DELETE_CONSIDERED_IDEMPOTENT set to false. Nobody has ever done this and nobody has ever complained. Just avoid using it as part of atomic commit protocols and you'll be fine

[alkis]

Thank you for the context @steveloughran. Lots of good info above.

Should we be adding an overload that takes FileStatus that pushes VersionIds in delete? Then caller can use it whenever FileStatuses are available.

To close, I want to mention that idempotency of operations is not only useful in the face of multiple (application) writers. It also plays a role when only one writer is involved. The reason is the distributed nature of the cloud storage and loadbalances/backends inside cloud storage implementation. For example:

1. writer lists a prefix, sees 2 files (a and b)
2. writer issues delete of 2 files to IP1 because of an intermitted connection issue at IP1
3. writers request times out
4. writer retries issuing deletes to IP2 and succeeds
5. writer creates file a
6. meanwhile IP1 network is back up and attempts to execute the deletes in (2) and deletes files a and b
7. writer observes the deletion of a which it just created

Had the request in (2) and (4) been idempotent, writer wouldn't observe a being deleted at (7) because the deletes would target specific versions of the files.

[steveloughran]

That scenario is a key part of the whole delete idempotency discussion. And as noted, trying to implement idempotency there on s3 requires permissions which apps are often not trusted with.

As for abfs, I'd worry more about rename resilience under load than deletion. If you've not hit problems there: you've not generated enough load. Look at HADOOP-18613 and HADOOP-18012 there.

Really you can't rely on delete or rename doing what you expect, so your commit protocol had better not rely on them. Though #6716 shows that even when I think I've done that, I can get caught out.
Be ware that step 2 of your failure condition is more likely to be due to a throttle response triggering backoff and retry rather than actually connectivity failures.

As for the "overload which takes FileStatus", look at openFile() work and followup issues in the s3a and abfs codebase which caught us out. In particular, any virtual FS on top (hive, viewFS, maybe the databricks one...) builds their own FileStatus instances because the path in them is absolute and includes the full URL. Which means the file status passed in is often of a different type from any FS specific implementation, lacks all version/etag info and whose path doesn't resolve.

it'd have to take path and filestatus separately.

Returning to this API, look at apache/iceberg#10233 for the latest integration; it is only given strings to map to paths, which is what the new API takes.

For a commit process with idempotency, are you looking for rename with versioning? or even a createFile() with if-matches condition on underlying files...something Azure supports.

adding if-matches into createFile() is possible if you want to design that, the spec would need to say "failure MAY happen at close() to support any store where the put/post only happens at end. And you need a way to pass that version of the file being overwritten (etag, version) down. Assuming this was etag only, we have that API though wrapper filesystems don't let pick it up (not in the base class; adding new field would break serialization across versions...).

If you want this *and are prepared to do the work, at least for stores which implement it (azure) then I'm happy to supervise your effort, especially if we can get the MSFT engineers involved. Create that JIRA and once you put up your first design I'll assign the JIRA to you.

It'd have target 3.4.2., but as getEtag() and openFile() have shipped for a while

string etag;
status = fs.getFileStatus(dest); (catch FNFE, set etag to "")
etag = status.getEtag()

FSDataOutputStream out = openFile(dest).
 must("fs.create.option.if-match", etag)
 .build()
...
out.close();

• catch FileAlreadyExistsException in open or close as a failure sign. if IllegalArgumentException is raised in build, that if-match tag is unknown/unsupported.

like I said, gets complex fast -but if you want something stable to work with current and future stores, it'd have to be the way. None of this needs new public FS APIs, just standard behaviour in stores which implement it.

go for it!