This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new b1c214c1170d [SPARK-58933][SQL] Resolve expressions in INSERT target
IDENTIFIER clauses
b1c214c1170d is described below
commit b1c214c1170df1b6bfa5afb66b1f9d071e6323ef
Author: Wenchen Fan <[email protected]>
AuthorDate: Fri Aug 28 10:27:06 2026 +0800
[SPARK-58933][SQL] Resolve expressions in INSERT target IDENTIFIER clauses
### What changes were proposed in this pull request?
This PR introduces a temporary binary `UnresolvedInsert` plan for INSERT
statements whose target
uses a dynamic `IDENTIFIER(...)` expression. Its target and input query are
ordinary children, so
analyzer rules can resolve parameters, functions, and SQL variables in the
target expression
through normal tree traversal.
After the expression is evaluated, its builder creates an
`UnresolvedInsertTarget` containing the
raw multipart identifier, target options, and write privileges. The
analyzer then converts this
marker to an `UnresolvedRelation` and lowers `UnresolvedInsert` to the
existing unary
`InsertIntoStatement`. The target is subsequently qualified and resolved by
the existing INSERT
analysis path.
`UnresolvedInsertTarget` is deliberately not a `NamedRelation`, so CTE
substitution cannot treat
an INSERT target as a readable relation. Static table names and literal
`IDENTIFIER` clauses still
parse directly to `InsertIntoStatement` and retain the existing behavior.
`QueryExecution` resolves only the dynamic target identifier before
transaction detection. This
allows transactional INSERTs to be recognized without adding a separate
target-resolution path or
changing the long-lived child contract of `InsertIntoStatement`.
`ResolveUnresolvedInsert` uses a dedicated tree pattern and rule ID, so the
fixed-point Resolution
batch skips plans that cannot contain the temporary carrier.
### Why are the changes needed?
`InsertIntoStatement.table` is not a logical-plan child, so normal analyzer
rules do not visit a
dynamic `IDENTIFIER` expression stored in that slot. Nested functions,
parameters, or SQL variables
can therefore remain unresolved.
Making `InsertIntoStatement` permanently binary would affect analyzer,
planner, lineage, and
command consumers that rely on its input query being its only child. The
temporary
`UnresolvedInsert` node limits the binary shape to identifier-expression
resolution, while
`UnresolvedInsertTarget` preserves the raw name until normal relation
resolution. This also avoids
qualifying an already-resolved identifier a second time.
### Does this PR introduce _any_ user-facing change?
Yes. INSERT statements can now resolve expressions inside a target
`IDENTIFIER` clause. For
example, the following statement now resolves and executes correctly:
```sql
INSERT INTO IDENTIFIER(
lower(regexp_replace(:table_name, 'PLACEHOLDER', 'TBL')))
REPLACE WHERE id = 1
VALUES (1, 'updated')
```
### How was this patch tested?
Added parser, analyzer, end-to-end, parse-metadata lineage and
classification, and
transactional-catalog coverage for dynamic INSERT targets, including
parameters, nested functions,
SQL variables, qualified multipart names, query-source and select-list
extraction, and a target
sharing its name with a CTE.
The following compile and test checks passed:
```bash
build/sbt catalyst/Test/compile
build/sbt 'sql/testOnly org.apache.spark.sql.ParametersSuite'
build/sbt 'sql/testOnly
org.apache.spark.sql.catalyst.parser.ParseSqlResultSuite'
build/sbt 'sql/testOnly
org.apache.spark.sql.connector.AppendDataTransactionSuite'
build/sbt pipelines/compile
build/sbt connect/compile
SPARK_GENERATE_GOLDEN_FILES=1 build/sbt \
'sql/testOnly org.apache.spark.sql.SQLQueryTestSuite -- -z explain.sql'
SPARK_GENERATE_GOLDEN_FILES=1 build/sbt \
'sql/testOnly org.apache.spark.sql.SQLQueryTestSuite -- -z
explain-aqe.sql'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
Closes #58204 from cloud-fan/fix-identifier-dml-table-resolution.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 3bc29d9d0ae031dfe0e793886ba652ca98ae7f29)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../spark/sql/catalyst/analysis/Analyzer.scala | 65 ++++++----
.../analysis/ResolveIdentifierClause.scala | 18 +--
.../catalyst/analysis/ResolveInlineTables.scala | 15 ++-
.../spark/sql/catalyst/analysis/parameters.scala | 13 +-
.../spark/sql/catalyst/analysis/unresolved.scala | 22 +++-
.../spark/sql/catalyst/parser/AstBuilder.scala | 47 +++----
.../sql/catalyst/plans/logical/statements.scala | 64 +++++++--
.../sql/catalyst/rules/RuleIdCollection.scala | 1 +
.../spark/sql/catalyst/trees/TreePatterns.scala | 1 +
.../sql/connect/pipelines/PipelinesHandler.scala | 3 +-
.../spark/sql/catalyst/parser/ParseSqlResult.scala | 8 +-
.../sql/catalyst/parser/SqlStatementCodes.scala | 2 +-
.../spark/sql/execution/QueryExecution.scala | 22 +++-
.../org/apache/spark/sql/ParametersSuite.scala | 144 +++++++++++----------
.../sql/catalyst/parser/ParseSqlResultSuite.scala | 14 ++
.../sql/connector/AppendDataTransactionSuite.scala | 24 ++++
.../graph/SqlGraphRegistrationContext.scala | 59 +++++----
17 files changed, 327 insertions(+), 195 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 af4f185d39aa..d9612dda2cfd 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
@@ -392,6 +392,36 @@ class Analyzer(
def getRelationResolution: RelationResolution = relationResolution
+ private def toUnresolvedRelation(target: UnresolvedInsertTarget):
UnresolvedRelation = {
+ val relation = withOrigin(target.origin) {
+ UnresolvedRelation(target.multipartIdentifier, target.options)
+ .requireWritePrivileges(target.writePrivileges)
+ }
+ relation.copyTagsFrom(target)
+ relation
+ }
+
+ /** Resolves only the dynamic identifier of an INSERT so transaction
detection can inspect it. */
+ private[sql] def resolveUnresolvedInsert(insert: UnresolvedInsert):
LogicalPlan = {
+ runWithSessionConf {
+ val identifierResolvedTarget = insert.table match {
+ case p: PlanWithUnresolvedIdentifier => execute(p)
+ case other => other
+ }
+ withOrigin(insert.origin) {
+ identifierResolvedTarget match {
+ case target: UnresolvedInsertTarget =>
+ insert.toInsertIntoStatement(toUnresolvedRelation(target))
+ case target if !target.fastEquals(insert.table) =>
+ val updated = insert.copy(table = target)
+ updated.copyTagsFrom(insert)
+ updated
+ case _ => insert
+ }
+ }
+ }
+ }
+
def executeAndCheck(plan: LogicalPlan, tracker: QueryPlanningTracker):
LogicalPlan = {
if (plan.analyzed) {
plan
@@ -569,6 +599,7 @@ class Analyzer(
override def batches: Seq[Batch] = earlyBatches ++ Seq(
Batch("Resolution", fixedPoint,
+ ResolveUnresolvedInsert ::
new ResolveCatalogs(catalogManager) ::
ResolveInsertInto ::
ResolveRelations ::
@@ -1353,6 +1384,15 @@ class Analyzer(
}
}
+ /** Lower an 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) {
+ case insert @ UnresolvedInsert(target: UnresolvedInsertTarget, _, _, _,
_, _, _, _, _) =>
+ insert.toInsertIntoStatement(toUnresolvedRelation(target))
+ }
+ }
+
/** Handle INSERT INTO for DSv2 */
object ResolveInsertInto extends ResolveInsertionBase {
override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsWithPruning(
@@ -1631,29 +1671,8 @@ class Analyzer(
}
def doApply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp {
- // `InsertIntoStatement.table` and `V2WriteCommand.table` are non-child
`LogicalPlan`
- // slots (`child = query`), so the default `resolveOperatorsUp` +
`mapExpressions`
- // traversal never resolves expressions placed inside them. For a
- // `PlanWithUnresolvedIdentifier`, `identifierExpr` (e.g. an
`UnresolvedAttribute`
- // referring to a SQL variable in `INSERT INTO IDENTIFIER(target_table)
...`) must
- // be resolved here before `ResolveIdentifierClause` can materialize the
relation.
- // Mirror the structural recursion into the non-child `.table` slot that
- // `BindParameters` and `ResolveIdentifierClause` already do for the
same shape
- // (SPARK-46625); unlike those rules, this one performs attribute
resolution rather
- // than parameter binding or placeholder materialization. Resolve
against `p` (whose
- // `children` are `Nil` on the INSERT / `OverwriteByExpression` path
built by
- // `buildWriteTableSlot`) so the IDENTIFIER expression cannot see query
output
- // columns -- only the last-resort variable resolution path fires. The
- // `!identifierExpr.resolved` guard makes the case idempotent under
bottom-up
- // traversal.
- case i: InsertIntoStatement
- if i.table.isInstanceOf[PlanWithUnresolvedIdentifier] &&
-
!i.table.asInstanceOf[PlanWithUnresolvedIdentifier].identifierExpr.resolved =>
- val p = i.table.asInstanceOf[PlanWithUnresolvedIdentifier]
- val resolvedExpr = resolveExpressionByPlanChildren(
- p.identifierExpr, p, includeLastResort = true)
- i.copy(table = p.copy(identifierExpr = resolvedExpr))
-
+ // V2WriteCommand.table is a non-child LogicalPlan slot, so recurse
explicitly when it
+ // contains an identifier expression that may refer to a SQL variable.
case w: V2WriteCommand
if w.table.isInstanceOf[PlanWithUnresolvedIdentifier] &&
!w.table.asInstanceOf[PlanWithUnresolvedIdentifier].identifierExpr.resolved =>
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 2bccc99f31d8..320cfc0a77c3 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
@@ -21,7 +21,7 @@ import scala.collection.mutable
import org.apache.spark.SparkException
import org.apache.spark.sql.catalyst.expressions.{Expression,
SubqueryExpression, VariableReference}
-import org.apache.spark.sql.catalyst.plans.logical.{CreateView,
InsertIntoStatement, LogicalPlan, V2WriteCommand}
+import org.apache.spark.sql.catalyst.plans.logical.{CreateView, LogicalPlan,
V2WriteCommand}
import org.apache.spark.sql.catalyst.rules.{Rule, RuleExecutor}
import org.apache.spark.sql.catalyst.trees.TreePattern._
import org.apache.spark.sql.errors.QueryCompilationErrors
@@ -71,22 +71,6 @@ class ResolveIdentifierClause(earlyBatches:
Seq[RuleExecutor[LogicalPlan]#Batch]
executor.execute(p.planBuilder.apply(
IdentifierResolution.evalIdentifierExpr(p.identifierExpr),
p.children))
- // `InsertIntoStatement.table` and `V2WriteCommand.table` are non-child
LogicalPlan slots
- // (`child = query`), so the standard `resolveOperatorsUp` traversal
never visits
- // placeholders inside them. Materialize them explicitly. Only
`InsertIntoStatement`
- // carries a parse-time placeholder today, but matching the
`V2WriteCommand` trait keeps
- // the rule consistent across the family.
- case i: InsertIntoStatement if
i.table.isInstanceOf[PlanWithUnresolvedIdentifier] =>
- val p = i.table.asInstanceOf[PlanWithUnresolvedIdentifier]
- if (p.identifierExpr.resolved && p.childrenResolved) {
- if (referredTempVars.isDefined) {
- referredTempVars.get ++= collectTemporaryVariablesInLogicalPlan(p)
- }
- i.copy(table = executor.execute(p.planBuilder.apply(
- IdentifierResolution.evalIdentifierExpr(p.identifierExpr),
p.children)))
- } else {
- i
- }
case w: V2WriteCommand if
w.table.isInstanceOf[PlanWithUnresolvedIdentifier] =>
val p = w.table.asInstanceOf[PlanWithUnresolvedIdentifier]
if (p.identifierExpr.resolved && p.childrenResolved) {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveInlineTables.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveInlineTables.scala
index 1c9bf9dd22bc..800b34627350 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveInlineTables.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveInlineTables.scala
@@ -18,7 +18,7 @@
package org.apache.spark.sql.catalyst.analysis
import org.apache.spark.sql.catalyst.expressions.EvalHelper
-import org.apache.spark.sql.catalyst.plans.logical.{InsertIntoStatement,
LogicalPlan, OverwriteByExpression, SubqueryAlias}
+import org.apache.spark.sql.catalyst.plans.logical.{InsertIntoStatement,
LogicalPlan, OverwriteByExpression, SubqueryAlias, UnresolvedInsert}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.CurrentOrigin
import org.apache.spark.sql.catalyst.trees.TreePattern.INLINE_TABLE_EVAL
@@ -34,6 +34,19 @@ object ResolveInlineTables extends Rule[LogicalPlan] with
EvalHelper {
// coercion will cast to the target column's collation.
// Preserve the inline table's origin so error messages point to the
VALUES clause,
// not the INSERT statement.
+ case i @ UnresolvedInsert(_, _, _, table: UnresolvedInlineTable, _, _,
_, _, _)
+ if table.expressionsResolved =>
+ CurrentOrigin.withOrigin(table.origin) {
+ i.copy(query = EvaluateUnresolvedInlineTable
+ .evaluateUnresolvedInlineTable(table, ignoreCollation = true))
+ }
+ case i @ UnresolvedInsert(
+ _, _, _, sa @ SubqueryAlias(_, table: UnresolvedInlineTable), _, _,
_, _, _)
+ if table.expressionsResolved =>
+ CurrentOrigin.withOrigin(table.origin) {
+ i.copy(query = sa.copy(child = EvaluateUnresolvedInlineTable
+ .evaluateUnresolvedInlineTable(table, ignoreCollation = true)))
+ }
case i @ InsertIntoStatement(_, _, _, table: UnresolvedInlineTable, _,
_, _, _, _)
if table.expressionsResolved =>
CurrentOrigin.withOrigin(table.origin) {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/parameters.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/parameters.scala
index deb63a7d9789..1dac9a7b9f7d 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/parameters.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/parameters.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.catalyst.analysis
import org.apache.spark.SparkException
import org.apache.spark.sql.catalyst.expressions.{Expression, LeafExpression,
SubqueryExpression, Unevaluable}
-import org.apache.spark.sql.catalyst.plans.logical.{InsertIntoStatement,
LogicalPlan, SupervisingCommand, V2WriteCommand}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan,
SupervisingCommand, V2WriteCommand}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.{COMMAND, PARAMETER,
PARAMETERIZED_QUERY, TreePattern, UNRESOLVED_WITH}
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase}
@@ -179,16 +179,9 @@ object BindParameters extends Rule[LogicalPlan] with
QueryErrorsBase {
p0.resolveOperatorsDownWithPruning(_.containsPattern(PARAMETER) && !stop) {
case p1 =>
stop = p1.isInstanceOf[ParameterizedQuery]
- // `InsertIntoStatement.table` and `V2WriteCommand.table` are
non-child LogicalPlan
- // slots, so the standard `resolveOperatorsDown` traversal never
visits parameter
- // markers inside them. Recurse explicitly so `INSERT ...
IDENTIFIER(:p)` and
- // `INSERT INTO IDENTIFIER(:p) REPLACE WHERE ...` resolve under the
legacy
- // parameter-substitution mode (SPARK-46625). The parser places the
placeholder only in
- // `InsertIntoStatement.table`; the `V2WriteCommand` trait match keeps
the rule
- // consistent for any analyzer-built node in the same shape.
+ // V2WriteCommand.table is a non-child LogicalPlan slot, so recurse
explicitly when it
+ // contains parameter markers.
val withBoundTable = p1 match {
- case i: InsertIntoStatement if i.table.containsPattern(PARAMETER) =>
- i.copy(table = bind(i.table)(f))
case w: V2WriteCommand if w.table.containsPattern(PARAMETER) =>
bind(w.table)(f) match {
case nr: NamedRelation => w.withNewTable(nr)
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 50b2970544dc..19a7cbc62ee8 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
@@ -63,12 +63,9 @@ trait UnresolvedUnaryNode extends UnaryNode with
UnresolvedNode
* Extends `NamedRelation` so it can occupy a `NamedRelation`-typed slot (e.g.
* `OverwriteByExpression.table`) directly at parse time, instead of wrapping
the whole command.
*
- * The parser always places this node inside the command's identifier slot (a
child slot for
- * DELETE/UPDATE/MERGE/CTAS/RTAS, or a non-child slot for
`InsertIntoStatement.table` and
- * `OverwriteByExpression.table` -- handled via explicit cases in
`ResolveIdentifierClause` and
- * `BindParameters`). It is never the substitution root of a `WITH ...
<command>` subtree, so
- * `CTEInChildren` semantics are not needed: any surrounding `WithCTE`
produced by
- * `CTESubstitution` targets the inner command directly.
+ * Depending on the command shape, the parser uses this node either as the
plan root or within the
+ * command's identifier slot. For INSERT, the slot is a child of
`UnresolvedInsert` until the
+ * identifier expression has been evaluated.
*/
case class PlanWithUnresolvedIdentifier(
identifierExpr: Expression,
@@ -123,6 +120,19 @@ case class ExpressionWithUnresolvedIdentifier(
}
}
+/**
+ * Holds the raw table identifier and lookup context for an unresolved INSERT
target.
+ *
+ * This is deliberately not a [[NamedRelation]]: while it is a child of
+ * [[org.apache.spark.sql.catalyst.plans.logical.UnresolvedInsert]], it must
not be interpreted as
+ * a readable relation or substituted with a same-named CTE. It is converted
to an
+ * [[UnresolvedRelation]] when the enclosing INSERT is lowered to
`InsertIntoStatement`.
+ */
+case class UnresolvedInsertTarget(
+ multipartIdentifier: Seq[String],
+ options: CaseInsensitiveStringMap,
+ writePrivileges: Set[TableWritePrivilege]) extends UnresolvedLeafNode
+
/**
* Holds the name of a relation that has yet to be looked up in a catalog.
*
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 609c9bd36be8..6c7ad866c6ef 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,10 +939,8 @@ class AstBuilder extends DataTypeAstBuilder
query: LogicalPlan,
queryAliasCtx: TableAliasContext): LogicalPlan = withOrigin(ctx) {
ctx match {
- // For all `InsertIntoStatement`-producing branches, build the `table`
slot directly via
- // `buildWriteTableSlot` so that any `PlanWithUnresolvedIdentifier`
lives *inside* the
- // command's identifier slot. This preserves the `CTEInChildren` shape
and lets
- // `CTESubstitution` place `WithCTE` on the command's children correctly
(SPARK-46625).
+ // 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)
@@ -1170,16 +1168,14 @@ class AstBuilder extends DataTypeAstBuilder
replaceCriteriaOpt = replaceCriteriaOpt)
}
- /**
- * Creates an [[InsertIntoStatement]] from [[InsertTableParams]].
- */
+ /** Creates an INSERT plan from [[InsertTableParams]]. */
private def createInsertIntoStatement(
insertParams: InsertTableParams,
tableSlot: LogicalPlan,
query: LogicalPlan,
overwrite: Boolean,
- withSchemaEvolution: Boolean): InsertIntoStatement = {
- InsertIntoStatement(
+ withSchemaEvolution: Boolean): LogicalPlan = {
+ val unresolvedInsert = UnresolvedInsert(
table = tableSlot,
partitionSpec = insertParams.partitionSpec,
userSpecifiedCols = insertParams.userSpecifiedCols,
@@ -1189,27 +1185,32 @@ 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 `table` slot of a write command. If the identifier reference is
a constant string,
- * returns an [[UnresolvedRelation]] directly; otherwise returns a
- * [[PlanWithUnresolvedIdentifier]] that materializes into an
[[UnresolvedRelation]] once the
- * identifier expression is resolved. Both branches produce a
[[NamedRelation]], which occupies
- * the `InsertIntoStatement.table` slot (a general `LogicalPlan` slot, since
`NamedRelation`
- * extends `LogicalPlan`).
- *
- * Placing the placeholder in the identifier slot (rather than wrapping the
entire write command)
- * preserves the `CTEInChildren` shape at parse time, so `CTESubstitution`
places `WithCTE` on the
- * command's children correctly. See SPARK-46625.
+ * 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.
*/
private def buildWriteTableSlot(
ctx: IdentifierReferenceContext,
optionsClause: Option[OptionsClauseContext],
- writePrivileges: Set[TableWritePrivilege]): NamedRelation = {
- withIdentClause(ctx, parts =>
- createUnresolvedRelation(ctx, parts, optionsClause, writePrivileges,
isStreaming = false))
- .asInstanceOf[NamedRelation]
+ writePrivileges: Set[TableWritePrivilege]): LogicalPlan = {
+ val options = withOrigin(ctx) { resolveOptions(optionsClause) }
+ withIdentClause(ctx, parts => withOrigin(ctx) {
+ UnresolvedInsertTarget(parts, options, writePrivileges)
+ })
}
/**
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 e10fa680ddda..75a06a39b841 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
@@ -20,10 +20,10 @@ package org.apache.spark.sql.catalyst.plans.logical
import org.apache.spark.sql.catalyst.analysis.{FieldName, FieldPosition,
UnresolvedException}
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression,
Unevaluable}
import org.apache.spark.sql.catalyst.trees.{LeafLike, UnaryLike}
+import org.apache.spark.sql.catalyst.trees.TreePattern._
import org.apache.spark.sql.connector.catalog.ColumnDefaultValue
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.types.DataType
-import org.apache.spark.util.collection.BitSet
/**
* A logical plan node that contains exactly what was parsed from SQL.
@@ -158,7 +158,55 @@ case class QualifiedColType(
}
/**
- * An INSERT INTO statement, as parsed from SQL.
+ * A temporary INSERT plan whose target identifier expression still needs
analyzer traversal.
+ *
+ * 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.
+ */
+case class UnresolvedInsert(
+ table: LogicalPlan,
+ partitionSpec: Map[String, Option[String]],
+ userSpecifiedCols: Seq[String],
+ query: LogicalPlan,
+ overwrite: Boolean,
+ ifPartitionNotExists: Boolean,
+ byName: Boolean = false,
+ replaceCriteriaOpt: Option[InsertReplaceCriteria] = None,
+ withSchemaEvolution: Boolean = false)
+ extends ParsedStatement with TransactionalWrite {
+
+ final override val nodePatterns: Seq[TreePattern] = Seq(UNRESOLVED_INSERT)
+
+ override def children: Seq[LogicalPlan] = Seq(table, query)
+
+ override def withCTEDefs(cteDefs: Seq[CTERelationDef]): LogicalPlan = {
+ copy(query = WithCTE(query, cteDefs))
+ }
+
+ def toInsertIntoStatement(target: LogicalPlan): InsertIntoStatement = {
+ val statement = InsertIntoStatement(
+ table = target,
+ partitionSpec = partitionSpec,
+ userSpecifiedCols = userSpecifiedCols,
+ query = query,
+ overwrite = overwrite,
+ ifPartitionNotExists = ifPartitionNotExists,
+ byName = byName,
+ replaceCriteriaOpt = replaceCriteriaOpt,
+ withSchemaEvolution = withSchemaEvolution)
+ statement.copyTagsFrom(this)
+ statement
+ }
+
+ override protected def withNewChildrenInternal(
+ newChildren: IndexedSeq[LogicalPlan]): UnresolvedInsert = {
+ copy(table = newChildren(0), query = newChildren(1))
+ }
+}
+
+/**
+ * An INSERT INTO statement whose target identifier is ready for relation
resolution.
*
* @param table the logical plan representing the table.
* @param userSpecifiedCols the user specified list of columns that belong
to the table.
@@ -212,18 +260,6 @@ case class InsertIntoStatement(
override def child: LogicalPlan = query
override protected def withNewChildInternal(newChild: LogicalPlan):
InsertIntoStatement =
copy(query = newChild)
-
- // `table` is a non-child LogicalPlan slot (`child = query`), so the default
tree-pattern
- // propagation in TreeNode/QueryPlan does not see patterns inside it. Add
`table`'s bits here
- // so that `containsPattern(...)` pruning correctly reports patterns living
in `table`
- // (e.g. `PARAMETER`, `PLAN_WITH_UNRESOLVED_IDENTIFIER`). The REPLACE
criteria condition is an
- // `Expression` child (`InsertReplaceCriteria` extends `Expression`), so its
bits already
- // propagate through the normal expression traversal.
- override protected def getDefaultTreePatternBits: BitSet = {
- val bits = super.getDefaultTreePatternBits
- bits.union(table.treePatternBits)
- bits
- }
}
sealed abstract class InsertReplaceCriteria extends Expression with
Unevaluable {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
index 9221637d4db5..cd1d14e1bd6c 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
@@ -73,6 +73,7 @@ object RuleIdCollection {
"org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveTempViews" ::
"org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveTranspose" ::
"org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveUnpivot" ::
+
"org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveUnresolvedInsert" ::
"org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveUpCast" ::
"org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveWindowFrame" ::
"org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveWindowOrder" ::
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
index 55b9f0621fd6..7c25ea5d9442 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
@@ -210,6 +210,7 @@ object TreePattern extends Enumeration {
val UNRESOLVED_EVENT_TIME_WATERMARK: Value = Value
val UNRESOLVED_HAVING: Value = Value
val UNRESOLVED_HINT: Value = Value
+ val UNRESOLVED_INSERT: Value = Value
val UNRESOLVED_QUALIFY: Value = Value
val UNRESOLVED_FUNC: Value = Value
val UNRESOLVED_PROCEDURE: Value = Value
diff --git
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
index ca41281a812b..718d1670bf8c 100644
---
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
+++
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelinesHandler.scala
@@ -31,7 +31,7 @@ import org.apache.spark.sql.{AnalysisException, Column}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute,
UnresolvedRelation}
import org.apache.spark.sql.catalyst.expressions.Expression
-import org.apache.spark.sql.catalyst.plans.logical.{Command, CreateNamespace,
CreateTable, CreateTableAsSelect, CreateView, DescribeRelation,
DescribeTablePartition, DropView, InsertIntoStatement, LogicalPlan,
RenameTable, ShowColumns, ShowCreateTable, ShowFunctions, ShowTableProperties,
ShowTables, ShowViews}
+import org.apache.spark.sql.catalyst.plans.logical.{Command, CreateNamespace,
CreateTable, CreateTableAsSelect, CreateView, DescribeRelation,
DescribeTablePartition, DropView, InsertIntoStatement, LogicalPlan,
RenameTable, ShowColumns, ShowCreateTable, ShowFunctions, ShowTableProperties,
ShowTables, ShowViews, UnresolvedInsert}
import org.apache.spark.sql.classic.ClassicConversions._
import org.apache.spark.sql.connect.common.DataTypeProtoConverter
import org.apache.spark.sql.connect.service.SessionHolder
@@ -173,6 +173,7 @@ private[connect] object PipelinesHandler extends Logging {
queryPlan.isInstanceOf[CreateTable] ||
queryPlan.isInstanceOf[CreateView] ||
queryPlan.isInstanceOf[InsertIntoStatement] ||
+ queryPlan.isInstanceOf[UnresolvedInsert] ||
queryPlan.isInstanceOf[RenameTable] ||
queryPlan.isInstanceOf[CreateNamespace] ||
queryPlan.isInstanceOf[DropView]
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 228da54c3c7d..3b3a0604d739 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
@@ -223,6 +223,9 @@ object ParseSqlResult {
case m: MergeIntoTable =>
visitPlan(m.targetTable, scope, TableRefRole.Target)(f)
visitPlan(m.sourceTable, scope, TableRefRole.Source)(f)
+ 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 =>
@@ -315,7 +318,7 @@ object ParseSqlResult {
}
def addTable(parts: Seq[String], scope: CteScope, role: TableRefRole):
Unit = {
- if (parts.nonEmpty && !isCteName(parts, scope)) {
+ if (parts.nonEmpty && (role == TableRefRole.Target || !isCteName(parts,
scope))) {
role match {
case TableRefRole.Target => targetTables += parts
case TableRefRole.Source => sourceTables += parts
@@ -342,6 +345,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: UnresolvedRelation => add(u.multipartIdentifier)
case u: UnresolvedTable => add(u.multipartIdentifier)
case u: UnresolvedView => add(u.multipartIdentifier)
@@ -386,6 +390,8 @@ object ParseSqlResult {
private def primaryQueryPlan(plan: LogicalPlan): LogicalPlan = plan match {
case UnresolvedWith(child, _, _) => primaryQueryPlan(child)
+ case UnresolvedInsert(_, _, _, query, _, _, _, _, _) =>
+ primaryQueryPlan(query)
case InsertIntoStatement(_, _, _, query, _, _, _, _, _) =>
primaryQueryPlan(query)
case c: CreateTableAsSelect => 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 cb2fae1d9af9..24cfdfcd2eb4 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 => Insert
+ case _: InsertIntoStatement | _: 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 51e54ef5f753..75ed0dcda60c 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
@@ -34,7 +34,7 @@ import org.apache.spark.sql.catalyst.{InternalRow,
QueryPlanningTracker}
import org.apache.spark.sql.catalyst.analysis.{Analyzer, LazyExpression,
NameParameterizedQuery, UnsupportedOperationChecker}
import org.apache.spark.sql.catalyst.expressions.codegen.ByteCodeStats
import org.apache.spark.sql.catalyst.plans.QueryPlan
-import org.apache.spark.sql.catalyst.plans.logical.{AppendData, Command,
CommandResult, CompoundBody, CreateTableAsSelect, LogicalPlan,
OverwriteByExpression, OverwritePartitionsDynamic, ReplaceTableAsSelect,
ReturnAnswer, Union, UnresolvedWith, WithCTE}
+import org.apache.spark.sql.catalyst.plans.logical.{AppendData, Command,
CommandResult, CompoundBody, CreateTableAsSelect, LogicalPlan,
OverwriteByExpression, OverwritePartitionsDynamic, ReplaceTableAsSelect,
ReturnAnswer, Union, UnresolvedInsert, UnresolvedWith, WithCTE}
import org.apache.spark.sql.catalyst.rules.{PlanChangeLogger, Rule,
RuleExecutor}
import org.apache.spark.sql.catalyst.transactions.TransactionUtils
import org.apache.spark.sql.catalyst.util.StringUtils.PlanStringConcat
@@ -99,6 +99,22 @@ class QueryExecution(
logical.exists(_.expressions.exists(_.exists(_.isInstanceOf[LazyExpression])))
}
+ // Legacy constant-only parameter substitution wraps the plan in a
ParameterizedQuery. Its
+ // parameters bind during analysis, after transaction selection, so INSERT
IDENTIFIER parameters
+ // 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)
+ }
+
+ logical match {
+ case unresolvedWith @ UnresolvedWith(insert: UnresolvedInsert, _, _) =>
+ unresolvedWith.copy(child = resolve(insert))
+ case insert: UnresolvedInsert => resolve(insert)
+ case other => other
+ }
+ }
+ private def logicalWithResolvedInsert: LogicalPlan =
lazyLogicalWithResolvedInsert.get
// 1. At the pre-Analyzed plan we look for nodes that implement the
TransactionalWrite trait.
// When a plan contains such a node we initiate a transaction. Note, we
should never start
@@ -116,7 +132,7 @@ class QueryExecution(
analyzerOpt.flatMap(_.catalogManager.transaction).orElse {
// Only begin a new transaction for outer QEs that lead to execution.
if (mode != CommandExecutionMode.SKIP) {
- val catalog = logical match {
+ val catalog = logicalWithResolvedInsert match {
case UnresolvedWith(TransactionalWrite(c), _, _) => Some(c)
case TransactionalWrite(c) => Some(c)
case _ => None
@@ -181,7 +197,7 @@ class QueryExecution(
SqlScriptingExecution.executeSqlScript(
session = sparkSession,
script = compoundBody)
- case _ => logical
+ case _ => logicalWithResolvedInsert
}
}
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 edca2c36f256..58cdc03607d0 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,10 +20,10 @@ 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}
+import org.apache.spark.sql.catalyst.analysis.{BindParameters,
CTESubstitution, ExpressionWithUnresolvedIdentifier, NameParameterizedQuery,
PlanWithUnresolvedIdentifier, 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, WithCTE}
+import org.apache.spark.sql.catalyst.plans.logical.{CacheTableAsSelect,
CTEInChildren, InsertIntoStatement, Limit, ReplaceTableAsSelect,
UnresolvedInsert, WithCTE}
import org.apache.spark.sql.catalyst.trees.SQLQueryContext
import org.apache.spark.sql.catalyst.trees.TreePattern.PARAMETER
import org.apache.spark.sql.catalyst.util.CharVarcharUtils
@@ -2474,9 +2474,8 @@ class ParametersSuite extends SharedSparkSession {
}
// SPARK-46625: WITH ... <write-with-IDENTIFIER> SELECT ... FROM cte
- // The placeholder is pushed into the command's identifier slot at parse
time, so
- // `CTESubstitution` sees the `CTEInChildren` directly and never produces
the invalid
- // `WithCTE(InsertIntoStatement, ...)` / `WithCTE(CreateTableAsSelect, ...)`
shape.
+ // The placeholder stays inside a CTEInChildren command at parse time, so
CTESubstitution never
+ // produces a WithCTE around the command itself.
private def assertNoWithCTEAroundCTEInChildren(df: DataFrame): Unit = {
df.queryExecution.analyzed.foreach {
case WithCTE(_: CTEInChildren, _) =>
@@ -2512,6 +2511,17 @@ class ParametersSuite extends SharedSparkSession {
}
}
+ test("INSERT target is not substituted by a same-named CTE") {
+ withTable("t_cte_target") {
+ sql("CREATE TABLE t_cte_target (a INT) USING PARQUET")
+ sql(
+ """WITH t_cte_target AS (SELECT 99 AS a)
+ |INSERT INTO IDENTIFIER(lower('T_CTE_TARGET'))
+ |SELECT 1 AS a""".stripMargin)
+ checkAnswer(spark.table("t_cte_target"), Row(1))
+ }
+ }
+
test("SPARK-46625: CREATE TABLE IDENTIFIER(:p) AS WITH ... SELECT ... FROM
cte") {
withTable("t_cte_ctas") {
val df = spark.sql(
@@ -2524,29 +2534,8 @@ class ParametersSuite extends SharedSparkSession {
}
}
- // SPARK-46625: legacy parameter-substitution mode triggers the
parameters.scala traversal
- // path. The placeholder lives in `InsertIntoStatement.table`, which is
*not* a child, so this
- // exercises the `InsertIntoStatement` special-case in `BindParameters.bind`
that recurses into
- // the `table` slot, and the `getDefaultTreePatternBits` override on
`InsertIntoStatement` that
- // exposes `table`'s tree-pattern bits for pruning.
- test("SPARK-46625: INSERT IDENTIFIER(:p) under legacy parameter
substitution") {
- withSQLConf(SQLConf.LEGACY_PARAMETER_SUBSTITUTION_CONSTANTS_ONLY.key ->
"true") {
- withTable("t_legacy_param") {
- sql("CREATE TABLE t_legacy_param (a INT) USING PARQUET")
- spark.sql(
- """WITH transformation AS (SELECT 11 AS a)
- |INSERT INTO IDENTIFIER(:tname)
- |SELECT * FROM transformation""".stripMargin,
- Map("tname" -> "t_legacy_param"))
- checkAnswer(spark.table("t_legacy_param"), Row(11))
- }
- }
- }
-
- // SPARK-46625: INSERT INTO REPLACE WHERE parses into `InsertIntoStatement`,
whose `table`
- // slot is a non-child `LogicalPlan`. `PlanWithUnresolvedIdentifier` sits in
the slot directly.
- // Verify on the parsed plan that the placeholder lives in
`InsertIntoStatement.table` rather
- // than wrapping the whole command -- running the analyzer fully would
require a v2 catalog.
+ // The temporary binary node exposes only a dynamic 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`.
@@ -2554,14 +2543,18 @@ class ParametersSuite extends SharedSparkSession {
"""WITH transformation AS (SELECT 99 AS a)
|INSERT INTO IDENTIFIER('some' || '_table') REPLACE WHERE a = 10
|SELECT * FROM transformation""".stripMargin)
- val insert = parsedPlan.collectFirst { case i: InsertIntoStatement => i
}.getOrElse(
- fail(s"Expected InsertIntoStatement in parsed plan:\n$parsedPlan"))
+ val insert = parsedPlan.collectFirst { case i: UnresolvedInsert => i
}.getOrElse(
+ fail(s"Expected UnresolvedInsert in parsed plan:\n$parsedPlan"))
assert(insert.table.isInstanceOf[PlanWithUnresolvedIdentifier],
- s"Expected InsertIntoStatement.table to be PlanWithUnresolvedIdentifier,
" +
+ s"Expected UnresolvedInsert.table to be PlanWithUnresolvedIdentifier, " +
s"got ${insert.table.getClass.getSimpleName}:\n$parsedPlan")
- // After CTESubstitution runs, the CTE defs should land on the command's
children (because
- // InsertIntoStatement is a CTEInChildren) -- never as
`WithCTE(InsertIntoStatement, _)`.
+ assert(insert.children === Seq(insert.table, insert.query))
+
val substituted = CTESubstitution.apply(parsedPlan)
+ val substitutedInsert = substituted.collectFirst { case i:
UnresolvedInsert => i }.getOrElse(
+ fail(s"Expected UnresolvedInsert after CTE substitution:\n$substituted"))
+ assert(!substitutedInsert.table.exists(_.isInstanceOf[WithCTE]),
+ s"CTE definitions must not wrap the target table:\n$substituted")
substituted.foreach {
case WithCTE(_: CTEInChildren, _) =>
fail(s"Found invalid WithCTE(CTEInChildren, _) shape after
CTESubstitution:\n$substituted")
@@ -2569,30 +2562,40 @@ class ParametersSuite extends SharedSparkSession {
}
}
- // SPARK-46625: Parameter inside `IDENTIFIER(:p)` on REPLACE WHERE lives in
- // `InsertIntoStatement.table`, which is a non-child slot. Verify that
- // `BindParameters.bind` reaches into the slot via the explicit
`InsertIntoStatement`
- // recursion (parameters.scala) and that the `getDefaultTreePatternBits`
override on
- // `InsertIntoStatement` exposes the PARAMETER bit for pruning. Done at the
rule level
- // because driving REPLACE WHERE through full analysis would require a v2
catalog.
- test("SPARK-46625: BindParameters recurses into InsertIntoStatement.table") {
+ test("dynamic INSERT targets use a temporary binary plan") {
+ val regularInsert = spark.sessionState.sqlParser
+ .parsePlan("INSERT INTO target_table SELECT 1")
+ assert(regularInsert.isInstanceOf[InsertIntoStatement])
+
+ val identifierInsert = spark.sessionState.sqlParser
+ .parsePlan("INSERT INTO IDENTIFIER(lower('TESTCAT.NS.TARGET_TABLE'))
SELECT 1")
+ .asInstanceOf[UnresolvedInsert]
+ assert(identifierInsert.table.isInstanceOf[PlanWithUnresolvedIdentifier])
+ assert(identifierInsert.children === Seq(identifierInsert.table,
identifierInsert.query))
+
+ val lowered =
spark.sessionState.analyzer.resolveUnresolvedInsert(identifierInsert)
+ .asInstanceOf[InsertIntoStatement]
+ assert(lowered.table.asInstanceOf[UnresolvedRelation].multipartIdentifier
===
+ Seq("testcat", "ns", "target_table"))
+ assert(lowered.children === Seq(lowered.query))
+ }
+
+ // BindParameters reaches a dynamic target through normal child traversal.
+ test("SPARK-46625: BindParameters visits UnresolvedInsert.table") {
val parsedPlan = spark.sessionState.sqlParser.parsePlan(
"""INSERT INTO IDENTIFIER(:tname) REPLACE WHERE a = 10
|SELECT 1 AS a""".stripMargin)
- val insert = parsedPlan.collectFirst { case i: InsertIntoStatement => i
}.getOrElse(
- fail(s"Expected InsertIntoStatement in parsed plan:\n$parsedPlan"))
- // Pruning prerequisite: the PARAMETER bit must be visible at the
InsertIntoStatement
- // level (it lives inside `table`, which is not a child); this exercises
the
- // `getDefaultTreePatternBits` override.
+ val insert = parsedPlan.collectFirst { case i: UnresolvedInsert => i
}.getOrElse(
+ fail(s"Expected UnresolvedInsert in parsed plan:\n$parsedPlan"))
assert(insert.containsPattern(PARAMETER),
- "InsertIntoStatement.getDefaultTreePatternBits must propagate `table`'s
PARAMETER bit")
+ "UnresolvedInsert must propagate the table child's PARAMETER bit")
val bound = BindParameters.apply(
NameParameterizedQuery(parsedPlan, Seq("tname"),
Seq(Literal("foo_table"))))
- val boundInsert = bound.collectFirst { case i: InsertIntoStatement => i
}.getOrElse(
- fail(s"Expected InsertIntoStatement in bound plan:\n$bound"))
+ val boundInsert = bound.collectFirst { case i: UnresolvedInsert => i
}.getOrElse(
+ fail(s"Expected UnresolvedInsert in bound plan:\n$bound"))
assert(!boundInsert.table.containsPattern(PARAMETER),
- s"Expected :tname inside InsertIntoStatement.table to be bound,
got:\n$boundInsert")
+ s"Expected :tname inside UnresolvedInsert.table to be bound,
got:\n$boundInsert")
}
test("BindParameters binds IDENTIFIER(:p) in the table slot for REPLACE
WHERE with a column " +
@@ -2600,31 +2603,37 @@ class ParametersSuite extends SharedSparkSession {
val parsedPlan = spark.sessionState.sqlParser.parsePlan(
"""INSERT INTO IDENTIFIER(:tname) (id, name) REPLACE WHERE id = 1
|SELECT 1 AS id, 'x' AS name""".stripMargin)
- val insert = parsedPlan.collectFirst { case i: InsertIntoStatement => i
}.getOrElse(
- fail(s"Expected InsertIntoStatement in parsed plan:\n$parsedPlan"))
+ val insert = parsedPlan.collectFirst { case i: UnresolvedInsert => i
}.getOrElse(
+ fail(s"Expected UnresolvedInsert in parsed plan:\n$parsedPlan"))
assert(insert.containsPattern(PARAMETER),
- "InsertIntoStatement must expose the table slot's PARAMETER bit for
pruning")
+ "UnresolvedInsert must expose the table child's PARAMETER bit for
pruning")
val bound = BindParameters.apply(
NameParameterizedQuery(parsedPlan, Seq("tname"),
Seq(Literal("foo_table"))))
- val boundInsert = bound.collectFirst { case i: InsertIntoStatement => i
}.getOrElse(
- fail(s"Expected InsertIntoStatement in bound plan:\n$bound"))
+ val boundInsert = bound.collectFirst { case i: UnresolvedInsert => i
}.getOrElse(
+ fail(s"Expected UnresolvedInsert in bound plan:\n$bound"))
assert(!boundInsert.table.containsPattern(PARAMETER),
- s"Expected :tname inside InsertIntoStatement.table to be bound,
got:\n$boundInsert")
+ s"Expected :tname inside UnresolvedInsert.table to be bound,
got:\n$boundInsert")
assert(boundInsert.userSpecifiedCols === Seq("id", "name"),
s"Expected the column list to survive table-slot binding, got: " +
s"${boundInsert.userSpecifiedCols}")
}
- // SPARK-46625 followup: `INSERT INTO IDENTIFIER(<sql-variable>) ...` places
a
- // `PlanWithUnresolvedIdentifier` in `InsertIntoStatement.table`, whose
`identifierExpr`
- // holds an `UnresolvedAttribute` for the variable name. That slot is a
non-child
- // `LogicalPlan`, so the default `ResolveReferences` traversal never
resolves the
- // attribute, `ResolveIdentifierClause` cannot fire (it waits on
`identifierExpr.resolved`),
- // and analysis fails. Verify that the explicit `InsertIntoStatement` case
added to
- // `ResolveReferences` rewrites the attribute to a `VariableReference` and
the insert
- // completes end-to-end.
+ test("ResolveFunctions resolves expressions in an UnresolvedInsert target") {
+ val parsedPlan = spark.sessionState.sqlParser.parsePlan(
+ """INSERT INTO IDENTIFIER(lower(regexp_replace(
+ | 'TESTCAT.PLACEHOLDER', 'PLACEHOLDER', 'TAB'))) REPLACE WHERE id = 1
+ |SELECT 1 AS id""".stripMargin)
+ val insert = parsedPlan.asInstanceOf[UnresolvedInsert]
+
+ val resolvedInsert = spark.sessionState.analyzer.ResolveFunctions(insert)
+ .asInstanceOf[UnresolvedInsert]
+ assert(resolvedInsert.table.asInstanceOf[PlanWithUnresolvedIdentifier]
+ .identifierExpr.resolved)
+ }
+
+ // Normal child traversal resolves a SQL variable in the target IDENTIFIER
expression.
test("SPARK-46625: INSERT INTO IDENTIFIER(<sql-variable>) resolves variable
in table slot") {
withTable("t_var_insert") {
sql("CREATE TABLE t_var_insert (a INT) USING PARQUET")
@@ -2639,13 +2648,8 @@ class ParametersSuite extends SharedSparkSession {
}
}
- // SPARK-46625 followup: when the SQL variable name in `IDENTIFIER(<name>)`
collides
- // with a query output column, the IDENTIFIER expression must still bind to
the
- // variable, not to the column. The `ResolveReferences` case for
`InsertIntoStatement`
- // resolves `identifierExpr` against the `PlanWithUnresolvedIdentifier`
itself (whose
- // `children` are `Nil` on this path), not against the surrounding
`InsertIntoStatement`
- // (whose child is `query`), so query output columns are out of scope and
only the
- // last-resort variable resolution path fires.
+ // The identifier leaf has no children, so a colliding query output column
is out of scope and
+ // variable resolution remains the last resort.
test("SPARK-46625: INSERT INTO IDENTIFIER(<sql-variable>) ignores colliding
query columns") {
withTable("t_shadow") {
sql("CREATE TABLE t_shadow (a INT) USING PARQUET")
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 2dd031fb2c20..ad43f28eeea1 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
@@ -117,6 +117,16 @@ class ParseSqlResultSuite extends SparkFunSuite {
assert(sourceTableRefs("DROP TABLE t").isEmpty)
}
+ test("dynamic INSERT targets retain query metadata") {
+ val sql = "INSERT INTO IDENTIFIER(lower('T')) SELECT a AS result FROM src"
+ val insert = obj(sql)
+ assert(insert \ "statement_identifier" === JString("INSERT"))
+ assert(insert \ "statement_code" === JInt(50))
+ assert(sourceTableRefs(sql) === Set(Seq("src")))
+ assert(insert \ "select_list" === JArray(List(
+ JObject("name" -> JArray(List(JString("result")))))))
+ }
+
test("CTE aliases only shadow references within their own scope") {
// The inner CTE named real_t must not hide the outer real table real_t.
assert(sourceTableRefs(
@@ -127,6 +137,10 @@ class ParseSqlResultSuite extends SparkFunSuite {
assert(sourceTableRefs(
"WITH a AS (SELECT * FROM b), b AS (SELECT 1 AS x) SELECT * FROM a") ===
Set(Seq("b")))
+
+ 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")))
}
test("positional markers inside BEGIN END are counted once") {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/AppendDataTransactionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/AppendDataTransactionSuite.scala
index e48d99061671..88235686d70c 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/AppendDataTransactionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/AppendDataTransactionSuite.scala
@@ -175,6 +175,30 @@ class AppendDataTransactionSuite extends
RowLevelOperationSuiteBase {
Row(4, 400, "finance")))
}
+ test("SQL INSERT INTO with function-valued IDENTIFIER and transactional
checks") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }""")
+
+ val baseTargetLoadCount = catalog.loadTableCalls.count {
+ case (context, _) => !context.writePrivileges().isEmpty
+ }
+ val (txn, txnTables) = executeTransaction {
+ sql(s"INSERT INTO IDENTIFIER(lower('$tableNameAsString'))
$targetOptionsClause " +
+ "VALUES (2, 200, 'software')")
+ }
+
+ assert(txn.currentState === Committed)
+ assert(txn.isClosed)
+ assert(txnTables.isEmpty)
+ assert(catalog.loadTableCalls.count {
+ case (context, _) => !context.writePrivileges().isEmpty
+ } === baseTargetLoadCount)
+ assertTargetLoadAndWriteOptions(txn,
java.util.Set.of(TableWritePrivilege.INSERT))
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Seq(Row(1, 100, "hr"), Row(2, 200, "software")))
+ }
+
for (isDynamic <- Seq(false, true))
test(s"SQL INSERT OVERWRITE with transactional checks - isDynamic:
$isDynamic") {
// create table with initial data; table is partitioned by dep
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 a8dd94040b91..daef65a00ccd 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
@@ -24,7 +24,7 @@ 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.expressions.Expression
-import org.apache.spark.sql.catalyst.plans.logical.{AutoCdcInto,
CreateFlowCommand, CreateMaterializedViewAsSelect, CreateStreamingTable,
CreateStreamingTableAsSelect, CreateStreamingTableAutoCdc, CreateView,
InsertIntoStatement, LogicalPlan}
+import org.apache.spark.sql.catalyst.plans.logical.{AutoCdcInto,
CreateFlowCommand, CreateMaterializedViewAsSelect, CreateStreamingTable,
CreateStreamingTableAsSelect, CreateStreamingTableAutoCdc, CreateView,
InsertIntoStatement, 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}
@@ -590,30 +590,9 @@ class SqlGraphRegistrationContext(
cf.flowOperation match {
case i: InsertIntoStatement =>
- validateInsertIntoFlow(i, queryOrigin)
- val flowTargetDatasetName = i.table match {
- case u: UnresolvedRelation =>
- IdentifierHelper.toTableIdentifier(u.multipartIdentifier)
- case _ =>
- throw SqlGraphElementRegistrationException(
- msg = "Unable to resolve target dataset name for INSERT INTO
flow",
- queryOrigin = queryOrigin
- )
- }
- graphRegistrationContext.registerFlow(
- UntypedFlow(
- identifier = flowIdentifier,
- destinationIdentifier =
qualifyDestinationIdentifier(flowTargetDatasetName),
- func = FlowAnalysis.createFlowFunctionFromLogicalPlan(i.query),
- sqlConf = context.getSqlConf,
- once = false,
- queryContext = QueryContext(
- currentCatalog = context.getCurrentCatalogOpt,
- currentDatabase = context.getCurrentDatabaseOpt
- ),
- origin = queryOrigin
- )
- )
+ handleInsert(i, flowIdentifier, queryOrigin)
+ case i: UnresolvedInsert =>
+ handleInsert(i.toInsertIntoStatement(i.table), flowIdentifier,
queryOrigin)
case a: AutoCdcInto =>
val flowTargetDatasetName =
IdentifierHelper.toTableIdentifier(a.targetTable)
graphRegistrationContext.registerFlow(
@@ -648,6 +627,36 @@ class SqlGraphRegistrationContext(
}
}
+ private def handleInsert(
+ insert: InsertIntoStatement,
+ flowIdentifier: TableIdentifier,
+ queryOrigin: QueryOrigin): Unit = {
+ validateInsertIntoFlow(insert, queryOrigin)
+ val flowTargetDatasetName = insert.table match {
+ case u: UnresolvedRelation =>
+ IdentifierHelper.toTableIdentifier(u.multipartIdentifier)
+ case _ =>
+ throw SqlGraphElementRegistrationException(
+ msg = "Unable to resolve target dataset name for INSERT INTO flow",
+ queryOrigin = queryOrigin
+ )
+ }
+ graphRegistrationContext.registerFlow(
+ UntypedFlow(
+ identifier = flowIdentifier,
+ destinationIdentifier =
qualifyDestinationIdentifier(flowTargetDatasetName),
+ func = FlowAnalysis.createFlowFunctionFromLogicalPlan(insert.query),
+ sqlConf = context.getSqlConf,
+ once = false,
+ queryContext = QueryContext(
+ currentCatalog = context.getCurrentCatalogOpt,
+ currentDatabase = context.getCurrentDatabaseOpt
+ ),
+ origin = queryOrigin
+ )
+ )
+ }
+
/** Qualifies a raw flow target dataset identifier against the current
catalog/database. */
private def qualifyDestinationIdentifier(
flowTargetDatasetIdentifier: TableIdentifier): TableIdentifier =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
