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 2024/08/19 10:22:45 UTC
(spark) branch master updated: [SPARK-49017][SQL] Insert statement fails when multiple parameters are being used
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 11b682cf5b7c [SPARK-49017][SQL] Insert statement fails when multiple parameters are being used
11b682cf5b7c is described below
commit 11b682cf5b7c5360a02410be288b7905eecc1d28
Author: Mihailo Milosevic <mi...@databricks.com>
AuthorDate: Mon Aug 19 12:22:33 2024 +0200
[SPARK-49017][SQL] Insert statement fails when multiple parameters are being used
### What changes were proposed in this pull request?
Fix for multiple parameters support.
### Why are the changes needed?
The use of multiple parameters with identifiers were broken
### Does this PR introduce _any_ user-facing change?
Yes, look at tests.
### How was this patch tested?
New tests added to ParametersSuite.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #47501 from mihailom-db/fixParameters.
Authored-by: Mihailo Milosevic <mi...@databricks.com>
Signed-off-by: Max Gekk <ma...@gmail.com>
---
.../analysis/ResolveIdentifierClause.scala | 4 +-
.../spark/sql/catalyst/analysis/unresolved.scala | 14 +++++-
.../spark/sql/catalyst/parser/AstBuilder.scala | 39 ++++++++++-----
.../sql/catalyst/analysis/AnalysisSuite.scala | 4 +-
.../spark/sql/execution/SparkSqlParser.scala | 4 +-
.../org/apache/spark/sql/ParametersSuite.scala | 57 ++++++++++++++++++++++
6 files changed, 102 insertions(+), 20 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveIdentifierClause.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveIdentifierClause.scala
index e0142c445ae8..0620f37fa0db 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveIdentifierClause.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveIdentifierClause.scala
@@ -36,8 +36,8 @@ class ResolveIdentifierClause(earlyBatches: Seq[RuleExecutor[LogicalPlan]#Batch]
override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
_.containsAnyPattern(UNRESOLVED_IDENTIFIER, UNRESOLVED_IDENTIFIER_WITH_CTE)) {
- case p: PlanWithUnresolvedIdentifier if p.identifierExpr.resolved =>
- executor.execute(p.planBuilder.apply(evalIdentifierExpr(p.identifierExpr)))
+ case p: PlanWithUnresolvedIdentifier if p.identifierExpr.resolved && p.childrenResolved =>
+ executor.execute(p.planBuilder.apply(evalIdentifierExpr(p.identifierExpr), p.children))
case u @ UnresolvedWithCTERelations(p, cteRelations) =>
this.apply(p) match {
case u @ UnresolvedRelation(Seq(table), _, _) =>
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
index f7a987368ec0..eb718eec4256 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala
@@ -60,9 +60,19 @@ trait UnresolvedUnaryNode extends UnaryNode with UnresolvedNode
*/
case class PlanWithUnresolvedIdentifier(
identifierExpr: Expression,
- planBuilder: Seq[String] => LogicalPlan)
- extends UnresolvedLeafNode {
+ children: Seq[LogicalPlan],
+ planBuilder: (Seq[String], Seq[LogicalPlan]) => LogicalPlan)
+ extends UnresolvedNode {
+
+ def this(identifierExpr: Expression, planBuilder: Seq[String] => LogicalPlan) = {
+ this(identifierExpr, Nil, (ident, _) => planBuilder(ident))
+ }
+
final override val nodePatterns: Seq[TreePattern] = Seq(UNRESOLVED_IDENTIFIER)
+
+ override protected def withNewChildrenInternal(
+ newChildren: IndexedSeq[LogicalPlan]): LogicalPlan =
+ copy(identifierExpr, newChildren, planBuilder)
}
/**
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 c6e0467b3aff..0bb4fc9c90d8 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
@@ -71,20 +71,34 @@ class AstBuilder extends DataTypeAstBuilder
builder: Seq[String] => LogicalPlan): LogicalPlan = {
val exprCtx = ctx.expression
if (exprCtx != null) {
- PlanWithUnresolvedIdentifier(withOrigin(exprCtx) { expression(exprCtx) }, builder)
+ PlanWithUnresolvedIdentifier(withOrigin(exprCtx) { expression(exprCtx) }, Nil,
+ (ident, _) => builder(ident))
} else {
builder.apply(visitMultipartIdentifier(ctx.multipartIdentifier))
}
}
+ protected def withIdentClause(
+ ctx: IdentifierReferenceContext,
+ otherPlans: Seq[LogicalPlan],
+ builder: (Seq[String], Seq[LogicalPlan]) => LogicalPlan): LogicalPlan = {
+ val exprCtx = ctx.expression
+ if (exprCtx != null) {
+ PlanWithUnresolvedIdentifier(withOrigin(exprCtx) { expression(exprCtx) }, otherPlans, builder)
+ } else {
+ builder.apply(visitMultipartIdentifier(ctx.multipartIdentifier), otherPlans)
+ }
+ }
+
protected def withFuncIdentClause(
ctx: FunctionNameContext,
- builder: Seq[String] => LogicalPlan): LogicalPlan = {
+ otherPlans: Seq[LogicalPlan],
+ builder: (Seq[String], Seq[LogicalPlan]) => LogicalPlan): LogicalPlan = {
val exprCtx = ctx.expression
if (exprCtx != null) {
- PlanWithUnresolvedIdentifier(withOrigin(exprCtx) { expression(exprCtx) }, builder)
+ PlanWithUnresolvedIdentifier(withOrigin(exprCtx) { expression(exprCtx) }, otherPlans, builder)
} else {
- builder.apply(getFunctionMultiparts(ctx))
+ builder.apply(getFunctionMultiparts(ctx), otherPlans)
}
}
@@ -453,12 +467,12 @@ class AstBuilder extends DataTypeAstBuilder
case table: InsertIntoTableContext =>
val (relationCtx, options, cols, partition, ifPartitionNotExists, byName)
= visitInsertIntoTable(table)
- withIdentClause(relationCtx, ident => {
+ withIdentClause(relationCtx, Seq(query), (ident, otherPlans) => {
val insertIntoStatement = InsertIntoStatement(
createUnresolvedRelation(relationCtx, ident, options),
partition,
cols,
- query,
+ otherPlans.head,
overwrite = false,
ifPartitionNotExists,
byName)
@@ -471,21 +485,21 @@ class AstBuilder extends DataTypeAstBuilder
case table: InsertOverwriteTableContext =>
val (relationCtx, options, cols, partition, ifPartitionNotExists, byName)
= visitInsertOverwriteTable(table)
- withIdentClause(relationCtx, ident => {
+ withIdentClause(relationCtx, Seq(query), (ident, otherPlans) => {
InsertIntoStatement(
createUnresolvedRelation(relationCtx, ident, options),
partition,
cols,
- query,
+ otherPlans.head,
overwrite = true,
ifPartitionNotExists,
byName)
})
case ctx: InsertIntoReplaceWhereContext =>
- withIdentClause(ctx.identifierReference, ident => {
+ withIdentClause(ctx.identifierReference, Seq(query), (ident, otherPlans) => {
OverwriteByExpression.byPosition(
createUnresolvedRelation(ctx.identifierReference, ident, Option(ctx.optionsClause())),
- query,
+ otherPlans.head,
expression(ctx.whereClause().booleanExpression()))
})
case dir: InsertOverwriteDirContext =>
@@ -1841,7 +1855,8 @@ class AstBuilder extends DataTypeAstBuilder
withFuncIdentClause(
func.functionName,
- ident => {
+ Nil,
+ (ident, _) => {
if (ident.length > 1) {
throw QueryParsingErrors.invalidTableValuedFunctionNameError(ident, ctx)
}
@@ -2585,7 +2600,7 @@ class AstBuilder extends DataTypeAstBuilder
} else {
// It's a function call
val funcCtx = ctx.functionName
- val func = withFuncIdentClause(
+ val func: Expression = withFuncIdentClause(
funcCtx,
arguments ++ filter ++ order.toSeq,
(ident, otherExprs) => {
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala
index 62856a96f7ee..1bb1412434c6 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala
@@ -1752,9 +1752,9 @@ class AnalysisSuite extends AnalysisTest with Matchers {
checkAnalysis(testRelation.select(ident2), testRelation.select($"a").analyze)
}
withClue("IDENTIFIER as table") {
- val ident = PlanWithUnresolvedIdentifier(name, _ => testRelation)
+ val ident = new PlanWithUnresolvedIdentifier(name, _ => testRelation)
checkAnalysis(ident.select($"a"), testRelation.select($"a").analyze)
- val ident2 = PlanWithUnresolvedIdentifier(replaceable, _ => testRelation)
+ val ident2 = new PlanWithUnresolvedIdentifier(replaceable, _ => testRelation)
checkAnalysis(ident2.select($"a"), testRelation.select($"a").analyze)
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 8566bd08ec51..8f27a0e8f673 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -571,7 +571,7 @@ class SparkSqlAstBuilder extends AstBuilder {
throw QueryParsingErrors.defineTempViewWithIfNotExistsError(ctx)
}
- withIdentClause(ctx.identifierReference(), ident => {
+ withIdentClause(ctx.identifierReference(), Seq(qPlan), (ident, otherPlans) => {
val tableIdentifier = ident.asTableIdentifier
if (tableIdentifier.database.isDefined) {
// Temporary view names should NOT contain database prefix like "database.table"
@@ -585,7 +585,7 @@ class SparkSqlAstBuilder extends AstBuilder {
visitCommentSpecList(ctx.commentSpec()),
properties,
Option(source(ctx.query)),
- qPlan,
+ otherPlans.head,
ctx.EXISTS != null,
ctx.REPLACE != null,
viewType = viewType)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/ParametersSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/ParametersSuite.scala
index 28ea65522fa9..268e876c282f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/ParametersSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/ParametersSuite.scala
@@ -625,6 +625,63 @@ class ParametersSuite extends QueryTest with SharedSparkSession with PlanTest {
comparePlans(expected, parameterizedSql)
}
+ test("SPARK-49017: bind named parameters with IDENTIFIER clause") {
+ withTable("testtab") {
+ // Create table
+ spark.sql("create table testtab (id int, name string)")
+
+ // Insert into table using single param
+ spark.sql("insert into identifier(:tab) values(1, 'test1')", Map("tab" -> "testtab"))
+
+ // Select from table using param
+ checkAnswer(spark.sql("select * from identifier(:tab)", Map("tab" -> "testtab")),
+ Seq(Row(1, "test1")))
+
+ // Insert into table using multiple params
+ spark.sql("insert into identifier(:tab) values(2, :name)",
+ Map("tab" -> "testtab", "name" -> "test2"))
+
+ // Select from table using param
+ checkAnswer(sql("select * from testtab"), Seq(Row(1, "test1"), Row(2, "test2")))
+
+ // Insert into table using multiple params and idents
+ sql("insert into testtab values(2, 'test3')")
+
+ // Select from table using param
+ checkAnswer(spark.sql("select identifier(:col) from identifier(:tab) where :name == name",
+ Map("tab" -> "testtab", "name" -> "test2", "col" -> "id")), Seq(Row(2)))
+ }
+ }
+
+ test("SPARK-49017: bind positional parameters with IDENTIFIER clause") {
+ withTable("testtab") {
+ // Create table
+ spark.sql("create table testtab (id int, name string)")
+
+ // Insert into table using single param
+ spark.sql("insert into identifier(?) values(1, 'test1')",
+ Array("testtab"))
+
+ // Select from table using param
+ checkAnswer(spark.sql("select * from identifier(?)", Array("testtab")),
+ Seq(Row(1, "test1")))
+
+ // Insert into table using multiple params
+ spark.sql("insert into identifier(?) values(2, ?)",
+ Array("testtab", "test2"))
+
+ // Select from table using param
+ checkAnswer(sql("select * from testtab"), Seq(Row(1, "test1"), Row(2, "test2")))
+
+ // Insert into table using multiple params and idents
+ sql("insert into testtab values(2, 'test3')")
+
+ // Select from table using param
+ checkAnswer(spark.sql("select identifier(?) from identifier(?) where ? == name",
+ Array("id", "testtab", "test2")), Seq(Row(2)))
+ }
+ }
+
test("SPARK-46999: bind parameters for nested IDENTIFIER clause") {
val query = sql(
"""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org