From b6f6218e5068343e5dff184cd52e0c58fcc95b7e Mon Sep 17 00:00:00 2001 From: Stevo Mitric Date: Fri, 11 Sep 2026 14:05:54 +0000 Subject: [PATCH 1/4] [SPARK-XXXXX][SQL] Support the TIME data type in the hll_sketch_agg function ### What changes were proposed in this pull request? Add support for the TIME data type to the `hll_sketch_agg` aggregate. TIME is physically stored as a long (nanoseconds since midnight), so it is hashed into the HllSketch exactly like the existing `LongType` path. `hll_union_agg` needs no change: it only ingests already-serialized BINARY sketches, so a sketch built from a TIME column merges through it unchanged. ### Why are the changes needed? Part of SPARK-57550 (extend support for the TIME data type). Approximate distinct counting over TIME columns is a natural, previously-missing capability. ### Does this PR introduce any user-facing change? Yes. `hll_sketch_agg(time_col[, lgConfigK])` is now accepted; previously it raised an analysis error for TIME inputs. ### How was this patch tested? New unit test in `DatasketchesHllSketchSuite` covering analyzer acceptance, cardinality estimation over TIME values, precision-insensitive de-duplication, and a `hll_union_agg` round-trip over sketches built from a TIME column. Co-authored-by: Isaac --- python/pyspark/sql/functions/builtin.py | 2 +- .../aggregate/datasketchesAggregates.scala | 10 +++-- .../DatasketchesHllSketchSuite.scala | 40 ++++++++++++++++++- 3 files changed, 47 insertions(+), 5 deletions(-) diff --git a/python/pyspark/sql/functions/builtin.py b/python/pyspark/sql/functions/builtin.py index 528073a105158..35c6862c63bf8 100644 --- a/python/pyspark/sql/functions/builtin.py +++ b/python/pyspark/sql/functions/builtin.py @@ -30011,7 +30011,7 @@ def hll_sketch_agg( Parameters ---------- col : :class:`~pyspark.sql.Column` or column name - A column that evaluates to an integer, long, string, or binary. + A column that evaluates to an integer, long, time, string, or binary. lgConfigK : :class:`~pyspark.sql.Column` or int, optional The log-base-2 of K, where K is the number of buckets or slots for the HllSketch. A column that evaluates to an integer. diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/datasketchesAggregates.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/datasketchesAggregates.scala index 0a06c54409c08..7ead098ee4ca1 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/datasketchesAggregates.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/datasketchesAggregates.scala @@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.trees.BinaryLike import org.apache.spark.sql.catalyst.util.CollationFactory import org.apache.spark.sql.errors.QueryExecutionErrors import org.apache.spark.sql.internal.types.StringTypeWithCollation -import org.apache.spark.sql.types.{AbstractDataType, BinaryType, BooleanType, DataType, IntegerType, LongType, StringType, TypeCollection} +import org.apache.spark.sql.types.{AbstractDataType, AnyTimeType, BinaryType, BooleanType, DataType, IntegerType, LongType, StringType, TimeType, TypeCollection} import org.apache.spark.unsafe.types.UTF8String @@ -51,7 +51,7 @@ import org.apache.spark.unsafe.types.UTF8String arguments = """ Arguments: * expr - The expression to aggregate into the HLL sketch. - An expression that evaluates to an integer, long, string, or binary. + An expression that evaluates to an integer, long, time, string, or binary. * lgConfigK - The log-base-2 of K, where K is the number of buckets for the sketch. An expression that evaluates to an integer. """, @@ -119,6 +119,7 @@ case class HllSketchAgg( TypeCollection( IntegerType, LongType, + AnyTimeType, StringTypeWithCollation(supportsTrimCollation = true), BinaryType), IntegerType) @@ -156,9 +157,12 @@ case class HllSketchAgg( // Spark SQL doesn't have equivalent types for ByteBuffer or char[] so leave those out. // We leave out support for Array types, as unique counting these aren't a common use case. // We leave out support for floating point types (such as DoubleType) due to imprecision. - // TODO: implement support for decimal/datetime/interval types + // TODO: implement support for decimal/date/timestamp/interval types case IntegerType => sketch.update(v.asInstanceOf[Int]) case LongType => sketch.update(v.asInstanceOf[Long]) + // TIME is physically stored as a long (nanoseconds since midnight), so it hashes exactly + // like LongType: equal times share the same nanos and therefore the same sketch entry. + case _: TimeType => sketch.update(v.asInstanceOf[Long]) case st: StringType => val collation = CollationFactory.fetchCollation(st.collationId) val str = v.asInstanceOf[UTF8String] diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/DatasketchesHllSketchSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/DatasketchesHllSketchSuite.scala index 4dcc3bf28d24e..82aaf04ea7093 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/DatasketchesHllSketchSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/DatasketchesHllSketchSuite.scala @@ -17,6 +17,8 @@ package org.apache.spark.sql.catalyst.expressions.aggregate +import java.time.LocalTime + import scala.collection.immutable.NumericRange import scala.util.Random @@ -26,7 +28,7 @@ import org.apache.datasketches.memory.Memory import org.apache.spark.{SparkFunSuite, SparkRuntimeException} import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.{BoundReference, HllSketchEstimate, HllUnion, Literal} -import org.apache.spark.sql.types.{BinaryType, DataType, IntegerType, LongType, StringType} +import org.apache.spark.sql.types.{BinaryType, DataType, IntegerType, LongType, StringType, TimeType} import org.apache.spark.unsafe.types.UTF8String @@ -88,6 +90,42 @@ class DatasketchesHllSketchSuite extends SparkFunSuite { binaryEstimateRange.contains(binaryRange.size.toLong)) } + test("Test hll_sketch_agg and hll_union_agg over the TIME type") { + // The analyzer admits TIME as a value to sketch. + assert( + new HllSketchAgg(BoundReference(0, TimeType(6), nullable = true), 12) + .checkInputDataTypes().isSuccess) + + // TIME is physically a long of nanos-of-day, so distinct-counting a TIME column behaves + // exactly like counting the underlying longs. + val timeRange = (0 until 1000).map(_.toLong * 1000000000L) // 0s..999s of the day, in nanos + val (estimate, estimateRange) = simulateUpdateMerge(TimeType(), timeRange) + assert(estimate == timeRange.size || estimateRange.contains(timeRange.size.toLong)) + + // Equal times (even written at different precisions) share the same nanos-of-day and are + // counted once. + val nineAm = LocalTime.of(9, 0, 0).toNanoOfDay + val noon = LocalTime.of(12, 0, 0).toNanoOfDay + val fivePm = LocalTime.of(17, 0, 0).toNanoOfDay + val aggFunc = new HllSketchAgg(BoundReference(0, TimeType(9), nullable = true), 12) + val buffer = Seq(noon, noon, noon, nineAm, nineAm) + .foldLeft(aggFunc.createAggregationBuffer())((buf, t) => aggFunc.update(buf, InternalRow(t))) + assert(estimateOf(aggFunc.eval(buffer).asInstanceOf[Array[Byte]]) == 2L) + + // A sketch built from a TIME column round-trips through hll_union_agg, which only ever sees the + // serialized BINARY sketch and so needs no TIME-specific handling of its own. + def timeSketch(values: Seq[Long]): Array[Byte] = { + val agg = new HllSketchAgg(BoundReference(0, TimeType(), nullable = true), 12) + val buf = values.foldLeft(agg.createAggregationBuffer())((b, v) => + agg.update(b, InternalRow(v))) + agg.eval(buf).asInstanceOf[Array[Byte]] + } + val merged = unionAgg( + Seq[Any](timeSketch(Seq(nineAm, noon)), timeSketch(Seq(noon, fivePm))), + allowDifferentLgConfigK = false) + assert(estimateOf(merged) == 3L) // distinct {09:00, 12:00, 17:00} + } + test("Test lgMaxK results in downsampling sketches with larger lgConfigK") { val aggFunc1 = new HllSketchAgg(BoundReference(0, IntegerType, nullable = true), 12) val sketch1 = aggFunc1.createAggregationBuffer() From 8f45f252bff4749cf353750667d9ec2350db2077 Mon Sep 17 00:00:00 2001 From: Stevo Mitric Date: Fri, 11 Sep 2026 16:48:28 +0000 Subject: [PATCH 2/4] [SPARK-59440][SQL][FOLLOWUP] Address review for hll_sketch_agg TIME support Follow-up to the initial commit, addressing code-review feedback: - Regenerate the hll.sql golden files (results and analyzer-results). Adding the TIME type to the input TypeCollection changes the DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE error text, which the ARRAY negative test asserts; the stale golden would have failed CI. - Update the remaining public doc surfaces to mention TIME: the SQL reference table (docs/sql-ref-sketch-aggregates.md) and all five hll_sketch_agg scaladocs in functions.scala. - Add end-to-end SQL coverage in hll.sql: hll_sketch_agg over a TIME column and an hll_union_agg round-trip over sketches built from TIME columns. - Strengthen DatasketchesHllSketchSuite: a sub-microsecond-differing case (so a regression that truncated nanoseconds before hashing would be caught) and a mixed-precision union. - Add a .. versionchanged:: note to the Python hll_sketch_agg docstring. Co-authored-by: Isaac --- docs/sql-ref-sketch-aggregates.md | 2 +- python/pyspark/sql/functions/builtin.py | 3 ++ .../org/apache/spark/sql/functions.scala | 16 +++++----- .../DatasketchesHllSketchSuite.scala | 26 ++++++++-------- .../sql-tests/analyzer-results/hll.sql.out | 30 ++++++++++++++++++- .../test/resources/sql-tests/inputs/hll.sql | 12 ++++++++ .../resources/sql-tests/results/hll.sql.out | 24 ++++++++++++++- 7 files changed, 90 insertions(+), 23 deletions(-) diff --git a/docs/sql-ref-sketch-aggregates.md b/docs/sql-ref-sketch-aggregates.md index fd07cdf928bdf..a761be1d495d3 100644 --- a/docs/sql-ref-sketch-aggregates.md +++ b/docs/sql-ref-sketch-aggregates.md @@ -98,7 +98,7 @@ hll_sketch_agg(expr [, lgConfigK]) | Argument | Type | Description | |----------|------|-------------| -| `expr` | INT, BIGINT, STRING, or BINARY | The expression whose distinct values will be counted | +| `expr` | INT, BIGINT, TIME, STRING, or BINARY | The expression whose distinct values will be counted | | `lgConfigK` | INT (optional) | Log-base-2 of K, where K is the number of buckets. Range: 4-21. Default: 12. Higher values provide more accuracy but use more memory. | Returns a BINARY containing the HLL sketch in updatable binary representation. diff --git a/python/pyspark/sql/functions/builtin.py b/python/pyspark/sql/functions/builtin.py index 35c6862c63bf8..b9adcaa4c4300 100644 --- a/python/pyspark/sql/functions/builtin.py +++ b/python/pyspark/sql/functions/builtin.py @@ -30008,6 +30008,9 @@ def hll_sketch_agg( .. versionadded:: 3.5.0 + .. versionchanged:: 4.4.0 + Supports the TIME type for the ``col`` argument. + Parameters ---------- col : :class:`~pyspark.sql.Column` or column name diff --git a/sql/api/src/main/scala/org/apache/spark/sql/functions.scala b/sql/api/src/main/scala/org/apache/spark/sql/functions.scala index 2845c670a7691..193018c953ca0 100644 --- a/sql/api/src/main/scala/org/apache/spark/sql/functions.scala +++ b/sql/api/src/main/scala/org/apache/spark/sql/functions.scala @@ -898,8 +898,8 @@ object functions { * configured with lgConfigK arg. * * @param e - * the column to compute the sketch on. A column that evaluates to an integral, a string or a - * binary. + * the column to compute the sketch on. A column that evaluates to an integral, a time, a + * string or a binary. * @param lgConfigK * the log-base-2 of K, where K is the number of buckets or slots for the HllSketch. A column * that evaluates to an integral. Must be a constant. @@ -916,8 +916,8 @@ object functions { * configured with lgConfigK arg. * * @param e - * the column to compute the sketch on. A column that evaluates to an integral, a string or a - * binary. + * the column to compute the sketch on. A column that evaluates to an integral, a time, a + * string or a binary. * @param lgConfigK * the log-base-2 of K, where K is the number of buckets or slots for the HllSketch. A column * that evaluates to an integral. Must be a constant. @@ -935,7 +935,7 @@ object functions { * * @param columnName * the name of the column to compute the sketch on. A column that evaluates to an integral, a - * string or a binary. + * time, a string or a binary. * @param lgConfigK * the log-base-2 of K, where K is the number of buckets or slots for the HllSketch. A column * that evaluates to an integral. Must be a constant. @@ -953,8 +953,8 @@ object functions { * configured with default lgConfigK value. * * @param e - * the column to compute the sketch on. A column that evaluates to an integral, a string or a - * binary. + * the column to compute the sketch on. A column that evaluates to an integral, a time, a + * string or a binary. * @group agg_funcs * @since 3.5.0 * @return @@ -969,7 +969,7 @@ object functions { * * @param columnName * the name of the column to compute the sketch on. A column that evaluates to an integral, a - * string or a binary. + * time, a string or a binary. * @group agg_funcs * @since 3.5.0 * @return diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/DatasketchesHllSketchSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/DatasketchesHllSketchSuite.scala index 82aaf04ea7093..ed037cbc19b11 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/DatasketchesHllSketchSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/DatasketchesHllSketchSuite.scala @@ -102,26 +102,28 @@ class DatasketchesHllSketchSuite extends SparkFunSuite { val (estimate, estimateRange) = simulateUpdateMerge(TimeType(), timeRange) assert(estimate == timeRange.size || estimateRange.contains(timeRange.size.toLong)) - // Equal times (even written at different precisions) share the same nanos-of-day and are - // counted once. val nineAm = LocalTime.of(9, 0, 0).toNanoOfDay val noon = LocalTime.of(12, 0, 0).toNanoOfDay val fivePm = LocalTime.of(17, 0, 0).toNanoOfDay - val aggFunc = new HllSketchAgg(BoundReference(0, TimeType(9), nullable = true), 12) - val buffer = Seq(noon, noon, noon, nineAm, nineAm) - .foldLeft(aggFunc.createAggregationBuffer())((buf, t) => aggFunc.update(buf, InternalRow(t))) - assert(estimateOf(aggFunc.eval(buffer).asInstanceOf[Array[Byte]]) == 2L) - - // A sketch built from a TIME column round-trips through hll_union_agg, which only ever sees the - // serialized BINARY sketch and so needs no TIME-specific handling of its own. - def timeSketch(values: Seq[Long]): Array[Byte] = { - val agg = new HllSketchAgg(BoundReference(0, TimeType(), nullable = true), 12) + + def timeSketch(precision: Int, values: Seq[Long]): Array[Byte] = { + val agg = new HllSketchAgg(BoundReference(0, TimeType(precision), nullable = true), 12) val buf = values.foldLeft(agg.createAggregationBuffer())((b, v) => agg.update(b, InternalRow(v))) agg.eval(buf).asInstanceOf[Array[Byte]] } + + // Repeated values are counted once (deduplication). + assert(estimateOf(timeSketch(9, Seq(noon, noon, noon, nineAm, nineAm))) == 2L) + + // The full nanos-of-day is hashed: times that differ only in sub-microsecond digits are + // distinct. A regression that truncated to micros before hashing would under-count these. + assert(estimateOf(timeSketch(9, Seq(noon, noon + 1L, noon + 2L))) == 3L) + + // Sketches built from TIME columns of different precisions round-trip through hll_union_agg, + // which only ever sees the serialized BINARY sketch and so needs no TIME-specific handling. val merged = unionAgg( - Seq[Any](timeSketch(Seq(nineAm, noon)), timeSketch(Seq(noon, fivePm))), + Seq[Any](timeSketch(3, Seq(nineAm, noon)), timeSketch(9, Seq(noon, fivePm))), allowDifferentLgConfigK = false) assert(estimateOf(merged) == 3L) // distinct {09:00, 12:00, 17:00} } diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/hll.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/hll.sql.out index 291f071ef06c2..22608c0e08475 100644 --- a/sql/core/src/test/resources/sql-tests/analyzer-results/hll.sql.out +++ b/sql/core/src/test/resources/sql-tests/analyzer-results/hll.sql.out @@ -190,6 +190,34 @@ Aggregate [hll_sketch_estimate(hll_union_agg(sketch#x, true, 0, 0)) AS hll_sketc +- LocalRelation [col#x] +-- !query +SELECT hll_sketch_estimate(hll_sketch_agg(col)) +FROM VALUES (TIME'12:00:00'), (TIME'12:00:00'), (TIME'09:00:00'), (TIME'17:00:00') tab(col) +-- !query analysis +Aggregate [hll_sketch_estimate(hll_sketch_agg(col#x, 12, 0, 0)) AS hll_sketch_estimate(hll_sketch_agg(col, 12))#xL] ++- SubqueryAlias tab + +- LocalRelation [col#x] + + +-- !query +SELECT hll_sketch_estimate(hll_union_agg(sketch, true)) + FROM (SELECT hll_sketch_agg(col) as sketch + FROM VALUES (TIME'12:00:00'), (TIME'09:00:00') AS tab(col) + UNION ALL + SELECT hll_sketch_agg(col) as sketch + FROM VALUES (TIME'12:00:00'), (TIME'17:00:00') AS tab(col)) +-- !query analysis +Aggregate [hll_sketch_estimate(hll_union_agg(sketch#x, true, 0, 0)) AS hll_sketch_estimate(hll_union_agg(sketch, true))#xL] ++- SubqueryAlias __auto_generated_subquery_name + +- Union false, false + :- Aggregate [hll_sketch_agg(col#x, 12, 0, 0) AS sketch#x] + : +- SubqueryAlias tab + : +- LocalRelation [col#x] + +- Aggregate [hll_sketch_agg(col#x, 12, 0, 0) AS sketch#x] + +- SubqueryAlias tab + +- LocalRelation [col#x] + + -- !query SELECT hll_sketch_agg(col) FROM VALUES (ARRAY(1, 2)), (ARRAY(3, 4)) tab(col) @@ -202,7 +230,7 @@ org.apache.spark.sql.catalyst.ExtendedAnalysisException "inputSql" : "\"col\"", "inputType" : "\"ARRAY\"", "paramIndex" : "first", - "requiredType" : "(\"INT\" or \"BIGINT\" or \"STRING\" or \"BINARY\")", + "requiredType" : "(\"INT\" or \"BIGINT\" or \"TIME\" or \"STRING\" or \"BINARY\")", "sqlExpr" : "\"hll_sketch_agg(col, 12)\"" }, "queryContext" : [ { diff --git a/sql/core/src/test/resources/sql-tests/inputs/hll.sql b/sql/core/src/test/resources/sql-tests/inputs/hll.sql index 35128da97fd61..552c30f2e0be8 100644 --- a/sql/core/src/test/resources/sql-tests/inputs/hll.sql +++ b/sql/core/src/test/resources/sql-tests/inputs/hll.sql @@ -59,6 +59,18 @@ SELECT hll_sketch_estimate(hll_union_agg(sketch, true)) SELECT hll_sketch_agg(col, 20) as sketch FROM VALUES (1) AS tab(col)); +-- TIME type: hll_sketch_agg counts distinct times, and the resulting sketches merge via +-- hll_union_agg (which only sees the serialized binary sketch). +SELECT hll_sketch_estimate(hll_sketch_agg(col)) +FROM VALUES (TIME'12:00:00'), (TIME'12:00:00'), (TIME'09:00:00'), (TIME'17:00:00') tab(col); + +SELECT hll_sketch_estimate(hll_union_agg(sketch, true)) + FROM (SELECT hll_sketch_agg(col) as sketch + FROM VALUES (TIME'12:00:00'), (TIME'09:00:00') AS tab(col) + UNION ALL + SELECT hll_sketch_agg(col) as sketch + FROM VALUES (TIME'12:00:00'), (TIME'17:00:00') AS tab(col)); + -- Negative test cases SELECT hll_sketch_agg(col) FROM VALUES (ARRAY(1, 2)), (ARRAY(3, 4)) tab(col); diff --git a/sql/core/src/test/resources/sql-tests/results/hll.sql.out b/sql/core/src/test/resources/sql-tests/results/hll.sql.out index 908221f0e7c40..46f59198fa06f 100644 --- a/sql/core/src/test/resources/sql-tests/results/hll.sql.out +++ b/sql/core/src/test/resources/sql-tests/results/hll.sql.out @@ -191,6 +191,28 @@ struct 1 +-- !query +SELECT hll_sketch_estimate(hll_sketch_agg(col)) +FROM VALUES (TIME'12:00:00'), (TIME'12:00:00'), (TIME'09:00:00'), (TIME'17:00:00') tab(col) +-- !query schema +struct +-- !query output +3 + + +-- !query +SELECT hll_sketch_estimate(hll_union_agg(sketch, true)) + FROM (SELECT hll_sketch_agg(col) as sketch + FROM VALUES (TIME'12:00:00'), (TIME'09:00:00') AS tab(col) + UNION ALL + SELECT hll_sketch_agg(col) as sketch + FROM VALUES (TIME'12:00:00'), (TIME'17:00:00') AS tab(col)) +-- !query schema +struct +-- !query output +3 + + -- !query SELECT hll_sketch_agg(col) FROM VALUES (ARRAY(1, 2)), (ARRAY(3, 4)) tab(col) @@ -205,7 +227,7 @@ org.apache.spark.sql.catalyst.ExtendedAnalysisException "inputSql" : "\"col\"", "inputType" : "\"ARRAY\"", "paramIndex" : "first", - "requiredType" : "(\"INT\" or \"BIGINT\" or \"STRING\" or \"BINARY\")", + "requiredType" : "(\"INT\" or \"BIGINT\" or \"TIME\" or \"STRING\" or \"BINARY\")", "sqlExpr" : "\"hll_sketch_agg(col, 12)\"" }, "queryContext" : [ { From f8b9cdfc8c1de23e1126ceeb7b19e5b5054f2dd5 Mon Sep 17 00:00:00 2001 From: Stevo Mitric Date: Sun, 13 Sep 2026 15:52:16 +0000 Subject: [PATCH 3/4] [SPARK-59440][SQL][FOLLOWUP] Keep AnyTimeType last in hll_sketch_agg input types Address review feedback: placing AnyTimeType before StringType in the ordered input TypeCollection would change ANSI implicit coercion for types the collection does not directly accept. canANSIStoreAssign is true for both (TIMESTAMP/TIMESTAMP_NTZ/DATE -> STRING) and (TIMESTAMP/TIMESTAMP_NTZ -> TIME), so an AnyTimeType member ahead of StringType would re-route TIMESTAMP/TIMESTAMP_NTZ to a TIME target (sketching only the nanos-of-day and silently under-counting distinct values) and DATE to a TIME target that has no cast rule (an error path), instead of their prior STRING coercion. Move AnyTimeType to the end of the TypeCollection so it is only ever a last-resort coercion target; a TIME argument is still accepted by the order-independent acceptsType short-circuit. Regenerate the hll.sql golden files, whose UNEXPECTED_INPUT_TYPE required-type list reflects the new member order. Co-authored-by: Isaac --- .../expressions/aggregate/datasketchesAggregates.scala | 10 ++++++++-- .../resources/sql-tests/analyzer-results/hll.sql.out | 2 +- .../src/test/resources/sql-tests/results/hll.sql.out | 2 +- 3 files changed, 10 insertions(+), 4 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/datasketchesAggregates.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/datasketchesAggregates.scala index 7ead098ee4ca1..19dd58dbaf9cd 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/datasketchesAggregates.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/datasketchesAggregates.scala @@ -116,12 +116,18 @@ case class HllSketchAgg( override def inputTypes: Seq[AbstractDataType] = Seq( + // AnyTimeType must stay last: a TIME argument is accepted by the order-independent + // acceptsType short-circuit regardless of position, but ANSI implicit coercion walks this + // collection in order for a type it does not directly accept. Keeping AnyTimeType after + // StringType preserves the pre-existing behavior where TIMESTAMP/TIMESTAMP_NTZ/DATE inputs + // coerce to STRING (canANSIStoreAssign to TIME is also true, so an earlier AnyTimeType would + // instead route them to TIME and silently under-count). TypeCollection( IntegerType, LongType, - AnyTimeType, StringTypeWithCollation(supportsTrimCollation = true), - BinaryType), + BinaryType, + AnyTimeType), IntegerType) override def dataType: DataType = BinaryType diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/hll.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/hll.sql.out index 22608c0e08475..ccaca0c9a6c5a 100644 --- a/sql/core/src/test/resources/sql-tests/analyzer-results/hll.sql.out +++ b/sql/core/src/test/resources/sql-tests/analyzer-results/hll.sql.out @@ -230,7 +230,7 @@ org.apache.spark.sql.catalyst.ExtendedAnalysisException "inputSql" : "\"col\"", "inputType" : "\"ARRAY\"", "paramIndex" : "first", - "requiredType" : "(\"INT\" or \"BIGINT\" or \"TIME\" or \"STRING\" or \"BINARY\")", + "requiredType" : "(\"INT\" or \"BIGINT\" or \"STRING\" or \"BINARY\" or \"TIME\")", "sqlExpr" : "\"hll_sketch_agg(col, 12)\"" }, "queryContext" : [ { diff --git a/sql/core/src/test/resources/sql-tests/results/hll.sql.out b/sql/core/src/test/resources/sql-tests/results/hll.sql.out index 46f59198fa06f..6dfc7864b6603 100644 --- a/sql/core/src/test/resources/sql-tests/results/hll.sql.out +++ b/sql/core/src/test/resources/sql-tests/results/hll.sql.out @@ -227,7 +227,7 @@ org.apache.spark.sql.catalyst.ExtendedAnalysisException "inputSql" : "\"col\"", "inputType" : "\"ARRAY\"", "paramIndex" : "first", - "requiredType" : "(\"INT\" or \"BIGINT\" or \"TIME\" or \"STRING\" or \"BINARY\")", + "requiredType" : "(\"INT\" or \"BIGINT\" or \"STRING\" or \"BINARY\" or \"TIME\")", "sqlExpr" : "\"hll_sketch_agg(col, 12)\"" }, "queryContext" : [ { From 5e95e1c6b34d06d2fb1d388e7ad2b938bde9c5b6 Mon Sep 17 00:00:00 2001 From: Stevo Mitric Date: Sun, 13 Sep 2026 16:11:12 +0000 Subject: [PATCH 4/4] [SPARK-59440][SQL][FOLLOWUP] Drop the explanatory comment on the input TypeCollection Co-authored-by: Isaac --- .../expressions/aggregate/datasketchesAggregates.scala | 6 ------ 1 file changed, 6 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/datasketchesAggregates.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/datasketchesAggregates.scala index 19dd58dbaf9cd..6069d4b8375bd 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/datasketchesAggregates.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/datasketchesAggregates.scala @@ -116,12 +116,6 @@ case class HllSketchAgg( override def inputTypes: Seq[AbstractDataType] = Seq( - // AnyTimeType must stay last: a TIME argument is accepted by the order-independent - // acceptsType short-circuit regardless of position, but ANSI implicit coercion walks this - // collection in order for a type it does not directly accept. Keeping AnyTimeType after - // StringType preserves the pre-existing behavior where TIMESTAMP/TIMESTAMP_NTZ/DATE inputs - // coerce to STRING (canANSIStoreAssign to TIME is also true, so an earlier AnyTimeType would - // instead route them to TIME and silently under-count). TypeCollection( IntegerType, LongType,