Split spooling partition output file into multiple files when size is large

[linzebing]

Description

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

Improvement.

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

trino-exchange plugin

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

In fault tolerant execution, if spooling file is large, we actually want it to be split into multiple reasonably sized files such that we can exploit more parallelism when reading the data. This PR splits large spooling files based on target size.

Suppose target size is 1GB, and previously we write a single file named 0.data of 3.2GB, after this change, we will write four files named 0_0.data, 0_1.data, 0_2.data, 0_3.data, in the form of {partitionId}_{partNum}.data, the sizes will be approximately 1GB, 1GB, 1GB, 0.2GB.

Related issues, pull requests, and links

Documentation

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

Release notes

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

# Section
* Fix some things. ({issue}`issuenumber`)

[arhimondr]

Looks good, a couple of comments

[arhimondr]

nit: Missed a new line after the Split partition output file into multiple files when size is large commit title

[arhimondr]

nit: I was looking at testLargePages and I realized that we are allocating large strings and storing them as static fields. That's >21MB of memory wasted: https://github.com/trinodb/trino/blob/master/plugin/trino-exchange/src/test/java/io/trino/plugin/exchange/AbstractTestExchangeManager.java#L53

Could you please convert these values into local variables in the test method?

[losipiuk]

I would suggest using softer naming: sink-target-file-size. Technically we can overshoot the size if a single row is larger than the configured value. "target" naming also allows for more elastic tweaks in implementation without mismanaging user expectations.
If you change name here update also in-code variable names.

[linzebing]

I added checkArgument to make sure max page storage size is no larger than max file size

[losipiuk]

Keeping track of currentFileSize feels like ExchangeStorageWriter responsibility. Consider moving it there and exposing long ExchangeStorageWriter.writtenBytes()

[linzebing]

That's reasonable, but that will require us to deduplicate logic in LocalFileSystemExchangeStorage, S3FileSystemExchangeStorage, and in the future Azure and GCS too. So I feel it might be better to just track it here.

[losipiuk]

LGTM