Skip to content

[SPARK-37728][SQL] Reading nested columns with ORC vectorized reader can cause ArrayIndexOutOfBoundsException #35002

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
yimin-yang wants to merge 2 commits into from
Closed

[SPARK-37728][SQL] Reading nested columns with ORC vectorized reader can cause ArrayIndexOutOfBoundsException #35002

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0805ece
fixed ArrayIndexOutOfBoundsException when reading nested columns from…
yimin-yang Dec 23, 2021
8dce12d
Merge branch 'apache:master' into fix-nested
yimin-yang Dec 28, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 5 additions & 8 deletions ...re/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcArrayColumnVector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -31,26 +32,22 @@
*/
public class OrcArrayColumnVector extends OrcColumnVector {
private final OrcColumnVector data;
private final long[] offsets;
private final long[] lengths;

OrcArrayColumnVector(
Copy link
Contributor

@c21 c21 Dec 27, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we don't need to store another copy of vector here. We can change OrcColumnVector.baseData from private to protected, and just use baseData to get offset and length. So how about defining the constructor like below:

OrcArrayColumnVector(
    DataType type,
    ListColumnVector vector,
    OrcColumnVector data) {

  super(type, vector);

  this.data = data;
}

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
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcColumnVector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
6 changes: 2 additions & 4 deletions ...re/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcColumnVectorUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 " +
Expand Down
13 changes: 5 additions & 8 deletions ...core/src/main/java/org/apache/spark/sql/execution/datasources/orc/OrcMapColumnVector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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(
Copy link
Contributor

@c21 c21 Dec 27, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similarly we can define OrcMapColumnVector constructor as below:

OrcMapColumnVector(
    DataType type,
    MapColumnVector vector,
    OrcColumnVector keys,
    OrcColumnVector values) {

  super(type, vector);

  this.keys = keys;
  this.values = values;
}

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
Expand Down
22 changes: 22 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -744,6 +744,28 @@ 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
assert(vectorizationEnabled)
checkAnswer(readDf, df)
}
}
}

test("SPARK-36594: ORC vectorized reader should properly check maximal number of fields") {
withTempPath { dir =>
val path = dir.getCanonicalPath
Expand Down