You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ma...@apache.org on 2026/06/29 22:25:11 UTC
(spark) branch master updated: [SPARK-57568][SQL] Support TimeType in Parquet/ORC aggregate push-down
This is an automated email from the ASF dual-hosted git repository.
MaxGekk 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 9c86d4e48908 [SPARK-57568][SQL] Support TimeType in Parquet/ORC aggregate push-down
9c86d4e48908 is described below
commit 9c86d4e48908137e92b5c5a25e9ce4ef0915a519
Author: Maxim Gekk <ma...@gmail.com>
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 <ma...@gmail.com>
Signed-off-by: Max Gekk <ma...@gmail.com>
---
.../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org