Support exchange spooling on GCS

[linzebing]

Description

Is this change a fix, improvement, new feature, refactoring, or other?

New feature.

Is this a change to the core query engine, a connector, client library, or the SPI interfaces? (be specific)

trino-exchange-filesystem

How would you describe this change to a non-technical end user or system administrator?

This PR adds support for exchange spooling on GCS. GCS is mostly S3-compatible, except for two minor incompatibilities.

An example exchange-manager.properties:

exchange-manager.name=filesystem
exchange.base-directories=gs://your-bucket-name
exchange.s3.region=us-west-1
exchange.s3.aws-access-key=your-google-access-key-id
exchange.s3.aws-secret-key=your-google-access-key-secret
exchange.s3.endpoint=https://storage.googleapis.com

Related issues, pull requests, and links

Documentation

( ) No documentation is needed.
( ) Sufficient documentation is included in this PR.
( ) Documentation PR is available with #prnumber.
(x) Documentation issue #issuenumber is filed, and can be handled later.
#12467

Release notes

( ) No release notes entries required.
(x) Release notes entries required with the following suggested text:

# Section
* Support exchange spooling on Google Cloud Storage.
* Dropped exchange spooling support for legacy S3 schemes s3n:// and s3a://.

[mosabua]

This will need docs .. please work with @colebow on adding this.

[arhimondr]

This can in theory grow indefinitely. In FileSystemExchange#close we call deleteRecursively in a loop for each task. This may result in huge spikes in number of threads (hundreds or even thousands). I would recommend going with a bounded executor with the number of threads set to desired concurrency, e.g.:

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                maximumConcurrency,
                maximumConcurrency,
                10, SECONDS,
                new LinkedBlockingQueue<>(),
                threadsNamed("gcs-delete-%s"));

I don't know what value do we want to pick for max concurrency though, maybe 50? 100?

Also it is a generally good idea to set executor.allowCoreThreadTimeOut(true); to let the inactive threads be reclaimed after a spike.

It might also be reasonable to improve FileSystemExchange#close to batch delete requests across multiple partitions.

[linzebing]

Yeah, I think it's better to batch delete requests. Changing it right now.

With batching, I think directly using cachedExecutor will be sufficient.

[linzebing]

I have changed to batch the batch deletes, such to minimize API calls to GCS.

Wonder if we should do the same for Azure and S3. Currently we are deleting a task output directory at a time. In theory, we can do something similar to GCS, collect all the objects into a list, and batch delete them. @arhimondr @losipiuk

[losipiuk]

I have changed to batch the batch deletes, such to minimize API calls to GCS.

Wonder if we should do the same for Azure and S3. Currently we are deleting a task output directory at a time. In theory, we can do something similar to GCS, collect all the objects into a list, and batch delete them. @arhimondr @losipiuk

Would make sense IMO. Good catch.

[arhimondr]

LGTM % comment

[linzebing]

Addressed comments. On batching deletes for S3 and Azure, decided to do it in a separate PR. It's a bit more complex than I thought as I need to deal with multiple buckets

[arhimondr]

nit: one parameter per line, static import listeningDecorator

[arhimondr]

Also set the core pool size to 100, otherwise it will keep running only a single thread until the queue is full

[linzebing]

Ah you are right. I'm using a SynchronousQueue here, which basically has a size of 0, and if concurrent tasks exceed 100, rejection will happen. Your suggestion above is better.