Implement verification for optimized parquet writer

[raunaqmorarka]

Description

Implements verification of file footer, row count and
checksum of columns.
Added a config parquet.optimized-writer.validation-percentage and
session property in hive connector to control the percentage of
written files that will be verified.
By default, we will verify 5% of written files.

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)

optimized parquet writer

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

Implements verification of files written by optimized parquet writer in hive connector.

Fixes #5356

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

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

# Hive
* Implement verification of files written by optimized parquet writer. ({issue}`13246`)

[sopel39]

cc @lukasz-stec

[skrzypo987]

lgtm % I'm not an expert in ORC format

[sopel39]

consider adding test cases for mismatch and equal cases for this and other methods

[sopel39]

ping

[lukasz-stec]

I added some comments.
I also have two high-level objections.

1. there seems to ba a lot of code copied between orc, rcfile and parquet write validation. It would be a lot cleaner to have it extracted to common place.
2. This changes modifies ParquetReader and related classes significantly, which in theory could be avoided (may be hard in practice though).

[lukasz-stec]

why the implementation is different than io.trino.orc.ValidationHash#mapSkipNullKeysHash?

[raunaqmorarka]

I'm not sure why there is even a isNull check in mapSkipNullKeysHash given that the null case is already handled by hash method.
Maybe @dain knows the reason behind that ?
The xor was avoided to avoid having a function which gives the same result if the key and value are flipped.

[sopel39]

lgtm % comments

[sopel39]

ping

[raunaqmorarka]

Optimized parquet writer verification inserts benchmark.pdf
Perf impact with 5% verification (current default) is around 2-3%
Perf impact with 100% verification would be around 45%.

[findepi]

Per #14047 (comment)
is this enabled in Hive connector only, and Iceberg/Delta (which also use the optimizer writer), do not run the verification?

[raunaqmorarka]

Per #14047 (comment)
is this enabled in Hive connector only, and Iceberg/Delta (which also use the optimizer writer), do not run the verification?

Right, this PR implements parquet writer verification only for hive connector. It should be straightforward to add it to delta lake and iceberg as well, but I don't know if we actually want to run it there given that we've already been using the new writer without verification in those connectors.

[findepi]

under what circumstances is this possible?

btw maybe we should throw here?
returning null will cause ParquetPageSource to close and not read anything anymore:

trino/plugin/trino-hive/src/main/java/io/trino/plugin/hive/parquet/ParquetPageSource.java

Lines 103 to 105 in f8e774a

	if (closed || page == null) {
	close();
	return null;

cc @alexjo2144 @homar

[raunaqmorarka]

This is just preserving the old code's behaviour

int batchSize = parquetReader.nextBatch();

if (closed || batchSize <= 0) {
      close();
      return null;
}

batchSize can be -1 on reaching end of list of row groups to read, which is okay.
batchSize can be 0 on encountering empty row groups. This should be encountered only for an empty file. It's possible that some buggy writer writes empty row groups in the middle of a file with data, in that case we will stop early.