[VL] Update Parquet unit tests (#12776)

Author: rui-mo

gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -391,27 +391,18 @@ class VeloxTestSettings extends BackendTestSettings {
   enableSuite[GlutenOrcV1SchemaPruningSuite]
   enableSuite[GlutenOrcV2SchemaPruningSuite]
   enableSuite[GlutenParquetColumnIndexSuite]
-    // Rewrite by just removing test timestamp.
-    .exclude("test reading unaligned pages - test all types")
-    // Rewrite by converting smaller integral value to timestamp.
-    .exclude("test reading unaligned pages - test all types (dict encode)")
   enableSuite[GlutenParquetCompressionCodecPrecedenceSuite]
   enableSuite[GlutenParquetDeltaByteArrayEncodingSuite]
   enableSuite[GlutenParquetDeltaEncodingInteger]
   enableSuite[GlutenParquetDeltaEncodingLong]
   enableSuite[GlutenParquetDeltaLengthByteArrayEncodingSuite]
   enableSuite[GlutenParquetEncodingSuite]
-    // Velox does not support rle encoding, but it can pass when native writer enabled.
-    .exclude("parquet v2 pages - rle encoding for boolean value columns")
   enableSuite[GlutenParquetFieldIdIOSuite]
   enableSuite[GlutenParquetFileFormatV1Suite]
   enableSuite[GlutenParquetFileFormatV2Suite]
   enableSuite[GlutenParquetV1FilterSuite]
     // Rewrite.
     .exclude("SPARK-23852: Broken Parquet push-down for partially-written stats")
-    // Rewrite for supported INT96 - timestamp.
-    .exclude("filter pushdown - timestamp")
-    .exclude("filter pushdown - date")
     // Exception bebaviour.
     .exclude("SPARK-25207: exception when duplicate fields in case-insensitive mode")
     // Ignore Spark's filter pushdown check.
@@ -427,9 +418,6 @@ class VeloxTestSettings extends BackendTestSettings {
   enableSuite[GlutenParquetV2FilterSuite]
     // Rewrite.
     .exclude("SPARK-23852: Broken Parquet push-down for partially-written stats")
-    // Rewrite for supported INT96 - timestamp.
-    .exclude("filter pushdown - timestamp")
-    .exclude("filter pushdown - date")
     // Exception bebaviour.
     .exclude("SPARK-25207: exception when duplicate fields in case-insensitive mode")
     // Ignore Spark's filter pushdown check.
@@ -443,6 +431,7 @@ class VeloxTestSettings extends BackendTestSettings {
     .exclude("filter pushdown - StringPredicate")
     .exclude("SPARK-38825: in and notIn filters")
   enableSuite[GlutenParquetInteroperabilitySuite]
+    // Disabled because Velox does not implement Spark’s legacy Impala INT96 timestamp conversion semantics.
     .exclude("parquet timestamp conversion")
   enableSuite[GlutenParquetIOSuite]
     // Velox doesn't write file metadata into parquet file.
@@ -451,35 +440,35 @@ class VeloxTestSettings extends BackendTestSettings {
     .exclude("SPARK-35640: read binary as timestamp should throw schema incompatible error")
     // Exception msg.
     .exclude("SPARK-35640: int as long should throw schema incompatible error")
-    // Velox parquet reader not allow offset zero.
-    .exclude("SPARK-40128 read DELTA_LENGTH_BYTE_ARRAY encoded strings")
   enableSuite[GlutenParquetV1PartitionDiscoverySuite]
   enableSuite[GlutenParquetV2PartitionDiscoverySuite]
   enableSuite[GlutenParquetProtobufCompatibilitySuite]
   enableSuite[GlutenParquetV1QuerySuite]
+    // No fix needed: INT read as LONG is allowed.
     .exclude("row group skipping doesn't overflow when reading into larger type")
     // Unsupport spark.sql.files.ignoreCorruptFiles.
     .exclude("Enabling/disabling ignoreCorruptFiles")
-    // decimal failed ut
+    // No fix needed: narrowing the scale should not be allowed.
     .exclude("SPARK-34212 Parquet should read decimals correctly")
     // new added in spark-3.3 and need fix later, random failure may caused by memory free
     .exclude("SPARK-39833: pushed filters with project without filter columns")
     .exclude("SPARK-39833: pushed filters with count()")
     // Rewrite because the filter after datasource is not needed.
     .exclude(
       "SPARK-26677: negated null-safe equality comparison should not filter matched row groups")
-    // Velox currently does not distinguish `isAdjustedToUTC` in Parquet.
+    // No fix needed: read TimestampLTZ as TimestampNTZ is allowed.
     .exclude("SPARK-36182: can't read TimestampLTZ as TimestampNTZ")
   enableSuite[GlutenParquetV2QuerySuite]
+    // No fix needed: INT read as LONG is allowed.
     .exclude("row group skipping doesn't overflow when reading into larger type")
     // Unsupport spark.sql.files.ignoreCorruptFiles.
     .exclude("Enabling/disabling ignoreCorruptFiles")
-    // decimal failed ut
+    // No fix needed: narrowing the scale should not be allowed.
     .exclude("SPARK-34212 Parquet should read decimals correctly")
     // Rewrite because the filter after datasource is not needed.
     .exclude(
       "SPARK-26677: negated null-safe equality comparison should not filter matched row groups")
-    // Velox currently does not distinguish `isAdjustedToUTC` in Parquet.
+    // No fix needed: read TimestampLTZ as TimestampNTZ is allowed.
     .exclude("SPARK-36182: can't read TimestampLTZ as TimestampNTZ")
   enableSuite[GlutenParquetV1SchemaPruningSuite]
   enableSuite[GlutenParquetV2SchemaPruningSuite]
@@ -526,8 +515,6 @@ class VeloxTestSettings extends BackendTestSettings {
   enableSuite[GlutenFileMetadataStructRowIndexSuite]
   enableSuite[GlutenParquetV1AggregatePushDownSuite]
   enableSuite[GlutenParquetV2AggregatePushDownSuite]
-    // TODO: Timestamp columns stats will lost if using int64 in parquet writer.
-    .exclude("aggregate push down - different data types")
   enableSuite[GlutenOrcV1AggregatePushDownSuite]
     .exclude("nested column: Count(nested sub-field) not push down")
   enableSuite[GlutenOrcV2AggregatePushDownSuite]

gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetColumnIndexSuite.scala

@@ -16,50 +16,6 @@
  */
 package org.apache.spark.sql.execution.datasources.parquet
 
-import org.apache.spark.sql.{DataFrame, GlutenSQLTestsBaseTrait}
+import org.apache.spark.sql.GlutenSQLTestsBaseTrait
 
-class GlutenParquetColumnIndexSuite extends ParquetColumnIndexSuite with GlutenSQLTestsBaseTrait {
-  private val actions: Seq[DataFrame => DataFrame] = Seq(
-    "_1 = 500",
-    "_1 = 500 or _1 = 1500",
-    "_1 = 500 or _1 = 501 or _1 = 1500",
-    "_1 = 500 or _1 = 501 or _1 = 1000 or _1 = 1500",
-    "_1 >= 500 and _1 < 1000",
-    "(_1 >= 500 and _1 < 1000) or (_1 >= 1500 and _1 < 1600)"
-  ).map(f => (df: DataFrame) => df.filter(f))
-
-  testGluten("test reading unaligned pages - test all types") {
-    val df = spark
-      .range(0, 2000)
-      .selectExpr(
-        "id as _1",
-        "cast(id as short) as _3",
-        "cast(id as int) as _4",
-        "cast(id as float) as _5",
-        "cast(id as double) as _6",
-        "cast(id as decimal(20,0)) as _7",
-        // We changed 1618161925000 to 1618161925 to avoid reaching the limitation of Velox:
-        // Timepoint is outside of supported year range.
-        "cast(cast(1618161925 + id * 60 * 60 * 24 as timestamp) as date) as _9"
-      )
-    checkUnalignedPages(df)(actions: _*)
-  }
-
-  testGluten("test reading unaligned pages - test all types (dict encode)") {
-    val df = spark
-      .range(0, 2000)
-      .selectExpr(
-        "id as _1",
-        "cast(id % 10 as byte) as _2",
-        "cast(id % 10 as short) as _3",
-        "cast(id % 10 as int) as _4",
-        "cast(id % 10 as float) as _5",
-        "cast(id % 10 as double) as _6",
-        "cast(id % 10 as decimal(20,0)) as _7",
-        "cast(id % 2 as boolean) as _8",
-        "cast(cast(1618161925 + (id % 10) * 60 * 60 * 24 as timestamp) as date) as _9",
-        "cast(1618161925 + (id % 10) as timestamp) as _10"
-      )
-    checkUnalignedPages(df)(actions: _*)
-  }
-}
+class GlutenParquetColumnIndexSuite extends ParquetColumnIndexSuite with GlutenSQLTestsBaseTrait {}

gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala

@@ -27,25 +27,16 @@ import org.apache.spark.sql.execution.datasources.{DataSourceStrategy, HadoopFsR
 import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
 import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy.{CORRECTED, LEGACY}
-import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType.INT96
 import org.apache.spark.sql.types._
 import org.apache.spark.tags.ExtendedSQLTest
 import org.apache.spark.util.Utils
 
 import org.apache.hadoop.fs.Path
-import org.apache.parquet.filter2.predicate.{FilterApi, FilterPredicate, Operators}
+import org.apache.parquet.filter2.predicate.{FilterApi, FilterPredicate}
 import org.apache.parquet.filter2.predicate.FilterApi._
-import org.apache.parquet.filter2.predicate.Operators.{Column => _, Eq, Gt, GtEq, Lt, LtEq, NotEq}
 import org.apache.parquet.hadoop.{ParquetFileReader, ParquetInputFormat, ParquetOutputFormat}
 import org.apache.parquet.hadoop.util.HadoopInputFile
 
-import java.sql.{Date, Timestamp}
-import java.time.LocalDate
-
-import scala.reflect.ClassTag
-import scala.reflect.runtime.universe.TypeTag
-
 abstract class GlutenParquetFilterSuite extends ParquetFilterSuite with GlutenSQLTestsBaseTrait {
   protected def checkFilterPredicate(
       predicate: Predicate,
@@ -66,44 +57,6 @@ abstract class GlutenParquetFilterSuite extends ParquetFilterSuite with GlutenSQ
       getWorkspaceFilePath("sql", "core", "src", "test", "resources").toString + "/" + name)
   }
 
-  testGluten("filter pushdown - timestamp") {
-    Seq(true, false).foreach {
-      java8Api =>
-        Seq(CORRECTED, LEGACY).foreach {
-          rebaseMode =>
-            val millisData = Seq(
-              "1000-06-14 08:28:53.123",
-              "1582-06-15 08:28:53.001",
-              "1900-06-16 08:28:53.0",
-              "2018-06-17 08:28:53.999")
-            // INT96 doesn't support pushdown
-            withSQLConf(
-              SQLConf.DATETIME_JAVA8API_ENABLED.key -> java8Api.toString,
-              SQLConf.PARQUET_INT96_REBASE_MODE_IN_WRITE.key -> rebaseMode.toString,
-              SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> INT96.toString
-            ) {
-              import testImplicits._
-              withTempPath {
-                file =>
-                  millisData
-                    .map(i => Tuple1(Timestamp.valueOf(i)))
-                    .toDF
-                    .write
-                    .format(dataSourceName)
-                    .save(file.getCanonicalPath)
-                  readParquetFile(file.getCanonicalPath) {
-                    df =>
-                      val schema = new SparkToParquetSchemaConverter(conf).convert(df.schema)
-                      assertResult(None) {
-                        createParquetFilters(schema).createFilter(sources.IsNull("_1"))
-                      }
-                  }
-              }
-            }
-        }
-    }
-  }
-
   testGluten("SPARK-12218: 'Not' is included in Parquet filter pushdown") {
     import testImplicits._
 
@@ -428,153 +381,4 @@ class GlutenParquetV2FilterSuite extends GlutenParquetFilterSuite with GlutenSQL
       }
     }
   }
-
-  /**
-   * Takes a sequence of products `data` to generate multi-level nested dataframes as new test data.
-   * It tests both non-nested and nested dataframes which are written and read back with Parquet
-   * datasource.
-   *
-   * This is different from [[ParquetTest.withParquetDataFrame]] which does not test nested cases.
-   */
-  private def withNestedParquetDataFrame[T <: Product: ClassTag: TypeTag](data: Seq[T])(
-      runTest: (DataFrame, String, Any => Any) => Unit): Unit =
-    withNestedParquetDataFrame(spark.createDataFrame(data))(runTest)
-
-  private def withNestedParquetDataFrame(inputDF: DataFrame)(
-      runTest: (DataFrame, String, Any => Any) => Unit): Unit = {
-    withNestedDataFrame(inputDF).foreach {
-      case (newDF, colName, resultFun) =>
-        withTempPath {
-          file =>
-            newDF.write.format(dataSourceName).save(file.getCanonicalPath)
-            readParquetFile(file.getCanonicalPath)(df => runTest(df, colName, resultFun))
-        }
-    }
-  }
-
-  testGluten("filter pushdown - date") {
-    implicit class StringToDate(s: String) {
-      def date: Date = Date.valueOf(s)
-    }
-
-    val data = Seq("1000-01-01", "2018-03-19", "2018-03-20", "2018-03-21")
-    import testImplicits._
-
-    // Velox backend does not support rebaseMode being LEGACY.
-    Seq(false, true).foreach {
-      java8Api =>
-        Seq(CORRECTED).foreach {
-          rebaseMode =>
-            withSQLConf(
-              SQLConf.DATETIME_JAVA8API_ENABLED.key -> java8Api.toString,
-              SQLConf.PARQUET_REBASE_MODE_IN_WRITE.key -> rebaseMode.toString) {
-              val dates = data.map(i => Tuple1(Date.valueOf(i))).toDF()
-              withNestedParquetDataFrame(dates) {
-                case (inputDF, colName, fun) =>
-                  implicit val df: DataFrame = inputDF
-
-                  def resultFun(dateStr: String): Any = {
-                    val parsed = if (java8Api) LocalDate.parse(dateStr) else Date.valueOf(dateStr)
-                    fun(parsed)
-                  }
-
-                  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 === "1000-01-01".date,
-                    classOf[Eq[_]],
-                    resultFun("1000-01-01"))
-                  logWarning(s"java8Api: $java8Api, rebaseMode, $rebaseMode")
-                  checkFilterPredicate(
-                    dateAttr <=> "1000-01-01".date,
-                    classOf[Eq[_]],
-                    resultFun("1000-01-01"))
-                  checkFilterPredicate(
-                    dateAttr =!= "1000-01-01".date,
-                    classOf[NotEq[_]],
-                    Seq("2018-03-19", "2018-03-20", "2018-03-21").map(i => Row.apply(resultFun(i))))
-
-                  checkFilterPredicate(
-                    dateAttr < "2018-03-19".date,
-                    classOf[Lt[_]],
-                    resultFun("1000-01-01"))
-                  checkFilterPredicate(
-                    dateAttr > "2018-03-20".date,
-                    classOf[Gt[_]],
-                    resultFun("2018-03-21"))
-                  checkFilterPredicate(
-                    dateAttr <= "1000-01-01".date,
-                    classOf[LtEq[_]],
-                    resultFun("1000-01-01"))
-                  checkFilterPredicate(
-                    dateAttr >= "2018-03-21".date,
-                    classOf[GtEq[_]],
-                    resultFun("2018-03-21"))
-
-                  checkFilterPredicate(
-                    Literal("1000-01-01".date) === dateAttr,
-                    classOf[Eq[_]],
-                    resultFun("1000-01-01"))
-                  checkFilterPredicate(
-                    Literal("1000-01-01".date) <=> dateAttr,
-                    classOf[Eq[_]],
-                    resultFun("1000-01-01"))
-                  checkFilterPredicate(
-                    Literal("2018-03-19".date) > dateAttr,
-                    classOf[Lt[_]],
-                    resultFun("1000-01-01"))
-                  checkFilterPredicate(
-                    Literal("2018-03-20".date) < dateAttr,
-                    classOf[Gt[_]],
-                    resultFun("2018-03-21"))
-                  checkFilterPredicate(
-                    Literal("1000-01-01".date) >= dateAttr,
-                    classOf[LtEq[_]],
-                    resultFun("1000-01-01"))
-                  checkFilterPredicate(
-                    Literal("2018-03-21".date) <= dateAttr,
-                    classOf[GtEq[_]],
-                    resultFun("2018-03-21"))
-
-                  checkFilterPredicate(
-                    !(dateAttr < "2018-03-21".date),
-                    classOf[GtEq[_]],
-                    resultFun("2018-03-21"))
-                  checkFilterPredicate(
-                    dateAttr < "2018-03-19".date || dateAttr > "2018-03-20".date,
-                    classOf[Operators.Or],
-                    Seq(Row(resultFun("1000-01-01")), Row(resultFun("2018-03-21"))))
-
-                  Seq(3, 20).foreach {
-                    threshold =>
-                      withSQLConf(
-                        SQLConf.PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD.key -> s"$threshold") {
-                        checkFilterPredicate(
-                          In(
-                            dateAttr,
-                            Array(
-                              "2018-03-19".date,
-                              "2018-03-20".date,
-                              "2018-03-21".date,
-                              "2018-03-22".date).map(Literal.apply)),
-                          if (threshold == 3) classOf[Operators.In[_]] else classOf[Operators.Or],
-                          Seq(
-                            Row(resultFun("2018-03-19")),
-                            Row(resultFun("2018-03-20")),
-                            Row(resultFun("2018-03-21")))
-                        )
-                      }
-                  }
-              }
-            }
-        }
-    }
-  }
 }