Support reading DECIMAL(18,2) columns from Parquet

[ash211]

We're seeing the below stacktrace on Parquet files with the following schema

! java.lang.UnsupportedOperationException: org.apache.parquet.column.values.dictionary.PlainValuesDictionary$PlainBinaryDictionary
! at org.apache.parquet.column.Dictionary.decodeToLong(Dictionary.java:52) ~[parquet-column-1.7.0.jar:1.7.0]
! at org.apache.spark.sql.execution.vectorized.OnHeapColumnVector.getLong(OnHeapColumnVector.java:274) ~[spark-sql_2.11-2.0.1.jar:2.0.1]
! at org.apache.spark.sql.execution.vectorized.ColumnVector.getDecimal(ColumnVector.java:588) ~[spark-sql_2.11-2.0.1.jar:2.0.1]
! at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source) ~[na:na]
! at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) ~[spark-sql_2.11-2.0.1.jar:2.0.1]
! at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:370) ~[spark-sql_2.11-2.0.1.jar:2.0.1]
! at org.apache.spark.sql.execution.SparkPlan$$anonfun$4.apply(SparkPlan.scala:246) ~[spark-sql_2.11-2.0.1.jar:2.0.1]
! at org.apache.spark.sql.execution.SparkPlan$$anonfun$4.apply(SparkPlan.scala:240) ~[spark-sql_2.11-2.0.1.jar:2.0.1]
! at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:803) ~[spark-core_2.11-2.0.1.jar:2.0.1]
! at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:803) ~[spark-core_2.11-2.0.1.jar:2.0.1]
! at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) ~[spark-core_2.11-2.0.1.jar:2.0.1]
! at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319) ~[spark-core_2.11-2.0.1.jar:2.0.1]
! at org.apache.spark.rdd.RDD.iterator(RDD.scala:283) ~[spark-core_2.11-2.0.1.jar:2.0.1]
! at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70) ~[spark-core_2.11-2.0.1.jar:2.0.1]
! at org.apache.spark.scheduler.Task.run(Task.scala:86) ~[spark-core_2.11-2.0.1.jar:2.0.1]
! at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274) ~[spark-core_2.11-2.0.1.jar:2.0.1]

Anonymized schema:

message parquetSchema {
  optional int32 COL_1 (INT_32);
  optional int32 COL_2 (INT_32);
  optional binary COL_3 (DECIMAL(18,2));
  optional binary COL_4 (UTF8);
  optional binary COL_5 (UTF8);
  optional int32 COL_6 (INT_32);
  optional int64 COL_7 (TIMESTAMP_MILLIS);
  optional int64 COL_8 (TIMESTAMP_MILLIS);
  optional int32 COL_9 (INT_32);
  optional binary COL_10 (UTF8);
  optional int64 COL_11 (TIMESTAMP_MILLIS);
  optional int64 COL_12 (TIMESTAMP_MILLIS);
}

COL_3 is the one causing problems.

[schlosna]

@ash211 haven't looked into this much as I'm on my phone but seems like it might be an issue in Parquet where it's mapping the column type to the wrong dictionary type -- it's getting a PlainBinaryDictionary when we want a PlainLongDictionary, see https://github.com/Parquet/parquet-mr/blob/master/parquet-column/src/main/java/parquet/column/Encoding.java

[pwoody]

I think this is because of Spark's assumption of the format when under a certain precision. It should be able to accept BINARY decimal values when under the max precision for Spark in particular.

…

Sent from my iPhone

On Jan 4, 2017, at 10:22 PM, David Schlosnagle ***@***.***> wrote: @ash211 haven't looked into this much as I'm on my phone but seems like it might be an issue in Parquet where it's mapping the column type to the wrong dictionary type -- it's getting a PlainBinaryDictionary when we want a PlainLongDictionary, see https://github.com/Parquet/parquet-mr/blob/master/parquet-column/src/main/java/parquet/column/Encoding.java — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub, or mute the thread.

[ash211]

Disabling the vectorized reader makes the data readable at least (at a presumed performance hit):

spark.sql.parquet.enableVectorizedReader=false

[robert3005]

Should open upstream