Skip to content

Minor cleanups and improvements to FileSystemExchange #14806

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
losipiuk merged 3 commits into from
Oct 29, 2022
Merged

Minor cleanups and improvements to FileSystemExchange #14806

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7e53ec9
Remove redundant code and comments
linzebing Oct 27, 2022
1ce05e1
Change misleading s3 scheme in TestFileSystemExchangeConfig
linzebing Oct 27, 2022
43d1802
Populate error when unsupported scheme is used
linzebing Oct 27, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions ...ilesystem/src/main/java/io/trino/plugin/exchange/filesystem/FileSystemExchangeModule.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,8 @@ protected void setup(Binder binder)

List<URI> baseDirectories = buildConfigObject(FileSystemExchangeConfig.class).getBaseDirectories();
if (baseDirectories.stream().map(URI::getScheme).distinct().count() != 1) {
throw new TrinoException(CONFIGURATION_INVALID, "Multiple schemes in exchange base directories");
binder.addError(new TrinoException(CONFIGURATION_INVALID, "Multiple schemes in exchange base directories"));
Comment thread
linzebing marked this conversation as resolved.
Outdated
return;
}
String scheme = baseDirectories.get(0).getScheme();
if (scheme == null || scheme.equals("file")) {
Expand All @@ -68,7 +69,7 @@ else if (ImmutableSet.of("abfs", "abfss").contains(scheme)) {
configBinder(binder).bindConfig(ExchangeAzureConfig.class);
}
else {
throw new TrinoException(NOT_SUPPORTED, format("Scheme %s is not supported as exchange spooling storage", scheme));
binder.addError(new TrinoException(NOT_SUPPORTED, format("Scheme %s is not supported as exchange spooling storage", scheme)));
Copy link
Copy Markdown
Member

@losipiuk losipiuk Oct 28, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what does it change really? Isn't exception caught up the stack anyway and handle same way?
We are throwing in line 52 and this one is no different, right?

Copy link
Copy Markdown
Member Author

@linzebing linzebing Oct 28, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for pointing out, updating line 52 too.

Copy link
Copy Markdown
Member

@losipiuk losipiuk Oct 28, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But still:
what does it change really?

Copy link
Copy Markdown
Member

@losipiuk losipiuk Oct 28, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually yeah - that way we may collect more than one error.

Copy link
Copy Markdown
Member Author

@linzebing linzebing Oct 28, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I tested and just realized that it doesn't make much of a difference in terms of error surfacing. Added return under line 52. I'm fine both ways.

}
}
}
2 changes: 0 additions & 2 deletions ...main/java/io/trino/plugin/exchange/filesystem/s3/BufferWriteAsyncResponseTransformer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,6 @@
* An implementation of {@link AsyncResponseTransformer} that writes data to the specified range of the given buffer.
* This class mimics the implementation of {@link ByteArrayAsyncResponseTransformer} but avoids memory copying.
*
* {@link AsyncResponseTransformer} that writes the data to the specified range of the given buffer
*
* @param <ResponseT> Response POJO type.
*/
public final class BufferWriteAsyncResponseTransformer<ResponseT>
Expand Down
14 changes: 0 additions & 14 deletions ...tem/src/main/java/io/trino/plugin/exchange/filesystem/s3/S3FileSystemExchangeStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,6 @@
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.S3AsyncClientBuilder;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.S3ClientBuilder;
import software.amazon.awssdk.services.s3.S3Configuration;
import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest;
import software.amazon.awssdk.services.s3.model.AbortMultipartUploadResponse;
Expand Down Expand Up @@ -458,18 +456,6 @@ private static AwsCredentialsProvider createAwsCredentialsProvider(ExchangeS3Con
return DefaultCredentialsProvider.create();
}

private S3Client createS3Client(AwsCredentialsProvider credentialsProvider, ClientOverrideConfiguration overrideConfig)
{
S3ClientBuilder clientBuilder = S3Client.builder()
.credentialsProvider(credentialsProvider)
.overrideConfiguration(overrideConfig);

region.ifPresent(clientBuilder::region);
endpoint.ifPresent(s3Endpoint -> clientBuilder.endpointOverride(URI.create(s3Endpoint)));

return clientBuilder.build();
}

private S3AsyncClient createS3AsyncClient(
AwsCredentialsProvider credentialsProvider,
ClientOverrideConfiguration overrideConfig,
Expand Down
4 changes: 2 additions & 2 deletions ...ystem/src/test/java/io/trino/plugin/exchange/filesystem/TestFileSystemExchangeConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ public void testDefaults()
public void testExplicitPropertyMappings()
{
Map<String, String> properties = ImmutableMap.<String, String>builder()
.put("exchange.base-directories", "s3n://exchange-spooling-test/")
.put("exchange.base-directories", "s3://exchange-spooling-test/")
.put("exchange.encryption-enabled", "false")
.put("exchange.max-page-storage-size", "32MB")
.put("exchange.sink-buffer-pool-min-size", "20")
Expand All @@ -62,7 +62,7 @@ public void testExplicitPropertyMappings()
.buildOrThrow();

FileSystemExchangeConfig expected = new FileSystemExchangeConfig()
.setBaseDirectories("s3n://exchange-spooling-test/")
.setBaseDirectories("s3://exchange-spooling-test/")
.setExchangeEncryptionEnabled(false)
.setMaxPageStorageSize(DataSize.of(32, MEGABYTE))
.setExchangeSinkBufferPoolMinSize(20)
Expand Down