[HUDI-3709] Fixing ParquetWriter impls not respecting Parquet Max File Size limit

[alexeykudinkin]

Tips

• Thank you very much for contributing to Apache Hudi.
• Please review https://hudi.apache.org/contribute/how-to-contribute before opening a pull request.

What is the purpose of the pull request

Currently writing t/h Spark DataSource connector, does not respect "hoodie.parquet.max.file.size" setting: in the snippet pasted below i'm trying to limit the file-size to 16Mb, while on disk i'm getting ~80Mb files.

The reason for that is that we rely on ParquetWriter to control the file size (canWrite method), that relies in turn on FileSystem to trace how much was actually written to FS.

The problem with this approach is that ParquetWriter is writing lazily: It creates instances of ParquetWriter which in turn cache the whole record group when write methods are invoked and only flushes the data to FS only when closing the Writer (ie when close) is invoked.

This PR instead rebases canWrite to rely on ParquetWriter::getDataSize which holistically reflects the size of the records both already written to FS as well as the ones kept in memory.

Brief change log

(for example:)

• Modify AnnotationLocation checkstyle rule in checkstyle.xml

Verify this pull request

(Please pick either of the following options)

This pull request is already covered by existing tests, such as (please describe tests).

Committer checklist

• Has a corresponding JIRA in PR title & commit

• Commit message is descriptive of the change

• CI is green

• Necessary doc changes done or have another open PR

• For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.

[nsivabalan]

LGTM. one minor clarification

[alexeykudinkin]

@hudi-bot run azure

[hudi-bot]

CI report:

• dd1c5ba Azure: FAILURE Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build

[nsivabalan]

I went through the code again wrt this patch and #5497. Probably we should bring back the WrapperFileSystem again so that we don't hit the parquetWriter to fetch the size. if we ensure we flush at regular intervals, wrapperFileSystem.getBytesWritten(Path file) should give us the right size of data that got written. this will also ensure we don't hit the disk or incur the cost due to column meta refresh within parquetWriter.

[nsivabalan]

@alexeykudinkin : let's jam on this when you are back.

[alexeykudinkin]

@nsivabalan we should not be interfering with the caching on the Parquet Writer level (by manually flushing), and checking the ParquetWriter for the currently accumulated buffer size is the right way to interface with it (as compared to intercept the FileSystem writes and accounting for how many bytes were written).

The issue inadvertently planted with this approach (addressed in #5497) was that the cost of the getDataSize was not factored in (assumed it to be O(1), while in reality it's O(N) of the written blocks)