This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 690c836511 [VL] Update Parquet unit tests (#12776)
690c836511 is described below
commit 690c83651183f0c5836a0a9b9e363cb57231d313
Author: Rui Mo <[email protected]>
AuthorDate: Thu Sep 17 17:07:39 2026 +0100
[VL] Update Parquet unit tests (#12776)
---
.../gluten/utils/velox/VeloxTestSettings.scala | 27 +--
.../parquet/GlutenParquetColumnIndexSuite.scala | 48 +----
.../parquet/GlutenParquetFilterSuite.scala | 198 +--------------------
.../gluten/utils/velox/VeloxTestSettings.scala | 30 +---
.../parquet/GlutenParquetColumnIndexSuite.scala | 48 +----
.../parquet/GlutenParquetFilterSuite.scala | 198 +--------------------
.../gluten/utils/velox/VeloxTestSettings.scala | 19 +-
.../parquet/GlutenParquetFilterSuite.scala | 198 +--------------------
.../gluten/utils/velox/VeloxTestSettings.scala | 19 +-
.../parquet/GlutenParquetFilterSuite.scala | 198 +--------------------
10 files changed, 28 insertions(+), 955 deletions(-)
diff --git
a/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 2346d5f7d3..e5c4ba13d1 100644
---
a/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/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,16 +440,15 @@ 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")
@@ -468,18 +456,19 @@ class VeloxTestSettings extends BackendTestSettings {
// 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]
diff --git
a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetColumnIndexSuite.scala
b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetColumnIndexSuite.scala
index 60e1ca04a2..4bb8e96455 100644
---
a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetColumnIndexSuite.scala
+++
b/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 {}
diff --git
a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
index a4f830e187..0e1a59084e 100644
---
a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
+++
b/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")))
- )
- }
- }
- }
- }
- }
- }
- }
}
diff --git
a/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 72ef076d04..a16b97c357 100644
---
a/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -342,27 +342,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.
@@ -378,9 +369,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.
@@ -394,6 +382,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.
@@ -402,16 +391,15 @@ 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")
@@ -419,18 +407,19 @@ class VeloxTestSettings extends BackendTestSettings {
// 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]
@@ -476,17 +465,12 @@ class VeloxTestSettings extends BackendTestSettings {
enableSuite[GlutenFileMetadataStructSuite]
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]
.exclude("nested column: Max(top level column) not push down")
.exclude("nested column: Count(nested sub-field) not push down")
enableSuite[GlutenParquetCodecSuite]
- // codec not supported in native
- .exclude("write and read - file source parquet - codec: lz4_raw")
- .exclude("write and read - file source parquet - codec: lz4raw")
enableSuite[GlutenOrcCodecSuite]
enableSuite[GlutenFileSourceStrategySuite]
// Plan comparison.
diff --git
a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetColumnIndexSuite.scala
b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetColumnIndexSuite.scala
index 60e1ca04a2..4bb8e96455 100644
---
a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetColumnIndexSuite.scala
+++
b/gluten-ut/spark35/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 {}
diff --git
a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
index 3c52ec82e9..0e1a59084e 100644
---
a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
+++
b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
@@ -26,26 +26,17 @@ import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.parseColumnPath
import org.apache.spark.sql.execution.datasources.{DataSourceStrategy,
HadoopFsRelation, LogicalRelation, PushableColumnAndNestedColumn}
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.LegacyBehaviorPolicy.{CORRECTED, LEGACY}
import org.apache.spark.sql.internal.SQLConf
-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")))
- )
- }
- }
- }
- }
- }
- }
- }
}
diff --git
a/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 0ae8965619..5693a36a62 100644
---
a/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -542,9 +542,6 @@ class VeloxTestSettings extends BackendTestSettings {
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.
@@ -560,9 +557,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.
@@ -576,6 +570,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.
@@ -584,18 +579,15 @@ 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")
// TODO: fix in Spark-4.0
.exclude("explode nested lists crossing a rowgroup boundary")
enableSuite[GlutenParquetV1PartitionDiscoverySuite]
enableSuite[GlutenParquetV2PartitionDiscoverySuite]
enableSuite[GlutenParquetProtobufCompatibilitySuite]
enableSuite[GlutenParquetV1QuerySuite]
- .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")
@@ -603,19 +595,14 @@ class VeloxTestSettings extends BackendTestSettings {
// 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.
- .exclude("SPARK-36182: can't read TimestampLTZ as TimestampNTZ")
enableSuite[GlutenParquetV2QuerySuite]
- .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.
- .exclude("SPARK-36182: can't read TimestampLTZ as TimestampNTZ")
enableSuite[GlutenParquetV1SchemaPruningSuite]
enableSuite[GlutenParquetV2SchemaPruningSuite]
enableSuite[GlutenParquetRebaseDatetimeV1Suite]
diff --git
a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
index b6d6cb8b50..f31f0df95c 100644
---
a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
+++
b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
@@ -28,26 +28,17 @@ import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.parseColumnPath
import org.apache.spark.sql.execution.datasources.{DataSourceStrategy,
HadoopFsRelation, LogicalRelation, PushableColumnAndNestedColumn}
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.LegacyBehaviorPolicy.{CORRECTED, LEGACY}
import org.apache.spark.sql.internal.SQLConf
-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,
@@ -68,44 +59,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._
@@ -429,153 +382,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")))
- )
- }
- }
- }
- }
- }
- }
- }
}
diff --git
a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 0890acd58f..0abaabf465 100644
---
a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -516,9 +516,6 @@ class VeloxTestSettings extends BackendTestSettings {
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.
@@ -534,9 +531,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.
@@ -550,6 +544,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.
@@ -558,8 +553,6 @@ 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")
// TODO: fix in Spark-4.0
.exclude("explode nested lists crossing a rowgroup boundary")
// TODO: fix on Spark-4.1
@@ -570,10 +563,9 @@ class VeloxTestSettings extends BackendTestSettings {
enableSuite[GlutenParquetV2PartitionDiscoverySuite]
enableSuite[GlutenParquetProtobufCompatibilitySuite]
enableSuite[GlutenParquetV1QuerySuite]
- .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")
@@ -581,19 +573,14 @@ class VeloxTestSettings extends BackendTestSettings {
// 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.
- .exclude("SPARK-36182: can't read TimestampLTZ as TimestampNTZ")
enableSuite[GlutenParquetV2QuerySuite]
- .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.
- .exclude("SPARK-36182: can't read TimestampLTZ as TimestampNTZ")
enableSuite[GlutenParquetV1SchemaPruningSuite]
enableSuite[GlutenParquetV2SchemaPruningSuite]
enableSuite[GlutenParquetRebaseDatetimeV1Suite]
diff --git
a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
index 5cbba475a6..4790b9c448 100644
---
a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
+++
b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/GlutenParquetFilterSuite.scala
@@ -28,26 +28,17 @@ import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits.parseColumnPath
import org.apache.spark.sql.execution.datasources.{DataSourceStrategy,
HadoopFsRelation, LogicalRelation, PushableColumnAndNestedColumn}
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.LegacyBehaviorPolicy.{CORRECTED, LEGACY}
import org.apache.spark.sql.internal.SQLConf
-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,
@@ -68,44 +59,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._
@@ -429,153 +382,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")))
- )
- }
- }
- }
- }
- }
- }
- }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
