Core: Delegate bulk and prefix operations in ResolvingFileIO #8169

bryanck · 2023-07-28T00:07:19Z

This PR adds implementing the SupportsPrefixOperations and SupportsBulkOperations interfaces in ResolvingFileIO. Currently prefix and bulk operations of the delegate FileIO aren't accessible. In the Spark delete orphan files action, this prevents bulk deletes from being used with ResolvingFileIO, for example. Currently all of the supported delegate FileIO classes implement those interfaces.

RussellSpitzer · 2023-08-02T21:52:17Z

core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java

+
+    PeekingIterator<String> iterator = Iterators.peekingIterator(originalIterator);
+    FileIO fileIO = io(iterator.peek());
+    if (!(fileIO instanceof SupportsPrefixOperations)) {


We check prefix, but then throw an exception warning we don't support bulk

RussellSpitzer · 2023-08-02T21:52:30Z

core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java

+          "FileIO doesn't support bulk operations: " + fileIO.getClass().getName());
+    }
+
+    ((SupportsBulkOperations) fileIO).deleteFiles(() -> iterator);


We didn't check this cast

We also should probably pass through the original iterable. Just in case this is a closable iterable.

core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java

RussellSpitzer · 2023-08-02T21:57:07Z

core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java

+  @Override
+  public Iterable<FileInfo> listPrefix(String prefix) {
+    FileIO fileIO = io(prefix);
+    if (!(fileIO instanceof SupportsPrefixOperations)) {


This breaks if the delegate can't actually list prefix

RussellSpitzer

This needs some tests but I think has some underlying issues. We can't expose that we support the various interfaces SupportsPrefix, SupportsBulk and then fail if the caller attempts to use them. If for example you were using a ResolvingFileIO and the underlying file IO didn't support prefix and bulk, this would break all usage of that IO.

When the underlying system doesn't support the interface we need to fall back to using the standard apis.

bryanck · 2023-08-02T22:02:26Z

Part of the assumption was that all of the supported delegates already (or will) implement those interfaces (HadoopFileIO, GCSFileIO, and S3FileIO), and my thought was that a FileIO implementation would be required to support those before being added here. So currently there won't be any typecasting issues.

RussellSpitzer · 2023-08-02T22:02:58Z

core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java

-public class ResolvingFileIO implements FileIO, HadoopConfigurable {
+/**
+ * FileIO implementation that uses location scheme to choose the correct FileIO implementation.
+ * Delegate FileIO implementations should support the mixin interfaces {@link


If we want to make this a requirement we should probably just add these as perquisites in the init code?

That makes sense to me, I'll add that.

bryanck · 2023-08-03T03:16:42Z

One note, if this change ends up being acceptable, it shouldn't be merged before #8168, as that adds the now-mandatory mixin interfaces to GCSFileIO.

RussellSpitzer · 2023-08-03T14:48:43Z

core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java

+    }
+
+    PeekingIterator<String> iterator = Iterators.peekingIterator(originalIterator);
+    FileIO fileIO = io(iterator.peek());


why don't we change the function io to return a FileIO which supportsPrefixOperations and BulkOperations if that's a requirement?

Thanks for the suggestion, I made this change and removed the typecasts.

RussellSpitzer · 2023-08-03T19:44:34Z

core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java

+
+    PeekingIterator<String> iterator = Iterators.peekingIterator(originalIterator);
+    SupportsBulkOperations fileIO = io(iterator.peek());
+    fileIO.deleteFiles(() -> iterator);


Did we resolve this? I think we probably want to pass the iterable here rather than the iterator to make sure we close everything properly

No not yet, I'm still thinking through the best solution. Passing the original iterable could cause an additional request to be made, i.e. one additional request for the peek, so I was hoping to avoid that.

RussellSpitzer · 2023-08-03T19:47:09Z

core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java

    String impl = implFromLocation(location);
-    FileIO io = ioInstances.get(impl);
+    T io = (T) ioInstances.get(impl);


Can we add this to the map type as well? I always forget if Java can do this, I know Scala can

I originally had it that way, but it required changing the class declaration and caused the revapi check to fail, I'm not sure what the tolerance is to make that change

Well we are forcing that, so the revapi change is appropriate

Ie, Runtime casting is just letting us sneak past the RevAPI static check

OK great. I made this change and added the revapi exception

After further thought, this isn't a great solution as it allows declaring a ResolvingFileIO with an incompatible generic type, e,g, new ResolvingFileIO<MyType>(), so I'll need to revert this change.

I don't believe there is another way to declare the map as something like Map<String, FileIO & SupportsPrefixOperations & SupportsBulkOperations>

You could always define a small inner abstract class, DelgateIO extends ....

Yes, then update S3FileIO, GCSFileIO, and HadoopFileIO to implement that, which makes it more clear what you need to implement if developing a new delegate class. I'll make that change.

I'll need to wait on the GCSFileIO PR to add prefix and bulk operations before that can implement this new interface, but we should wait for that anyway before merging this.

nastra · 2023-08-04T10:38:49Z

core/src/test/java/org/apache/iceberg/io/TestResolvingIO.java

@@ -19,8 +19,16 @@
 package org.apache.iceberg.io;

 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatCode;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;


I believe these imports are wrong and should be ...Assertions.assertThatThrownBy rather than AssertionsForClassTypes

Thanks, I fixed that

This reverts commit f7b4530.

RussellSpitzer · 2023-08-04T16:10:03Z

core/src/test/java/org/apache/iceberg/io/TestResolvingIO.java

+
+  @Test
+  public void testEnsureInterfaceImplementation() {
+    ResolvingFileIO testResolvingFileIO = spy(new ResolvingFileIO());


I think this is ok, but what I was hoping we would have coverage for was that all delegates properly worked. This may be very difficult, I'm just looking for a way to confirm that the class will work at runtime. If we can force all the checks at compile time I'd be ok with that as well.

We wouldn't be able to test the actual delegates without having the dependencies here (i.e. iceberg-aws and iceberg-gcp). I'll see if a service pattern could work where the delegate types register themselves with ResolvingFileIO, then we can have the compile time check.

I looked into using Java SPI, but I don't think it is a good idea as it could cause problems with shadow jars. Also it isn't much different than this solution. We can't register with ResolvingFileIO from the delegate class as the delegate class might not be loaded at all yet.

One option, somewhat similar to #7976 would be to remove SupportsPrefixOperations and then provide a default implementation for bulk delete for the orphan file case so the delegate doesn't need the mixin interface. Though I was hoping we could have the prefix operations also as they won't be accessible at all.

I added tests for each of the delegate types to ensure they load. Though that doesn't prevent a new class from being added to the list that doesn't implement the interface.

bryanck · 2023-08-04T20:30:48Z

I'd like to close this PR in favor of #7976, which focuses on the immediate problem of orphan files not doing batch deletes. There are opportunities to improve FileIO resolution/delegation but this PR wasn't created with that fully in mind, so a new PR to address that would make more sense to me.

bryanck · 2023-08-04T21:32:25Z

Apologies for taking up people's time on this, hopefully I or someone else will have an improved PR to address some of the issues raised.

Core: Delegate bulk and prefix operations in ResolvingFileIO

3ac5713

github-actions bot added the core label Jul 28, 2023

check for empty iterator

2aace57

RussellSpitzer reviewed Aug 2, 2023

View reviewed changes

core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java Show resolved Hide resolved

RussellSpitzer reviewed Aug 2, 2023

View reviewed changes

RussellSpitzer requested changes Aug 2, 2023

View reviewed changes

RussellSpitzer reviewed Aug 2, 2023

View reviewed changes

bryanck added 2 commits August 2, 2023 19:46

Ensure delegate FileIO implement necessary interfaces

2d129e9

add test for interface assertion

df3e850

simplify test

2009bd2

RussellSpitzer reviewed Aug 3, 2023

View reviewed changes

bryanck added 2 commits August 3, 2023 08:13

return composite type and remove type casts

19230d5

avoid updating class declaration

9d73070

RussellSpitzer reviewed Aug 3, 2023

View reviewed changes

use composite type in cache and add revapi exception

f7b4530

nastra reviewed Aug 4, 2023

View reviewed changes

bryanck added 3 commits August 4, 2023 05:20

Revert "use composite type in cache and add revapi exception"

f6ea325

This reverts commit f7b4530.

fix test import

d4c591f

add delegate interface - waiting on GCSFileIO

8e6d9a1

github-actions bot added the AWS label Aug 4, 2023

bryanck added 2 commits August 4, 2023 06:31

test fix

839da9d

move delegate interface to API

065dd0b

github-actions bot added the API label Aug 4, 2023

RussellSpitzer reviewed Aug 4, 2023

View reviewed changes

bryanck mentioned this pull request Aug 4, 2023

Core: Extend ResolvingFileIO to support BulkOperations #7976

Merged

tests for file IO loading

076aa43

github-actions bot added the GCP label Aug 4, 2023

bryanck added 2 commits August 4, 2023 10:56

import fix

ce32d5c

test fix

109d385

bryanck closed this Aug 4, 2023

nastra mentioned this pull request Aug 21, 2023

Core: Extend ResolvingFileIO to support prefix-based operations #8334

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Core: Delegate bulk and prefix operations in ResolvingFileIO #8169

Core: Delegate bulk and prefix operations in ResolvingFileIO #8169

bryanck commented Jul 28, 2023

RussellSpitzer Aug 2, 2023

RussellSpitzer Aug 2, 2023

RussellSpitzer Aug 2, 2023

RussellSpitzer Aug 2, 2023

RussellSpitzer left a comment

bryanck commented Aug 2, 2023 •

edited

Loading

RussellSpitzer Aug 2, 2023

bryanck Aug 2, 2023

bryanck commented Aug 3, 2023

RussellSpitzer Aug 3, 2023

bryanck Aug 3, 2023

RussellSpitzer Aug 3, 2023

bryanck Aug 3, 2023

RussellSpitzer Aug 3, 2023

bryanck Aug 3, 2023

RussellSpitzer Aug 3, 2023 •

edited

Loading

bryanck Aug 4, 2023

bryanck Aug 4, 2023 •

edited

Loading

bryanck Aug 4, 2023 •

edited

Loading

RussellSpitzer Aug 4, 2023

bryanck Aug 4, 2023

bryanck Aug 4, 2023

nastra Aug 4, 2023

bryanck Aug 4, 2023

RussellSpitzer Aug 4, 2023

bryanck Aug 4, 2023

bryanck Aug 4, 2023

bryanck Aug 4, 2023

bryanck commented Aug 4, 2023

bryanck commented Aug 4, 2023

Core: Delegate bulk and prefix operations in ResolvingFileIO #8169

Core: Delegate bulk and prefix operations in ResolvingFileIO #8169

Conversation

bryanck commented Jul 28, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

RussellSpitzer left a comment

Choose a reason for hiding this comment

bryanck commented Aug 2, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bryanck commented Aug 3, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

RussellSpitzer Aug 3, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bryanck Aug 4, 2023 • edited Loading

Choose a reason for hiding this comment

bryanck Aug 4, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bryanck commented Aug 4, 2023

bryanck commented Aug 4, 2023

bryanck commented Aug 2, 2023 •

edited

Loading

RussellSpitzer Aug 3, 2023 •

edited

Loading

bryanck Aug 4, 2023 •

edited

Loading

bryanck Aug 4, 2023 •

edited

Loading