[WIP] Implement exchange spooling by linzebing · Pull Request #10376 · trinodb/trino

linzebing · 2021-12-21T23:40:56Z

This PR adds trino-exchange plugin to Trino, which contains a local file system implementation as well as a S3-based implementation.

kokosing · 2021-12-22T10:20:05Z

Is "spooling" and "spilling" a different things? Do we reuse one for the other?

losipiuk · 2021-12-22T10:30:12Z

Is "spooling" and "spilling" a different things? Do we reuse one for the other?

Different. Not very related. Spooling is a single word name for the implementation of persistent exchange buffers (concept added in task level retries PR) which dump data to external filesystem.

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/SecretKeySerializer.java

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/FileSystemExchange.java

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/FileSystemExchangeManager.java

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/FileSystemExchangeSink.java

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/ExchangeStorageWriter.java

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/FileSystemExchangeStorage.java

...in/trino-exchange/src/main/java/io/trino/plugin/exchange/s3/S3FileSystemExchangeStorage.java

plugin/trino-exchange/src/test/java/io/trino/plugin/exchange/FileSystemExchangeQueryRunner.java

plugin/trino-hive/src/test/java/io/trino/plugin/hive/AbstractHiveConnectorTest.java

plugin/trino-hive/src/test/java/io/trino/plugin/hive/HiveQueryRunner.java

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/FileSystemExchangeSink.java

linzebing · 2021-12-30T05:53:24Z

I realized that it might be better to implement parallel read illustrated using a producer-consumer model, instead of async read via double buffering. I will test both.

plugin/trino-exchange/src/test/java/io/trino/plugin/exchange/containers/Minio.java

plugin/trino-exchange/src/test/java/io/trino/plugin/exchange/containers/S3MinioStorage.java

...rino-exchange/src/test/java/io/trino/plugin/exchange/s3/TestS3FileSystemExchangeManager.java

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/s3/ExchangeS3Config.java

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/FileSystemExchangeSource.java

losipiuk · 2021-12-30T14:55:14Z

Thanks. Looks good. I did not reread everything. I assume changes in the old commits were (mostly) addressing of comments. Please point me to a specific piece of code if you feel I should look at it. It would be nice if github exposed force-push timeline for PRs and allowed to get a diff including all the changes since I last reviewed that.
We had sth like that in my previous job and it worked great.

arhimondr · 2022-01-04T22:59:24Z

Resolves #9936

arhimondr · 2022-01-08T20:11:24Z

@linzebing

S3 minimum upload buffer size is 5MB. When running MinIO-based tests for suites like AbstractTestFaultTolerantExecutionAggregations, a few tests will throw out of memory errors because of running out of heap space.

Running out of heap space usually indicates a memory accounting problem.

Also the number of partitions configured for those tests is only 5: 9a5c5d7#diff-c3957ed97dd34166ebb4937df48e597924e5c1345b2668f807014f834154a4d9R28. A single task shouldn't allocate more than 5MB * 5 * 2 = 50MB of buffer space. The number of nodes in a cluster is set to 4. With fault tolerant execution enabled only a single task per query is run on a node. So in theory the buffer space for a single query shouldn't exceed ~200MB. With two queries running concurrently it get's to 400MB. In theory it could cause additional memory pressure, however somehow it feels not very likely. I would recommend taking a heapdump (tests are configured to generate heap dump automatically when running out of heap) to check if everything is properly accounted and there's no apparent leaks.

Also I found a different memory issue related to S3 (#10464) and I disabled streaming upload for MinIO based tests for now (b555618). I wonder if it could be anyhow related?

linzebing · 2022-01-08T21:16:22Z

@arhimondr I will try to get a heap dump

losipiuk · 2022-01-10T14:41:58Z

Also I found a different memory issue related to S3 (#10464) and I disabled streaming upload for MinIO based tests for now (b555618). I wonder if it could be anyhow related?

Oh - so this is the reason test I run for #10432 stopped failing with OO today after I rebased :D.

plugin/trino-exchange/pom.xml

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/ExchangeStorageWriter.java

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/FileSystemExchangeManager.java

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/FileSystemExchangeSinkHandle.java

...-hive/src/test/java/io/trino/plugin/hive/TestHiveFaultTolerantExecutionJoinQueriesMinio.java

plugin/trino-hive/pom.xml

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/ExchangeStorageWriter.java

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/FileSystemExchangeSource.java

...in/trino-exchange/src/main/java/io/trino/plugin/exchange/s3/S3FileSystemExchangeStorage.java

arhimondr · 2022-01-18T20:44:57Z

...in/trino-exchange/src/main/java/io/trino/plugin/exchange/s3/S3FileSystemExchangeStorage.java

+
+            try {
+                List<CompletedPart> completedParts = uploadFutures.stream()
+                        .map(CompletableFuture::join)


This will cause the close method to block for potentially a long time (as it will have to wait for all parts to finish upload). Currently if you look at the call stack it is not designed to block for a very long time. The call to ExchangeSink#finish is done from the OutputBuffer#noMorePages that is executed from a rather tiny executor that can easily run out of threads.

This is actually my mistake. Now after carefully looking at the implementation of other output buffers it looks like ExchangeSink#finish and ExchangeSink#abort should be asynchronous and return a Future. I'm going to address this issue in the main PR.

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/FileSystemExchangeSink.java

arhimondr · 2022-01-18T20:55:43Z

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/FileSystemExchangeSink.java

+                }
+            }
+            try {
+                exchangeStorage.createEmptyFile(outputDirectory.resolve(COMMITTED_MARKER_FILE_NAME));


Same here, I wonder if this should be non blocking

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/FileSystemExchangeSink.java

arhimondr added 10 commits December 10, 2021 17:04

Rename ExchangeClient into DirectExchangeClient

ba28b30

Add external exchange interface

166e33f

Add reference implementation of extrenal exchange plugin

920529f

Integrate external exchange with exchange operator

125a958

Integrate external exchange with output buffers

eebfd9a

Support task level retries in DeduplicationExchangeClientBuffer

cae3211

Rename TaskSource to SplitAssignment

262fbb6

Implement task level failure recovery

08e06d7

Add integration tests for fault tolerant execution

e54367a

Simplify assertions in TestDirectExchangeClient

1d3881f

cla-bot bot added the cla-signed label Dec 21, 2021

linzebing added the WIP label Dec 21, 2021

linzebing requested review from arhimondr and losipiuk December 21, 2021 23:41

losipiuk reviewed Dec 22, 2021

View reviewed changes

plugin/trino-exchange/src/main/java/io/trino/plugin/exchange/SecretKeySerializer.java Outdated Show resolved Hide resolved