Make Literals foldable, ensure Parquet predicates pushdown

dataset/src/main/scala/frameless/functions/Lit.scala

@@ -59,6 +59,7 @@ private[frameless] case class Lit[T <: AnyVal](
 
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): Expression = this
 
-  // see https://github.com/typelevel/frameless/pull/721#issuecomment-1581137730 for why true and not catalystExpr.foldable (InvokeLike <3.3.1 SPARK-40380)
-  override val foldable: Boolean = true
+  // see https://github.com/typelevel/frameless/pull/721#issuecomment-1581137730 (InvokeLike <3.3.1 SPARK-40380)
+  // for why this does not push down on 3.2, 3.3.1 and higher _do_ pushdown
+  override val foldable: Boolean = catalystExpr.foldable
 }

dataset/src/test/scala/frameless/sql/rules/FramelessLitPushDownTests.scala

@@ -12,43 +12,51 @@ import java.time.Instant
 class FramelessLitPushDownTests extends SQLRulesSuite {
   private val now: Long = currentTimestamp()
 
-  test("java.sql.Timestamp push-down") {
-    val expected = java.sql.Timestamp.from(microsToInstant(now))
-    val expectedStructure = X1(SQLTimestamp(now))
-    val expectedPushDownFilters = List(IsNotNull("a"), GreaterThanOrEqual("a", expected))
-
-    predicatePushDownTest[SQLTimestamp](
-      expectedStructure,
-      expectedPushDownFilters,
-      { case e @ expressions.GreaterThanOrEqual(_, _: Lit[_]) => e },
-      _ >= expectedStructure.a
-    )
-  }
-
-  test("java.time.Instant push-down") {
-    val expected = java.sql.Timestamp.from(microsToInstant(now))
-    val expectedStructure = X1(microsToInstant(now))
-    val expectedPushDownFilters = List(IsNotNull("a"), GreaterThanOrEqual("a", expected))
-
-    predicatePushDownTest[Instant](
-      expectedStructure,
-      expectedPushDownFilters,
-      { case e @ expressions.GreaterThanOrEqual(_, _: Lit[_]) => e },
-      _ >= expectedStructure.a
-    )
-  }
-
-  test("struct push-down") {
-    type Payload = X4[Int, Int, Int, Int]
-    val expectedStructure = X1(X4(1, 2, 3, 4))
-    val expected = new GenericRowWithSchema(Array(1, 2, 3, 4), TypedExpressionEncoder[Payload].schema)
-    val expectedPushDownFilters = List(IsNotNull("a"), EqualTo("a", expected))
-
-    predicatePushDownTest[Payload](
-      expectedStructure,
-      expectedPushDownFilters,
-      { case e @ expressions.EqualTo(_, _: Lit[_]) => e },
-      _ === expectedStructure.a
-    )
-  }
+  val invokeShouldFoldOnHigherThan3_2 = not3_2 _
+
+  test("java.sql.Timestamp push-down") (
+    invokeShouldFoldOnHigherThan3_2 {
+      val expected = java.sql.Timestamp.from(microsToInstant(now))
+      val expectedStructure = X1(SQLTimestamp(now))
+      val expectedPushDownFilters = List(IsNotNull("a"), GreaterThanOrEqual("a", expected))
+
+      predicatePushDownTest[SQLTimestamp](
+        expectedStructure,
+        expectedPushDownFilters,
+        { case e @ expressions.GreaterThanOrEqual(_, _: Lit[_]) => e },
+        _ >= expectedStructure.a
+      )
+    }
+  )
+
+  test("java.time.Instant push-down") (
+    invokeShouldFoldOnHigherThan3_2 {
+      val expected = java.sql.Timestamp.from(microsToInstant(now))
+      val expectedStructure = X1(microsToInstant(now))
+      val expectedPushDownFilters = List(IsNotNull("a"), GreaterThanOrEqual("a", expected))
+
+      predicatePushDownTest[Instant](
+        expectedStructure,
+        expectedPushDownFilters,
+        { case e @ expressions.GreaterThanOrEqual(_, _: Lit[_]) => e },
+        _ >= expectedStructure.a
+      )
+    }
+  )
+
+  test("struct push-down") (
+    invokeShouldFoldOnHigherThan3_2 {
+      type Payload = X4[Int, Int, Int, Int]
+      val expectedStructure = X1(X4(1, 2, 3, 4))
+      val expected = new GenericRowWithSchema(Array(1, 2, 3, 4), TypedExpressionEncoder[Payload].schema)
+      val expectedPushDownFilters = List(IsNotNull("a"), EqualTo("a", expected))
+
+      predicatePushDownTest[Payload](
+        expectedStructure,
+        expectedPushDownFilters,
+        { case e @ expressions.EqualTo(_, _: Lit[_]) => e },
+        _ === expectedStructure.a
+      )
+    }
+  )
 }

dataset/src/test/scala/frameless/sql/rules/SQLRulesSuite.scala

@@ -2,12 +2,12 @@ package frameless.sql.rules
 
 import frameless._
 import frameless.sql._
+import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.expressions.Expression
 import org.apache.spark.sql.sources.Filter
 import org.apache.spark.sql.catalyst.plans.logical
 import org.apache.spark.sql.execution.FileSourceScanExec
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
-
 import org.scalatest.Assertion
 import org.scalatest.matchers.should.Matchers
 
@@ -17,6 +17,27 @@ trait SQLRulesSuite extends TypedDatasetSuite with Matchers { self =>
     s"$tmpDir/${self.getClass.getName}"
   }
 
+  lazy val sparkFullVersion = {
+    val pos = classOf[Expression].getPackage.getSpecificationVersion
+    if (pos eq null) // DBR is always null
+      SparkSession.active.version // taking a running spark version string, hence lazy
+    else
+      pos
+  }
+
+  lazy val sparkVersion = {
+    sparkFullVersion.split('.').take(2).mkString(".")
+  }
+
+  /**
+   * Don't run this test on 3.2
+   */
+  def not3_2[T](thunk: => T): Any =
+    if (sparkVersion != "3.2")
+      thunk
+    else
+      ()
+
   def withDataset[A: TypedEncoder: CatalystOrdered](payload: A)(f: TypedDataset[A] => Assertion): Assertion = {
     TypedDataset.create(Seq(payload)).write.mode("overwrite").parquet(path)
     f(TypedDataset.createUnsafe[A](session.read.parquet(path)))