You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/28 16:40:03 UTC
(spark) branch master updated: [SPARK-58933][SQL] Simplify and fix parsed INSERT target handling
This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 900b44e22f03 [SPARK-58933][SQL] Simplify and fix parsed INSERT target handling
900b44e22f03 is described below
commit 900b44e22f03bdbf991038dbc97a48dab5c7ecb9
Author: Wenchen Fan <we...@databricks.com>
AuthorDate: Sat Aug 29 00:39:44 2026 +0800
[SPARK-58933][SQL] Simplify and fix parsed INSERT target handling
### What changes were proposed in this pull request?
This follow-up to #58204 simplifies and corrects parsed INSERT target handling:
* Make the SQL parser always produce `UnresolvedInsert` for table-target INSERT statements. A
static target is represented by `UnresolvedInsertTarget`, while a dynamic `IDENTIFIER`
expression remains in `PlanWithUnresolvedIdentifier`. The analyzer lowers either form to
`InsertIntoStatement` once the target identifier is ready for relation resolution.
* Update parser-oriented consumers, including SQL statement classification, `parse_sql` lineage
collection, and pipeline flow registration, to handle the single parsed INSERT shape.
`InsertIntoStatement` handling remains where plans can be created programmatically or have
already been lowered by analysis.
* Scope the `parse_sql` CTE-shadow exemption specifically to `UnresolvedInsertTarget`. DELETE,
UPDATE, and MERGE targets continue to follow CTE substitution semantics.
* Run early dynamic INSERT target resolution inside the analysis planning tracker and report its
failures through `QueryPlanningTracker.setAnalysisFailed`.
### Why are the changes needed?
The parser previously produced `InsertIntoStatement` for static targets and `UnresolvedInsert` for
dynamic targets. Parser consumers therefore had to understand both shapes even though
`UnresolvedInsert` is only an intermediate node and is lowered immediately after its target is
ready. Using one parsed representation makes that boundary explicit and removes duplicated
matching logic.
The target-role exemption added by #58204 was also broader than required. A CTE-shadowed DELETE,
UPDATE, or MERGE target could be reported as a catalog target even though CTE substitution replaces
that relation. In addition, early dynamic target resolution was absent from analysis timing and
failure reporting.
### Does this PR introduce _any_ user-facing change?
Yes, within unreleased master only. `parse_sql` no longer reports CTE-shadowed DELETE, UPDATE, or
MERGE targets as catalog target-table references. There is no change relative to a released Spark
version.
### How was this patch tested?
Added and updated regression coverage, then ran:
```bash
build/sbt \
"catalyst/testOnly org.apache.spark.sql.catalyst.parser.PlanParserSuite org.apache.spark.sql.catalyst.parser.DDLParserSuite org.apache.spark.sql.catalyst.parser.IdentifierClauseParserSuite" \
"sql/testOnly org.apache.spark.sql.ParametersSuite org.apache.spark.sql.catalyst.parser.ParseSqlResultSuite org.apache.spark.sql.execution.QueryExecutionSuite org.apache.spark.sql.execution.command.v2.CreateFlowCommandSuite" \
pipelines/compile
```
All 284 Catalyst tests and 204 SQL tests passed, and the pipelines module compiled successfully.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
Closes #58378 from cloud-fan/fix-identifier-dml-followup.
Authored-by: Wenchen Fan <we...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/catalyst/analysis/Analyzer.scala | 4 +-
.../spark/sql/catalyst/parser/AstBuilder.scala | 72 ++++++++--------------
.../sql/catalyst/plans/logical/statements.scala | 9 +--
.../spark/sql/catalyst/parser/DDLParserSuite.scala | 72 ++++++++++------------
.../parser/IdentifierClauseParserSuite.scala | 6 +-
.../sql/catalyst/parser/PlanParserSuite.scala | 30 ++++-----
.../spark/sql/catalyst/parser/ParseSqlResult.scala | 10 +--
.../sql/catalyst/parser/SqlStatementCodes.scala | 2 +-
.../spark/sql/execution/QueryExecution.scala | 13 +++-
.../sql-tests/analyzer-results/explain-aqe.sql.out | 2 +-
.../sql-tests/analyzer-results/explain.sql.out | 2 +-
.../sql-tests/results/explain-aqe.sql.out | 3 +-
.../resources/sql-tests/results/explain.sql.out | 3 +-
.../org/apache/spark/sql/ParametersSuite.scala | 14 +++--
.../sql/catalyst/parser/ParseSqlResultSuite.scala | 10 +++
.../spark/sql/execution/QueryExecutionSuite.scala | 40 ++++++++++++
.../command/v2/CreateFlowCommandSuite.scala | 12 ++--
.../graph/SqlGraphRegistrationContext.scala | 24 ++++----
18 files changed, 181 insertions(+), 147 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 ce9736aa8a3d..952f17738d23 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
@@ -401,7 +401,7 @@ class Analyzer(
relation
}
- /** Resolves only the dynamic identifier of an INSERT so transaction detection can inspect it. */
+ /** Lowers a parsed INSERT enough for transaction detection to inspect its target. */
private[sql] def resolveUnresolvedInsert(insert: UnresolvedInsert): LogicalPlan = {
runWithSessionConf {
val identifierResolvedTarget = insert.table match {
@@ -1379,7 +1379,7 @@ class Analyzer(
}
}
- /** Lower an INSERT as soon as its target identifier expression has been evaluated. */
+ /** Lower a parsed INSERT as soon as its target identifier expression has been evaluated. */
object ResolveUnresolvedInsert extends Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
_.containsPattern(UNRESOLVED_INSERT), ruleId) {
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 67ae96dd4f9e..966e3e51cc8d 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
@@ -939,28 +939,22 @@ class AstBuilder extends DataTypeAstBuilder
query: LogicalPlan,
queryAliasCtx: TableAliasContext): LogicalPlan = withOrigin(ctx) {
ctx match {
- // A dynamic target is exposed only by the temporary UnresolvedInsert plan. Once its
- // identifier expression has been evaluated, analysis continues with InsertIntoStatement.
case table: InsertIntoTableContext =>
val insertParams = visitInsertIntoTable(table)
- val privileges = Set(TableWritePrivilege.INSERT)
- createInsertIntoStatement(
+ createUnresolvedInsert(
insertParams = insertParams,
- tableSlot = buildWriteTableSlot(
- insertParams.relationCtx, insertParams.options, privileges),
query = query,
overwrite = false,
- withSchemaEvolution = table.EVOLUTION() != null)
+ withSchemaEvolution = table.EVOLUTION() != null,
+ writePrivileges = Set(TableWritePrivilege.INSERT))
case table: InsertOverwriteTableContext =>
val insertParams = visitInsertOverwriteTable(table)
- val privileges = Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE)
- createInsertIntoStatement(
+ createUnresolvedInsert(
insertParams = insertParams,
- tableSlot = buildWriteTableSlot(
- insertParams.relationCtx, insertParams.options, privileges),
query = query,
overwrite = true,
- withSchemaEvolution = table.EVOLUTION() != null)
+ withSchemaEvolution = table.EVOLUTION() != null,
+ writePrivileges = Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
case ctx: InsertIntoReplaceBooleanCondContext =>
// Although REPLACE WHERE and REPLACE ON share a unified grammar rule, they have
// different SQL semantics:
@@ -971,17 +965,14 @@ class AstBuilder extends DataTypeAstBuilder
val isInsertReplaceWhere = ctx.WHERE() != null
if (isInsertReplaceWhere) {
val insertParams = visitInsertIntoReplaceWhere(ctx)
- val privileges = Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE)
- createInsertIntoStatement(
+ createUnresolvedInsert(
insertParams = insertParams,
- tableSlot = buildWriteTableSlot(
- insertParams.relationCtx, insertParams.options, privileges),
query = query,
overwrite = true,
- withSchemaEvolution = ctx.EVOLUTION() != null)
+ withSchemaEvolution = ctx.EVOLUTION() != null,
+ writePrivileges = Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
} else {
val insertParams = visitInsertIntoReplaceOn(ctx)
- val privileges = Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE)
val finalQuery = {
val queryAliasOpt =
getTableAliasWithoutColumnAlias(queryAliasCtx, "INSERT REPLACE ON")
@@ -991,24 +982,21 @@ class AstBuilder extends DataTypeAstBuilder
}
}.getOrElse(query)
}
- createInsertIntoStatement(
+ createUnresolvedInsert(
insertParams = insertParams,
- tableSlot = buildWriteTableSlot(
- insertParams.relationCtx, insertParams.options, privileges),
query = finalQuery,
overwrite = true,
- withSchemaEvolution = ctx.EVOLUTION() != null)
+ withSchemaEvolution = ctx.EVOLUTION() != null,
+ writePrivileges = Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
}
case ctx: InsertIntoReplaceUsingContext =>
val insertParams = visitInsertIntoReplaceUsing(ctx)
- val privileges = Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE)
- createInsertIntoStatement(
+ createUnresolvedInsert(
insertParams = insertParams,
- tableSlot = buildWriteTableSlot(
- insertParams.relationCtx, insertParams.options, privileges),
query = query,
overwrite = true,
- withSchemaEvolution = ctx.EVOLUTION() != null)
+ withSchemaEvolution = ctx.EVOLUTION() != null,
+ writePrivileges = Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE))
case dir: InsertOverwriteDirContext =>
val (isLocal, storage, provider) = visitInsertOverwriteDir(dir)
InsertIntoDir(isLocal, storage, provider, query, overwrite = true)
@@ -1168,15 +1156,16 @@ class AstBuilder extends DataTypeAstBuilder
replaceCriteriaOpt = replaceCriteriaOpt)
}
- /** Creates an INSERT plan from [[InsertTableParams]]. */
- private def createInsertIntoStatement(
+ /** Creates an unresolved INSERT plan from [[InsertTableParams]]. */
+ private def createUnresolvedInsert(
insertParams: InsertTableParams,
- tableSlot: LogicalPlan,
query: LogicalPlan,
overwrite: Boolean,
- withSchemaEvolution: Boolean): LogicalPlan = {
- val unresolvedInsert = UnresolvedInsert(
- table = tableSlot,
+ withSchemaEvolution: Boolean,
+ writePrivileges: Set[TableWritePrivilege]): UnresolvedInsert = {
+ UnresolvedInsert(
+ table = buildWriteTableSlot(
+ insertParams.relationCtx, insertParams.options, writePrivileges),
partitionSpec = insertParams.partitionSpec,
userSpecifiedCols = insertParams.userSpecifiedCols,
query = query,
@@ -1185,23 +1174,12 @@ class AstBuilder extends DataTypeAstBuilder
byName = insertParams.byName,
replaceCriteriaOpt = insertParams.replaceCriteriaOpt,
withSchemaEvolution = withSchemaEvolution)
-
- tableSlot match {
- case target: UnresolvedInsertTarget =>
- val relation = CurrentOrigin.withOrigin(target.origin) {
- new UnresolvedRelation(target.multipartIdentifier, target.options, isStreaming = false)
- .requireWritePrivileges(target.writePrivileges)
- }
- relation.copyTagsFrom(target)
- unresolvedInsert.toInsertIntoStatement(relation)
- case _ => unresolvedInsert
- }
}
/**
- * Build the target of an INSERT. A dynamic IDENTIFIER remains inside `UnresolvedInsert` so its
- * expression receives normal analyzer traversal. The builder produces `UnresolvedInsertTarget`
- * rather than `UnresolvedRelation` so CTE substitution cannot interpret the target as a source.
+ * Build the target of an INSERT. An `IDENTIFIER` expression receives normal analyzer traversal.
+ * A resolved identifier is kept in `UnresolvedInsertTarget` rather than `UnresolvedRelation` so
+ * CTE substitution cannot interpret the target as a source.
*/
private def buildWriteTableSlot(
ctx: IdentifierReferenceContext,
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala
index 75a06a39b841..85327eccae22 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala
@@ -158,11 +158,12 @@ case class QualifiedColType(
}
/**
- * A temporary INSERT plan whose target identifier expression still needs analyzer traversal.
+ * An INSERT statement as parsed from SQL.
*
- * The target is exposed as a child only until it becomes an `UnresolvedInsertTarget`. The analyzer
- * then converts this node to the unary [[InsertIntoStatement]], before resolving the target as a
- * relation.
+ * The target is exposed as a child until the analyzer converts this node to the unary
+ * [[InsertIntoStatement]]. A static target is represented by `UnresolvedInsertTarget`, while a
+ * dynamic target keeps its identifier expression in `PlanWithUnresolvedIdentifier` until that
+ * expression is resolved.
*/
case class UnresolvedInsert(
table: LogicalPlan,
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
index d85cb0b93133..eac592f28ae3 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala
@@ -46,11 +46,7 @@ class DDLParserSuite extends AnalysisTest {
// We don't care the write privileges in this suite.
val parsed = parsePlan(sql).transform {
case u: UnresolvedRelation => u.clearWritePrivileges
- case i: InsertIntoStatement =>
- i.table match {
- case u: UnresolvedRelation => i.copy(table = u.clearWritePrivileges)
- case _ => i
- }
+ case u: UnresolvedInsertTarget => UnresolvedRelation(u.multipartIdentifier, u.options)
case o: OverwriteByExpression =>
o.table match {
case u: UnresolvedRelation => o.copy(table = u.clearWritePrivileges)
@@ -1629,7 +1625,7 @@ class DDLParserSuite extends AnalysisTest {
"INSERT INTO testcat.ns1.ns2.tbl SELECT * FROM source"
).foreach { sql =>
parseCompare(sql,
- InsertIntoStatement(
+ UnresolvedInsert(
UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
Map.empty,
Nil,
@@ -1644,7 +1640,7 @@ class DDLParserSuite extends AnalysisTest {
"INSERT INTO testcat.ns1.ns2.tbl (a, b) SELECT * FROM source"
).foreach { sql =>
parseCompare(sql,
- InsertIntoStatement(
+ UnresolvedInsert(
UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
Map.empty,
Seq("a", "b"),
@@ -1655,7 +1651,7 @@ class DDLParserSuite extends AnalysisTest {
test("insert table: append from another catalog") {
parseCompare("INSERT INTO TABLE testcat.ns1.ns2.tbl SELECT * FROM testcat2.db.tbl",
- InsertIntoStatement(
+ UnresolvedInsert(
UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
Map.empty,
Nil,
@@ -1670,7 +1666,7 @@ class DDLParserSuite extends AnalysisTest {
|PARTITION (p1 = 3, p2)
|SELECT * FROM source
""".stripMargin,
- InsertIntoStatement(
+ UnresolvedInsert(
UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
Map("p1" -> Some("3"), "p2" -> None),
Nil,
@@ -1685,7 +1681,7 @@ class DDLParserSuite extends AnalysisTest {
|PARTITION (p1 = 3, p2) (a, b)
|SELECT * FROM source
""".stripMargin,
- InsertIntoStatement(
+ UnresolvedInsert(
UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
Map("p1" -> Some("3"), "p2" -> None),
Seq("a", "b"),
@@ -1699,7 +1695,7 @@ class DDLParserSuite extends AnalysisTest {
"INSERT OVERWRITE testcat.ns1.ns2.tbl SELECT * FROM source"
).foreach { sql =>
parseCompare(sql,
- InsertIntoStatement(
+ UnresolvedInsert(
UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
Map.empty,
Nil,
@@ -1714,7 +1710,7 @@ class DDLParserSuite extends AnalysisTest {
"INSERT OVERWRITE testcat.ns1.ns2.tbl (a, b) SELECT * FROM source"
).foreach { sql =>
parseCompare(sql,
- InsertIntoStatement(
+ UnresolvedInsert(
UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
Map.empty,
Seq("a", "b"),
@@ -1730,7 +1726,7 @@ class DDLParserSuite extends AnalysisTest {
|PARTITION (p1 = 3, p2)
|SELECT * FROM source
""".stripMargin,
- InsertIntoStatement(
+ UnresolvedInsert(
UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
Map("p1" -> Some("3"), "p2" -> None),
Nil,
@@ -1745,7 +1741,7 @@ class DDLParserSuite extends AnalysisTest {
|PARTITION (p1 = 3, p2) (a, b)
|SELECT * FROM source
""".stripMargin,
- InsertIntoStatement(
+ UnresolvedInsert(
UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
Map("p1" -> Some("3"), "p2" -> None),
Seq("a", "b"),
@@ -1760,7 +1756,7 @@ class DDLParserSuite extends AnalysisTest {
|PARTITION (p1 = 3) IF NOT EXISTS
|SELECT * FROM source
""".stripMargin,
- InsertIntoStatement(
+ UnresolvedInsert(
UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
Map("p1" -> Some("3")),
Nil,
@@ -1810,7 +1806,7 @@ class DDLParserSuite extends AnalysisTest {
"INSERT INTO testcat.ns1.ns2.tbl BY NAME SELECT * FROM source"
).foreach { sql =>
parseCompare(sql,
- InsertIntoStatement(
+ UnresolvedInsert(
UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
Map.empty,
Nil,
@@ -1823,7 +1819,7 @@ class DDLParserSuite extends AnalysisTest {
"INSERT OVERWRITE testcat.ns1.ns2.tbl BY NAME SELECT * FROM source"
).foreach { sql =>
parseCompare(sql,
- InsertIntoStatement(
+ UnresolvedInsert(
UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
Map.empty,
Nil,
@@ -1846,7 +1842,7 @@ class DDLParserSuite extends AnalysisTest {
test("insert table: REPLACE WHERE with BY NAME") {
parseCompare(
"INSERT INTO testcat.ns1.ns2.tbl BY NAME REPLACE WHERE a > 5 SELECT * FROM source",
- InsertIntoStatement(
+ UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Nil,
@@ -1861,7 +1857,7 @@ class DDLParserSuite extends AnalysisTest {
test("insert table: REPLACE WHERE without BY NAME") {
parseCompare(
"INSERT INTO testcat.ns1.ns2.tbl REPLACE WHERE a > 5 SELECT * FROM source",
- InsertIntoStatement(
+ UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Nil,
@@ -1910,7 +1906,7 @@ class DDLParserSuite extends AnalysisTest {
test("insert table: REPLACE WHERE with column list") {
parseCompare(
"INSERT INTO testcat.ns1.ns2.tbl (a, b) REPLACE WHERE a > 5 SELECT * FROM source",
- InsertIntoStatement(
+ UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Seq("a", "b"),
@@ -1961,7 +1957,7 @@ class DDLParserSuite extends AnalysisTest {
parseCompare(
"INSERT WITH SCHEMA EVOLUTION INTO testcat.ns1.ns2.tbl (a, b) " +
"REPLACE WHERE a > 5 SELECT * FROM source",
- InsertIntoStatement(
+ UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Seq("a", "b"),
@@ -1978,7 +1974,7 @@ class DDLParserSuite extends AnalysisTest {
parseCompare(
"INSERT INTO testcat.ns1.ns2.tbl WITH (key = 'value') (a, b) " +
"REPLACE WHERE a > 5 SELECT * FROM source",
- InsertIntoStatement(
+ UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl"), opts),
partitionSpec = Map.empty,
userSpecifiedCols = Seq("a", "b"),
@@ -2010,7 +2006,7 @@ class DDLParserSuite extends AnalysisTest {
parseCompare(
sql = insertSQLStmt,
- expected = InsertIntoStatement(
+ expected = UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = userSpecifiedCols,
@@ -2031,7 +2027,7 @@ class DDLParserSuite extends AnalysisTest {
parseCompare(
sql = insertSQLStmt,
- expected = InsertIntoStatement(
+ expected = UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = userSpecifiedCols,
@@ -2059,7 +2055,7 @@ class DDLParserSuite extends AnalysisTest {
parseCompare(
sql = insertSQLStmt,
- expected = InsertIntoStatement(
+ expected = UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Seq.empty,
@@ -2085,7 +2081,7 @@ class DDLParserSuite extends AnalysisTest {
parseCompare(
sql = insertSQLStmt,
- expected = InsertIntoStatement(
+ expected = UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Seq.empty,
@@ -2105,7 +2101,7 @@ class DDLParserSuite extends AnalysisTest {
parseCompare(
sql = insertSQLStmt,
- expected = InsertIntoStatement(
+ expected = UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Seq.empty,
@@ -2127,7 +2123,7 @@ class DDLParserSuite extends AnalysisTest {
parseCompare(
sql = insertSQLStmt,
- expected = InsertIntoStatement(
+ expected = UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Seq.empty,
@@ -2147,7 +2143,7 @@ class DDLParserSuite extends AnalysisTest {
parseCompare(
sql = insertSQLStmt,
- expected = InsertIntoStatement(
+ expected = UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Seq.empty,
@@ -2167,7 +2163,7 @@ class DDLParserSuite extends AnalysisTest {
val table = "testcat.ns1.ns2.tbl"
parseCompare(
sql = s"INSERT INTO $table REPLACE ON col1 = col2 SELECT * FROM source",
- expected = InsertIntoStatement(
+ expected = UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Seq.empty,
@@ -2201,7 +2197,7 @@ class DDLParserSuite extends AnalysisTest {
s"""INSERT INTO $table AS t
|REPLACE ON (t.a = s.b)
|(SELECT * FROM source) AS s""".stripMargin,
- expected = InsertIntoStatement(
+ expected = UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Seq.empty,
@@ -2228,7 +2224,7 @@ class DDLParserSuite extends AnalysisTest {
s"""INSERT INTO $table AS t
|REPLACE USING (col1, col2)
|(SELECT * FROM source) AS s""".stripMargin,
- expected = InsertIntoStatement(
+ expected = UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Seq.empty,
@@ -2248,7 +2244,7 @@ class DDLParserSuite extends AnalysisTest {
"""INSERT INTO testcat.ns1.ns2.tbl
|REPLACE WHERE a > 5
|(SELECT * FROM source) AS s""".stripMargin,
- expected = InsertIntoStatement(
+ expected = UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Nil,
@@ -2264,7 +2260,7 @@ class DDLParserSuite extends AnalysisTest {
parseCompare(
sql = s"INSERT INTO $table AS t REPLACE ON t.a = s.a AND t.b = s.b " +
"(SELECT * FROM source) AS s",
- expected = InsertIntoStatement(
+ expected = UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Seq.empty,
@@ -2289,7 +2285,7 @@ class DDLParserSuite extends AnalysisTest {
val table = "testcat.ns1.ns2.tbl"
parseCompare(
sql = s"INSERT INTO $table AS t REPLACE USING (id) SELECT * FROM source",
- expected = InsertIntoStatement(
+ expected = UnresolvedInsert(
table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")),
partitionSpec = Map.empty,
userSpecifiedCols = Seq.empty,
@@ -3384,8 +3380,8 @@ class DDLParserSuite extends AnalysisTest {
}
test("SPARK-33474: Support typed literals as partition spec values") {
- def insertPartitionPlan(part: String, optimizeInsertIntoCmds: Boolean): InsertIntoStatement = {
- InsertIntoStatement(
+ def insertPartitionPlan(part: String, optimizeInsertIntoCmds: Boolean): UnresolvedInsert = {
+ UnresolvedInsert(
UnresolvedRelation(Seq("t")),
Map("part" -> Some(part)),
Seq.empty[String],
@@ -3536,7 +3532,7 @@ class DDLParserSuite extends AnalysisTest {
UnresolvedAttribute("DEFAULT"))))))
parseCompare(
"INSERT INTO t PARTITION(part = date'2019-01-02') VALUES ('a', DEFAULT)",
- InsertIntoStatement(
+ UnresolvedInsert(
UnresolvedRelation(Seq("t")),
Map("part" -> Some("2019-01-02")),
userSpecifiedCols = Seq.empty[String],
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/IdentifierClauseParserSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/IdentifierClauseParserSuite.scala
index d62cdf1f3c15..a08e2daf4d49 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/IdentifierClauseParserSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/IdentifierClauseParserSuite.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.parser
import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, ExpressionWithUnresolvedIdentifier, UnresolvedAttribute, UnresolvedExtractValue, UnresolvedFunction, UnresolvedInlineTable, UnresolvedStar}
import org.apache.spark.sql.catalyst.expressions.{Alias, Cast, LambdaFunction, Literal, UnresolvedNamedLambdaVariable}
import org.apache.spark.sql.catalyst.parser.CatalystSqlParser.parsePlan
-import org.apache.spark.sql.catalyst.plans.logical.{InsertIntoStatement, OneRowRelation, Pivot, Project, SubqueryAlias, Unpivot}
+import org.apache.spark.sql.catalyst.plans.logical.{OneRowRelation, Pivot, Project, SubqueryAlias, Unpivot, UnresolvedInsert}
import org.apache.spark.sql.catalyst.util.EvaluateUnresolvedInlineTable
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
@@ -179,13 +179,13 @@ class IdentifierClauseParserSuite extends AnalysisTest {
test("Partition spec with IDENTIFIER() for partition column name") {
val plan = parsePlan(
"INSERT INTO partition_spec_test PARTITION (IDENTIFIER('c2') = 'value1') VALUES (1)")
- .asInstanceOf[InsertIntoStatement]
+ .asInstanceOf[UnresolvedInsert]
val values = EvaluateUnresolvedInlineTable.evaluate(
UnresolvedInlineTable(Seq("col1"), Seq(Seq(Literal(1)))))
comparePlans(
plan,
- InsertIntoStatement(
+ UnresolvedInsert(
plan.table,
Map("c2" -> Some("value1")),
Nil,
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
index a0fd03ff00c0..5af549bf3c1b 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
@@ -22,7 +22,7 @@ import scala.annotation.nowarn
import org.apache.spark.SparkThrowable
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier}
-import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, RelationChanges, RelationTimeTravel, UnresolvedAlias, UnresolvedAttribute, UnresolvedFunction, UnresolvedGenerator, UnresolvedInlineTable, UnresolvedRelation, UnresolvedStar, UnresolvedSubqueryColumnAliases, UnresolvedTableValuedFunction, UnresolvedTVFAliases}
+import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, RelationChanges, RelationTimeTravel, UnresolvedAlias, UnresolvedAttribute, UnresolvedFunction, UnresolvedGenerator, UnresolvedInlineTable, UnresolvedInsertTarget, UnresolvedRelation, UnresolvedStar, UnresolvedSubqueryColumnAliases, UnresolvedTableValuedFunction, UnresolvedTVFAliases}
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.logical._
@@ -47,11 +47,7 @@ class PlanParserSuite extends AnalysisTest {
// We don't care the write privileges in this suite.
val parsed = parsePlan(sqlCommand).transform {
case u: UnresolvedRelation => u.clearWritePrivileges
- case i: InsertIntoStatement =>
- i.table match {
- case u: UnresolvedRelation => i.copy(table = u.clearWritePrivileges)
- case _ => i
- }
+ case u: UnresolvedInsertTarget => UnresolvedRelation(u.multipartIdentifier, u.options)
}
comparePlans(parsed, plan, checkAnalysis = false)
}
@@ -60,6 +56,11 @@ class PlanParserSuite extends AnalysisTest {
super.parseException(parsePlan)(sqlText)
}
+ private def unresolvedInsertInto(tableName: String, query: LogicalPlan): LogicalPlan = {
+ UnresolvedInsert(
+ table(tableName), Map.empty, Nil, query, overwrite = false, ifPartitionNotExists = false)
+ }
+
private def cte(
plan: LogicalPlan,
allowRecursion: Boolean,
@@ -359,8 +360,8 @@ class PlanParserSuite extends AnalysisTest {
parameters = Map("error" -> "'from'", "hint" -> ""))
assertEqual(
"from a insert into tbl1 select * insert into tbl2 select * where s < 10",
- table("a").select(star()).insertInto("tbl1").union(
- table("a").where($"s" < 10).select(star()).insertInto("tbl2")))
+ unresolvedInsertInto("tbl1", table("a").select(star())).union(
+ unresolvedInsertInto("tbl2", table("a").where($"s" < 10).select(star()))))
assertEqual(
"select * from (from a select * select *)",
table("a").select(star())
@@ -458,7 +459,7 @@ class PlanParserSuite extends AnalysisTest {
partition: Map[String, Option[String]],
overwrite: Boolean = false,
ifPartitionNotExists: Boolean = false): LogicalPlan =
- InsertIntoStatement(table("s"), partition, Nil, plan, overwrite, ifPartitionNotExists)
+ UnresolvedInsert(table("s"), partition, Nil, plan, overwrite, ifPartitionNotExists)
// Single inserts
assertEqual(s"insert overwrite table s $sql",
@@ -473,7 +474,7 @@ class PlanParserSuite extends AnalysisTest {
// Multi insert
val plan2 = table("t").where($"x" > 5).select(star())
assertEqual("from t insert into s select * limit 1 insert into u select * where x > 5",
- plan.limit(1).insertInto("s").union(plan2.insertInto("u")))
+ unresolvedInsertInto("s", plan.limit(1)).union(unresolvedInsertInto("u", plan2)))
}
test("aggregation") {
@@ -579,11 +580,10 @@ class PlanParserSuite extends AnalysisTest {
|select *
|where s < 10
""".stripMargin,
- Union(from
+ Union(unresolvedInsertInto("t2", from
.generate(jsonTuple, alias = Some("jtup"), outputNames = Seq("q", "z"))
- .select(star())
- .insertInto("t2"),
- from.where($"s" < 10).select(star()).insertInto("t3")))
+ .select(star())),
+ unresolvedInsertInto("t3", from.where($"s" < 10).select(star()))))
// Unresolved generator.
val expected = table("t")
@@ -1696,7 +1696,7 @@ class PlanParserSuite extends AnalysisTest {
assertEqual(
"INSERT INTO s SELECT /*+ REPARTITION(100), COALESCE(500), COALESCE(10) */ * FROM t",
- InsertIntoStatement(table("s"), Map.empty, Nil,
+ UnresolvedInsert(table("s"), Map.empty, Nil,
UnresolvedHint("REPARTITION", Seq(Literal(100)),
UnresolvedHint("COALESCE", Seq(Literal(500)),
UnresolvedHint("COALESCE", Seq(Literal(10)),
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala
index 3b3a0604d739..fbc4b96f9dd7 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResult.scala
@@ -226,8 +226,6 @@ object ParseSqlResult {
case i: UnresolvedInsert =>
visitPlan(i.table, scope, TableRefRole.Target)(f)
visitPlan(i.query, nextScope, TableRefRole.Source)(f)
- case i: InsertIntoStatement =>
- visitPlan(i.query, nextScope, TableRefRole.Source)(f)
case c: CreateTableAsSelect =>
visitPlan(c.name, scope, TableRefRole.Target)(f)
visitPlan(c.query, nextScope, TableRefRole.Source)(f)
@@ -271,8 +269,6 @@ object ParseSqlResult {
visitPlan(ctePlan, bodyScope, TableRefRole.Source)(f)
definitionScope += normalized
}
- case i: InsertIntoStatement =>
- visitPlan(i.table, scope, TableRefRole.Target)(f)
case c: CacheTable if c.multipartIdentifier.isEmpty =>
visitPlan(c.table, scope, TableRefRole.Target)(f)
case c: CompoundBody =>
@@ -318,7 +314,7 @@ object ParseSqlResult {
}
def addTable(parts: Seq[String], scope: CteScope, role: TableRefRole): Unit = {
- if (parts.nonEmpty && (role == TableRefRole.Target || !isCteName(parts, scope))) {
+ if (parts.nonEmpty && !isCteName(parts, scope)) {
role match {
case TableRefRole.Target => targetTables += parts
case TableRefRole.Source => sourceTables += parts
@@ -345,7 +341,7 @@ object ParseSqlResult {
foreachExpressionDeep(p)(visitExpr)
def add(parts: Seq[String]): Unit = addTable(parts, scope, role)
p match {
- case u: UnresolvedInsertTarget => add(u.multipartIdentifier)
+ case u: UnresolvedInsertTarget => addTarget(u.multipartIdentifier)
case u: UnresolvedRelation => add(u.multipartIdentifier)
case u: UnresolvedTable => add(u.multipartIdentifier)
case u: UnresolvedView => add(u.multipartIdentifier)
@@ -392,8 +388,6 @@ object ParseSqlResult {
case UnresolvedWith(child, _, _) => primaryQueryPlan(child)
case UnresolvedInsert(_, _, _, query, _, _, _, _, _) =>
primaryQueryPlan(query)
- case InsertIntoStatement(_, _, _, query, _, _, _, _, _) =>
- primaryQueryPlan(query)
case c: CreateTableAsSelect => primaryQueryPlan(c.query)
case r: ReplaceTableAsSelect => primaryQueryPlan(r.query)
case c: CreateView => primaryQueryPlan(c.query)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala
index 24cfdfcd2eb4..1a350baa60b4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala
@@ -132,7 +132,7 @@ object SqlStatementCodes {
def classify(plan: LogicalPlan): SqlStatementClassification = plan match {
case UnresolvedWith(child, _, _) => classify(child)
case _: CompoundBody => BeginEnd
- case _: InsertIntoStatement | _: UnresolvedInsert => Insert
+ case _: UnresolvedInsert => Insert
case _: DeleteFromTable | _: DeleteFromTableWithFilters => DeleteWhere
case _: UpdateTable => UpdateWhere
case _: MergeIntoTable => Merge
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
index 75ed0dcda60c..8be200dfc5bb 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
@@ -104,7 +104,18 @@ class QueryExecution(
// are intentionally not handled here and remain a known limitation.
private val lazyLogicalWithResolvedInsert = LazyTry {
def resolve(insert: UnresolvedInsert): LogicalPlan = {
- analyzerOpt.getOrElse(sparkSession.sessionState.analyzer).resolveUnresolvedInsert(insert)
+ try {
+ executePhase(QueryPlanningTracker.ANALYSIS) {
+ QueryPlanningTracker.withTracker(tracker) {
+ analyzerOpt.getOrElse(sparkSession.sessionState.analyzer)
+ .resolveUnresolvedInsert(insert)
+ }
+ }
+ } catch {
+ case NonFatal(e) =>
+ tracker.setAnalysisFailed(logical)
+ throw e
+ }
}
logical match {
diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/explain-aqe.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/explain-aqe.sql.out
index 3f16d4f75651..dc679f0a121c 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/explain-aqe.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/explain-aqe.sql.out
@@ -208,7 +208,7 @@ ExplainCommand 'Aggregate ['key], ['key, unresolvedalias('MIN('val))], Formatted
-- !query
EXPLAIN EXTENDED INSERT INTO TABLE explain_temp5 SELECT * FROM explain_temp4
-- !query analysis
-ExplainCommand 'InsertIntoStatement 'UnresolvedRelation [explain_temp5], [__required_write_privileges__=INSERT], false, false, false, false, false, ExtendedMode
+ExplainCommand 'UnresolvedInsert false, false, false, false, ExtendedMode
-- !query
diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/explain.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/explain.sql.out
index 3f16d4f75651..dc679f0a121c 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/explain.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/explain.sql.out
@@ -208,7 +208,7 @@ ExplainCommand 'Aggregate ['key], ['key, unresolvedalias('MIN('val))], Formatted
-- !query
EXPLAIN EXTENDED INSERT INTO TABLE explain_temp5 SELECT * FROM explain_temp4
-- !query analysis
-ExplainCommand 'InsertIntoStatement 'UnresolvedRelation [explain_temp5], [__required_write_privileges__=INSERT], false, false, false, false, false, ExtendedMode
+ExplainCommand 'UnresolvedInsert false, false, false, false, ExtendedMode
-- !query
diff --git a/sql/core/src/test/resources/sql-tests/results/explain-aqe.sql.out b/sql/core/src/test/resources/sql-tests/results/explain-aqe.sql.out
index 69ad095bc1a5..3288d7754d85 100644
--- a/sql/core/src/test/resources/sql-tests/results/explain-aqe.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/explain-aqe.sql.out
@@ -1183,7 +1183,8 @@ EXPLAIN EXTENDED INSERT INTO TABLE explain_temp5 SELECT * FROM explain_temp4
struct<plan:string>
-- !query output
== Parsed Logical Plan ==
-'InsertIntoStatement 'UnresolvedRelation [explain_temp5], [__required_write_privileges__=INSERT], false, false, false, false, false
+'UnresolvedInsert false, false, false, false
+:- 'UnresolvedInsertTarget [explain_temp5], [], {INSERT}
+- 'Project [*]
+- 'UnresolvedRelation [explain_temp4], [], false
diff --git a/sql/core/src/test/resources/sql-tests/results/explain.sql.out b/sql/core/src/test/resources/sql-tests/results/explain.sql.out
index 6ab9234a24c7..a251e33f4677 100644
--- a/sql/core/src/test/resources/sql-tests/results/explain.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/explain.sql.out
@@ -1075,7 +1075,8 @@ EXPLAIN EXTENDED INSERT INTO TABLE explain_temp5 SELECT * FROM explain_temp4
struct<plan:string>
-- !query output
== Parsed Logical Plan ==
-'InsertIntoStatement 'UnresolvedRelation [explain_temp5], [__required_write_privileges__=INSERT], false, false, false, false, false
+'UnresolvedInsert false, false, false, false
+:- 'UnresolvedInsertTarget [explain_temp5], [], {INSERT}
+- 'Project [*]
+- 'UnresolvedRelation [explain_temp4], [], false
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 58cdc03607d0..e3c98d01db25 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
@@ -20,7 +20,7 @@ package org.apache.spark.sql
import java.time.{Instant, LocalDate, LocalDateTime, ZoneId}
import org.apache.spark.sql.catalyst.ExtendedAnalysisException
-import org.apache.spark.sql.catalyst.analysis.{BindParameters, CTESubstitution, ExpressionWithUnresolvedIdentifier, NameParameterizedQuery, PlanWithUnresolvedIdentifier, UnresolvedRelation}
+import org.apache.spark.sql.catalyst.analysis.{BindParameters, CTESubstitution, ExpressionWithUnresolvedIdentifier, NameParameterizedQuery, PlanWithUnresolvedIdentifier, UnresolvedInsertTarget, UnresolvedRelation}
import org.apache.spark.sql.catalyst.expressions.Literal
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.catalyst.plans.logical.{CacheTableAsSelect, CTEInChildren, InsertIntoStatement, Limit, ReplaceTableAsSelect, UnresolvedInsert, WithCTE}
@@ -2534,11 +2534,11 @@ class ParametersSuite extends SharedSparkSession {
}
}
- // The temporary binary node exposes only a dynamic target to analyzer traversal and keeps CTE
- // definitions on the input query.
+ // The unresolved binary node exposes the target to analyzer traversal and keeps CTE definitions
+ // on the input query.
test("SPARK-46625: WITH ... INSERT INTO IDENTIFIER(:p) REPLACE WHERE ... parser") {
// Use a non-literal-string expression so `withIdentClause` produces
- // `PlanWithUnresolvedIdentifier` rather than short-circuiting to `UnresolvedRelation`.
+ // `PlanWithUnresolvedIdentifier` rather than short-circuiting to `UnresolvedInsertTarget`.
val parsedPlan = spark.sessionState.sqlParser.parsePlan(
"""WITH transformation AS (SELECT 99 AS a)
|INSERT INTO IDENTIFIER('some' || '_table') REPLACE WHERE a = 10
@@ -2562,10 +2562,12 @@ class ParametersSuite extends SharedSparkSession {
}
}
- test("dynamic INSERT targets use a temporary binary plan") {
+ test("parsed INSERT targets use an unresolved binary plan") {
val regularInsert = spark.sessionState.sqlParser
.parsePlan("INSERT INTO target_table SELECT 1")
- assert(regularInsert.isInstanceOf[InsertIntoStatement])
+ .asInstanceOf[UnresolvedInsert]
+ assert(regularInsert.table.isInstanceOf[UnresolvedInsertTarget])
+ assert(regularInsert.children === Seq(regularInsert.table, regularInsert.query))
val identifierInsert = spark.sessionState.sqlParser
.parsePlan("INSERT INTO IDENTIFIER(lower('TESTCAT.NS.TARGET_TABLE')) SELECT 1")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResultSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResultSuite.scala
index ad43f28eeea1..397584cb0c8d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResultSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/catalyst/parser/ParseSqlResultSuite.scala
@@ -141,6 +141,16 @@ class ParseSqlResultSuite extends SparkFunSuite {
val insert = "WITH t AS (SELECT * FROM src) INSERT INTO t SELECT * FROM t"
assert(targetTableRefs(insert) === Set(Seq("t")))
assert(sourceTableRefs(insert) === Set(Seq("src")))
+
+ val otherDml = Seq(
+ "WITH t AS (SELECT * FROM src) DELETE FROM t WHERE a = 1",
+ "WITH t AS (SELECT * FROM src) UPDATE t SET a = 1",
+ "WITH t AS (SELECT * FROM src) " +
+ "MERGE INTO t USING t AS s ON t.a = s.a WHEN MATCHED THEN DELETE")
+ otherDml.foreach { sql =>
+ assert(targetTableRefs(sql).isEmpty, sql)
+ assert(sourceTableRefs(sql) === Set(Seq("src")), sql)
+ }
}
test("positional markers inside BEGIN END are counted once") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala
index fd047d54b957..ab3b4c3df78d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala
@@ -542,6 +542,46 @@ class QueryExecutionSuite extends SharedSparkSession {
mockCallback.assertAnalyzed()
}
+ test("dynamic INSERT target analysis is tracked") {
+ withTable("target_table") {
+ sql("CREATE TABLE target_table (id INT) USING parquet")
+
+ val tracker = new QueryPlanningTracker
+ val plan = spark.sessionState.sqlParser.parsePlan(
+ "INSERT INTO IDENTIFIER(lower('TARGET_TABLE')) VALUES (1)")
+ new QueryExecution(spark, plan, tracker).assertAnalyzed()
+
+ val identifierResolution = tracker.rules.collectFirst {
+ case (name, summary) if name.endsWith("ResolveIdentifierClause") => summary
+ }.getOrElse(fail("ResolveIdentifierClause was not tracked"))
+ assert(identifierResolution.numEffectiveInvocations === 1)
+ }
+ }
+
+ test("dynamic INSERT target analysis failure is reported") {
+ var failedPlan: LogicalPlan = null
+ val callback = new QueryPlanningTrackerCallback {
+ override def analysisFailed(
+ tracker: QueryPlanningTracker,
+ parsedPlan: LogicalPlan): Unit = failedPlan = parsedPlan
+
+ override def analyzed(tracker: QueryPlanningTracker, analyzedPlan: LogicalPlan): Unit =
+ fail("analysis should fail")
+
+ override def readyForExecution(tracker: QueryPlanningTracker): Unit =
+ fail("query should not be ready for execution")
+ }
+ val plan = spark.sessionState.sqlParser.parsePlan(
+ "INSERT INTO IDENTIFIER(1) VALUES (1)")
+ val qe = new QueryExecution(
+ spark,
+ plan,
+ new QueryPlanningTracker(Some(callback)))
+
+ intercept[AnalysisException](qe.assertAnalyzed())
+ assert(failedPlan eq plan)
+ }
+
test("SPARK-51265: IncrementalExecution should set the command execution code correctly") {
withTempView("s") {
val streamDf = spark.readStream.format("rate").load()
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CreateFlowCommandSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CreateFlowCommandSuite.scala
index 5298434cfe6f..cf079565b0c1 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CreateFlowCommandSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CreateFlowCommandSuite.scala
@@ -17,14 +17,15 @@
package org.apache.spark.sql.execution.command.v2
-import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, UnresolvedIdentifier, UnresolvedRelation}
+import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, UnresolvedIdentifier, UnresolvedInsertTarget}
import org.apache.spark.sql.catalyst.plans.logical.{
CreateFlowCommand,
- InsertIntoStatement
+ UnresolvedInsert
}
import org.apache.spark.sql.connector.catalog.TableWritePrivilege
import org.apache.spark.sql.execution.SparkSqlParser
import org.apache.spark.sql.execution.command.v1.CommandSuiteBase
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
/**
* The class contains tests for the `CREATE FLOW ...` command
@@ -59,9 +60,10 @@ class CreateFlowCommandSuite extends CommandSuiteBase with AnalysisTest {
val plan = parser.parsePlan("CREATE FLOW f AS INSERT OVERWRITE a PARTITION (col1) BY NAME " +
"SELECT col1, col2 FROM b")
val cmd = plan.asInstanceOf[CreateFlowCommand]
- assert(cmd.right == InsertIntoStatement(
- table = UnresolvedRelation(Seq("a"))
- .requireWritePrivileges(Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE)),
+ assert(cmd.right == UnresolvedInsert(
+ table = UnresolvedInsertTarget(
+ Seq("a"), CaseInsensitiveStringMap.empty(),
+ Set(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE)),
partitionSpec = Map("col1" -> None),
userSpecifiedCols = Seq.empty,
query = parser.parsePlan("SELECT col1, col2 FROM b"),
diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/SqlGraphRegistrationContext.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/SqlGraphRegistrationContext.scala
index daef65a00ccd..eda9ba2ea155 100644
--- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/SqlGraphRegistrationContext.scala
+++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/SqlGraphRegistrationContext.scala
@@ -22,9 +22,9 @@ import org.apache.spark.{SparkException, SparkRuntimeException}
import org.apache.spark.sql.{AnalysisException, SparkSession}
import org.apache.spark.sql.Column
import org.apache.spark.sql.catalyst.{QueryPlanningTracker, TableIdentifier}
-import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute, UnresolvedRelation}
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute, UnresolvedInsertTarget}
import org.apache.spark.sql.catalyst.expressions.Expression
-import org.apache.spark.sql.catalyst.plans.logical.{AutoCdcInto, CreateFlowCommand, CreateMaterializedViewAsSelect, CreateStreamingTable, CreateStreamingTableAsSelect, CreateStreamingTableAutoCdc, CreateView, InsertIntoStatement, LogicalPlan, UnresolvedInsert}
+import org.apache.spark.sql.catalyst.plans.logical.{AutoCdcInto, CreateFlowCommand, CreateMaterializedViewAsSelect, CreateStreamingTable, CreateStreamingTableAsSelect, CreateStreamingTableAutoCdc, CreateView, LogicalPlan, UnresolvedInsert}
import org.apache.spark.sql.catalyst.util.StringUtils
import org.apache.spark.sql.classic.ClassicConversions._
import org.apache.spark.sql.execution.command.{CreateViewCommand, SetCatalogCommand, SetCommand, SetNamespaceCommand}
@@ -589,10 +589,8 @@ class SqlGraphRegistrationContext(
.identifier
cf.flowOperation match {
- case i: InsertIntoStatement =>
- handleInsert(i, flowIdentifier, queryOrigin)
case i: UnresolvedInsert =>
- handleInsert(i.toInsertIntoStatement(i.table), flowIdentifier, queryOrigin)
+ handleInsert(i, flowIdentifier, queryOrigin)
case a: AutoCdcInto =>
val flowTargetDatasetName = IdentifierHelper.toTableIdentifier(a.targetTable)
graphRegistrationContext.registerFlow(
@@ -628,12 +626,12 @@ class SqlGraphRegistrationContext(
}
private def handleInsert(
- insert: InsertIntoStatement,
+ insert: UnresolvedInsert,
flowIdentifier: TableIdentifier,
queryOrigin: QueryOrigin): Unit = {
validateInsertIntoFlow(insert, queryOrigin)
val flowTargetDatasetName = insert.table match {
- case u: UnresolvedRelation =>
+ case u: UnresolvedInsertTarget =>
IdentifierHelper.toTableIdentifier(u.multipartIdentifier)
case _ =>
throw SqlGraphElementRegistrationException(
@@ -669,34 +667,34 @@ class SqlGraphRegistrationContext(
.identifier
private def validateInsertIntoFlow(
- insertIntoStatement: InsertIntoStatement,
+ insert: UnresolvedInsert,
queryOrigin: QueryOrigin
): Unit = {
- if (insertIntoStatement.partitionSpec.nonEmpty) {
+ if (insert.partitionSpec.nonEmpty) {
throw SqlGraphElementRegistrationException(
msg = "Partition spec may not be specified for flow target.",
queryOrigin = queryOrigin
)
}
- if (insertIntoStatement.userSpecifiedCols.nonEmpty) {
+ if (insert.userSpecifiedCols.nonEmpty) {
throw SqlGraphElementRegistrationException(
msg = "Column schema may not be specified for flow target.",
queryOrigin = queryOrigin
)
}
- if (insertIntoStatement.overwrite) {
+ if (insert.overwrite) {
throw SqlGraphElementRegistrationException(
msg = "INSERT OVERWRITE flows not supported.",
queryOrigin = queryOrigin
)
}
- if (insertIntoStatement.ifPartitionNotExists) {
+ if (insert.ifPartitionNotExists) {
throw SqlGraphElementRegistrationException(
msg = "IF NOT EXISTS not supported for flows.",
queryOrigin = queryOrigin
)
}
- if (!insertIntoStatement.byName) {
+ if (!insert.byName) {
throw SqlGraphElementRegistrationException(
msg = "Only INSERT INTO by name flows supported.",
queryOrigin = queryOrigin
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org