This is an automated email from the ASF dual-hosted git repository.
uros-b 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 fc7caa407355 [SPARK-58711][SQL] Use StructType.getFieldIndex in
Parquet and ORC aggregate push down
fc7caa407355 is described below
commit fc7caa4073557dc0e9e8d27a8f9de28da259908c
Author: Uros <[email protected]>
AuthorDate: Tue Aug 18 14:39:37 2026 +0000
[SPARK-58711][SQL] Use StructType.getFieldIndex in Parquet and ORC
aggregate push down
### What changes were proposed in this pull request?
Uses `StructType.getFieldIndex` for the column lookups in the Parquet and
ORC aggregate push down paths:
```scala
index = dataSchema.getFieldIndex(colName).getOrElse(-1) // was
fieldNames.toList.indexOf(colName)
if (partitionSchema.getFieldIndex(colName).isDefined) { // was
fields.map(_.name).toSet.contains(colName)
```
Three sites in `ParquetUtils` and one in `OrcUtils`.
### Why are the changes needed?
`fieldNames` is a `def` that allocates a fresh `Array` on every call, and
the Parquet sites then convert it to a `List`, so each lookup allocated twice
and scanned linearly; the `partitionSchema` check built a whole `Set` per call.
`getFieldIndex` reads the cached `nameToIndex` map.
To be accurate about the size of this: **it is a readability and allocation
cleanup, not a measurable speedup.** Aggregate push down is off by default, and
each surrounding loop iteration reads Parquet or ORC footer statistics, which
dominates any name lookup. The `OrcUtils` site is included because the two push
down implementations are deliberate mirrors that share `AggregatePushDownUtils`.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing aggregate push down tests (`FileSourceAggregatePushDownSuite`)
cover both paths.
`getOrElse(-1)` matches `indexOf`'s not-found contract, and the exact-match
`getFieldIndex` is used rather than `getFieldIndexCaseInsensitive`, preserving
case sensitivity. The one case where the two forms would differ is a schema
with duplicate column names, where `indexOf` returns the first match and a
name-keyed map the last; `FileTable.schema` runs
`checkSchemaColumnNameDuplication` on both the data and partition schemas
before any scan is built, and both `ParquetTable` and `OrcTa [...]
Note: open PR #57030 touches the same two `ParquetUtils` lines. There is no
logical conflict, and I am happy to rebase if that one lands first.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57929 from uros-b/perf-parquetutils-fieldindex.
Authored-by: Uros <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
---
.../org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala | 2 +-
.../spark/sql/execution/datasources/parquet/ParquetUtils.scala | 8 ++++----
2 files changed, 5 insertions(+), 5 deletions(-)
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 cf5544f92301..dd51f0eaf029 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
@@ -589,7 +589,7 @@ object OrcUtils extends Logging {
// Get column statistics with column name.
def getColumnStatistics(columnName: String): ColumnStatistics = {
- val columnIndex = dataSchema.fieldNames.indexOf(columnName)
+ val columnIndex = dataSchema.getFieldIndex(columnName).getOrElse(-1)
columnsStatistics.get(columnIndex).getStatistics
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala
index 55daa3aa8907..af143955ed70 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala
@@ -344,7 +344,7 @@ object ParquetUtils extends Logging {
agg match {
case max: Max if V2ColumnUtils.extractV2Column(max.column).isDefined
=>
val colName = V2ColumnUtils.extractV2Column(max.column).get
- index = dataSchema.fieldNames.toList.indexOf(colName)
+ index = dataSchema.getFieldIndex(colName).getOrElse(-1)
schemaName = "max(" + colName + ")"
val currentMax = getCurrentBlockMaxOrMin(filePath, blockMetaData,
index, true)
if (value == None ||
currentMax.asInstanceOf[Comparable[Any]].compareTo(value) > 0) {
@@ -352,7 +352,7 @@ object ParquetUtils extends Logging {
}
case min: Min if V2ColumnUtils.extractV2Column(min.column).isDefined
=>
val colName = V2ColumnUtils.extractV2Column(min.column).get
- index = dataSchema.fieldNames.toList.indexOf(colName)
+ index = dataSchema.getFieldIndex(colName).getOrElse(-1)
schemaName = "min(" + colName + ")"
val currentMin = getCurrentBlockMaxOrMin(filePath, blockMetaData,
index, false)
if (value == None ||
currentMin.asInstanceOf[Comparable[Any]].compareTo(value) < 0) {
@@ -363,12 +363,12 @@ object ParquetUtils extends Logging {
schemaName = "count(" + colName + ")"
rowCount += block.getRowCount
var isPartitionCol = false
- if (partitionSchema.fields.map(_.name).toSet.contains(colName)) {
+ if (partitionSchema.getFieldIndex(colName).isDefined) {
isPartitionCol = true
}
isCount = true
if (!isPartitionCol) {
- index = dataSchema.fieldNames.toList.indexOf(colName)
+ index = dataSchema.getFieldIndex(colName).getOrElse(-1)
// Count(*) includes the null values, but Count(colName) doesn't.
rowCount -= getNumNulls(filePath, blockMetaData, index)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
