This is an automated email from the ASF dual-hosted git repository.
MaxGekk pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 5e2c8eff2e00 [SPARK-57557][SQL] Support the TIME data type in quantile
aggregates
5e2c8eff2e00 is described below
commit 5e2c8eff2e009270eee5b556ec45012f7721e830
Author: Vaibhav Garg <[email protected]>
AuthorDate: Wed Jul 1 11:35:38 2026 +0200
[SPARK-57557][SQL] Support the TIME data type in quantile aggregates
### What changes were proposed in this pull request?
This PR adds support for the TIME data type (`TimeType`) as an input to the
core quantile aggregate expressions. SPARK-57557 is a sub-task of the umbrella
SPARK-57550 (extend TIME type support across Spark).
The following expressions now accept TIME columns and return TIME:
- `percentile`, `median`, `percentile_cont`, `percentile_disc` (exact
quantiles, `percentiles.scala`)
- `percentile_approx` / `approx_percentile` (`ApproximatePercentile`)
- `histogram_numeric` (`HistogramNumeric`)
TIME's internal representation is a `Long` (nanoseconds of day), so it
flows through the existing value↔double conversion paths exactly like the other
ordered, `Long`-backed types. The change is mechanical:
- `TimeType` (via `AnyTimeType`) added to each expression's `inputTypes` /
`TypeCollection`
- A `TimeType` branch added to the update (`Long` → `Double`) and eval
(`Double` → `Long`) conversions
- The return type uses `child.dataType`, preserving the input precision
(e.g. `median` over a `TIME(3)` column returns `TIME(3)`)
Design notes (called out so they can be challenged in review):
- **Scope:** This PR covers only the core quantile aggregates above. The
Apache DataSketches aggregates (HLL, Theta, and the KLL quantile sketches) are
intentionally deferred and can be added in a follow-up.
- **Interpolation:** For the exact `percentile`/`median` (continuous
distribution), interpolated results mirror existing numeric/interval behavior —
e.g. the median of two TIME values returns their midpoint, which may be a value
not present in the data; sub-nanosecond fractions are truncated.
- **Template mirrored:** For the exact `percentile`/`median` and
`histogram_numeric`, TIMESTAMP is not currently supported either, so the
implementation mirrors DAY-TIME INTERVAL (also `Long`-internal). For
`percentile_approx` and `histogram_numeric`, TIMESTAMP was already supported
and TIME mirrors it directly.
### Why are the changes needed?
The TIME data type was added by SPIP SPARK-51162 and shipped in Spark
4.1.0, but the quantile aggregate functions did not accept it. Calling
`percentile`, `median`, `percentile_approx`, or `histogram_numeric` on a TIME
column failed with a type-check error, even though the operation is
well-defined: TIME is internally a `Long` (nanoseconds of day) and quantile
computation over an ordered numeric value applies directly. This PR closes that
gap.
### Does this PR introduce any user-facing change?
Yes. `percentile`, `median`, `percentile_cont`, `percentile_disc`,
`percentile_approx` / `approx_percentile`, and `histogram_numeric` now accept
TIME columns and return TIME values. Previously these expressions rejected TIME
input with a type-check error.
### How was this patch tested?
All scoped suites were run locally and passed:
- `catalyst/testOnly
org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentileSuite
org.apache.spark.sql.catalyst.expressions.aggregate.PercentileSuite
org.apache.spark.sql.catalyst.expressions.aggregate.HistogramNumericSuite` —
all tests passed (each suite has a new `test("SPARK-57557: ...")` covering TIME
input, return type, precision preservation, and median interpolation)
- `sql/testOnly org.apache.spark.sql.ApproximatePercentileQuerySuite` — 20
tests passed (new `SPARK-57557: percentile_approx supports TIME type` test,
scalar and array-of-percentiles paths)
- `SPARK_GENERATE_GOLDEN_FILES=1 sql/testOnly
org.apache.spark.sql.SQLQueryTestSuite -- -z percentiles.sql` — golden files
regenerated and manually reviewed; a new TIME section was added to
`percentiles.sql`
- One pre-existing `PercentileSuite` assertion (the accepted-types error
message) was updated to include TIME
The full lint/scalastyle and complete test matrix run on GitHub Actions.
### Was this patch authored or co-authored using generative AI tooling?
Yes. Generated using Kiro (Claude Opus 4.8)
Closes #56889 from vboo123/SPARK-57557.
Lead-authored-by: Vaibhav Garg <[email protected]>
Co-authored-by: Vaibhav Garg <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
(cherry picked from commit 762c578cee28a9dceac98c90bee6bb78da37fc85)
Signed-off-by: Max Gekk <[email protected]>
---
.../aggregate/ApproximatePercentile.scala | 15 ++++----
.../expressions/aggregate/HistogramNumeric.scala | 9 ++---
.../expressions/aggregate/percentiles.scala | 16 +++++----
.../aggregate/ApproximatePercentileSuite.scala | 19 +++++++++-
.../aggregate/HistogramNumericSuite.scala | 18 ++++++++++
.../expressions/aggregate/PercentileSuite.scala | 23 +++++++++---
.../sql-tests/analyzer-results/percentiles.sql.out | 42 ++++++++++++++++++++++
.../resources/sql-tests/inputs/percentiles.sql | 14 ++++++++
.../sql-tests/results/percentiles.sql.out | 31 ++++++++++++++++
.../sql/ApproximatePercentileQuerySuite.scala | 20 ++++++++++-
10 files changed, 183 insertions(+), 24 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala
index 8ad062ab0e2f..19e30c53b436 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala
@@ -55,8 +55,8 @@ import org.apache.spark.util.ArrayImplicits._
// scalastyle:off line.size.limit
@ExpressionDescription(
usage = """
- _FUNC_(col, percentage [, accuracy]) - Returns the approximate
`percentile` of the numeric or
- ansi interval column `col` which is the smallest value in the ordered
`col` values (sorted
+ _FUNC_(col, percentage [, accuracy]) - Returns the approximate
`percentile` of the numeric,
+ ansi interval or TIME column `col` which is the smallest value in the
ordered `col` values (sorted
from least to greatest) such that no more than `percentage` of `col`
values is less than
the value or equal to that value. The value of percentage must be
between 0.0 and 1.0.
The `accuracy` parameter (default: 10000) is a positive numeric literal
which controls
@@ -102,10 +102,10 @@ case class ApproximatePercentile(
private lazy val accuracy: Long = accuracyNum.longValue
override def inputTypes: Seq[AbstractDataType] = {
- // Support NumericType, DateType, TimestampType and TimestampNTZType since
their internal types
- // are all numeric, and can be easily cast to double for processing.
+ // Support NumericType, DateType, TimestampType, TimestampNTZType and
TimeType since their
+ // internal types are all numeric, and can be easily cast to double for
processing.
Seq(TypeCollection(NumericType, DateType, TimestampType, TimestampNTZType,
- YearMonthIntervalType, DayTimeIntervalType),
+ YearMonthIntervalType, DayTimeIntervalType, AnyTimeType),
TypeCollection(DoubleType, ArrayType(DoubleType, containsNull = false)),
IntegralType)
}
@@ -183,7 +183,7 @@ case class ApproximatePercentile(
// Convert the value to a double value
val doubleValue = child.dataType match {
case DateType | _: YearMonthIntervalType =>
value.asInstanceOf[Int].toDouble
- case TimestampType | TimestampNTZType | _: DayTimeIntervalType =>
+ case TimestampType | TimestampNTZType | _: DayTimeIntervalType | _:
TimeType =>
value.asInstanceOf[Long].toDouble
case n: NumericType =>
PhysicalNumericType.numeric(n)
@@ -205,7 +205,8 @@ case class ApproximatePercentile(
val doubleResult = buffer.getPercentiles(percentages)
val result = child.dataType match {
case DateType | _: YearMonthIntervalType => doubleResult.map(_.toInt)
- case TimestampType | TimestampNTZType | _: DayTimeIntervalType =>
doubleResult.map(_.toLong)
+ case TimestampType | TimestampNTZType | _: DayTimeIntervalType | _:
TimeType =>
+ doubleResult.map(_.toLong)
case ByteType => doubleResult.map(_.toByte)
case ShortType => doubleResult.map(_.toShort)
case IntegerType => doubleResult.map(_.toInt)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/HistogramNumeric.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/HistogramNumeric.scala
index 142f4a4eae4c..dadf7ac53c59 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/HistogramNumeric.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/HistogramNumeric.scala
@@ -78,11 +78,11 @@ case class HistogramNumeric(
private lazy val propagateInputType: Boolean =
SQLConf.get.histogramNumericPropagateInputType
override def inputTypes: Seq[AbstractDataType] = {
- // Support NumericType, DateType, TimestampType and TimestampNTZType,
YearMonthIntervalType,
- // DayTimeIntervalType since their internal types are all numeric,
+ // Support NumericType, DateType, TimestampType, TimestampNTZType,
TimeType,
+ // YearMonthIntervalType, DayTimeIntervalType since their internal types
are all numeric,
// and can be easily cast to double for processing.
Seq(TypeCollection(NumericType, DateType, TimestampType, TimestampNTZType,
- YearMonthIntervalType, DayTimeIntervalType), IntegerType)
+ YearMonthIntervalType, DayTimeIntervalType, AnyTimeType), IntegerType)
}
override def checkInputDataTypes(): TypeCheckResult = {
@@ -163,7 +163,8 @@ case class HistogramNumeric(
coord.x.toInt
case FloatType => coord.x.toFloat
case ShortType => coord.x.toShort
- case _: DayTimeIntervalType | LongType | TimestampType |
TimestampNTZType =>
+ case _: DayTimeIntervalType | LongType | TimestampType |
TimestampNTZType
+ | _: TimeType =>
coord.x.toLong
case d: DecimalType =>
val bigDecimal = BigDecimal
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/percentiles.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/percentiles.scala
index 6dfa1b499df2..ac351db9e471 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/percentiles.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/percentiles.scala
@@ -65,6 +65,7 @@ abstract class PercentileBase
val resultType = child.dataType match {
case _: YearMonthIntervalType => YearMonthIntervalType()
case _: DayTimeIntervalType => DayTimeIntervalType()
+ case t: TimeType => t
case _ => DoubleType
}
if (returnPercentileArray) ArrayType(resultType, false) else resultType
@@ -75,7 +76,7 @@ abstract class PercentileBase
case _: ArrayType => ArrayType(DoubleType, false)
case _ => DoubleType
}
- Seq(NumericAndAnsiInterval, percentageExpType, IntegralType)
+ Seq(TypeCollection(NumericAndAnsiInterval, AnyTimeType),
percentageExpType, IntegralType)
}
// Check the inputTypes are valid, and the percentageExpression satisfies:
@@ -175,6 +176,7 @@ abstract class PercentileBase
val results = child.dataType match {
case _: YearMonthIntervalType => percentiles.map(_.toInt)
case _: DayTimeIntervalType => percentiles.map(_.toLong)
+ case _: TimeType => percentiles.map(_.toLong)
case _ => percentiles
}
if (percentiles.isEmpty) {
@@ -255,8 +257,8 @@ abstract class PercentileBase
@ExpressionDescription(
usage =
"""
- _FUNC_(col, percentage [, frequency]) - Returns the exact percentile
value of numeric
- or ANSI interval column `col` at the given percentage. The value of
percentage must be
+ _FUNC_(col, percentage [, frequency]) - Returns the exact percentile
value of numeric,
+ ANSI interval or TIME column `col` at the given percentage. The value
of percentage must be
between 0.0 and 1.0. The value of frequency should be positive integral
_FUNC_(col, array(percentage1 [, percentage2]...) [, frequency]) -
Returns the exact
@@ -328,7 +330,7 @@ case class Percentile(
}
@ExpressionDescription(
- usage = "_FUNC_(col) - Returns the median of numeric or ANSI interval column
`col`.",
+ usage = "_FUNC_(col) - Returns the median of numeric, ANSI interval or TIME
column `col`.",
examples = """
Examples:
> SELECT _FUNC_(col) FROM VALUES (0), (10) AS tab(col);
@@ -353,7 +355,7 @@ case class Median(child: Expression)
/**
* Return a percentile value based on a continuous distribution of
- * numeric or ANSI interval column at the given percentage (specified in ORDER
BY clause).
+ * numeric, ANSI interval or TIME column at the given percentage (specified in
ORDER BY clause).
* The value of percentage must be between 0.0 and 1.0.
*/
case class PercentileCont(left: Expression, right: Expression, reverse:
Boolean = false)
@@ -480,7 +482,7 @@ case class PercentileDisc(
// scalastyle:off line.size.limit
@ExpressionDescription(
usage = "_FUNC_(percentage) WITHIN GROUP (ORDER BY col) - Return a
percentile value based on " +
- "a continuous distribution of numeric or ANSI interval column `col` at the
given " +
+ "a continuous distribution of numeric, ANSI interval or TIME column `col`
at the given " +
"`percentage` (specified in ORDER BY clause).",
examples = """
Examples:
@@ -506,7 +508,7 @@ object PercentileContBuilder extends ExpressionBuilder {
// scalastyle:off line.size.limit
@ExpressionDescription(
usage = "_FUNC_(percentage) WITHIN GROUP (ORDER BY col) - Return a
percentile value based on " +
- "a discrete distribution of numeric or ANSI interval column `col` at the
given " +
+ "a discrete distribution of numeric, ANSI interval or TIME column `col` at
the given " +
"`percentage` (specified in ORDER BY clause).",
examples = """
Examples:
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentileSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentileSuite.scala
index 48dd7764f5ad..1710a1b897a6 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentileSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentileSuite.scala
@@ -31,7 +31,7 @@ import
org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile
import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
import org.apache.spark.sql.catalyst.util.{ArrayData, QuantileSummaries}
import org.apache.spark.sql.catalyst.util.QuantileSummaries.Stats
-import org.apache.spark.sql.types.{ArrayType, Decimal, DecimalType,
DoubleType, FloatType, IntegerType, IntegralType, LongType}
+import org.apache.spark.sql.types.{ArrayType, Decimal, DecimalType,
DoubleType, FloatType, IntegerType, IntegralType, LongType, TimeType}
import org.apache.spark.util.SizeEstimator
class ApproximatePercentileSuite extends SparkFunSuite {
@@ -135,6 +135,23 @@ class ApproximatePercentileSuite extends SparkFunSuite {
}
}
+ test("SPARK-57557: ApproximatePercentile supports TIME type") {
+ // The result type mirrors the input TIME type, preserving precision.
+ assert(new ApproximatePercentile(
+ BoundReference(0, TimeType(), nullable = false), Literal(0.5)).dataType
=== TimeType())
+ assert(new ApproximatePercentile(
+ BoundReference(0, TimeType(3), nullable = false), Literal(0.5)).dataType
=== TimeType(3))
+
+ // TIME's internal value is a Long (nanos-of-day); update/merge/eval
round-trips it as a Long.
+ val agg = new ApproximatePercentile(
+ BoundReference(0, TimeType(), nullable = false), Literal(0.5))
+ val buffer = agg.createAggregationBuffer()
+ (1L to 1000L).foreach(v => agg.update(buffer, InternalRow(v)))
+ val result = agg.eval(buffer)
+ assert(result.isInstanceOf[Long])
+ assert(Math.abs(result.asInstanceOf[Long] - 500L) < 5L)
+ }
+
test("class ApproximatePercentile, low level interface, update, merge,
eval...") {
val childExpression = Cast(BoundReference(0, IntegerType, nullable =
true), DoubleType)
val inputAggregationBufferOffset = 1
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/HistogramNumericSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/HistogramNumericSuite.scala
index 82e8277b42b9..98bf07ff76a3 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/HistogramNumericSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/HistogramNumericSuite.scala
@@ -199,6 +199,24 @@ class HistogramNumericSuite extends SparkFunSuite with
SQLHelper {
assert(agg.eval(buffer) != null)
}
+ test("SPARK-57557: HistogramNumeric supports TIME type") {
+ // With propagateInputType, the histogram's 'x' field preserves the input
TIME type and the
+ // bin centers are returned as Long nanos-of-day.
+ withSQLConf(SQLConf.HISTOGRAM_NUMERIC_PROPAGATE_INPUT_TYPE.key -> "true") {
+ val agg = new HistogramNumeric(BoundReference(0, TimeType(), nullable =
true), Literal(5))
+ val xType = agg.dataType match {
+ case ArrayType(StructType(Array(
+ StructField("x", t, _, _), StructField("y", _, _, _))), _) => t
+ }
+ assert(xType === TimeType())
+ val buffer = agg.createAggregationBuffer()
+ Seq(0L, 10L, 20L).foreach(v => agg.update(buffer, InternalRow(v)))
+ val result = agg.eval(buffer).asInstanceOf[GenericArrayData]
+ assert(result.numElements() > 0)
+ assert(result.getStruct(0, 2).get(0, TimeType()).isInstanceOf[Long])
+ }
+ }
+
test("class HistogramNumeric, exercise many different numeric input types") {
val inputs = Seq(
(Literal(null),
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileSuite.scala
index d3401613dcd8..92b036cbf94d 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileSuite.scala
@@ -113,6 +113,21 @@ class PercentileSuite extends SparkFunSuite {
}
}
+ test("SPARK-57557: Percentile and median support TIME type") {
+ // The return type mirrors the input TIME type, preserving precision.
+ assert(new Percentile(BoundReference(0, TimeType(), nullable = false),
Literal(0.5))
+ .dataType === TimeType())
+ assert(Median(BoundReference(0, TimeType(3), nullable = false)).dataType
=== TimeType(3))
+
+ // The exact median of two TIME values interpolates (continuous
distribution): the midpoint is
+ // returned as the internal Long value (nanos-of-day), e.g. median(0L,
10L) -> 5L.
+ val agg = new Percentile(BoundReference(0, TimeType(), nullable = false),
Literal(0.5))
+ val buffer = agg.createAggregationBuffer()
+ agg.update(buffer, InternalRow(0L))
+ agg.update(buffer, InternalRow(10L))
+ assert(agg.eval(buffer) === 5L)
+ }
+
test("class Percentile, low level interface, update, merge, eval...") {
val childExpression = Cast(BoundReference(0, IntegerType, nullable =
true), DoubleType)
val inputAggregationBufferOffset = 1
@@ -175,8 +190,8 @@ class PercentileSuite extends SparkFunSuite {
errorSubClass = "UNEXPECTED_INPUT_TYPE",
messageParameters = Map(
"paramIndex" -> ordinalNumber(0),
- "requiredType" -> ("(\"NUMERIC\" or \"INTERVAL DAY TO SECOND\" " +
- "or \"INTERVAL YEAR TO MONTH\")"),
+ "requiredType" -> ("((\"NUMERIC\" or \"INTERVAL DAY TO SECOND\" " +
+ "or \"INTERVAL YEAR TO MONTH\") or \"TIME\")"),
"inputSql" -> "\"a\"",
"inputType" -> toSQLType(dataType)
)
@@ -198,8 +213,8 @@ class PercentileSuite extends SparkFunSuite {
errorSubClass = "UNEXPECTED_INPUT_TYPE",
messageParameters = Map(
"paramIndex" -> ordinalNumber(0),
- "requiredType" -> ("(\"NUMERIC\" or \"INTERVAL DAY TO SECOND\" " +
- "or \"INTERVAL YEAR TO MONTH\")"),
+ "requiredType" -> ("((\"NUMERIC\" or \"INTERVAL DAY TO SECOND\" " +
+ "or \"INTERVAL YEAR TO MONTH\") or \"TIME\")"),
"inputSql" -> "\"a\"",
"inputType" -> toSQLType(dataType)
)
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/percentiles.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/percentiles.sql.out
index eb8102afa47e..c55b1ce4afc6 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/percentiles.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/percentiles.sql.out
@@ -1188,3 +1188,45 @@ org.apache.spark.sql.catalyst.ExtendedAnalysisException
SET spark.sql.legacy.percentileDiscCalculation = false
-- !query analysis
SetCommand (spark.sql.legacy.percentileDiscCalculation,Some(false))
+
+
+-- !query
+CREATE OR REPLACE TEMPORARY VIEW aggr_time AS SELECT * FROM VALUES
+ (TIME '00:00:00'), (TIME '06:00:00'), (TIME '12:00:00'), (TIME '18:00:00')
+AS aggr_time(t)
+-- !query analysis
+CreateViewCommand `aggr_time`, SELECT * FROM VALUES
+ (TIME '00:00:00'), (TIME '06:00:00'), (TIME '12:00:00'), (TIME '18:00:00')
+AS aggr_time(t), false, true, LocalTempView, UNSUPPORTED, true
+ +- Project [t#x]
+ +- SubqueryAlias aggr_time
+ +- LocalRelation [t#x]
+
+
+-- !query
+SELECT
+ median(t),
+ percentile(t, 0.5),
+ percentile_cont(0.5) WITHIN GROUP (ORDER BY t),
+ percentile_disc(0.5) WITHIN GROUP (ORDER BY t)
+FROM aggr_time
+-- !query analysis
+Aggregate [median(t#x) AS median(t)#x, percentile(t#x, cast(0.5 as double), 1,
0, 0, false) AS percentile(t, 0.5, 1)#x, percentile_cont(t#x, cast(0.5 as
double), false) AS percentile_cont(0.5) WITHIN GROUP (ORDER BY t)#x,
percentile_disc(t#x, cast(0.5 as double), false, 0, 0, false) AS
percentile_disc(0.5) WITHIN GROUP (ORDER BY t)#x]
++- SubqueryAlias aggr_time
+ +- View (`aggr_time`, [t#x])
+ +- Project [cast(t#x as time(6)) AS t#x]
+ +- Project [t#x]
+ +- SubqueryAlias aggr_time
+ +- LocalRelation [t#x]
+
+
+-- !query
+SELECT percentile_approx(t, array(0.25, 0.5, 0.75)) FROM aggr_time
+-- !query analysis
+Aggregate [percentile_approx(t#x, cast(array(0.25, cast(0.5 as decimal(2,2)),
0.75) as array<double>), 10000, 0, 0) AS percentile_approx(t, array(0.25, 0.5,
0.75), 10000)#x]
++- SubqueryAlias aggr_time
+ +- View (`aggr_time`, [t#x])
+ +- Project [cast(t#x as time(6)) AS t#x]
+ +- Project [t#x]
+ +- SubqueryAlias aggr_time
+ +- LocalRelation [t#x]
diff --git a/sql/core/src/test/resources/sql-tests/inputs/percentiles.sql
b/sql/core/src/test/resources/sql-tests/inputs/percentiles.sql
index 4b3e8708222a..fe698b1145dd 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/percentiles.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/percentiles.sql
@@ -427,3 +427,17 @@ SELECT
FROM values (12, 0.25), (13, 0.25), (22, 0.25) as v(a, b);
SET spark.sql.legacy.percentileDiscCalculation = false;
+
+-- SPARK-57557: percentile, median and percentile_cont/disc over the TIME type
+CREATE OR REPLACE TEMPORARY VIEW aggr_time AS SELECT * FROM VALUES
+ (TIME '00:00:00'), (TIME '06:00:00'), (TIME '12:00:00'), (TIME '18:00:00')
+AS aggr_time(t);
+
+SELECT
+ median(t),
+ percentile(t, 0.5),
+ percentile_cont(0.5) WITHIN GROUP (ORDER BY t),
+ percentile_disc(0.5) WITHIN GROUP (ORDER BY t)
+FROM aggr_time;
+
+SELECT percentile_approx(t, array(0.25, 0.5, 0.75)) FROM aggr_time;
diff --git a/sql/core/src/test/resources/sql-tests/results/percentiles.sql.out
b/sql/core/src/test/resources/sql-tests/results/percentiles.sql.out
index 55aaa8ee7378..b6c1d721acff 100644
--- a/sql/core/src/test/resources/sql-tests/results/percentiles.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/percentiles.sql.out
@@ -1188,3 +1188,34 @@ SET spark.sql.legacy.percentileDiscCalculation = false
struct<key:string,value:string>
-- !query output
spark.sql.legacy.percentileDiscCalculation false
+
+
+-- !query
+CREATE OR REPLACE TEMPORARY VIEW aggr_time AS SELECT * FROM VALUES
+ (TIME '00:00:00'), (TIME '06:00:00'), (TIME '12:00:00'), (TIME '18:00:00')
+AS aggr_time(t)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT
+ median(t),
+ percentile(t, 0.5),
+ percentile_cont(0.5) WITHIN GROUP (ORDER BY t),
+ percentile_disc(0.5) WITHIN GROUP (ORDER BY t)
+FROM aggr_time
+-- !query schema
+struct<median(t):time(6),percentile(t, 0.5, 1):time(6),percentile_cont(0.5)
WITHIN GROUP (ORDER BY t):time(6),percentile_disc(0.5) WITHIN GROUP (ORDER BY
t):time(6)>
+-- !query output
+09:00:00 09:00:00 09:00:00 06:00:00
+
+
+-- !query
+SELECT percentile_approx(t, array(0.25, 0.5, 0.75)) FROM aggr_time
+-- !query schema
+struct<percentile_approx(t, array(0.25, 0.5, 0.75), 10000):array<time(6)>>
+-- !query output
+[00:00:00,06:00:00,12:00:00]
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/ApproximatePercentileQuerySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/ApproximatePercentileQuerySuite.scala
index d9ff495fbbd8..ae24a21538f1 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/ApproximatePercentileQuerySuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/ApproximatePercentileQuerySuite.scala
@@ -18,13 +18,14 @@
package org.apache.spark.sql
import java.sql.{Date, Timestamp}
-import java.time.{Duration, LocalDateTime, Period}
+import java.time.{Duration, LocalDateTime, LocalTime, Period}
import
org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile
import
org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY
import
org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.PercentileDigest
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.TimeType
import org.apache.spark.tags.SlowSQLTest
/**
@@ -116,6 +117,23 @@ class ApproximatePercentileQuerySuite extends
SharedSparkSession {
}
}
+ test("SPARK-57557: percentile_approx supports TIME type") {
+ withTempView(table) {
+ spark.sql(
+ s"""SELECT * FROM VALUES
+ | (TIME '01:00:00'), (TIME '02:00:00'), (TIME '03:00:00'),
+ | (TIME '04:00:00'), (TIME '05:00:00') AS tab(c)
+ """.stripMargin).createOrReplaceTempView(table)
+ val scalarDf = spark.sql(s"SELECT percentile_approx(c, 0.5) FROM $table")
+ // The result type is TIME, mirroring the input column type.
+ assert(scalarDf.schema.head.dataType === TimeType())
+ checkAnswer(scalarDf, Row(LocalTime.of(3, 0)))
+ checkAnswer(
+ spark.sql(s"SELECT percentile_approx(c, array(0.2, 0.5, 0.8D)) FROM
$table"),
+ Row(Seq(LocalTime.of(1, 0), LocalTime.of(3, 0), LocalTime.of(4, 0))))
+ }
+ }
+
test("percentile_approx, multiple records with the minimum value in a
partition") {
withTempView(table) {
spark.sparkContext.makeRDD(Seq(1, 1, 2, 1, 1, 3, 1, 1, 4, 1, 1, 5),
4).toDF("col")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
