Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 20 additions & 5 deletions spark/src/test/scala/org/apache/comet/CometFuzzTestSuite.scala
Original file line number Diff line number Diff line change
Expand Up @@ -274,15 +274,14 @@ class CometFuzzTestSuite extends CometFuzzTestBase {
SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> outputTimestampType.toString,
SQLConf.SESSION_LOCAL_TIMEZONE.key -> defaultTimezone) {

// TODO test with MapType
// https://github.com/apache/datafusion-comet/issues/2945
val schema = StructType(
Seq(
StructField("c0", DataTypes.DateType),
StructField("c1", DataTypes.createArrayType(DataTypes.DateType)),
StructField(
"c2",
DataTypes.createStructType(Array(StructField("c3", DataTypes.DateType))))))
DataTypes.createStructType(Array(StructField("c3", DataTypes.DateType)))),
StructField("c4", MapType(DateType, DateType))))

ParquetGenerator.makeParquetFile(
random,
Expand All @@ -309,10 +308,26 @@ class CometFuzzTestSuite extends CometFuzzTestBase {
val columns =
df.schema.fields
.filter(f => DataTypeSupport.hasTemporalType(f.dataType))
.map(_.name)

for (col <- columns) {
checkSparkAnswer(s"SELECT $col FROM t1 ORDER BY $col")
// Maps are not orderable; the answer helper compares unordered results too.
val (_, cometPlan) = col.dataType match {
case _: MapType =>
val query = s"SELECT ${col.name} FROM t1"
if (int96TimestampConversion) {
checkSparkAnswer(query)
} else {
checkSparkAnswerAndOperator(query)
}
case _ =>
checkSparkAnswer(s"SELECT ${col.name} FROM t1 ORDER BY ${col.name}")
}
// INT96 timestamp conversion intentionally disables the Comet extension.
if (!int96TimestampConversion) {
assert(
collectNativeScans(cometPlan).size == 1,
s"Expected a Comet scan for ${col.name} in $tz:\n$cometPlan")
}
}
}
}
Expand Down
Loading