[SPARK-26797][SQL][WIP][test-maven] Start using the new logical types API of Parquet 1.11.1 instead of the deprecated one

dev/run-tests.py

@@ -343,7 +343,8 @@ def get_hive_profiles(hive_version):
 def build_spark_maven(extra_profiles):
     # Enable all of the profiles for the build:
     build_profiles = extra_profiles + modules.root.build_profile_flags
-    mvn_goals = ["clean", "package", "-DskipTests"]
+    mvn_goals = ["dependency:purge-local-repository", "-Dinclude=org.apache.parquet",
+                 "clean", "package", "-DskipTests"]
     profiles_and_goals = build_profiles + mvn_goals
 
     print("[info] Building Spark using Maven with these arguments: ", " ".join(profiles_and_goals))

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala

@@ -47,6 +47,8 @@ object DateTimeUtils {
   final val JULIAN_DAY_OF_EPOCH = 2440588
 
   final val TimeZoneUTC = TimeZone.getTimeZone("UTC")
+  // for why ".normalized", see https://stackoverflow.com/a/39507023/2965879
+  final val ZoneIdUTC = ZoneId.of("UTC").normalized()
 
   val TIMEZONE_OPTION = "timeZone"
 

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

@@ -32,6 +32,7 @@
 import org.apache.parquet.column.values.ValuesReader;
 import org.apache.parquet.io.api.Binary;
 import org.apache.parquet.schema.DecimalMetadata;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
 import org.apache.parquet.schema.OriginalType;
 import org.apache.parquet.schema.PrimitiveType;
 
@@ -46,6 +47,10 @@
 import org.apache.spark.sql.types.DecimalType;
 
 import static org.apache.parquet.column.ValuesType.REPETITION_LEVEL;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.DateLogicalTypeAnnotation;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.TimestampLogicalTypeAnnotation;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit.MICROS;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit.MILLIS;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64;
 import static org.apache.spark.sql.execution.datasources.parquet.SpecificParquetRecordReaderBase.ValuesReaderIntIterator;
@@ -104,7 +109,7 @@ public class VectorizedColumnReader {
 
   private final PageReader pageReader;
   private final ColumnDescriptor descriptor;
-  private final OriginalType originalType;
+  private final LogicalTypeAnnotation logicalTypeAnnotation;
   // The timezone conversion to apply to int96 timestamps. Null if no conversion.
   private final ZoneId convertTz;
   private static final ZoneId UTC = ZoneOffset.UTC;
@@ -136,15 +141,15 @@ private boolean canReadAsBinaryDecimal(DataType dt) {
 
   public VectorizedColumnReader(
       ColumnDescriptor descriptor,
-      OriginalType originalType,
+      LogicalTypeAnnotation logicalTypeAnnotation,
       PageReader pageReader,
       ZoneId convertTz,
       String datetimeRebaseMode,
       String int96RebaseMode) throws IOException {
     this.descriptor = descriptor;
     this.pageReader = pageReader;
     this.convertTz = convertTz;
-    this.originalType = originalType;
+    this.logicalTypeAnnotation = logicalTypeAnnotation;
     this.maxDefLevel = descriptor.getMaxDefinitionLevel();
 
     DictionaryPage dictionaryPage = pageReader.readDictionaryPage();
@@ -192,13 +197,14 @@ private boolean isLazyDecodingSupported(PrimitiveType.PrimitiveTypeName typeName
     boolean isSupported = false;
     switch (typeName) {
       case INT32:
-        isSupported = originalType != OriginalType.DATE || "CORRECTED".equals(datetimeRebaseMode);
+        isSupported = (!(logicalTypeAnnotation instanceof DateLogicalTypeAnnotation)
+                       || "CORRECTED".equals(datetimeRebaseMode));
         break;
       case INT64:
-        if (originalType == OriginalType.TIMESTAMP_MICROS) {
+        if (isTimestampWithUnit(logicalTypeAnnotation, MICROS)) {
           isSupported = "CORRECTED".equals(datetimeRebaseMode);
         } else {
-          isSupported = originalType != OriginalType.TIMESTAMP_MILLIS;
+          isSupported = !(logicalTypeAnnotation instanceof TimestampLogicalTypeAnnotation);
         }
         break;
       case FLOAT:
@@ -278,6 +284,7 @@ void readBatch(int total, WritableColumnVector column) throws IOException {
           // Column vector supports lazy decoding of dictionary values so just set the dictionary.
           // We can't do this if rowId != 0 AND the column doesn't have a dictionary (i.e. some
           // non-dictionary encoded values have already been added).
+          // TODO: replace OriginalType with something from LogicalTypeAnnotation
           PrimitiveType primitiveType = descriptor.getPrimitiveType();
           if (primitiveType.getOriginalType() == OriginalType.DECIMAL &&
               primitiveType.getDecimalMetadata().getPrecision() <= Decimal.MAX_INT_DIGITS() &&
@@ -398,14 +405,22 @@ private void decodeDictionaryIds(
       case INT64:
         if (column.dataType() == DataTypes.LongType ||
             canReadAsLongDecimal(column.dataType()) ||
-            (originalType == OriginalType.TIMESTAMP_MICROS &&
+            (isTimestampWithUnit(logicalTypeAnnotation, MICROS) &&
               "CORRECTED".equals(datetimeRebaseMode))) {
           for (int i = rowId; i < rowId + num; ++i) {
             if (!column.isNullAt(i)) {
               column.putLong(i, dictionary.decodeToLong(dictionaryIds.getDictId(i)));
             }
           }
-        } else if (originalType == OriginalType.TIMESTAMP_MILLIS) {
+        } else if (isTimestampWithUnit(logicalTypeAnnotation, MICROS)) {
+          final boolean failIfRebase = "EXCEPTION".equals(datetimeRebaseMode);
+          for (int i = rowId; i < rowId + num; ++i) {
+            if (!column.isNullAt(i)) {
+              long julianMicros = dictionary.decodeToLong(dictionaryIds.getDictId(i));
+              column.putLong(i, rebaseMicros(julianMicros, failIfRebase));
+            }
+          }
+        } else if (isTimestampWithUnit(logicalTypeAnnotation, MILLIS)) {
           if ("CORRECTED".equals(datetimeRebaseMode)) {
             for (int i = rowId; i < rowId + num; ++i) {
               if (!column.isNullAt(i)) {
@@ -423,14 +438,6 @@ private void decodeDictionaryIds(
               }
             }
           }
-        } else if (originalType == OriginalType.TIMESTAMP_MICROS) {
-          final boolean failIfRebase = "EXCEPTION".equals(datetimeRebaseMode);
-          for (int i = rowId; i < rowId + num; ++i) {
-            if (!column.isNullAt(i)) {
-              long julianMicros = dictionary.decodeToLong(dictionaryIds.getDictId(i));
-              column.putLong(i, rebaseMicros(julianMicros, failIfRebase));
-            }
-          }
         } else {
           throw constructConvertNotSupportedException(descriptor, column);
         }
@@ -592,7 +599,7 @@ private void readLongBatch(int rowId, int num, WritableColumnVector column) thro
       defColumn.readLongs(
         num, column, rowId, maxDefLevel, (VectorizedValuesReader) dataColumn,
         DecimalType.is32BitDecimalType(column.dataType()));
-    } else if (originalType == OriginalType.TIMESTAMP_MICROS) {
+    } else if (isTimestampWithUnit(logicalTypeAnnotation, MICROS)) {
       if ("CORRECTED".equals(datetimeRebaseMode)) {
         defColumn.readLongs(
           num, column, rowId, maxDefLevel, (VectorizedValuesReader) dataColumn, false);
@@ -601,7 +608,7 @@ private void readLongBatch(int rowId, int num, WritableColumnVector column) thro
         defColumn.readLongsWithRebase(
           num, column, rowId, maxDefLevel, (VectorizedValuesReader) dataColumn, failIfRebase);
       }
-    } else if (originalType == OriginalType.TIMESTAMP_MILLIS) {
+    } else if (isTimestampWithUnit(logicalTypeAnnotation, MILLIS)) {
       if ("CORRECTED".equals(datetimeRebaseMode)) {
         for (int i = 0; i < num; i++) {
           if (defColumn.readInteger() == maxDefLevel) {
@@ -626,6 +633,13 @@ private void readLongBatch(int rowId, int num, WritableColumnVector column) thro
     }
   }
 
+  private boolean isTimestampWithUnit(
+    LogicalTypeAnnotation logicalTypeAnnotation,
+    LogicalTypeAnnotation.TimeUnit timeUnit) {
+    return (logicalTypeAnnotation instanceof TimestampLogicalTypeAnnotation) &&
+      ((TimestampLogicalTypeAnnotation) logicalTypeAnnotation).getUnit() == timeUnit;
+  }
+
   private void readFloatBatch(int rowId, int num, WritableColumnVector column) throws IOException {
     // This is where we implement support for the valid type conversions.
     // TODO: support implicit cast to double?

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

@@ -332,7 +332,7 @@ private void checkEndOfRowGroup() throws IOException {
       if (missingColumns[i]) continue;
       columnReaders[i] = new VectorizedColumnReader(
         columns.get(i),
-        types.get(i).getOriginalType(),
+        types.get(i).getLogicalTypeAnnotation(),
         pages.getPageReader(columns.get(i)),
         convertTz,
         datetimeRebaseMode,

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala

@@ -272,20 +272,6 @@ class ParquetFileFormat
 
       lazy val footerFileMetaData =
         ParquetFileReader.readFooter(sharedConf, filePath, SKIP_ROW_GROUPS).getFileMetaData
-      // Try to push down filters when filter push-down is enabled.
-      val pushed = if (enableParquetFilterPushDown) {
-        val parquetSchema = footerFileMetaData.getSchema
-        val parquetFilters = new ParquetFilters(parquetSchema, pushDownDate, pushDownTimestamp,
-          pushDownDecimal, pushDownStringStartWith, pushDownInFilterThreshold, isCaseSensitive)
-        filters
-          // Collects all converted Parquet filter predicates. Notice that not all predicates can be
-          // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
-          // is used here.
-          .flatMap(parquetFilters.createFilter(_))
-          .reduceOption(FilterApi.and)
-      } else {
-        None
-      }
 
       // PARQUET_INT96_TIMESTAMP_CONVERSION says to apply timezone conversions to int96 timestamps'
       // *only* if the file was created by something other than "parquet-mr", so check the actual
@@ -302,6 +288,22 @@ class ParquetFileFormat
           None
         }
 
+      // Try to push down filters when filter push-down is enabled.
+      val pushed = if (enableParquetFilterPushDown) {
+        val parquetSchema = footerFileMetaData.getSchema
+        val parquetFilters = new ParquetFilters(parquetSchema, pushDownDate, pushDownTimestamp,
+          pushDownDecimal, pushDownStringStartWith, pushDownInFilterThreshold, isCaseSensitive,
+          convertTz.orNull)
+        filters
+          // Collects all converted Parquet filter predicates. Notice that not all predicates can be
+          // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
+          // is used here.
+          .flatMap(parquetFilters.createFilter(_))
+          .reduceOption(FilterApi.and)
+      } else {
+        None
+      }
+
       val datetimeRebaseMode = DataSourceUtils.datetimeRebaseMode(
         footerFileMetaData.getKeyValueMetaData.get,
         datetimeRebaseModeInRead)