apache / apache/parquet-java

GroupColumnIO.getFirst() doesn't check for empty groups

未关闭
#2,044 0 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看
Component: Java Component: Parquet Priority: Major Type: bug
主要语言
Java
星标
3.1k
派生
1.6k
平均合并
3 天 12 小时
30 天内合并 PR
33

描述

The following Spark snippet reproduces this issue with Spark 2.1 (with parquet-mr 1.8.1) and Spark 2.2-SNAPSHOT (with parquet-mr 1.8.2):

```Java
import org.apache.spark.sql.types._

val path = "/tmp/parquet-test"

case class Inner(f00: Int)
case class Outer(f0: Inner, f1: Int)

val df = Seq(Outer(Inner(1), 1)).toDF()

df.printSchema()
// root
// |-- f0: struct (nullable = true)
// | |-- f00: integer (nullable = false)
// |-- f1: integer (nullable = false)

df.write.mode("overwrite").parquet(path)

val requestedSchema =
new StructType().
add("f0", new StructType().
// This nested field name differs from the original one
add("f01", IntegerType)).
add("f1", IntegerType)

println(requestedSchema.treeString)
// root
// |-- f0: struct (nullable = true)
// | |-- f01: integer (nullable = true)
// |-- f1: integer (nullable = true)

spark.read.schema(requestedSchema).parquet(path).show()
```

In the above snippet, `requestedSchema` is compatible with the schema of the written Parquet file, but the following exception is thrown:

```
org.apache.parquet.io.ParquetDecodingException: Can not read value at 0 in block -1 in file file:/tmp/parquet-test/part-00007-d2b0bec1-7be5-4b51-8d53-3642680bc9c2.snappy.parquet
at org.apache.parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:243)
at org.apache.parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:227)
at org.apache.spark.sql.execution.datasources.RecordReaderIterator.hasNext(RecordReaderIterator.scala:39)
at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:109)
at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:184)
at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:109)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:377)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:231)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:225)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:826)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:826)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:99)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:322)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.IndexOutOfBoundsException: Index: 0, Size: 0
at java.util.ArrayList.rangeCheck(ArrayList.java:653)
at java.util.ArrayList.get(ArrayList.java:429)
at org.apache.parquet.io.GroupColumnIO.getFirst(GroupColumnIO.java:102)
at org.apache.parquet.io.GroupColumnIO.getFirst(GroupColumnIO.java:102)
at org.apache.parquet.io.PrimitiveColumnIO.getFirst(PrimitiveColumnIO.java:102)
at org.apache.parquet.io.PrimitiveColumnIO.isFirst(PrimitiveColumnIO.java:97)
at org.apache.parquet.io.RecordReaderImplementation.(RecordReaderImplementation.java:277)
at org.apache.parquet.io.MessageColumnIO$1.visit(MessageColumnIO.java:135)
at org.apache.parquet.io.MessageColumnIO$1.visit(MessageColumnIO.java:101)
at org.apache.parquet.filter2.compat.FilterCompat$NoOpFilter.accept(FilterCompat.java:154)
at org.apache.parquet.io.MessageColumnIO.getRecordReader(MessageColumnIO.java:101)
at org.apache.parquet.hadoop.InternalParquetRecordReader.checkRead(InternalParquetRecordReader.java:140)
at org.apache.parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:214)
... 21 more
```

According to this stack trace, it seems that `GroupColumnIO.getFirst()` [doesn't check for empty groups](https://github.com/apache/parquet-mr/blob/apache-parquet-1.8.2/parquet-column/src/main/java/org/apache/parquet/io/GroupColumnIO.java#L103) properly.

I haven't tried parquet-mr 1.9.0 but it probably suffers from the same issue.

**Reporter**: [Cheng Lian](https://issues.apache.org/jira/secure/ViewProfile.jspa?name=lian+cheng) / @liancheng

**Note**: *This issue was originally created as [PARQUET-893](https://issues.apache.org/jira/browse/PARQUET-893). Please see the [migration documentation](https://issues.apache.org/jira/browse/PARQUET-2502) for further details.*

贡献指南

这个仓库没有索引到贡献指南

调研方向

该失败指向 parquet-column/src/main/java/org/apache/parquet/io/GroupColumnIO.java 中的 getFirst(),堆栈跟踪中包含 RecordReaderImplementation。首先复现提供的 Spark schema 示例并检查 column IO 路径;完成的标准是读取不匹配的嵌套 schema 时不再抛出 IndexOutOfBoundsException。

由索引模型根据 Issue 内容生成。

评估

技术栈
java
领域
data-engineering
Issue 类型
缺陷
难度
4/5
预计耗时
3-5 天
活跃度
停滞
描述清晰度
基本清楚
新手友好度
42/100

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。