[SPARK-37728][SQL][3.2] Reading nested columns with ORC vectorized re…

…ader can cause ArrayIndexOutOfBoundsException

### What changes were proposed in this pull request?

This is a backport of apache#35002 .

When an OrcColumnarBatchReader is created, method initBatch will be called only once. In method initBatch:

`orcVectorWrappers[i] = OrcColumnVectorUtils.toOrcColumnVector(dt, wrap.batch().cols[colId]);`

When the second argument of toOrcColumnVector is a ListColumnVector/MapColumnVector, orcVectorWrappers[i] is initialized with the ListColumnVector or MapColumnVector's offsets and lengths.

However, when method nextBatch of OrcColumnarBatchReader is called, method ensureSize of ColumnVector (and its subclasses, like MultiValuedColumnVector) could be called, then the ListColumnVector/MapColumnVector's offsets and lengths could refer to new array objects. This could result in the ArrayIndexOutOfBoundsException.

This PR makes OrcArrayColumnVector.getArray and OrcMapColumnVector.getMap always get offsets and lengths from the underlying ColumnVector, which can resolve this issue.

### Why are the changes needed?

Bugfix

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

Pass the CIs with the newly added test case.

Closes apache#35038 from yym1995/branch-3.2.

Lead-authored-by: Yimin <yimin.y@outlook.com>
Co-authored-by: Yimin Yang <26797163+yym1995@users.noreply.github.com>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcArrayColumnVector.java

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.execution.datasources.orc;
 
 import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.ListColumnVector;
 
 import org.apache.spark.sql.types.ArrayType;
 import org.apache.spark.sql.types.DataType;
@@ -31,26 +32,22 @@
  */
 public class OrcArrayColumnVector extends OrcColumnVector {
   private final OrcColumnVector data;
-  private final long[] offsets;
-  private final long[] lengths;
 
   OrcArrayColumnVector(
       DataType type,
       ColumnVector vector,
-      OrcColumnVector data,
-      long[] offsets,
-      long[] lengths) {
+      OrcColumnVector data) {
 
     super(type, vector);
 
     this.data = data;
-    this.offsets = offsets;
-    this.lengths = lengths;
   }
 
   @Override
   public ColumnarArray getArray(int rowId) {
-    return new ColumnarArray(data, (int) offsets[rowId], (int) lengths[rowId]);
+    int offsets = (int) ((ListColumnVector) baseData).offsets[rowId];
+    int lengths = (int) ((ListColumnVector) baseData).lengths[rowId];
+    return new ColumnarArray(data, offsets, lengths);
   }
 
   @Override

sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcColumnVector.java

@@ -29,7 +29,7 @@
  * this column vector is used to adapt Hive ColumnVector with Spark ColumnarVector.
  */
 public abstract class OrcColumnVector extends org.apache.spark.sql.vectorized.ColumnVector {
-  private final ColumnVector baseData;
+  protected final ColumnVector baseData;
   private int batchSize;
 
   OrcColumnVector(DataType type, ColumnVector vector) {

sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcColumnVectorUtils.java

@@ -53,15 +53,13 @@ static OrcColumnVector toOrcColumnVector(DataType type, ColumnVector vector) {
       ListColumnVector listVector = (ListColumnVector) vector;
       OrcColumnVector dataVector = toOrcColumnVector(
         ((ArrayType) type).elementType(), listVector.child);
-      return new OrcArrayColumnVector(
-        type, vector, dataVector, listVector.offsets, listVector.lengths);
+      return new OrcArrayColumnVector(type, vector, dataVector);
     } else if (vector instanceof MapColumnVector) {
       MapColumnVector mapVector = (MapColumnVector) vector;
       MapType mapType = (MapType) type;
       OrcColumnVector keysVector = toOrcColumnVector(mapType.keyType(), mapVector.keys);
       OrcColumnVector valuesVector = toOrcColumnVector(mapType.valueType(), mapVector.values);
-      return new OrcMapColumnVector(
-        type, vector, keysVector, valuesVector, mapVector.offsets, mapVector.lengths);
+      return new OrcMapColumnVector(type, vector, keysVector, valuesVector);
     } else {
       throw new IllegalArgumentException(
         String.format("OrcColumnVectorUtils.toOrcColumnVector should not take %s as type " +

sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcMapColumnVector.java

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.execution.datasources.orc;
 
 import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.MapColumnVector;
 
 import org.apache.spark.sql.types.DataType;
 import org.apache.spark.sql.types.Decimal;
@@ -32,28 +33,24 @@
 public class OrcMapColumnVector extends OrcColumnVector {
   private final OrcColumnVector keys;
   private final OrcColumnVector values;
-  private final long[] offsets;
-  private final long[] lengths;
 
   OrcMapColumnVector(
       DataType type,
       ColumnVector vector,
       OrcColumnVector keys,
-      OrcColumnVector values,
-      long[] offsets,
-      long[] lengths) {
+      OrcColumnVector values) {
 
     super(type, vector);
 
     this.keys = keys;
     this.values = values;
-    this.offsets = offsets;
-    this.lengths = lengths;
   }
 
   @Override
   public ColumnarMap getMap(int ordinal) {
-    return new ColumnarMap(keys, values, (int) offsets[ordinal], (int) lengths[ordinal]);
+    int offsets = (int) ((MapColumnVector) baseData).offsets[ordinal];
+    int lengths = (int) ((MapColumnVector) baseData).lengths[ordinal];
+    return new ColumnarMap(keys, values, offsets, lengths);
   }
 
   @Override

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala

@@ -34,7 +34,9 @@ import org.apache.orc.mapreduce.OrcInputFormat
 import org.apache.spark.{SparkConf, SparkException}
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.execution.FileSourceScanExec
 import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation, RecordReaderIterator}
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.sql.types._
@@ -713,6 +715,27 @@ abstract class OrcQuerySuite extends OrcQueryTest with SharedSparkSession {
       }
     }
   }
+
+  test("SPARK-37728: Reading nested columns with ORC vectorized reader should not " +
+    "cause ArrayIndexOutOfBoundsException") {
+    withTempPath { dir =>
+      val path = dir.getCanonicalPath
+      val df = spark.range(100).map { _ =>
+        val arrayColumn = (0 until 50).map(_ => (0 until 1000).map(k => k.toString))
+        arrayColumn
+      }.toDF("record").repartition(1)
+      df.write.format("orc").save(path)
+
+      withSQLConf(SQLConf.ORC_VECTORIZED_READER_NESTED_COLUMN_ENABLED.key -> "true") {
+        val readDf = spark.read.orc(path)
+        val vectorizationEnabled = readDf.queryExecution.executedPlan.find {
+          case scan @ (_: FileSourceScanExec | _: BatchScanExec) => scan.supportsColumnar
+          case _ => false
+        }.isDefined
+        checkAnswer(readDf, df)
+      }
+    }
+  }
 }
 
 class OrcV1QuerySuite extends OrcQuerySuite {