From 349c6bcde67190323980f1dcac4c5a24cb92ad81 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Wed, 22 Jul 2026 12:56:07 +0800 Subject: [PATCH] [SPARK-58252][SQL] Throw GROUP_BY_POS_OUT_OF_RANGE for out-of-range ordinal in pipe AGGREGATE GROUP BY An out-of-range GROUP BY ordinal in the SQL pipe AGGREGATE operator (e.g. `|> aggregate sum(y) group by 5`) leaked a raw java.lang.IndexOutOfBoundsException from `inputs(index - 1)` instead of a proper AnalysisException. This adds the same bounds guard the regular GROUP BY ordinal resolver uses, throwing GROUP_BY_POS_OUT_OF_RANGE (sqlState 42805). The ordinal node is also created with the literal's origin so the error's query context points at the ordinal itself, matching regular GROUP BY. Generated-by: Claude Code (Opus 4.8) --- .../sql/catalyst/analysis/Analyzer.scala | 13 +++-- .../sql/catalyst/parser/AstBuilder.scala | 7 ++- .../analyzer-results/pipe-operators.sql.out | 44 +++++++++++++++++ .../sql-tests/inputs/pipe-operators.sql | 8 ++++ .../sql-tests/results/pipe-operators.sql.out | 48 +++++++++++++++++++ 5 files changed, 115 insertions(+), 5 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index f8e79f9d8da95..2579c824204c9 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala @@ -2229,11 +2229,18 @@ class Analyzer( private def resolvePipeAggregateExpressionOrdinal( expr: NamedExpression, inputs: Seq[Attribute]): NamedExpression = expr match { - case UnresolvedPipeAggregateOrdinal(index) => + case ordinal @ UnresolvedPipeAggregateOrdinal(index) => // In this case, the user applied the SQL pipe aggregate operator ("|> AGGREGATE") and used // ordinals in its GROUP BY clause. This expression then refers to the i-th attribute of the - // child operator (one-based). Here we resolve the ordinal to the corresponding attribute. - inputs(index - 1) + // child operator (one-based). Here we resolve the ordinal to the corresponding attribute, or + // throw GROUP_BY_POS_OUT_OF_RANGE if it is outside the range of the child's attributes. + withPosition(ordinal) { + if (index > 0 && index <= inputs.size) { + inputs(index - 1) + } else { + throw QueryCompilationErrors.groupByPositionRangeError(index, inputs.size) + } + } case other => other } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala index 0d3f678da456a..0bae1eca1f685 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala @@ -7913,9 +7913,12 @@ class AstBuilder extends DataTypeAstBuilder // grouping and aggregate expressions, respectively. This will let the // [[ResolveOrdinalInOrderByAndGroupBy]] rule detect the ordinal in the aggregate list // and replace it with the corresponding attribute from the child operator. - case UnresolvedOrdinal(v: Int) => + case ordinal @ UnresolvedOrdinal(v: Int) => newGroupingExpressions += UnresolvedOrdinal(newAggregateExpressions.length + 1) - newAggregateExpressions += UnresolvedPipeAggregateOrdinal(v) + // Preserve the ordinal's origin so an out-of-range error points at the position + // itself, not the whole pipe statement (matching regular GROUP BY). + newAggregateExpressions += + CurrentOrigin.withOrigin(ordinal.origin) { UnresolvedPipeAggregateOrdinal(v) } case e: Expression => newGroupingExpressions += e newAggregateExpressions += UnresolvedAlias(e, None) diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out index 73e03af34181c..94fbe261ca29c 100644 --- a/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out +++ b/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out @@ -3492,6 +3492,50 @@ org.apache.spark.sql.catalyst.parser.ParseException } +-- !query +select 3 as x, 4 as y +|> aggregate sum(y) group by 5 +-- !query analysis +org.apache.spark.sql.AnalysisException +{ + "errorClass" : "GROUP_BY_POS_OUT_OF_RANGE", + "sqlState" : "42805", + "messageParameters" : { + "index" : "5", + "size" : "2" + }, + "queryContext" : [ { + "objectType" : "", + "objectName" : "", + "startIndex" : 52, + "stopIndex" : 52, + "fragment" : "5" + } ] +} + + +-- !query +select 3 as x, 4 as y +|> aggregate sum(y) group by 0 +-- !query analysis +org.apache.spark.sql.AnalysisException +{ + "errorClass" : "GROUP_BY_POS_OUT_OF_RANGE", + "sqlState" : "42805", + "messageParameters" : { + "index" : "0", + "size" : "2" + }, + "queryContext" : [ { + "objectType" : "", + "objectName" : "", + "startIndex" : 52, + "stopIndex" : 52, + "fragment" : "0" + } ] +} + + -- !query table courseSales |> aggregate sum(earnings) group by rollup(course, `year`) diff --git a/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql b/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql index 7347ad1385a3c..916b114f17f57 100644 --- a/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql +++ b/sql/core/src/test/resources/sql-tests/inputs/pipe-operators.sql @@ -1128,6 +1128,14 @@ table other select 3 as x, 4 as y |> aggregate group by all; +-- The AGGREGATE GROUP BY ordinal position is out of range. +select 3 as x, 4 as y +|> aggregate sum(y) group by 5; + +-- The AGGREGATE GROUP BY ordinal position is not positive. +select 3 as x, 4 as y +|> aggregate sum(y) group by 0; + -- GROUP BY ROLLUP is not supported yet. table courseSales |> aggregate sum(earnings) group by rollup(course, `year`) diff --git a/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out b/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out index 939583e1ddb71..66cf72698a2d1 100644 --- a/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out +++ b/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out @@ -3109,6 +3109,54 @@ org.apache.spark.sql.catalyst.parser.ParseException } +-- !query +select 3 as x, 4 as y +|> aggregate sum(y) group by 5 +-- !query schema +struct<> +-- !query output +org.apache.spark.sql.AnalysisException +{ + "errorClass" : "GROUP_BY_POS_OUT_OF_RANGE", + "sqlState" : "42805", + "messageParameters" : { + "index" : "5", + "size" : "2" + }, + "queryContext" : [ { + "objectType" : "", + "objectName" : "", + "startIndex" : 52, + "stopIndex" : 52, + "fragment" : "5" + } ] +} + + +-- !query +select 3 as x, 4 as y +|> aggregate sum(y) group by 0 +-- !query schema +struct<> +-- !query output +org.apache.spark.sql.AnalysisException +{ + "errorClass" : "GROUP_BY_POS_OUT_OF_RANGE", + "sqlState" : "42805", + "messageParameters" : { + "index" : "0", + "size" : "2" + }, + "queryContext" : [ { + "objectType" : "", + "objectName" : "", + "startIndex" : 52, + "stopIndex" : 52, + "fragment" : "0" + } ] +} + + -- !query table courseSales |> aggregate sum(earnings) group by rollup(course, `year`)