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 254cbd96fa60 [SPARK-57568][SQL] Support TimeType in Parquet/ORC
aggregate push-down
254cbd96fa60 is described below
commit 254cbd96fa606792cbdb2a33ae60037190ab3143
Author: Maxim Gekk <[email protected]>
AuthorDate: Tue Jun 30 00:24:57 2026 +0200
[SPARK-57568][SQL] Support TimeType in Parquet/ORC aggregate push-down
### What changes were proposed in this pull request?
Enable MIN/MAX/COUNT aggregate push-down over `TIME` columns for the
Parquet and ORC data sources, computed from file-footer statistics.
- Add `TimeType` to the MIN/MAX type allow-list in
`AggregatePushDownUtils.getSchemaForPushedAggregation`. This is shared,
engine-agnostic code called by both `ParquetScanBuilder` and `OrcScanBuilder`,
so the single change enables push-down eligibility for both engines at once.
- Add a `TimeType` case to `OrcUtils.getMinMaxFromColumnStatistics`. ORC
stores `TIME` as a `LONG`, so its statistics are `IntegerColumnStatistics`; the
min/max value is wrapped in a `LongWritable` and converted back to the Spark
`TimeType` by `OrcDeserializer`.
- No Parquet reader change is needed: `TIME` is stored as Parquet `INT64`,
so the existing `INT64` branch in `ParquetUtils.createAggInternalRowFromFooter`
feeds the footer stat into a `ParquetRowConverter` built from the footer
`PrimitiveType`, which carries the `TIME(MICROS)`/`TIME(NANOS)` logical
annotation and maps to `TimeType`.
The columnar conversion
(`AggregatePushDownUtils.convertAggregatesRowToBatch` via
`RowToColumnConverter`) already supports `TimeType` (SPARK-54203), so the
columnar path works for `TIME`.
### Why are the changes needed?
This is a sub-task of SPARK-57550 (extending support for the `TIME` data
type). Aggregate push-down lets Parquet/ORC answer `MIN`/`MAX`/`COUNT` from
footer statistics without reading and aggregating `TIME` data at the Spark
layer.
### Does this PR introduce _any_ user-facing change?
No. This is an internal optimization on the aggregate push-down path; query
results are unchanged.
### How was this patch tested?
Added tests to the shared `FileSourceAggregatePushDownSuite` trait, which
is extended by `ParquetV1/V2AggregatePushDownSuite` and
`OrcV1/V2AggregatePushDownSuite`, so each test exercises all four engines:
- Positive: `MIN`/`MAX`/`COUNT(col)`/`COUNT(*)` push-down over a `TIME`
column at precisions 0, 6, 7, and 9, covering both the Parquet micros
(precision <= 6) and nanos (precision >= 7) storage paths, with a null row so
`COUNT(col)` and `COUNT(*)` differ.
- Negative: a data filter on the `TIME` column, an aggregate over a
non-column expression, and push-down disabled by config -- all asserting the
aggregate is not pushed.
Ran:
```
build/sbt 'sql/testOnly *ParquetV1AggregatePushDownSuite
*ParquetV2AggregatePushDownSuite *OrcV1AggregatePushDownSuite
*OrcV2AggregatePushDownSuite'
```
All 92 tests pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor (Claude Opus 4.8)
Closes #56846 from MaxGekk/time-aggr-pushdown.
Authored-by: Maxim Gekk <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
(cherry picked from commit 9c86d4e48908137e92b5c5a25e9ce4ef0915a519)
Signed-off-by: Max Gekk <[email protected]>
---
.../datasources/AggregatePushDownUtils.scala | 9 +-
.../sql/execution/datasources/orc/OrcUtils.scala | 4 +
.../FileSourceAggregatePushDownSuite.scala | 128 ++++++++++++++++++++-
3 files changed, 138 insertions(+), 3 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/AggregatePushDownUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/AggregatePushDownUtils.scala
index 97ee3cd661b3..db3a34db05ac 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/AggregatePushDownUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/AggregatePushDownUtils.scala
@@ -24,7 +24,7 @@ import
org.apache.spark.sql.connector.expressions.aggregate.{AggregateFunc, Aggr
import org.apache.spark.sql.execution.RowToColumnConverter
import org.apache.spark.sql.execution.datasources.v2.V2ColumnUtils
import org.apache.spark.sql.execution.vectorized.{OffHeapColumnVector,
OnHeapColumnVector}
-import org.apache.spark.sql.types.{BooleanType, ByteType, DateType,
DoubleType, FloatType, IntegerType, LongType, ShortType, StructField,
StructType}
+import org.apache.spark.sql.types.{BooleanType, ByteType, DateType,
DoubleType, FloatType, IntegerType, LongType, ShortType, StructField,
StructType, TimeType}
import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
/**
@@ -67,6 +67,8 @@ object AggregatePushDownUtils {
val structField = getStructFieldForCol(columnName)
structField.dataType match {
+ // Push down min/max only for primitive types whose footer statistics
are
+ // well-defined and directly comparable. Notable exclusions:
// not push down complex type
// not push down Timestamp because INT96 sort order is undefined,
// Parquet doesn't return statistics for INT96
@@ -74,8 +76,11 @@ object AggregatePushDownUtils {
// (https://issues.apache.org/jira/browse/PARQUET-1685), Parquet Binary
// could be Spark StringType, BinaryType or DecimalType.
// not push down for ORC with same reason.
+ // TIME is included because it is INT64/LONG-backed with a defined sort
+ // order, so real min/max statistics exist in the footer for both
Parquet
+ // and ORC.
case BooleanType | ByteType | ShortType | IntegerType
- | LongType | FloatType | DoubleType | DateType =>
+ | LongType | FloatType | DoubleType | DateType | _: TimeType =>
finalSchema = finalSchema.add(structField.copy(s"$aggType(" +
structField.name + ")"))
true
case _ =>
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
index c1ed3864b63a..e20216d69863 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
@@ -506,6 +506,10 @@ object OrcUtils extends Logging {
case ShortType => new ShortWritable(value.toShort)
case IntegerType => new IntWritable(value.toInt)
case LongType => new LongWritable(value)
+ // ORC stores TIME as LONG (nanos-of-day), so its stats are
+ // IntegerColumnStatistics. OrcDeserializer converts the
LongWritable
+ // back to the Spark TimeType.
+ case _: TimeType => new LongWritable(value)
case _ => throw new IllegalArgumentException(
s"getMinMaxFromColumnStatistics should not take type $dataType "
+
"for IntegerColumnStatistics")
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceAggregatePushDownSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceAggregatePushDownSuite.scala
index c95842fc7306..ac677876c238 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceAggregatePushDownSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceAggregatePushDownSuite.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.execution.datasources
import java.sql.{Date, Timestamp}
+import java.time.LocalTime
import org.apache.spark.{SparkConf, SparkUnsupportedOperationException}
import org.apache.spark.sql.{DataFrame, ExplainSuiteHelper, Row}
@@ -29,7 +30,7 @@ import
org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
import org.apache.spark.sql.functions.min
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType,
DateType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType,
ShortType, StringType, StructField, StructType, TimestampType}
+import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType,
DateType, Decimal, DecimalType, DoubleType, FloatType, IntegerType, LongType,
ShortType, StringType, StructField, StructType, TimestampType, TimeType}
import org.apache.spark.tags.SlowSQLTest
/**
@@ -571,6 +572,131 @@ trait FileSourceAggregatePushDownSuite
Seq(nullRow), Seq(nullRowWithOutTSAndBinary), Seq(zeroCount))
}
+ private def testTimeAggPushDown(
+ precision: Int,
+ times: Seq[LocalTime],
+ expectedMin: LocalTime,
+ expectedMax: LocalTime): Unit = {
+ val schema = StructType(Seq(StructField("TimeCol", TimeType(precision))))
+ // One null row in addition to the non-null `times`, so COUNT(TimeCol)
excludes it but
+ // COUNT(*) includes it.
+ val rows = times.map(Row(_)) :+ Row(null)
+ val rdd = sparkContext.parallelize(rows)
+ withTempPath { file =>
+ spark.createDataFrame(rdd,
schema).write.format(format).save(file.getCanonicalPath)
+ withTempView("time_test") {
+ spark.read.format(format).load(file.getCanonicalPath)
+ .createOrReplaceTempView("time_test")
+ Seq("false", "true").foreach { enableVectorizedReader =>
+ withSQLConf(aggPushDownEnabledKey -> "true",
+ vectorizedReaderEnabledKey -> enableVectorizedReader) {
+ val df = sql(
+ "SELECT min(TimeCol), max(TimeCol), count(TimeCol), count(*)
FROM time_test")
+ checkPushedInfo(df,
+ "PushedAggregation: [MIN(TimeCol), MAX(TimeCol), COUNT(TimeCol),
COUNT(*)]")
+ checkAnswer(df,
+ Seq(Row(expectedMin, expectedMax, times.length, times.length +
1)))
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57568: aggregate push down - TIME of different precisions") {
+ // Parquet stores TIME with precision 0..6 as INT64 TIME(MICROS) and
precision 7..9 as
+ // INT64 TIME(NANOS); ORC stores TIME as the raw nanos-of-day LONG. To
keep a single shared
+ // expectation valid across both engines, each value is chosen to be
exactly representable at
+ // its column's precision, so no engine-specific truncation can diverge.
+ // precision 0 (seconds)
+ testTimeAggPushDown(0,
+ Seq(LocalTime.of(1, 2, 3), LocalTime.of(23, 59, 59), LocalTime.of(10,
30, 0)),
+ LocalTime.of(1, 2, 3), LocalTime.of(23, 59, 59))
+ // precision 6 (microseconds)
+ testTimeAggPushDown(6,
+ Seq(LocalTime.of(1, 2, 3, 123456000), LocalTime.of(23, 59, 59,
999999000),
+ LocalTime.of(10, 30, 0, 500000)),
+ LocalTime.of(1, 2, 3, 123456000), LocalTime.of(23, 59, 59, 999999000))
+ // precision 7 (hundreds of nanoseconds)
+ testTimeAggPushDown(7,
+ Seq(LocalTime.of(1, 2, 3, 123456700), LocalTime.of(23, 59, 59,
999999900),
+ LocalTime.of(10, 30, 0, 100)),
+ LocalTime.of(1, 2, 3, 123456700), LocalTime.of(23, 59, 59, 999999900))
+ // precision 9 (nanoseconds)
+ testTimeAggPushDown(9,
+ Seq(LocalTime.of(1, 2, 3, 123456789), LocalTime.of(23, 59, 59,
999999999),
+ LocalTime.of(10, 30, 0, 1)),
+ LocalTime.of(1, 2, 3, 123456789), LocalTime.of(23, 59, 59, 999999999))
+ }
+
+ test("SPARK-57568: aggregate push down - TIME over an empty file") {
+ // Aggregating a TIME column over zero rows: MIN/MAX return NULL and COUNT
returns 0 on every
+ // engine. This pins the no-data path, which the precision test above
(always >= 1 row) does
+ // not reach. An all-null but non-empty file is intentionally not asserted
here: that is
+ // pre-existing, type-agnostic behavior shared by all push-down types
(Parquet rejects MIN/MAX
+ // push-down on an all-null block while ORC returns NULL), unchanged by
this PR.
+ Seq(6, 9).foreach { precision =>
+ val schema = StructType(Seq(StructField("TimeCol", TimeType(precision))))
+ val rdd = sparkContext.parallelize(Seq.empty[Row])
+ withTempPath { file =>
+ spark.createDataFrame(rdd,
schema).write.format(format).save(file.getCanonicalPath)
+ withTempView("time_empty") {
+ spark.read.format(format).load(file.getCanonicalPath)
+ .createOrReplaceTempView("time_empty")
+ Seq("false", "true").foreach { enableVectorizedReader =>
+ withSQLConf(aggPushDownEnabledKey -> "true",
+ vectorizedReaderEnabledKey -> enableVectorizedReader) {
+ val df = sql(
+ "SELECT min(TimeCol), max(TimeCol), count(TimeCol), count(*)
FROM time_empty")
+ checkPushedInfo(df,
+ "PushedAggregation: [MIN(TimeCol), MAX(TimeCol),
COUNT(TimeCol), COUNT(*)]")
+ checkAnswer(df, Seq(Row(null, null, 0, 0)))
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-57568: aggregate not push down - TIME with filter or
expression") {
+ val schema = StructType(Seq(StructField("TimeCol", TimeType(6))))
+ val rows = Seq(
+ Row(LocalTime.of(1, 2, 3)),
+ Row(LocalTime.of(23, 59, 59)),
+ Row(LocalTime.of(10, 30, 0)))
+ val rdd = sparkContext.parallelize(rows)
+ withTempPath { file =>
+ spark.createDataFrame(rdd,
schema).write.format(format).save(file.getCanonicalPath)
+ withTempView("time_neg_test") {
+ spark.read.format(format).load(file.getCanonicalPath)
+ .createOrReplaceTempView("time_neg_test")
+ withSQLConf(aggPushDownEnabledKey -> "true") {
+ // A data filter on a non-partition column prevents push down.
+ val withFilter =
+ sql("SELECT min(TimeCol) FROM time_neg_test WHERE TimeCol >
TIME'05:00:00'")
+ checkPushedInfo(withFilter, "PushedAggregation: []")
+ checkAnswer(withFilter, Seq(Row(LocalTime.of(10, 30, 0))))
+
+ // Aggregating over an expression (not a plain column) prevents push
down. The two CASE
+ // branches differ so the optimizer cannot fold the expression back
into a column.
+ val withExpr = sql(
+ "SELECT max(CASE WHEN TimeCol > TIME'05:00:00' THEN TimeCol ELSE
TIME'00:00:00' END) " +
+ "FROM time_neg_test")
+ checkPushedInfo(withExpr, "PushedAggregation: []")
+ checkAnswer(withExpr, Seq(Row(LocalTime.of(23, 59, 59))))
+
+ // Aggregate push down disabled by config.
+ withSQLConf(aggPushDownEnabledKey -> "false") {
+ val disabled =
+ sql("SELECT min(TimeCol), max(TimeCol), count(TimeCol) FROM
time_neg_test")
+ checkPushedInfo(disabled, "PushedAggregation: []")
+ checkAnswer(disabled,
+ Seq(Row(LocalTime.of(1, 2, 3), LocalTime.of(23, 59, 59), 3)))
+ }
+ }
+ }
+ }
+ }
+
test("column name case sensitivity") {
Seq("false", "true").foreach { enableVectorizedReader =>
withSQLConf(aggPushDownEnabledKey -> "true",
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
