This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new cc5c0c38df8e [SPARK-57848][SQL] Support the TIME data type in
approx_top_k
cc5c0c38df8e is described below
commit cc5c0c38df8e8038621c7b56ebd61449fa084f91
Author: Rakesh Raushan <[email protected]>
AuthorDate: Mon Jul 20 22:21:26 2026 +0800
[SPARK-57848][SQL] Support the TIME data type in approx_top_k
### What changes were proposed in this pull request?
Add support for TimeType to approx_top_k. After this change approx_top_k
accepts TimeType and returns correct top-k time values and columns.
### Why are the changes needed?
The TIME data type was added by SPIP
[SPARK-51162](https://issues.apache.org/jira/browse/SPARK-51162) and shipped in
Spark 4.1.0, but the approx_top_k aggregate functions did not accept it.
Calling approx_top_k on a TIME column failed with a type-check error, even
though the operation is well-defined. This PR closes that gap.
### Does this PR introduce _any_ user-facing change?
Yes, Now users can use approx_top_k over TimeType columns as well.
### How was this patch tested?
UTs added.
### Was this patch authored or co-authored using generative AI tooling?
Used claude code for UTs.
Closes #57013 from iRakson/SPARK-57848.
Authored-by: Rakesh Raushan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../aggregate/ApproxTopKAggregates.scala | 11 +-
.../org/apache/spark/sql/ApproxTopKSuite.scala | 159 ++++++++++++++++++++-
2 files changed, 164 insertions(+), 6 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproxTopKAggregates.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproxTopKAggregates.scala
index b09f242ccbac..2690a52b2d31 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproxTopKAggregates.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproxTopKAggregates.scala
@@ -234,7 +234,8 @@ object ApproxTopK {
itemType match {
case _: BooleanType | _: ByteType | _: ShortType | _: IntegerType |
_: LongType | _: FloatType | _: DoubleType | _: DateType |
- _: TimestampType | _: TimestampNTZType | _: StringType | _:
DecimalType => true
+ _: TimestampType | _: TimestampNTZType | _: StringType |
+ _: DecimalType | _: TimeType => true
// BinaryType is not supported now, as ItemsSketch seems cannot count
the frequency correctly
case _ => false
}
@@ -255,7 +256,7 @@ object ApproxTopK {
new ItemsSketch[Boolean](maxMapSize).asInstanceOf[ItemsSketch[Any]]
case _: ByteType | _: ShortType | _: IntegerType | _: FloatType | _:
DateType =>
new ItemsSketch[Number](maxMapSize).asInstanceOf[ItemsSketch[Any]]
- case _: LongType | _: TimestampType | _: TimestampNTZType =>
+ case _: LongType | _: TimestampType | _: TimestampNTZType | _: TimeType
=>
new ItemsSketch[Long](maxMapSize).asInstanceOf[ItemsSketch[Any]]
case _: DoubleType =>
new ItemsSketch[Double](maxMapSize).asInstanceOf[ItemsSketch[Any]]
@@ -275,7 +276,7 @@ object ApproxTopK {
case _: BooleanType => new
ArrayOfBooleansSerDe().asInstanceOf[ArrayOfItemsSerDe[Any]]
case _: ByteType | _: ShortType | _: IntegerType | _: FloatType | _:
DateType =>
new ArrayOfNumbersSerDe().asInstanceOf[ArrayOfItemsSerDe[Any]]
- case _: LongType | _: TimestampType | _: TimestampNTZType =>
+ case _: LongType | _: TimestampType | _: TimestampNTZType | _: TimeType
=>
new ArrayOfLongsSerDe().asInstanceOf[ArrayOfItemsSerDe[Any]]
case _: DoubleType =>
new ArrayOfDoublesSerDe().asInstanceOf[ArrayOfItemsSerDe[Any]]
@@ -378,6 +379,8 @@ class ApproxTopKAggregateBuffer[T](val sketch:
ItemsSketch[T], private var nullC
sketch.asInstanceOf[ItemsSketch[Long]].update(v.asInstanceOf[Long])
case _: TimestampNTZType =>
sketch.asInstanceOf[ItemsSketch[Long]].update(v.asInstanceOf[Long])
+ case _: TimeType =>
+ sketch.asInstanceOf[ItemsSketch[Long]].update(v.asInstanceOf[Long])
case st: StringType =>
val orig = v.asInstanceOf[UTF8String]
if (UnsafeRowUtils.isBinaryStable(st)) {
@@ -453,7 +456,7 @@ class ApproxTopKAggregateBuffer[T](val sketch:
ItemsSketch[T], private var nullC
val item: Any = itemDataType match {
case _: BooleanType | _: ByteType | _: ShortType | _: IntegerType |
_: LongType | _: FloatType | _: DoubleType | _: DecimalType |
- _: DateType | _: TimestampType | _: TimestampNTZType =>
+ _: DateType | _: TimestampType | _: TimestampNTZType | _:
TimeType =>
curFrequentItem.getItem
case _: StringType =>
curFrequentItem.getItem match {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/ApproxTopKSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/ApproxTopKSuite.scala
index 99efbd29c753..98aba80682e6 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/ApproxTopKSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/ApproxTopKSuite.scala
@@ -18,13 +18,13 @@
package org.apache.spark.sql
import java.sql.{Date, Timestamp}
-import java.time.LocalDateTime
+import java.time.{LocalDateTime, LocalTime}
import org.apache.spark.{SparkArithmeticException, SparkRuntimeException}
import org.apache.spark.sql.catalyst.ExtendedAnalysisException
import org.apache.spark.sql.errors.DataTypeErrors.toSQLType
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{BooleanType, ByteType, DataType, DateType,
DecimalType, DoubleType, FloatType, IntegerType, LongType, ShortType,
StringType, TimestampNTZType, TimestampType}
+import org.apache.spark.sql.types.{BooleanType, ByteType, DataType, DateType,
DecimalType, DoubleType, FloatType, IntegerType, LongType, ShortType,
StringType, TimestampNTZType, TimestampType, TimeType}
class ApproxTopKSuite extends SharedSparkSession {
@@ -63,6 +63,11 @@ class ApproxTopKSuite extends SharedSparkSession {
"TIMESTAMP_NTZ'2025-01-01 00:00:00', TIMESTAMP_NTZ'2025-01-02 00:00:00'",
Seq(Row(LocalDateTime.of(2025, 1, 1, 0, 0), 3),
Row(LocalDateTime.of(2025, 1, 2, 0, 0), 1))), // Timestamp_ntz
+ ("TIME'00:00:00', TIME'00:00:00', TIME'06:00:00', TIME'06:00:00',
TIME'06:00:00', " +
+ "TIME'08:30:00.123456', TIME'10:00:00', TIME'10:00:00', TIME'11:00:00'",
+ Seq(Row(LocalTime.of(6, 0, 0), 3), Row(LocalTime.of(0, 0, 0), 2),
+ Row(LocalTime.of(10, 0, 0), 2), Row(LocalTime.of(11, 0, 0), 1),
+ Row(LocalTime.of(8, 30, 0, 123456000), 1))), // Time
("CAST(0.0 AS DECIMAL(4, 1)), CAST(0.0 AS DECIMAL(4, 1)), " +
"CAST(0.0 AS DECIMAL(4, 1)), CAST(1.0 AS DECIMAL(4, 1)), " +
"CAST(1.0 AS DECIMAL(4, 1)), CAST(2.0 AS DECIMAL(4, 1))",
@@ -554,6 +559,22 @@ class ApproxTopKSuite extends SharedSparkSession {
)
)
+ // TIME has no common type with DATE/TIMESTAMP/TIMESTAMP_NTZ (unlike those
three, which all
+ // mutually widen and so only fail at approx_top_k_combine runtime, not at
UNION analysis), so
+ // it cannot join mixedDateTimeTypes: "among different ... types - fail at
combine" below
+ // asserts every pair in that Seq unions successfully and only fails at the
runtime combine.
+ // The two TIME entries below (different precisions) do widen against each
other at UNION
+ // (findWiderDateTimeType(TimeType(p1), TimeType(p2)) = TimeType(max(p1,
p2))), mirroring
+ // mixedDateTimeTypes's own internal widening -- see
+ // "SPARK-57848: among different time precisions - fail at combine" below.
+ val mixedTimeTypes: Seq[(DataType, String, Seq[String])] = Seq(
+ (TimeType(), "TIME(6)",
+ Seq("TIME'12:00:00'", "TIME'12:00:00'", "TIME'13:00:00'")),
+ (TimeType(3), "TIME(3)",
+ Seq("CAST('12:00:00.123' AS TIME(3))", "CAST('12:00:00.123' AS TIME(3))",
+ "CAST('13:00:00.123' AS TIME(3))"))
+ )
+
// positive tests for approx_top_k_combine on every types
gridTest("SPARK-52798: same type, same size, specified combine size -
success")(itemsWithTopK) {
case (input, expected) =>
@@ -664,6 +685,28 @@ class ApproxTopKSuite extends SharedSparkSession {
checkMixedTypeError(mixedDateTimeTypes)
}
+ // TIME counterpart to the test above: TIME(3) and TIME(6) widen to TIME(6)
at UNION via
+ // findWiderDateTimeType, so the UNION succeeds, but the two sketches still
carry their
+ // original, different itemDataType (time(3) vs time(6)), so combine fails
at runtime.
+ test("SPARK-57848: among different time precisions - fail at combine") {
+ for (i <- 0 until mixedTimeTypes.size - 1) {
+ for (j <- i + 1 until mixedTimeTypes.size) {
+ val (type1, _, seq1) = mixedTimeTypes(i)
+ val (type2, _, seq2) = mixedTimeTypes(j)
+ withView("accumulation1", "accumulation2", "unioned") {
+ setupMixedTypeAccumulation(seq1, seq2)
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ sql("SELECT approx_top_k_combine(acc, 30) as com FROM
unioned;").collect()
+ },
+ condition = "APPROX_TOP_K_SKETCH_TYPE_NOT_MATCH",
+ parameters = Map("type1" -> toSQLType(type1), "type2" ->
toSQLType(type2))
+ )
+ }
+ }
+ }
+ }
+
// enumerate all combinations of number and datetime types
gridTest("SPARK-52798: number vs datetime - fail on UNION")(
for {
@@ -776,6 +819,118 @@ class ApproxTopKSuite extends SharedSparkSession {
)
}
+ // TIME has no common type with DATE/TIMESTAMP/TIMESTAMP_NTZ, so mixing it
with any of them
+ // fails at UNION analysis, unlike combining two of those three which fails
at combine runtime.
+ gridTest("SPARK-57848: time vs datetime - fail on UNION")(
+ for {
+ (type1, typeName1, seq1) <- mixedTimeTypes
+ (type2, typeName2, seq2) <- mixedDateTimeTypes
+ } yield ((type1, typeName1, seq1), (type2, typeName2, seq2))) {
+ case ((_, type1, seq1), (_, type2, seq2)) =>
+ checkError(
+ exception = intercept[ExtendedAnalysisException] {
+ withView("accumulation1", "accumulation2", "unioned") {
+ setupMixedTypeAccumulation(seq1, seq2)
+ }
+ },
+ condition = "INCOMPATIBLE_COLUMN_TYPE",
+ parameters = Map(
+ "tableOrdinalNumber" -> "second",
+ "columnOrdinalNumber" -> "first",
+ "dataType2" -> ("\"STRUCT<sketch: BINARY NOT NULL, maxItemsTracked:
INT NOT NULL, " +
+ "itemDataType: " + type1 + ", itemDataTypeDDL: STRING NOT
NULL>\""),
+ "operator" -> "UNION",
+ "hint" -> "",
+ "dataType1" -> ("\"STRUCT<sketch: BINARY NOT NULL, maxItemsTracked:
INT NOT NULL, " +
+ "itemDataType: " + type2 + ", itemDataTypeDDL: STRING NOT NULL>\"")
+ ),
+ queryContext = Array(
+ ExpectedContext(
+ "SELECT acc from accumulation1 UNION ALL SELECT acc FROM
accumulation2", 0, 68))
+ )
+ }
+
+ // TIME has no common type with any number type either, so it fails at
UNION, same as
+ // number vs datetime above.
+ gridTest("SPARK-57848: time vs number - fail on UNION")(
+ for {
+ (type1, typeName1, seq1) <- mixedTimeTypes
+ (type2, typeName2, seq2) <- mixedNumberTypes
+ } yield ((type1, typeName1, seq1), (type2, typeName2, seq2))) {
+ case ((_, type1, seq1), (_, type2, seq2)) =>
+ checkError(
+ exception = intercept[ExtendedAnalysisException] {
+ withView("accumulation1", "accumulation2", "unioned") {
+ setupMixedTypeAccumulation(seq1, seq2)
+ }
+ },
+ condition = "INCOMPATIBLE_COLUMN_TYPE",
+ parameters = Map(
+ "tableOrdinalNumber" -> "second",
+ "columnOrdinalNumber" -> "first",
+ "dataType2" -> ("\"STRUCT<sketch: BINARY NOT NULL, maxItemsTracked:
INT NOT NULL, " +
+ "itemDataType: " + type1 + ", itemDataTypeDDL: STRING NOT
NULL>\""),
+ "operator" -> "UNION",
+ "hint" -> "",
+ "dataType1" -> ("\"STRUCT<sketch: BINARY NOT NULL, maxItemsTracked:
INT NOT NULL, " +
+ "itemDataType: " + type2 + ", itemDataTypeDDL: STRING NOT NULL>\"")
+ ),
+ queryContext = Array(
+ ExpectedContext(
+ "SELECT acc from accumulation1 UNION ALL SELECT acc FROM
accumulation2", 0, 68))
+ )
+ }
+
+ gridTest("SPARK-57848: time vs string - fail at combine")(mixedTimeTypes) {
+ case (type1, _, seq1) =>
+ withView("accumulation1", "accumulation2", "unioned") {
+ setupMixedTypeAccumulation(
+ seq1, Seq("'a'", "'b'", "'c'", "'c'", "'c'", "'c'", "'d'", "'d'"))
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ sql("SELECT approx_top_k_combine(acc, 30) as com FROM
unioned;").collect()
+ },
+ condition = "APPROX_TOP_K_SKETCH_TYPE_NOT_MATCH",
+ parameters = Map("type1" -> toSQLType(type1), "type2" ->
toSQLType(StringType))
+ )
+ }
+ }
+
+ gridTest("SPARK-57848: time vs boolean - fail at UNION")(mixedTimeTypes) {
+ case (_, type1, seq1) =>
+ val seq2 = Seq("(true)", "(true)", "(false)", "(false)")
+ checkError(
+ exception = intercept[ExtendedAnalysisException] {
+ withView("accumulation1", "accumulation2", "unioned") {
+ setupMixedTypeAccumulation(seq1, seq2)
+ }
+ },
+ condition = "INCOMPATIBLE_COLUMN_TYPE",
+ parameters = Map(
+ "tableOrdinalNumber" -> "second",
+ "columnOrdinalNumber" -> "first",
+ "dataType2" -> ("\"STRUCT<sketch: BINARY NOT NULL, maxItemsTracked:
INT NOT NULL, " +
+ "itemDataType: " + type1 + ", itemDataTypeDDL: STRING NOT
NULL>\""),
+ "operator" -> "UNION",
+ "hint" -> "",
+ "dataType1" -> ("\"STRUCT<sketch: BINARY NOT NULL, maxItemsTracked:
INT NOT NULL, " +
+ "itemDataType: BOOLEAN, itemDataTypeDDL: STRING NOT NULL>\"")
+ ),
+ queryContext = Array(
+ ExpectedContext(
+ "SELECT acc from accumulation1 UNION ALL SELECT acc FROM
accumulation2", 0, 68))
+ )
+ }
+
+ test("SPARK-57848: approx_top_k over time column") {
+ val res = sql(
+ "SELECT approx_top_k(expr, 10, 100) FROM VALUES (TIME '06:00:00'), (TIME
'06:00:00'), " +
+ "(TIME '08:00:00'), (TIME '08:00:00'), (TIME '08:00:00'), (TIME
'10:00:00') AS tab(expr);"
+ )
+ checkAnswer(res, Row(Seq(Row(LocalTime.of(8, 0, 0), 3),
Row(LocalTime.of(6, 0, 0), 2),
+ Row(LocalTime.of(10, 0, 0), 1))))
+ }
+
test("SPARK-52798: string vs boolean - fail at combine") {
withSQLConf("spark.sql.ansi.enabled" -> "true") {
val seq1 = Seq("'a'", "'b'", "'c'", "'c'", "'c'", "'c'", "'d'", "'d'")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
