Skip to content
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

Jump to bottom

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

Closed
wants to merge 8 commits into from
Closed

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
3403ca4
[SPARK-37728][SQL] Reading nested columns with ORC vectorized reader …
yimin-yang Dec 28, 2021
1708b68
update
yimin-yang Dec 29, 2021
6dd4f46
Merge branch 'apache:branch-3.2' into branch-3.2
yimin-yang Dec 29, 2021
de4fd40
update
yimin-yang Dec 29, 2021
62b98ce
Merge branch 'branch-3.2' of github.com:yym1995/spark into branch-3.2
yimin-yang Dec 29, 2021
5f2e85d
update
yimin-yang Dec 30, 2021
be7a155
update
yimin-yang Dec 31, 2021
294b02b
update
yimin-yang Dec 31, 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(
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(
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
23 changes: 23 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 @@ -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._
Expand Down Expand Up @@ -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
Copy link
Member

@dongjoon-hyun dongjoon-hyun Dec 31, 2021 •
edited
Loading

Choose a reason for hiding this comment

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

Why did you remove, assert(vectorizationEnabled), here, which is differently from the master branch? It looks like a removal of important test coverage again.

Since this test case is about Reading nested columns with ORC vectorized reader ..., you should not remove it. Did I miss something here?

Copy link
Contributor Author

@yimin-yang yimin-yang Jan 4, 2022 •
edited
Loading

Choose a reason for hiding this comment

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

Why did you remove, assert(vectorizationEnabled), here, which is differently from the master branch? It looks like a removal of important test coverage again.

Since this test case is about Reading nested columns with ORC vectorized reader ..., you should not remove it. Did I miss something here?

Because when testing with OrcV2QuerySuite, method supportColumnarReads in
https://github.com/apache/spark/blob/branch-3.2/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcPartitionReaderFactory.scala will be called. resultSchema.forall(_.dataType.isInstanceOf[AtomicType]) will return false in that case. Therefore, I removed assert(vectorizationEnabled).

To fix this issue, I think #33626 should also be backported to branch-3.2.

Copy link
Member

Choose a reason for hiding this comment

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

Got it. Thank you for the background, @yym1995 .

checkAnswer(readDf, df)
}
}
}
}

class OrcV1QuerySuite extends OrcQuerySuite {
Expand Down