[SPARK-31489][SPARK-31488][SQL] Translate date values of pushed down filters to java.sql.Date

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/CatalystTypeConverters.scala

@@ -86,7 +86,7 @@ object CatalystTypeConverters {
    * @tparam ScalaOutputType The type of Scala values returned when converting Catalyst to Scala.
    * @tparam CatalystType The internal Catalyst type used to represent values of this Scala type.
    */
-  private abstract class CatalystTypeConverter[ScalaInputType, ScalaOutputType, CatalystType]
+  private[sql] abstract class CatalystTypeConverter[ScalaInputType, ScalaOutputType, CatalystType]
     extends Serializable {
 
     /**
@@ -301,7 +301,7 @@ object CatalystTypeConverters {
       row.getUTF8String(column).toString
   }
 
-  private object DateConverter extends CatalystTypeConverter[Date, Date, Any] {
+  private[sql] object DateConverter extends CatalystTypeConverter[Date, Date, Any] {
     override def toCatalystImpl(scalaValue: Date): Int = DateTimeUtils.fromJavaDate(scalaValue)
     override def toScala(catalystValue: Any): Date =
       if (catalystValue == null) null else DateTimeUtils.toJavaDate(catalystValue.asInstanceOf[Int])

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.catalyst
 
 import java.sql.{Date, Timestamp}
+import java.time.LocalDate
 
 import scala.language.implicitConversions
 
@@ -146,6 +147,7 @@ package object dsl {
     implicit def doubleToLiteral(d: Double): Literal = Literal(d)
     implicit def stringToLiteral(s: String): Literal = Literal.create(s, StringType)
     implicit def dateToLiteral(d: Date): Literal = Literal(d)
+    implicit def localDateToLiteral(d: LocalDate): Literal = Literal(d)
     implicit def bigDecimalToLiteral(d: BigDecimal): Literal = Literal(d.underlying())
     implicit def bigDecimalToLiteral(d: java.math.BigDecimal): Literal = Literal(d)
     implicit def decimalToLiteral(d: Decimal): Literal = Literal(d)

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

@@ -26,8 +26,8 @@ import org.apache.hadoop.fs.Path
 import org.apache.spark.internal.Logging
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql._
-import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow, QualifiedTableName}
-import org.apache.spark.sql.catalyst.CatalystTypeConverters.convertToScala
+import org.apache.spark.sql.catalyst.{InternalRow, QualifiedTableName}
+import org.apache.spark.sql.catalyst.CatalystTypeConverters.{convertToScala, DateConverter}
 import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.encoders.RowEncoder
@@ -448,47 +448,54 @@ object DataSourceStrategy {
     }
   }
 
+  private def translateFilterValues(catalystValue: Any, dataType: DataType): Any = {
+    dataType match {
+      case DateType => DateConverter.toScala(catalystValue)
+      case t => convertToScala(catalystValue, t)
+    }
+  }
+
   private def translateLeafNodeFilter(predicate: Expression): Option[Filter] = predicate match {
     case expressions.EqualTo(PushableColumn(name), Literal(v, t)) =>
-      Some(sources.EqualTo(name, convertToScala(v, t)))
+      Some(sources.EqualTo(name, translateFilterValues(v, t)))
     case expressions.EqualTo(Literal(v, t), PushableColumn(name)) =>
-      Some(sources.EqualTo(name, convertToScala(v, t)))
+      Some(sources.EqualTo(name, translateFilterValues(v, t)))
 
     case expressions.EqualNullSafe(PushableColumn(name), Literal(v, t)) =>
-      Some(sources.EqualNullSafe(name, convertToScala(v, t)))
+      Some(sources.EqualNullSafe(name, translateFilterValues(v, t)))
     case expressions.EqualNullSafe(Literal(v, t), PushableColumn(name)) =>
-      Some(sources.EqualNullSafe(name, convertToScala(v, t)))
+      Some(sources.EqualNullSafe(name, translateFilterValues(v, t)))
 
     case expressions.GreaterThan(PushableColumn(name), Literal(v, t)) =>
-      Some(sources.GreaterThan(name, convertToScala(v, t)))
+      Some(sources.GreaterThan(name, translateFilterValues(v, t)))
     case expressions.GreaterThan(Literal(v, t), PushableColumn(name)) =>
-      Some(sources.LessThan(name, convertToScala(v, t)))
+      Some(sources.LessThan(name, translateFilterValues(v, t)))
 
     case expressions.LessThan(PushableColumn(name), Literal(v, t)) =>
-      Some(sources.LessThan(name, convertToScala(v, t)))
+      Some(sources.LessThan(name, translateFilterValues(v, t)))
     case expressions.LessThan(Literal(v, t), PushableColumn(name)) =>
-      Some(sources.GreaterThan(name, convertToScala(v, t)))
+      Some(sources.GreaterThan(name, translateFilterValues(v, t)))
 
     case expressions.GreaterThanOrEqual(PushableColumn(name), Literal(v, t)) =>
-      Some(sources.GreaterThanOrEqual(name, convertToScala(v, t)))
+      Some(sources.GreaterThanOrEqual(name, translateFilterValues(v, t)))
     case expressions.GreaterThanOrEqual(Literal(v, t), PushableColumn(name)) =>
-      Some(sources.LessThanOrEqual(name, convertToScala(v, t)))
+      Some(sources.LessThanOrEqual(name, translateFilterValues(v, t)))
 
     case expressions.LessThanOrEqual(PushableColumn(name), Literal(v, t)) =>
-      Some(sources.LessThanOrEqual(name, convertToScala(v, t)))
+      Some(sources.LessThanOrEqual(name, translateFilterValues(v, t)))
     case expressions.LessThanOrEqual(Literal(v, t), PushableColumn(name)) =>
-      Some(sources.GreaterThanOrEqual(name, convertToScala(v, t)))
+      Some(sources.GreaterThanOrEqual(name, translateFilterValues(v, t)))
 
     case expressions.InSet(e @ PushableColumn(name), set) =>
-      val toScala = CatalystTypeConverters.createToScalaConverter(e.dataType)
+      val toScala = translateFilterValues(_, e.dataType)
       Some(sources.In(name, set.toArray.map(toScala)))
 
     // Because we only convert In to InSet in Optimizer when there are more than certain
     // items. So it is possible we still get an In expression here that needs to be pushed
     // down.
     case expressions.In(e @ PushableColumn(name), list) if list.forall(_.isInstanceOf[Literal]) =>
       val hSet = list.map(_.eval(EmptyRow))
-      val toScala = CatalystTypeConverters.createToScalaConverter(e.dataType)
+      val toScala = translateFilterValues(_, e.dataType)
       Some(sources.In(name, hSet.toArray.map(toScala)))
 
     case expressions.IsNull(PushableColumn(name)) =>

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

@@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.datasources.parquet
 import java.math.{BigDecimal => JBigDecimal}
 import java.nio.charset.StandardCharsets
 import java.sql.{Date, Timestamp}
+import java.time.LocalDate
 
 import org.apache.parquet.filter2.predicate.{FilterApi, FilterPredicate, Operators}
 import org.apache.parquet.filter2.predicate.FilterApi._
@@ -1561,6 +1562,63 @@ abstract class ParquetFilterSuite extends QueryTest with ParquetTest with Shared
       }
     }
   }
+
+  test("filter pushdown - local date") {
+    implicit class StringToDate(s: String) {
+      def date: LocalDate = LocalDate.parse(s)
+    }
+
+    val data = Seq("2018-03-18", "2018-03-19", "2018-03-20", "2018-03-21").map(_.date)
+    import testImplicits._
+    withSQLConf(SQLConf.DATETIME_JAVA8API_ENABLED.key -> "true") {
+      withNestedDataFrame(data.map(i => Tuple1(i)).toDF()) { case (inputDF, colName, resultFun) =>
+        withParquetDataFrame(inputDF) { implicit df =>
+          val dateAttr: Expression = df(colName).expr
+          assert(df(colName).expr.dataType === DateType)
+
+          checkFilterPredicate(dateAttr.isNull, classOf[Eq[_]], Seq.empty[Row])
+          checkFilterPredicate(dateAttr.isNotNull, classOf[NotEq[_]],
+            data.map(i => Row.apply(resultFun(i))))
+
+          checkFilterPredicate(dateAttr === "2018-03-18".date, classOf[Eq[_]],
+            resultFun("2018-03-18".date))
+          checkFilterPredicate(dateAttr <=> "2018-03-18".date, classOf[Eq[_]],
+            resultFun("2018-03-18".date))
+          checkFilterPredicate(dateAttr =!= "2018-03-18".date, classOf[NotEq[_]],
+            Seq("2018-03-19", "2018-03-20", "2018-03-21").map(i => Row.apply(resultFun(i.date))))
+
+          checkFilterPredicate(dateAttr < "2018-03-19".date, classOf[Lt[_]],
+            resultFun("2018-03-18".date))
+          checkFilterPredicate(dateAttr > "2018-03-20".date, classOf[Gt[_]],
+            resultFun("2018-03-21".date))
+          checkFilterPredicate(dateAttr <= "2018-03-18".date, classOf[LtEq[_]],
+            resultFun("2018-03-18".date))
+          checkFilterPredicate(dateAttr >= "2018-03-21".date, classOf[GtEq[_]],
+            resultFun("2018-03-21".date))
+
+          checkFilterPredicate(Literal("2018-03-18".date) === dateAttr, classOf[Eq[_]],
+            resultFun("2018-03-18".date))
+          checkFilterPredicate(Literal("2018-03-18".date) <=> dateAttr, classOf[Eq[_]],
+            resultFun("2018-03-18".date))
+          checkFilterPredicate(Literal("2018-03-19".date) > dateAttr, classOf[Lt[_]],
+            resultFun("2018-03-18".date))
+          checkFilterPredicate(Literal("2018-03-20".date) < dateAttr, classOf[Gt[_]],
+            resultFun("2018-03-21".date))
+          checkFilterPredicate(Literal("2018-03-18".date) >= dateAttr, classOf[LtEq[_]],
+            resultFun("2018-03-18".date))
+          checkFilterPredicate(Literal("2018-03-21".date) <= dateAttr, classOf[GtEq[_]],
+            resultFun("2018-03-21".date))
+
+          checkFilterPredicate(!(dateAttr < "2018-03-21".date), classOf[GtEq[_]],
+            resultFun("2018-03-21".date))
+          checkFilterPredicate(
+            dateAttr < "2018-03-19".date || dateAttr > "2018-03-20".date,
+            classOf[Operators.Or],
+            Seq(Row(resultFun("2018-03-18".date)), Row(resultFun("2018-03-21".date))))
+        }
+      }
+    }
+  }
 }
 
 class ParquetV1FilterSuite extends ParquetFilterSuite {

sql/core/v1.2/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala

@@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.datasources.orc
 import java.math.MathContext
 import java.nio.charset.StandardCharsets
 import java.sql.{Date, Timestamp}
+import java.time.LocalDate
 
 import scala.collection.JavaConverters._
 
@@ -450,5 +451,31 @@ class OrcFilterSuite extends OrcTest with SharedSparkSession {
       ).get.toString
     }
   }
+
+  test("filter pushdown - local date") {
+    val dates = Seq("2017-08-18", "2017-08-19", "2017-08-20", "2017-08-21").map { day =>
+      LocalDate.parse(day)
+    }
+    withSQLConf(SQLConf.DATETIME_JAVA8API_ENABLED.key -> "true") {
+      withOrcDataFrame(dates.map(Tuple1(_))) { implicit df =>
+        checkFilterPredicate($"_1".isNull, PredicateLeaf.Operator.IS_NULL)
+
+        checkFilterPredicate($"_1" === dates(0), PredicateLeaf.Operator.EQUALS)
+        checkFilterPredicate($"_1" <=> dates(0), PredicateLeaf.Operator.NULL_SAFE_EQUALS)
+
+        checkFilterPredicate($"_1" < dates(1), PredicateLeaf.Operator.LESS_THAN)
+        checkFilterPredicate($"_1" > dates(2), PredicateLeaf.Operator.LESS_THAN_EQUALS)
+        checkFilterPredicate($"_1" <= dates(0), PredicateLeaf.Operator.LESS_THAN_EQUALS)
+        checkFilterPredicate($"_1" >= dates(3), PredicateLeaf.Operator.LESS_THAN)
+
+        checkFilterPredicate(Literal(dates(0)) === $"_1", PredicateLeaf.Operator.EQUALS)
+        checkFilterPredicate(Literal(dates(0)) <=> $"_1", PredicateLeaf.Operator.NULL_SAFE_EQUALS)
+        checkFilterPredicate(Literal(dates(1)) > $"_1", PredicateLeaf.Operator.LESS_THAN)
+        checkFilterPredicate(Literal(dates(2)) < $"_1", PredicateLeaf.Operator.LESS_THAN_EQUALS)
+        checkFilterPredicate(Literal(dates(0)) >= $"_1", PredicateLeaf.Operator.LESS_THAN_EQUALS)
+        checkFilterPredicate(Literal(dates(3)) <= $"_1", PredicateLeaf.Operator.LESS_THAN)
+      }
+    }
+  }
 }
 

sql/core/v2.3/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala

@@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.datasources.orc
 import java.math.MathContext
 import java.nio.charset.StandardCharsets
 import java.sql.{Date, Timestamp}
+import java.time.LocalDate
 
 import scala.collection.JavaConverters._
 
@@ -451,5 +452,31 @@ class OrcFilterSuite extends OrcTest with SharedSparkSession {
       ).get.toString
     }
   }
+
+  test("filter pushdown - local date") {
+    val dates = Seq("2017-08-18", "2017-08-19", "2017-08-20", "2017-08-21").map { day =>
+      LocalDate.parse(day)
+    }
+    withSQLConf(SQLConf.DATETIME_JAVA8API_ENABLED.key -> "true") {
+      withOrcDataFrame(dates.map(Tuple1(_))) { implicit df =>
+        checkFilterPredicate($"_1".isNull, PredicateLeaf.Operator.IS_NULL)
+
+        checkFilterPredicate($"_1" === dates(0), PredicateLeaf.Operator.EQUALS)
+        checkFilterPredicate($"_1" <=> dates(0), PredicateLeaf.Operator.NULL_SAFE_EQUALS)
+
+        checkFilterPredicate($"_1" < dates(1), PredicateLeaf.Operator.LESS_THAN)
+        checkFilterPredicate($"_1" > dates(2), PredicateLeaf.Operator.LESS_THAN_EQUALS)
+        checkFilterPredicate($"_1" <= dates(0), PredicateLeaf.Operator.LESS_THAN_EQUALS)
+        checkFilterPredicate($"_1" >= dates(3), PredicateLeaf.Operator.LESS_THAN)
+
+        checkFilterPredicate(Literal(dates(0)) === $"_1", PredicateLeaf.Operator.EQUALS)
+        checkFilterPredicate(Literal(dates(0)) <=> $"_1", PredicateLeaf.Operator.NULL_SAFE_EQUALS)
+        checkFilterPredicate(Literal(dates(1)) > $"_1", PredicateLeaf.Operator.LESS_THAN)
+        checkFilterPredicate(Literal(dates(2)) < $"_1", PredicateLeaf.Operator.LESS_THAN_EQUALS)
+        checkFilterPredicate(Literal(dates(0)) >= $"_1", PredicateLeaf.Operator.LESS_THAN_EQUALS)
+        checkFilterPredicate(Literal(dates(3)) <= $"_1", PredicateLeaf.Operator.LESS_THAN)
+      }
+    }
+  }
 }