Support reading BYTE_STREAM_SPLIT encoding in parquet

[manupatteri]

Description

Support reading BYTE_STREAM_SPLIT encoding in parquet

Related issues, pull requests, and links

Fixes #8357

Documentation

(x) No documentation is needed.
( ) 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, Hudi, Delta Lake, Iceberg
* Support reading BYTE_STREAM_SPLIT encoding in parquet. ({issue}`8357`)

[manupatteri]

@raunaqmorarka could you please review it again? I thought I can resolve conflicts after basic review is done. Please let me know if I should fix it first.

[raunaqmorarka]

Please resolve the merge conflicts and squash commits.

[manupatteri]

@raunaqmorarka I am working on a testcase for float. Appreciate if you can review on rest of changes.

[mosabua]

@raunaqmorarka can you please follow up for @manupatteri

[manupatteri]

@raunaqmorarka @mosabua I have added tests for float as well. Kindly review it again and let me know your comments.

[manupatteri]

@raunaqmorarka
When I used the common static method, "initialize" in init method in DoubleApacheParquetValueDecoder and FloatApacheParquetValueDecoder, I was getting below exception with byte_stream_split testcase.

org.apache.parquet.io.ParquetDecodingException: Invalid ByteStreamSplit stream, num values upper bound (w/ nulls): 0, num encoded values: 100

at org.apache.parquet.column.values.bytestreamsplit.ByteStreamSplitValuesReader.initFromPage(ByteStreamSplitValuesReader.java:77)
at io.trino.parquet.reader.decoders.ApacheParquetValueDecoders.initialize(ApacheParquetValueDecoders.java:142)
at io.trino.parquet.reader.decoders.ApacheParquetValueDecoders$FloatApacheParquetValueDecoder.init(ApacheParquetValueDecoders.java:120)
at io.trino.parquet.reader.AbstractColumnReader.createValueDecoder(AbstractColumnReader.java:109)
at io.trino.parquet.reader.flat.FlatColumnReader.readFlatPageV1(FlatColumnReader.java:298)
at io.trino.parquet.reader.flat.FlatColumnReader.readPage(FlatColumnReader.java:261)
at io.trino.parquet.reader.flat.FlatColumnReader.seekToNextPage(FlatColumnReader.java:245)
at io.trino.parquet.reader.flat.FlatColumnReader.seek(FlatColumnReader.java:116)
at io.trino.parquet.reader.flat.FlatColumnReader.readPrimitive(FlatColumnReader.java:87)
at io.trino.parquet.reader.ParquetReader.readPrimitive(ParquetReader.java:463)
at io.trino.parquet.reader.ParquetReader.readColumnChunk(ParquetReader.java:553)
at io.trino.parquet.reader.ParquetReader.readBlock(ParquetReader.java:536)
at io.trino.parquet.reader.ParquetReader.lambda$nextPage$3(ParquetReader.java:251)
at io.trino.parquet.reader.ParquetBlockFactory$ParquetBlockLoader.load(ParquetBlockFactory.java:72)
at io.trino.spi.block.LazyBlock$LazyData.load(LazyBlock.java:312)
at io.trino.spi.block.LazyBlock$LazyData.getFullyLoadedBlock(LazyBlock.java:291)
at io.trino.spi.block.LazyBlock.getLoadedBlock(LazyBlock.java:186)
at io.trino.parquet.reader.TestByteStreamSplitEncoding.readAndCompare(TestByteStreamSplitEncoding.java:88)
at io.trino.parquet.reader.TestByteStreamSplitEncoding.testReadFloatDouble(TestByteStreamSplitEncoding.java:57)
at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
at java.base/java.lang.reflect.Method.invoke(Method.java:580)
at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:104)
at org.testng.internal.Invoker.invokeMethod(Invoker.java:645)
at org.testng.internal.Invoker.invokeTestMethod(Invoker.java:851)
at org.testng.internal.Invoker.invokeTestMethods(Invoker.java:1177)
at org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:129)
at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:112)
at org.testng.TestRunner.privateRun(TestRunner.java:756)
at org.testng.TestRunner.run(TestRunner.java:610)
at org.testng.SuiteRunner.runTest(SuiteRunner.java:387)
at org.testng.SuiteRunner.runSequentially(SuiteRunner.java:382)
at org.testng.SuiteRunner.privateRun(SuiteRunner.java:340)
at org.testng.SuiteRunner.run(SuiteRunner.java:289)
at org.testng.SuiteRunnerWorker.runSuite(SuiteRunnerWorker.java:52)
at org.testng.SuiteRunnerWorker.run(SuiteRunnerWorker.java:86)
at org.testng.TestNG.runSuitesSequentially(TestNG.java:1293)
at org.testng.TestNG.runSuitesLocally(TestNG.java:1218)
at org.testng.TestNG.runSuites(TestNG.java:1133)
at org.testng.TestNG.run(TestNG.java:1104)
at com.intellij.rt.testng.IDEARemoteTestNG.run(IDEARemoteTestNG.java:65)
at com.intellij.rt.testng.RemoteTestNGStarter.main(RemoteTestNGStarter.java:105)

To fix that I had to introduce changes in "initialize" method(to use correct valueCount in ValuesReader::initFromPage call)