Dictionary decoding for int64 timestamps (apache#86)

Author: pwoody

sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java

@@ -244,6 +244,7 @@ private void decodeDictionaryIds(int rowId, int num, ColumnVector column,
 
       case INT64:
         if (column.dataType() == DataTypes.LongType ||
+            column.dataType() == DataTypes.TimestampType ||
             DecimalType.is64BitDecimalType(column.dataType())) {
           for (int i = rowId; i < rowId + num; ++i) {
             if (!column.isNullAt(i)) {

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala

@@ -811,6 +811,27 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSQLContext {
       }
     }
   }
+
+  test("Timestamp INT64 Dictionary encoding") {
+    val data = (1 to 1000).map { i =>
+      if (i < 500) {
+        Row(new java.sql.Timestamp(10))
+      } else {
+        Row(new java.sql.Timestamp(i))
+      }
+    }
+    val schema = StructType(List(StructField("time", TimestampType, false)).toArray)
+    withSQLConf(ParquetOutputFormat.DICTIONARY_PAGE_SIZE -> "64",
+        ParquetOutputFormat.PAGE_SIZE -> "128",
+        SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") {
+      withTempPath { file =>
+        val df = spark.createDataFrame(sparkContext.parallelize(data), schema)
+        df.coalesce(1).write.parquet(file.getCanonicalPath)
+        val df2 = spark.read.parquet(file.getCanonicalPath)
+        checkAnswer(df2, df.collect().toSeq)
+      }
+    }
+  }
 }
 
 class JobCommitFailureParquetOutputCommitter(outputPath: Path, context: TaskAttemptContext)