Reduce code duplication in Arrow code

[nastra]

No description provided.

[nastra]

@rymurr it is probably easier looking at the modified files directly instead of looking at the diff when reviewing.

[nastra]

Results on branch master

Benchmark                                                                 Mode  Cnt  Score   Error  Units
VectorizedReadFlatParquetDataBenchmark.readDatesIcebergVectorized5k         ss    5  1.933 ± 0.171   s/op
VectorizedReadFlatParquetDataBenchmark.readDatesSparkVectorized5k           ss    5  1.548 ± 0.040   s/op
VectorizedReadFlatParquetDataBenchmark.readDecimalsIcebergVectorized5k      ss    5  8.692 ± 0.245   s/op
VectorizedReadFlatParquetDataBenchmark.readDecimalsSparkVectorized5k        ss    5  7.726 ± 0.163   s/op
VectorizedReadFlatParquetDataBenchmark.readDoublesIcebergVectorized5k       ss    5  2.564 ± 0.074   s/op
VectorizedReadFlatParquetDataBenchmark.readDoublesSparkVectorized5k         ss    5  2.419 ± 0.094   s/op
VectorizedReadFlatParquetDataBenchmark.readFloatsIcebergVectorized5k        ss    5  2.466 ± 0.128   s/op
VectorizedReadFlatParquetDataBenchmark.readFloatsSparkVectorized5k          ss    5  2.274 ± 0.033   s/op
VectorizedReadFlatParquetDataBenchmark.readIntegersIcebergVectorized5k      ss    5  2.385 ± 0.171   s/op
VectorizedReadFlatParquetDataBenchmark.readIntegersSparkVectorized5k        ss    5  2.437 ± 0.125   s/op
VectorizedReadFlatParquetDataBenchmark.readLongsIcebergVectorized5k         ss    5  2.490 ± 0.115   s/op
VectorizedReadFlatParquetDataBenchmark.readLongsSparkVectorized5k           ss    5  2.654 ± 0.182   s/op
VectorizedReadFlatParquetDataBenchmark.readStringsIcebergVectorized5k       ss    5  4.884 ± 0.381   s/op
VectorizedReadFlatParquetDataBenchmark.readStringsSparkVectorized5k         ss    5  4.366 ± 0.439   s/op
VectorizedReadFlatParquetDataBenchmark.readTimestampsIcebergVectorized5k    ss    5  1.729 ± 0.150   s/op
VectorizedReadFlatParquetDataBenchmark.readTimestampsSparkVectorized5k      ss    5  1.896 ± 0.174   s/op

Results on branch arrow-refactoring

Benchmark                                                                 Mode  Cnt  Score   Error  Units
VectorizedReadFlatParquetDataBenchmark.readDatesIcebergVectorized5k         ss    5  1.611 ± 0.035   s/op
VectorizedReadFlatParquetDataBenchmark.readDatesSparkVectorized5k           ss    5  1.496 ± 0.072   s/op
VectorizedReadFlatParquetDataBenchmark.readDecimalsIcebergVectorized5k      ss    5  8.461 ± 0.251   s/op
VectorizedReadFlatParquetDataBenchmark.readDecimalsSparkVectorized5k        ss    5  8.250 ± 0.083   s/op
VectorizedReadFlatParquetDataBenchmark.readDoublesIcebergVectorized5k       ss    5  2.872 ± 0.050   s/op
VectorizedReadFlatParquetDataBenchmark.readDoublesSparkVectorized5k         ss    5  2.781 ± 0.129   s/op
VectorizedReadFlatParquetDataBenchmark.readFloatsIcebergVectorized5k        ss    5  2.752 ± 0.216   s/op
VectorizedReadFlatParquetDataBenchmark.readFloatsSparkVectorized5k          ss    5  2.261 ± 0.101   s/op
VectorizedReadFlatParquetDataBenchmark.readIntegersIcebergVectorized5k      ss    5  2.743 ± 0.123   s/op
VectorizedReadFlatParquetDataBenchmark.readIntegersSparkVectorized5k        ss    5  2.652 ± 0.221   s/op
VectorizedReadFlatParquetDataBenchmark.readLongsIcebergVectorized5k         ss    5  2.754 ± 0.818   s/op
VectorizedReadFlatParquetDataBenchmark.readLongsSparkVectorized5k           ss    5  2.682 ± 0.136   s/op
VectorizedReadFlatParquetDataBenchmark.readStringsIcebergVectorized5k       ss    5  4.611 ± 0.066   s/op
VectorizedReadFlatParquetDataBenchmark.readStringsSparkVectorized5k         ss    5  3.901 ± 0.143   s/op
VectorizedReadFlatParquetDataBenchmark.readTimestampsIcebergVectorized5k    ss    5  1.661 ± 0.062   s/op
VectorizedReadFlatParquetDataBenchmark.readTimestampsSparkVectorized5k      ss    5  1.572 ± 0.056   s/op

Detailed results are attached below:
master_detailed_results_new.txt
refactoring_detailed_results_new.txt

[rymurr]

One minor comment @nastra . Looks great though, the vectorised code is a lot easier to read! I love deleting code!

[rymurr]

Any reason this is cast to IntVector and the others are directly manipulating the data buffer?

[nastra]

Dictionary encoded vectors are always represented as IntVector, but since BaseDictEncodedReader uses the more generic FieldVector we need to do a cast to IntVector here. Fwiw, here's how it was done in the original code:

iceberg/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedDictionaryEncodedParquetValuesReader.java

Line 55 in 6bcca16

intVector.set(idx, currentValue);

and

iceberg/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java

Line 136 in 6bcca16

vectorizedColumnIterator.nextBatchDictionaryIds((IntVector) vec, nullabilityHolder);

[rymurr]

Looks awesome @nastra

[rymurr]

@rdblue: @nastra suggested merging this w/o squashing as each commit atomically refactors a single class. WDYT?

[rdblue]

I'm fine either way. Whatever you'd like to do.

[rymurr]

note - rebasing to keep each atomic class refactor as a separate (revertible) commit