From c59153aab50ef4fd4d12d5f2aa898d13c97ae73c Mon Sep 17 00:00:00 2001 From: Hemanth Boyina Date: Mon, 21 Sep 2026 18:52:38 +0530 Subject: [PATCH 1/2] [SPARK-59609][SQL] Fix Parquet/ORC aggregate push-down reading statistics from the wrong column --- .../execution/datasources/orc/OrcUtils.scala | 37 ++++++--- .../datasources/parquet/ParquetUtils.scala | 76 ++++++++++++------- .../v2/orc/OrcPartitionReaderFactory.scala | 4 +- .../FileSourceAggregatePushDownSuite.scala | 55 ++++++++++++++ 4 files changed, 130 insertions(+), 42 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 6e49959acc2b..65f51403a3c0 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 @@ -41,7 +41,7 @@ import org.apache.spark.internal.Logging import org.apache.spark.internal.LogKeys.PATH import org.apache.spark.sql.{SPARK_VERSION_METADATA_KEY, SparkSession} import org.apache.spark.sql.catalyst.{FileSourceOptions, InternalRow} -import org.apache.spark.sql.catalyst.analysis.caseSensitiveResolution +import org.apache.spark.sql.catalyst.analysis.{caseInsensitiveResolution, caseSensitiveResolution} import org.apache.spark.sql.catalyst.expressions.JoinedRow import org.apache.spark.sql.catalyst.parser.CatalystSqlParser import org.apache.spark.sql.catalyst.util.{quoteIdentifier, CaseInsensitiveMap, CharVarcharUtils} @@ -585,7 +585,9 @@ object OrcUtils extends Logging { partitionSchema: StructType, aggregation: Aggregation, aggSchema: StructType, - partitionValues: InternalRow): InternalRow = { + partitionValues: InternalRow, + isCaseSensitive: Boolean, + conf: Configuration): InternalRow = { var columnsStatistics: OrcColumnStatistics = null try { columnsStatistics = OrcFooterReader.readStatistics(reader) @@ -595,10 +597,23 @@ object OrcUtils extends Logging { s"ORC aggregate push down by setting 'spark.sql.orc.aggregatePushdown' to false.", e) } - // Get column statistics with column name. - def getColumnStatistics(columnName: String): ColumnStatistics = { - val columnIndex = dataSchema.getFieldIndex(columnName).getOrElse(-1) - columnsStatistics.get(columnIndex).getStatistics + // Resolve to the file's own ordinal, mirroring `requestedColumnIds`; -1 if absent. + val orcFieldNames = reader.getSchema.getFieldNames.asScala + val forcePositionalEvolution = OrcConf.FORCE_POSITIONAL_EVOLUTION.getBoolean(conf) + def fileColumnIndex(columnName: String): Int = { + if (forcePositionalEvolution || orcFieldNames.forall(_.startsWith("_col"))) { + val index = dataSchema.getFieldIndex(columnName).getOrElse(-1) + if (index >= 0 && index < orcFieldNames.length) index else -1 + } else if (isCaseSensitive) { + orcFieldNames.indexWhere(caseSensitiveResolution(_, columnName)) + } else { + orcFieldNames.indexWhere(caseInsensitiveResolution(_, columnName)) + } + } + + def getColumnStatistics(columnName: String): Option[ColumnStatistics] = { + val columnIndex = fileColumnIndex(columnName) + if (columnIndex >= 0) Some(columnsStatistics.get(columnIndex).getStatistics) else None } // Get Min/Max statistics and store as ORC `WritableComparable` format. @@ -657,14 +672,14 @@ object OrcUtils extends Logging { aggregation.aggregateExpressions.zipWithIndex.map { case (max: Max, index) if V2ColumnUtils.extractV2Column(max.column).isDefined => val columnName = V2ColumnUtils.extractV2Column(max.column).get - val statistics = getColumnStatistics(columnName) val dataType = schemaWithoutGroupBy(index).dataType - getMinMaxFromColumnStatistics(statistics, dataType, isMax = true) + getColumnStatistics(columnName) + .map(getMinMaxFromColumnStatistics(_, dataType, isMax = true)).orNull case (min: Min, index) if V2ColumnUtils.extractV2Column(min.column).isDefined => val columnName = V2ColumnUtils.extractV2Column(min.column).get - val statistics = getColumnStatistics(columnName) val dataType = schemaWithoutGroupBy.apply(index).dataType - getMinMaxFromColumnStatistics(statistics, dataType, isMax = false) + getColumnStatistics(columnName) + .map(getMinMaxFromColumnStatistics(_, dataType, isMax = false)).orNull case (count: Count, _) if V2ColumnUtils.extractV2Column(count.column).isDefined => val columnName = V2ColumnUtils.extractV2Column(count.column).get val isPartitionColumn = partitionSchema.fields.map(_.name).contains(columnName) @@ -675,7 +690,7 @@ object OrcUtils extends Logging { val nonNullRowsCount = if (isPartitionColumn) { columnsStatistics.getStatistics.getNumberOfValues } else { - getColumnStatistics(columnName).getNumberOfValues + getColumnStatistics(columnName).map(_.getNumberOfValues).getOrElse(0L) } new LongWritable(nonNullRowsCount) case (_: CountStar, _) => 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 af143955ed70..3b2296698fb3 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 @@ -255,18 +255,18 @@ object ParquetUtils extends Logging { aggSchema: StructType, partitionValues: InternalRow, datetimeRebaseSpec: RebaseSpec): InternalRow = { + // if there are group by columns, we will build result row first, + // and then append group by columns values (partition columns values) to the result row. + val schemaWithoutGroupBy = + AggregatePushDownUtils.getSchemaWithoutGroupingExpression(aggSchema, aggregation) + val (primitiveTypes, values) = getPushedDownAggResult( - footer, filePath, dataSchema, partitionSchema, aggregation) + footer, filePath, partitionSchema, aggregation, schemaWithoutGroupBy) val builder = Types.buildMessage primitiveTypes.foreach(t => builder.addField(t)) val parquetSchema = builder.named("root") - // if there are group by columns, we will build result row first, - // and then append group by columns values (partition columns values) to the result row. - val schemaWithoutGroupBy = - AggregatePushDownUtils.getSchemaWithoutGroupingExpression(aggSchema, aggregation) - val schemaConverter = new ParquetToSparkSchemaConverter val converter = new ParquetRowConverter( schemaConverter, @@ -276,8 +276,11 @@ object ParquetUtils extends Logging { datetimeRebaseSpec, RebaseSpec(LegacyBehaviorPolicy.CORRECTED), NoopUpdater) + // Reset fields to null so a column absent from the file reads as null, not the zero default. + converter.start() val primitiveTypeNames = primitiveTypes.map(_.getPrimitiveTypeName) primitiveTypeNames.zipWithIndex.foreach { + case (_, i) if values(i) == null => case (PrimitiveType.PrimitiveTypeName.BOOLEAN, i) => val v = values(i).asInstanceOf[Boolean] converter.getConverter(i).asPrimitiveConverter.addBoolean(v) @@ -323,54 +326,62 @@ object ParquetUtils extends Logging { private[sql] def getPushedDownAggResult( footer: ParquetMetadata, filePath: String, - dataSchema: StructType, partitionSchema: StructType, - aggregation: Aggregation) + aggregation: Aggregation, + aggSchema: StructType) : (Array[PrimitiveType], Array[Any]) = { val footerFileMetaData = footer.getFileMetaData - val fields = footerFileMetaData.getSchema.getFields + // Resolve by name in the file's own schema; a positional lookup breaks under mergeSchema. + val fileSchema = footerFileMetaData.getSchema + def fileFieldIndex(colName: String): Int = + if (fileSchema.containsField(colName)) fileSchema.getFieldIndex(colName) else -1 val blocks = footer.getBlocks val primitiveTypeBuilder = mutable.ArrayBuilder.make[PrimitiveType] val valuesBuilder = mutable.ArrayBuilder.make[Any] + lazy val sparkToParquet = new SparkToParquetSchemaConverter(SQLConf.get) - aggregation.aggregateExpressions.foreach { agg => + aggregation.aggregateExpressions.zipWithIndex.foreach { case (agg, aggIndex) => var value: Any = None var rowCount = 0L var isCount = false - var index = 0 + var index = -1 var schemaName = "" blocks.forEach { block => val blockMetaData = block.getColumns agg match { case max: Max if V2ColumnUtils.extractV2Column(max.column).isDefined => val colName = V2ColumnUtils.extractV2Column(max.column).get - index = dataSchema.getFieldIndex(colName).getOrElse(-1) + index = fileFieldIndex(colName) schemaName = "max(" + colName + ")" - val currentMax = getCurrentBlockMaxOrMin(filePath, blockMetaData, index, true) - if (value == None || currentMax.asInstanceOf[Comparable[Any]].compareTo(value) > 0) { - value = currentMax + if (index >= 0) { + val currentMax = getCurrentBlockMaxOrMin(filePath, blockMetaData, index, true) + if (value == None || currentMax.asInstanceOf[Comparable[Any]].compareTo(value) > 0) { + value = currentMax + } } case min: Min if V2ColumnUtils.extractV2Column(min.column).isDefined => val colName = V2ColumnUtils.extractV2Column(min.column).get - index = dataSchema.getFieldIndex(colName).getOrElse(-1) + index = fileFieldIndex(colName) schemaName = "min(" + colName + ")" - val currentMin = getCurrentBlockMaxOrMin(filePath, blockMetaData, index, false) - if (value == None || currentMin.asInstanceOf[Comparable[Any]].compareTo(value) < 0) { - value = currentMin + if (index >= 0) { + val currentMin = getCurrentBlockMaxOrMin(filePath, blockMetaData, index, false) + if (value == None || currentMin.asInstanceOf[Comparable[Any]].compareTo(value) < 0) { + value = currentMin + } } case count: Count if V2ColumnUtils.extractV2Column(count.column).isDefined => val colName = V2ColumnUtils.extractV2Column(count.column).get schemaName = "count(" + colName + ")" rowCount += block.getRowCount - var isPartitionCol = false - if (partitionSchema.getFieldIndex(colName).isDefined) { - isPartitionCol = true - } + val isPartitionCol = partitionSchema.getFieldIndex(colName).isDefined isCount = true if (!isPartitionCol) { - index = dataSchema.getFieldIndex(colName).getOrElse(-1) - // Count(*) includes the null values, but Count(colName) doesn't. - rowCount -= getNumNulls(filePath, blockMetaData, index) + index = fileFieldIndex(colName) + if (index >= 0) { + rowCount -= getNumNulls(filePath, blockMetaData, index) + } else { + rowCount -= block.getRowCount + } } case _: CountStar => schemaName = "count(*)" @@ -381,14 +392,21 @@ object ParquetUtils extends Logging { } if (isCount) { valuesBuilder += rowCount - primitiveTypeBuilder += Types.required(PrimitiveTypeName.INT64).named(schemaName); - } else { + primitiveTypeBuilder += Types.required(PrimitiveTypeName.INT64).named(schemaName) + } else if (index >= 0) { valuesBuilder += value - val field = fields.get(index) + val field = fileSchema.getFields.get(index) primitiveTypeBuilder += Types.required(field.asPrimitiveType.getPrimitiveTypeName) .as(field.getLogicalTypeAnnotation) .length(field.asPrimitiveType.getTypeLength) .named(schemaName) + } else { + // Absent column: null result; the type comes from the requested Spark type. + valuesBuilder += null + val sparkType = aggSchema(aggIndex).dataType + val parquetType = sparkToParquet.convertField( + StructField(schemaName, sparkType, nullable = true), inShredded = false) + primitiveTypeBuilder += parquetType.asPrimitiveType() } } (primitiveTypeBuilder.result(), valuesBuilder.result()) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcPartitionReaderFactory.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcPartitionReaderFactory.scala index 6b75b1473bd2..5678615f2467 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcPartitionReaderFactory.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcPartitionReaderFactory.scala @@ -194,7 +194,7 @@ case class OrcPartitionReaderFactory( Utils.tryWithResource(createORCReader(filePath, conf)._1) { reader => OrcUtils.createAggInternalRowFromFooter( reader, filePath.toString, dataSchema, partitionSchema, aggregation.get, - readDataSchema, file.partitionValues) + readDataSchema, file.partitionValues, isCaseSensitive, conf) } } @@ -222,7 +222,7 @@ case class OrcPartitionReaderFactory( Utils.tryWithResource(createORCReader(filePath, conf)._1) { reader => val row = OrcUtils.createAggInternalRowFromFooter( reader, filePath.toString, dataSchema, partitionSchema, aggregation.get, - readDataSchema, file.partitionValues) + readDataSchema, file.partitionValues, isCaseSensitive, conf) AggregatePushDownUtils.convertAggregatesRowToBatch(row, readDataSchema, offHeap = false) } } 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 ac677876c238..a2853ad0db92 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 @@ -773,6 +773,33 @@ class ParquetV2AggregatePushDownSuite extends ParquetAggregatePushDownSuite { } } + test("SPARK-59609: aggregate push-down reads statistics from the correct column " + + "after schema merging") { + Seq("false", "true").foreach { enableVectorizedReader => + withTempPath { dir => + val path = dir.getCanonicalPath + spark.sql("SELECT 1 AS id, 10 AS value, 100 AS other") + .union(spark.sql("SELECT 2 AS id, 20 AS value, 200 AS other")) + .coalesce(1).write.parquet(path + "/with_value") + // Missing `value`; after merging, `other` sits where `value` is in the merged schema. + spark.sql("SELECT 3 AS id, 300 AS other") + .union(spark.sql("SELECT 4 AS id, 400 AS other")) + .coalesce(1).write.parquet(path + "/without_value") + withTempView("t") { + spark.read.option("mergeSchema", "true").option("recursiveFileLookup", "true") + .parquet(path).createOrReplaceTempView("t") + withSQLConf( + aggPushDownEnabledKey -> "true", + vectorizedReaderEnabledKey -> enableVectorizedReader) { + checkAnswer( + sql("SELECT COUNT(*), COUNT(value), MIN(value), MAX(value) FROM t"), + Row(4, 2, 10, 20)) + } + } + } + } + } + // The error originates in the executor-side partition reader, so it may be wrapped in a // higher-level exception. Walk the cause chain to find the structured Spark exception. private def interceptAggPushDownError(query: String): SparkUnsupportedOperationException = { @@ -806,4 +833,32 @@ class OrcV2AggregatePushDownSuite extends OrcAggregatePushDownSuite { override protected def sparkConf: SparkConf = super.sparkConf.set(SQLConf.USE_V1_SOURCE_LIST, "") + + // ORC has no mergeSchema, but a divergent physical column order exercises the same bug. + test("SPARK-59609: aggregate push-down reads statistics from the correct column " + + "when the file column order differs from the read schema") { + Seq("false", "true").foreach { enableVectorizedReader => + withTempPath { dir => + val path = dir.getCanonicalPath + spark.sql("SELECT 1 AS id, 10 AS value, 100 AS other") + .union(spark.sql("SELECT 2 AS id, 20 AS value, 200 AS other")) + .coalesce(1).write.orc(path + "/order1") + // Columns laid out as (id, other, value), so `value` sits at a different physical ordinal. + spark.sql("SELECT 3 AS id, 300 AS other, 30 AS value") + .union(spark.sql("SELECT 4 AS id, 400 AS other, 40 AS value")) + .coalesce(1).write.orc(path + "/order2") + withTempView("t") { + spark.read.schema("id INT, value INT, other INT") + .option("recursiveFileLookup", "true").orc(path).createOrReplaceTempView("t") + withSQLConf( + aggPushDownEnabledKey -> "true", + vectorizedReaderEnabledKey -> enableVectorizedReader) { + checkAnswer( + sql("SELECT COUNT(*), COUNT(value), MIN(value), MAX(value) FROM t"), + Row(4, 4, 10, 40)) + } + } + } + } + } } From a50a139b203ce872dea0c01df7eb2ce02e91257b Mon Sep 17 00:00:00 2001 From: Hemanth Boyina Date: Tue, 22 Sep 2026 11:39:21 +0530 Subject: [PATCH 2/2] fix the review comments --- .../datasources/parquet/ParquetUtils.scala | 12 ++++++---- .../ParquetPartitionReaderFactory.scala | 8 +++---- .../FileSourceAggregatePushDownSuite.scala | 23 +++++++++++++++++++ 3 files changed, 35 insertions(+), 8 deletions(-) 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 3b2296698fb3..642dcd912dbf 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 @@ -37,6 +37,7 @@ import org.apache.spark.internal.Logging import org.apache.spark.internal.LogKeys.{CLASS_NAME, CONFIG} import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.analysis.{caseInsensitiveResolution, caseSensitiveResolution} import org.apache.spark.sql.catalyst.expressions.JoinedRow import org.apache.spark.sql.catalyst.expressions.variant.VariantExpressionEvalUtils import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec @@ -249,11 +250,11 @@ object ParquetUtils extends Logging { private[sql] def createAggInternalRowFromFooter( footer: ParquetMetadata, filePath: String, - dataSchema: StructType, partitionSchema: StructType, aggregation: Aggregation, aggSchema: StructType, partitionValues: InternalRow, + isCaseSensitive: Boolean, datetimeRebaseSpec: RebaseSpec): InternalRow = { // if there are group by columns, we will build result row first, // and then append group by columns values (partition columns values) to the result row. @@ -261,7 +262,7 @@ object ParquetUtils extends Logging { AggregatePushDownUtils.getSchemaWithoutGroupingExpression(aggSchema, aggregation) val (primitiveTypes, values) = getPushedDownAggResult( - footer, filePath, partitionSchema, aggregation, schemaWithoutGroupBy) + footer, filePath, partitionSchema, aggregation, schemaWithoutGroupBy, isCaseSensitive) val builder = Types.buildMessage primitiveTypes.foreach(t => builder.addField(t)) @@ -328,13 +329,16 @@ object ParquetUtils extends Logging { filePath: String, partitionSchema: StructType, aggregation: Aggregation, - aggSchema: StructType) + aggSchema: StructType, + isCaseSensitive: Boolean) : (Array[PrimitiveType], Array[Any]) = { val footerFileMetaData = footer.getFileMetaData // Resolve by name in the file's own schema; a positional lookup breaks under mergeSchema. val fileSchema = footerFileMetaData.getSchema + val resolver = if (isCaseSensitive) caseSensitiveResolution else caseInsensitiveResolution def fileFieldIndex(colName: String): Int = - if (fileSchema.containsField(colName)) fileSchema.getFieldIndex(colName) else -1 + (0 until fileSchema.getFieldCount) + .indexWhere(i => resolver(fileSchema.getFieldName(i), colName)) val blocks = footer.getBlocks val primitiveTypeBuilder = mutable.ArrayBuilder.make[PrimitiveType] val valuesBuilder = mutable.ArrayBuilder.make[Any] diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetPartitionReaderFactory.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetPartitionReaderFactory.scala index 99f978826cc3..6760a86ae748 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetPartitionReaderFactory.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetPartitionReaderFactory.scala @@ -141,8 +141,8 @@ case class ParquetPartitionReaderFactory( if (openedFooter.footer != null && !openedFooter.footer.getBlocks.isEmpty) { ParquetUtils.createAggInternalRowFromFooter(openedFooter.footer, - file.urlEncodedPath, dataSchema, partitionSchema, aggregation.get, - readDataSchema, file.partitionValues, + file.urlEncodedPath, partitionSchema, aggregation.get, + readDataSchema, file.partitionValues, isCaseSensitive, getDatetimeRebaseSpec(openedFooter.footer.getFileMetaData)) } else { null @@ -187,8 +187,8 @@ case class ParquetPartitionReaderFactory( if (openedFooter.footer != null && !openedFooter.footer.getBlocks.isEmpty) { val row = ParquetUtils.createAggInternalRowFromFooter(openedFooter.footer, - file.urlEncodedPath, dataSchema, partitionSchema, aggregation.get, - readDataSchema, file.partitionValues, + file.urlEncodedPath, partitionSchema, aggregation.get, + readDataSchema, file.partitionValues, isCaseSensitive, getDatetimeRebaseSpec(openedFooter.footer.getFileMetaData)) AggregatePushDownUtils.convertAggregatesRowToBatch( row, readDataSchema, enableOffHeapColumnVector && Option(TaskContext.get()).isDefined) 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 a2853ad0db92..042046a2452c 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 @@ -800,6 +800,29 @@ class ParquetV2AggregatePushDownSuite extends ParquetAggregatePushDownSuite { } } + test("SPARK-59609: aggregate push-down resolves the column case-insensitively") { + Seq("false", "true").foreach { enableVectorizedReader => + withTempPath { dir => + val path = dir.getCanonicalPath + // Physical column is `value`, but the read schema names it `VALUE`. + spark.sql("SELECT 1 AS id, 10 AS value") + .union(spark.sql("SELECT 2 AS id, 20 AS value")) + .coalesce(1).write.parquet(path) + withTempView("t") { + spark.read.schema("id INT, VALUE INT").parquet(path).createOrReplaceTempView("t") + withSQLConf( + SQLConf.CASE_SENSITIVE.key -> "false", + aggPushDownEnabledKey -> "true", + vectorizedReaderEnabledKey -> enableVectorizedReader) { + checkAnswer( + sql("SELECT COUNT(VALUE), MIN(VALUE), MAX(VALUE) FROM t"), + Row(2, 10, 20)) + } + } + } + } + } + // The error originates in the executor-side partition reader, so it may be wrapped in a // higher-level exception. Walk the cause chain to find the structured Spark exception. private def interceptAggPushDownError(query: String): SparkUnsupportedOperationException = {