fix parquet splitters

Author: onursatici

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

@@ -313,7 +313,7 @@ class ParquetFileFormat
         val splits = ParquetFileFormat.fileSplits.get(root,
           new Callable[ParquetFileSplitter] {
             override def call(): ParquetFileSplitter =
-              createParquetFileSplits(root, hadoopConf, schema, sparkSession)
+              createParquetFileSplits(root, hadoopConf, sparkSession)
           })
         root -> splits.buildSplitter(filters)
       }.toMap
@@ -331,11 +331,11 @@ class ParquetFileFormat
   private def createParquetFileSplits(
     root: Path,
     hadoopConf: Configuration,
-    schema: StructType,
     sparkSession: SparkSession): ParquetFileSplitter = {
     getMetadataForPath(root, hadoopConf)
       .map { meta =>
-        new ParquetMetadataFileSplitter(root, meta.getBlocks.asScala, schema, sparkSession)
+        new ParquetMetadataFileSplitter(
+          root, meta.getBlocks.asScala, meta.getFileMetaData.getSchema, sparkSession)
       }
       .getOrElse(ParquetDefaultFileSplitter)
   }

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

@@ -25,15 +25,15 @@ import scala.concurrent.{ExecutionContext, Future}
 import com.google.common.cache.{Cache, CacheBuilder}
 import org.apache.hadoop.fs.{FileStatus, Path}
 import org.apache.hadoop.mapreduce.lib.input.FileSplit
-import org.apache.parquet.filter2.predicate.{FilterApi, FilterPredicate}
+import org.apache.parquet.filter2.predicate.FilterApi
 import org.apache.parquet.filter2.statisticslevel.StatisticsFilter
 import org.apache.parquet.hadoop.metadata.BlockMetaData
+import org.apache.parquet.schema.MessageType
 import org.roaringbitmap.RoaringBitmap
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.sources.Filter
-import org.apache.spark.sql.types.StructType
 import org.apache.spark.util.ThreadUtils
 
 
@@ -54,14 +54,18 @@ object ParquetDefaultFileSplitter extends ParquetFileSplitter {
 class ParquetMetadataFileSplitter(
     val root: Path,
     val blocks: Seq[BlockMetaData],
-    val schema: StructType,
+    val parquetSchema: MessageType,
     val session: SparkSession)
   extends ParquetFileSplitter
   with Logging {
 
-  private val parquetFilters = new ParquetFilters(
-    session.sessionState.conf.parquetFilterPushDownDate,
-    session.sessionState.conf.isParquetINT96AsTimestamp)
+  val sqlConf = session.sessionState.conf
+  val parquetFilters = new ParquetFilters(
+    sqlConf.parquetFilterPushDownDate,
+    sqlConf.parquetFilterPushDownTimestamp,
+    sqlConf.parquetFilterPushDownDecimal,
+    sqlConf.parquetFilterPushDownStringStartWith,
+    sqlConf.parquetFilterPushDownInFilterThreshold)
 
   private val referencedFiles = blocks.map(bmd => new Path(root, bmd.getPath)).toSet
 
@@ -106,7 +110,7 @@ class ParquetMetadataFileSplitter(
   private def applyParquetFilter(
       filters: Seq[Filter],
       blocks: Seq[BlockMetaData]): Seq[BlockMetaData] = {
-    val predicates = filters.flatMap(parquetFilters.createFilter(schema, _))
+    val predicates = filters.flatMap(parquetFilters.createFilter(parquetSchema, _))
     if (predicates.nonEmpty) {
       // Asynchronously build bitmaps
       Future {
@@ -127,7 +131,7 @@ class ParquetMetadataFileSplitter(
         .filter(filterSets.getIfPresent(_) == null)
         .flatMap { filter =>
           val bitmap = new RoaringBitmap
-          parquetFilters.createFilter(schema, filter).map((filter, _, bitmap))
+          parquetFilters.createFilter(parquetSchema, filter).map((filter, _, bitmap))
         }
       var i = 0
       val blockLen = blocks.size

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

@@ -33,6 +33,7 @@ import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
 import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName._
 
 import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.catalyst.util.DateTimeUtils.SQLDate
 import org.apache.spark.sql.sources
 import org.apache.spark.unsafe.types.UTF8String
 
@@ -67,27 +68,8 @@ private[parquet] class ParquetFilters(
 
   import ParquetColumns._
 
-  private val makeInSet: PartialFunction[DataType, (String, Set[Any]) => FilterPredicate] = {
-    case IntegerType =>
-      (n: String, v: Set[Any]) =>
-        FilterApi.userDefined(intColumn(n), SetInFilter(v.asInstanceOf[Set[java.lang.Integer]]))
-    case LongType =>
-      (n: String, v: Set[Any]) =>
-        FilterApi.userDefined(longColumn(n), SetInFilter(v.asInstanceOf[Set[java.lang.Long]]))
-    case FloatType =>
-      (n: String, v: Set[Any]) =>
-        FilterApi.userDefined(floatColumn(n), SetInFilter(v.asInstanceOf[Set[java.lang.Float]]))
-    case DoubleType =>
-      (n: String, v: Set[Any]) =>
-        FilterApi.userDefined(doubleColumn(n), SetInFilter(v.asInstanceOf[Set[java.lang.Double]]))
-    case StringType =>
-      (n: String, v: Set[Any]) =>
-        FilterApi.userDefined(binaryColumn(n),
-          SetInFilter(v.map(s => Binary.fromString(s.asInstanceOf[String]))))
-    case BinaryType =>
-      (n: String, v: Set[Any]) =>
-        FilterApi.userDefined(binaryColumn(n),
-          SetInFilter(v.map(e => Binary.fromReusedByteArray(e.asInstanceOf[Array[Byte]]))))
+  private def dateToDays(date: Date): SQLDate = {
+    DateTimeUtils.fromJavaDate(date)
   }
 
   private def decimalToInt32(decimal: JBigDecimal): Integer = decimal.unscaledValue().intValue()
@@ -231,6 +213,7 @@ private[parquet] class ParquetFilters(
       (n: String, v: Any) =>
         FilterApi.lt(intColumn(n), dateToDays(v.asInstanceOf[Date]).asInstanceOf[Integer])
     case ParquetTimestampMicrosType if pushDownTimestamp =>
+      (n: String, v: Any) => FilterApi.lt(
         longColumn(n),
         DateTimeUtils.fromJavaTimestamp(v.asInstanceOf[Timestamp]).asInstanceOf[JLong])
     case ParquetTimestampMillisType if pushDownTimestamp =>
@@ -487,9 +470,6 @@ private[parquet] class ParquetFilters(
           .map(FilterApi.not)
           .map(LogicalInverseRewriter.rewrite)
 
-      case sources.In(name, values) if canMakeFilterOn(name) =>
-        makeInSet.lift(nameToType(name)).map(_(name, values.toSet))
-
       case sources.In(name, values) if canMakeFilterOn(name, values.head)
         && values.distinct.length <= pushDownInFilterThreshold =>
         values.distinct.flatMap { v =>