ColumnarBatch, ParquetPartitionReaderFactory and spark.sql.columnVector.offheap.enabled

Author: jaceklaskowski

docs/SQLConf.md

@@ -665,7 +665,11 @@ Used when:
 
 ## <span id="numShufflePartitions"><span id="SHUFFLE_PARTITIONS"> numShufflePartitions
 
-The value of [spark.sql.shuffle.partitions](configuration-properties.md#spark.sql.shuffle.partitions) configuration property or...FIXME
+[spark.sql.shuffle.partitions](configuration-properties.md#spark.sql.shuffle.partitions)
+
+## <span id="COLUMN_VECTOR_OFFHEAP_ENABLED"><span id="offHeapColumnVectorEnabled"> offHeapColumnVectorEnabled
+
+[spark.sql.columnVector.offheap.enabled](configuration-properties.md#spark.sql.columnVector.offheap.enabled)
 
 ## <span id="rangeExchangeSampleSizePerPartition"><span id="RANGE_EXCHANGE_SAMPLE_SIZE_PER_PARTITION"> rangeExchangeSampleSizePerPartition
 

docs/configuration-properties.md

@@ -116,6 +116,23 @@ Default: `true`
 
 Use [SQLConf.ENABLE_FULL_OUTER_SHUFFLED_HASH_JOIN_CODEGEN](SQLConf.md#ENABLE_FULL_OUTER_SHUFFLED_HASH_JOIN_CODEGEN) to access the property
 
+## <span id="spark.sql.columnVector.offheap.enabled"> columnVector.offheap.enabled
+
+**spark.sql.columnVector.offheap.enabled**
+
+**(internal)** Enables [OffHeapColumnVector](OffHeapColumnVector.md) (`true`) or [OnHeapColumnVector](OnHeapColumnVector.md) (`false`) in [ColumnarBatch](vectorized-query-execution/ColumnarBatch.md)
+
+Default: `false`
+
+Use [SQLConf.offHeapColumnVectorEnabled](SQLConf.md#offHeapColumnVectorEnabled) for the current value
+
+Used when:
+
+* `RowToColumnarExec` is requested to `doExecuteColumnar`
+* `DefaultCachedBatchSerializer` is requested to `vectorTypes` and `convertCachedBatchToColumnarBatch`
+* `ParquetFileFormat` is requested to [vectorTypes](datasources/parquet/ParquetFileFormat.md#vectorTypes) and [buildReaderWithPartitionValues](datasources/parquet/ParquetFileFormat.md#buildReaderWithPartitionValues)
+* `ParquetPartitionReaderFactory` is [created](datasources/parquet/ParquetPartitionReaderFactory.md#enableOffHeapColumnVector)
+
 ## <span id="spark.sql.files.maxPartitionBytes"><span id="FILES_MAX_PARTITION_BYTES"> files.maxPartitionBytes
 
 **spark.sql.files.maxPartitionBytes**
@@ -1187,14 +1204,6 @@ Default: `true`
 
 Use [SQLConf.wholeStageSplitConsumeFuncByOperator](SQLConf.md#wholeStageSplitConsumeFuncByOperator) method to access the current value.
 
-## <span id="spark.sql.columnVector.offheap.enabled"> spark.sql.columnVector.offheap.enabled
-
-**(internal)** Enables [OffHeapColumnVector](OffHeapColumnVector.md) in [ColumnarBatch](vectorized-query-execution/ColumnarBatch.md) (`true`) or not (`false`). When `false`, [OnHeapColumnVector](OnHeapColumnVector.md) is used instead.
-
-Default: `false`
-
-Use [SQLConf.offHeapColumnVectorEnabled](SQLConf.md#offHeapColumnVectorEnabled) method to access the current value.
-
 ## <span id="spark.sql.columnNameOfCorruptRecord"> spark.sql.columnNameOfCorruptRecord
 
 ## <span id="spark.sql.constraintPropagation.enabled"> spark.sql.constraintPropagation.enabled

docs/connector/Batch.md

@@ -4,7 +4,7 @@
 
 ## Contract
 
-### <span id="createReaderFactory"> createReaderFactory
+### <span id="createReaderFactory"> Creating PartitionReaderFactory
 
 ```java
 PartitionReaderFactory createReaderFactory()

docs/datasources/parquet/ParquetPartitionReaderFactory.md

@@ -1,6 +1,6 @@
 # ParquetPartitionReaderFactory
 
-`ParquetPartitionReaderFactory` is a [FilePartitionReaderFactory](../FilePartitionReaderFactory.md).
+`ParquetPartitionReaderFactory` is a [FilePartitionReaderFactory](../FilePartitionReaderFactory.md) for [ParquetScan](ParquetScan.md#createReaderFactory) for batch queries.
 
 ## Creating Instance
 
@@ -18,6 +18,13 @@
 
 * `ParquetScan` is requested to [create a PartitionReaderFactory](ParquetScan.md#createReaderFactory)
 
+## <span id="enableOffHeapColumnVector"><span id="spark.sql.columnVector.offheap.enabled"> columnVector.offheap.enabled
+
+`ParquetPartitionReaderFactory` uses [spark.sql.columnVector.offheap.enabled](../../configuration-properties.md#spark.sql.columnVector.offheap.enabled) configuration property when requested for the following:
+
+* [Create a Vectorized Reader](#createParquetVectorizedReader) (and create a [VectorizedParquetRecordReader](VectorizedParquetRecordReader.md#useOffHeap))
+* [Build a Columnar Reader](#buildColumnarReader) (and `convertAggregatesRowToBatch`)
+
 ## <span id="supportColumnarReads"> supportColumnarReads
 
 ```scala
@@ -51,7 +58,7 @@ buildColumnarReader(
 
 In the end, `buildColumnarReader` returns a [PartitionReader](../../connector/PartitionReader.md) that returns [ColumnarBatch](../../vectorized-query-execution/ColumnarBatch.md)es (when [requested for records](../../connector/PartitionReader.md#get)).
 
-## <span id="buildReader"> Building PartitionReader
+## <span id="buildReader"> Building Partition Reader
 
 ```scala
 buildReader(
@@ -95,9 +102,38 @@ createVectorizedReader(
 
 In the end, `createVectorizedReader` requests the [VectorizedParquetRecordReader](VectorizedParquetRecordReader.md) to [initBatch](VectorizedParquetRecordReader.md#initBatch) (with the [partitionSchema](#partitionSchema) and the [partitionValues](../PartitionedFile.md#partitionValues) of the given [PartitionedFile](../PartitionedFile.md)) and returns it.
 
-`createVectorizedReader` is used when:
+---
+
+`createVectorizedReader` is used when `ParquetPartitionReaderFactory` is requested for the following:
+
+* [Build a partition reader (for a file)](#buildReader) (with [enableVectorizedReader](#enableVectorizedReader) enabled)
+* [Build a columnar partition reader (for a file)](#buildColumnarReader)
+
+### <span id="createParquetVectorizedReader"> createParquetVectorizedReader
+
+```scala
+createParquetVectorizedReader(
+  partitionValues: InternalRow,
+  pushed: Option[FilterPredicate],
+  convertTz: Option[ZoneId],
+  datetimeRebaseSpec: RebaseSpec,
+  int96RebaseSpec: RebaseSpec): VectorizedParquetRecordReader
+```
+
+`createParquetVectorizedReader` creates a [VectorizedParquetRecordReader](VectorizedParquetRecordReader.md) (with [capacity](#capacity)).
+
+`createParquetVectorizedReader` creates a [RecordReaderIterator](../RecordReaderIterator.md) (for the `VectorizedParquetRecordReader`).
 
-* `ParquetPartitionReaderFactory` is requested to [buildReader](#buildReader) and [buildColumnarReader](#buildColumnarReader)
+`createParquetVectorizedReader` prints out the following DEBUG message to the logs (with the [partitionSchema](#partitionSchema) and the given `partitionValues`):
+
+```text
+Appending [partitionSchema] [partitionValues]
+```
+
+In the end, `createParquetVectorizedReader` returns the `VectorizedParquetRecordReader`.
+
+??? note "Unused RecordReaderIterator?"
+    It appears that the `RecordReaderIterator` is created but not used. _Feeling confused_.
 
 ## <span id="buildReaderBase"> buildReaderBase
 
@@ -116,6 +152,8 @@ buildReaderBase[T](
 
 `buildReaderBase`...FIXME
 
+---
+
 `buildReaderBase` is used when:
 
 * `ParquetPartitionReaderFactory` is requested to [createRowBaseReader](#createRowBaseReader) and [createVectorizedReader](#createVectorizedReader)

docs/datasources/parquet/ParquetScan.md

@@ -21,7 +21,7 @@
 
 * `ParquetScanBuilder` is requested to [build a Scan](ParquetScanBuilder.md#build)
 
-## <span id="createReaderFactory"> createReaderFactory
+## <span id="createReaderFactory"> Creating PartitionReaderFactory
 
 ```scala
 createReaderFactory(): PartitionReaderFactory

docs/datasources/parquet/VectorizedParquetRecordReader.md

@@ -62,21 +62,24 @@ void initBatch(
 
 `initBatch` creates a [batch schema](../../types/index.md) that is [sparkSchema](SpecificParquetRecordReaderBase.md#sparkSchema) and the input `partitionColumns` schema (if available).
 
-`initBatch` requests [OffHeapColumnVector](../../OffHeapColumnVector.md#allocateColumns) or [OnHeapColumnVector](../../OnHeapColumnVector.md#allocateColumns) to allocate column vectors per the input `memMode`, i.e. [OFF_HEAP](#OFF_HEAP) or [ON_HEAP](#ON_HEAP) memory modes, respectively. `initBatch` records the allocated column vectors as the internal [WritableColumnVectors](#columnVectors).
+`initBatch` requests [OffHeapColumnVector](../../OffHeapColumnVector.md#allocateColumns) or [OnHeapColumnVector](../../OnHeapColumnVector.md#allocateColumns) to allocate column vectors per the input `memMode` (i.e., [OFF_HEAP](#OFF_HEAP) or [ON_HEAP](#ON_HEAP) memory modes, respectively). `initBatch` records the allocated column vectors as the internal [WritableColumnVectors](#columnVectors).
 
-!!! note
+!!! note "spark.sql.columnVector.offheap.enabled"
     [OnHeapColumnVector](../../OnHeapColumnVector.md) is used based on [spark.sql.columnVector.offheap.enabled](../../configuration-properties.md#spark.sql.columnVector.offheap.enabled) configuration property.
 
-`initBatch` creates a [ColumnarBatch](../../vectorized-query-execution/ColumnarBatch.md) (with the [allocated WritableColumnVectors](#columnVectors)) and records it as the internal [ColumnarBatch](#columnarBatch).
+`initBatch` creates a [ColumnarBatch](#columnarBatch) (with the [allocated WritableColumnVectors](#columnVectors)).
 
-`initBatch` does some additional maintenance to the [columnVectors](#columnVectors).
+`initBatch` does some additional maintenance to the [WritableColumnVectors](#columnVectors).
+
+---
 
 `initBatch` is used when:
 
 * `VectorizedParquetRecordReader` is requested to [resultBatch](#resultBatch)
 * `ParquetFileFormat` is requested to [build a data reader (with partition column values appended)](ParquetFileFormat.md#buildReaderWithPartitionValues)
 * `ParquetPartitionReaderFactory` is requested to [createVectorizedReader](ParquetPartitionReaderFactory.md#createVectorizedReader)
 
+<!---
 ## Review Me
 
 `VectorizedParquetRecordReader` uses <<OFF_HEAP, OFF_HEAP>> memory mode when [spark.sql.columnVector.offheap.enabled](../../configuration-properties.md#spark.sql.columnVector.offheap.enabled) internal configuration property is enabled (`true`).
@@ -212,3 +215,4 @@ NOTE: `getCurrentValue` is part of the Hadoop https://hadoop.apache.org/docs/r2.
 * `NewHadoopRDD` is requested to compute a partition (`compute`)
 
 * `RecordReaderIterator` is requested for the [next internal row](../RecordReaderIterator.md#next)
+-->

docs/vectorized-query-execution/ColumnarBatch.md

@@ -20,12 +20,13 @@ tags:
 
 `ColumnarBatch` is created when:
 
+* `ArrowConverters` utility is requested to `fromBatchIterator`
 * `RowToColumnarExec` unary physical operator is requested to `doExecuteColumnar`
 * [InMemoryTableScanExec](../physical-operators/InMemoryTableScanExec.md) leaf physical operator is requested for a [RDD[ColumnarBatch]](../physical-operators/InMemoryTableScanExec.md#columnarInputRDD)
 * `MapInPandasExec` unary physical operator is requested to `doExecute`
-* `OrcColumnarBatchReader` and `VectorizedParquetRecordReader` are requested to `initBatch`
+* `OrcColumnarBatchReader` is requested to `initBatch`
 * `PandasGroupUtils` utility is requested to `executePython`
-* `ArrowConverters` utility is requested to `fromBatchIterator`
+* `VectorizedParquetRecordReader` is requested to [init a batch](../datasources/parquet/VectorizedParquetRecordReader.md#initBatch)
 
 ## <span id="row"> ColumnarBatchRow