This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 4b8c2c62314d [SPARK-58933][4.2][SQL] Resolve expressions in INSERT
target IDENTIFIER clauses
4b8c2c62314d is described below
commit 4b8c2c62314ddac52c7bf85fefec1f3ed7fad5e5
Author: Wenchen Fan <[email protected]>
AuthorDate: Fri Aug 28 16:15:06 2026 +0800
[SPARK-58933][4.2][SQL] Resolve expressions in INSERT target IDENTIFIER
clauses
### What changes were proposed in this pull request?
This backports #58204 (`3bc29d9d0ae031dfe0e793886ba652ca98ae7f29`) to
branch-4.2.
It introduces a temporary binary `UnresolvedInsert` plan for INSERT
statements whose target uses a
dynamic `IDENTIFIER(...)` expression. This lets normal analyzer traversal
resolve parameters,
functions, and SQL variables in the target before lowering the plan to the
existing write plan.
For branch-4.2, `INSERT ... REPLACE WHERE` is lowered to that branch's
`OverwriteByExpression` representation. The transactional regression test
is also expressed using
the older transaction test harness. The parse-metadata changes from the
original commit are omitted
because the corresponding `ParseSqlResult` and `SqlStatementCodes`
infrastructure is not present on
branch-4.2.
### Why are the changes needed?
`InsertIntoStatement.table` is not a logical-plan child, so analyzer rules
do not normally visit a
dynamic target expression stored there. Nested functions, parameters, or
SQL variables can
therefore remain unresolved.
### Does this PR introduce _any_ user-facing change?
Yes. INSERT statements on branch-4.2 can now resolve expressions inside a
target `IDENTIFIER`
clause, matching the behavior fixed on master by #58204.
### How was this patch tested?
The backported parser, analyzer, end-to-end, and transactional-catalog
tests were included and
adapted to branch-4.2. The following checks passed locally with Java 17:
```bash
build/sbt -java-home /usr/lib/jvm/java-17-openjdk-amd64 \
catalyst/Test/compile \
'sql/testOnly org.apache.spark.sql.ParametersSuite' \
'sql/testOnly org.apache.spark.sql.connector.AppendDataTransactionSuite' \
pipelines/compile connect/compile
```
`ParametersSuite`: 161 tests passed. `AppendDataTransactionSuite`: 18 tests
passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
Closes #58373 from cloud-fan/SPARK-58933-branch-4.2.
Authored-by: Wenchen Fan <[email protected]>
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 | 83 ++++++------
.../sql/catalyst/plans/logical/statements.scala | 92 +++++++++++--
.../sql/catalyst/rules/RuleIdCollection.scala | 1 +
.../spark/sql/catalyst/trees/TreePatterns.scala | 1 +
.../sql/connect/pipelines/PipelinesHandler.scala | 3 +-
.../spark/sql/execution/QueryExecution.scala | 22 +++-
.../org/apache/spark/sql/ParametersSuite.scala | 145 +++++++++++----------
.../sql/connector/AppendDataTransactionSuite.scala | 17 +++
.../graph/SqlGraphRegistrationContext.scala | 46 ++++---
14 files changed, 334 insertions(+), 209 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 de0e8a171962..03434579bc30 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
@@ -360,6 +360,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.toInsertPlan(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
@@ -537,6 +567,7 @@ class Analyzer(
override def batches: Seq[Batch] = earlyBatches ++ Seq(
Batch("Resolution", fixedPoint,
+ ResolveUnresolvedInsert ::
new ResolveCatalogs(catalogManager) ::
ResolveInsertInto ::
ResolveRelations ::
@@ -1313,6 +1344,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.toInsertPlan(toUnresolvedRelation(target))
+ }
+ }
+
/** Handle INSERT INTO for DSv2 */
object ResolveInsertInto extends ResolveInsertionBase {
override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsWithPruning(
@@ -1575,29 +1615,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 cfa6f3358806..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` and
- // `OverwriteByExpression` carry 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 31c835986e20..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). Today only the
`OverwriteByExpression`
- // variant of `V2WriteCommand` is parser-built with a placeholder in
`table`; the trait
- // match keeps the rule consistent for any future 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 a5b467d0f081..fdab9bec5e66 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 e6de6a1c9bc6..f7c4d6c80f06 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
@@ -933,11 +933,8 @@ class AstBuilder extends DataTypeAstBuilder
query: LogicalPlan,
queryAliasCtx: TableAliasContext): LogicalPlan = withOrigin(ctx) {
ctx match {
- // For all `InsertIntoStatement` / `OverwriteByExpression`-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 the
unary write plan.
case table: InsertIntoTableContext =>
val insertParams = visitInsertIntoTable(table)
val privileges = Set(TableWritePrivilege.INSERT)
@@ -976,27 +973,22 @@ class AstBuilder extends DataTypeAstBuilder
if (ctx.tableAlias() != null && ctx.tableAlias().strictIdentifier()
!= null) {
throw
QueryParsingErrors.insertReplaceWhereTableAliasNotAllowed(ctx.tableAlias())
}
- val options = Option(ctx.optionsClause())
+ val insertParams = InsertTableParams(
+ relationCtx = ctx.identifierReference,
+ options = Option(ctx.optionsClause()),
+ userSpecifiedCols = Seq.empty,
+ partitionSpec = Map.empty,
+ ifPartitionNotExists = false,
+ byName = ctx.NAME() != null,
+ replaceCriteriaOpt =
Some(InsertReplaceWhere(expression(ctx.replaceCondition))))
val privileges = Set(TableWritePrivilege.INSERT,
TableWritePrivilege.DELETE)
- // `PlanWithUnresolvedIdentifier` is a `NamedRelation`, so it can
occupy
- // `OverwriteByExpression.table` directly; the materialization
happens in
- // `ResolveIdentifierClause` via its `OverwriteByExpression`
special-case.
- val table = buildWriteTableSlot(ctx.identifierReference, options,
privileges)
- val deleteExpr = expression(ctx.replaceCondition)
- val isByName = ctx.NAME() != null
- if (isByName) {
- OverwriteByExpression.byName(
- table,
- df = query,
- deleteExpr,
- withSchemaEvolution = ctx.EVOLUTION() != null)
- } else {
- OverwriteByExpression.byPosition(
- table,
- query = query,
- deleteExpr,
- withSchemaEvolution = ctx.EVOLUTION() != null)
- }
+ createInsertIntoStatement(
+ insertParams = insertParams,
+ tableSlot = buildWriteTableSlot(
+ insertParams.relationCtx, insertParams.options, privileges),
+ query = query,
+ overwrite = true,
+ withSchemaEvolution = ctx.EVOLUTION() != null)
} else {
val insertParams = visitInsertIntoReplaceOn(ctx)
val privileges = Set(TableWritePrivilege.INSERT,
TableWritePrivilege.DELETE)
@@ -1150,16 +1142,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,
@@ -1169,27 +1159,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.toInsertPlan(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]], so the result
- * fits `NamedRelation`-typed slots (e.g. `OverwriteByExpression.table`) as
well as the more
- * general `LogicalPlan` slot of `InsertIntoStatement.table`.
- *
- * 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 4fbe71ed7d3e..99a8548ab443 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
@@ -17,13 +17,17 @@
package org.apache.spark.sql.catalyst.plans.logical
-import org.apache.spark.sql.catalyst.analysis.{FieldName, FieldPosition,
UnresolvedException}
+import org.apache.spark.sql.catalyst.analysis.{
+ FieldName,
+ FieldPosition,
+ NamedRelation,
+ 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 +162,71 @@ 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 appropriate unary write plan, 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
+ }
+
+ /** Converts this temporary node to the appropriate unary write plan. */
+ def toInsertPlan(target: LogicalPlan): LogicalPlan = replaceCriteriaOpt
match {
+ case Some(InsertReplaceWhere(condition)) =>
+ val relation = target.asInstanceOf[NamedRelation]
+ val overwrite = if (byName) {
+ OverwriteByExpression.byName(
+ relation, query, condition, withSchemaEvolution =
withSchemaEvolution)
+ } else {
+ OverwriteByExpression.byPosition(
+ relation, query, condition, withSchemaEvolution =
withSchemaEvolution)
+ }
+ overwrite.copyTagsFrom(this)
+ overwrite
+ case _ => toInsertIntoStatement(target)
+ }
+
+ 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.
@@ -211,16 +279,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`).
- override protected def getDefaultTreePatternBits: BitSet = {
- val bits = super.getDefaultTreePatternBits
- bits.union(table.treePatternBits)
- bits
- }
}
sealed abstract class InsertReplaceCriteria extends Expression with
Unevaluable {
@@ -249,3 +307,11 @@ case class InsertReplaceOn(
newChildren: IndexedSeq[Expression]): InsertReplaceOn =
copy(cond = newChildren.head)
}
+
+/** Rows are matched based on the specified boolean condition. */
+case class InsertReplaceWhere(cond: Expression) extends InsertReplaceCriteria {
+ override def children: Seq[Expression] = Seq(cond)
+ override protected def withNewChildrenInternal(
+ newChildren: IndexedSeq[Expression]): InsertReplaceWhere =
+ copy(cond = newChildren.head)
+}
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 a890d43f0672..f5cdba11b7f3 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 a18adc7ad833..4427530a778c 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
@@ -194,6 +194,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 a7249ef70e49..7623a81415c6 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/execution/QueryExecution.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
index d843af844d94..adaae69119cd 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}
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 c32cecbb9a82..aa7e90491fa6 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, Limit, OverwriteByExpression, 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,30 +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 goes through
`OverwriteByExpression`, whose `table`
- // slot is typed `NamedRelation`. `PlanWithUnresolvedIdentifier` extends
`NamedRelation` so the
- // placeholder sits in the slot directly. Verify on the parsed plan that the
placeholder lives
- // in `OverwriteByExpression.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`.
@@ -2555,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 overwrite = parsedPlan.collectFirst { case o: OverwriteByExpression =>
o }.getOrElse(
- fail(s"Expected OverwriteByExpression in parsed plan:\n$parsedPlan"))
- assert(overwrite.table.isInstanceOf[PlanWithUnresolvedIdentifier],
- s"Expected OverwriteByExpression.table to be
PlanWithUnresolvedIdentifier, " +
- s"got ${overwrite.table.getClass.getSimpleName}:\n$parsedPlan")
- // After CTESubstitution runs, the CTE defs should land on the command's
children (because
- // OverwriteByExpression is a CTEInChildren) -- never as
`WithCTE(OverwriteByExpression, _)`.
+ 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 UnresolvedInsert.table to be PlanWithUnresolvedIdentifier, " +
+ s"got ${insert.table.getClass.getSimpleName}:\n$parsedPlan")
+ 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")
@@ -2570,40 +2562,56 @@ class ParametersSuite extends SharedSparkSession {
}
}
- // SPARK-46625: Parameter inside `IDENTIFIER(:p)` on REPLACE WHERE lives in
- // `OverwriteByExpression.table`, which is a non-child slot. Verify that
- // `BindParameters.bind` reaches into the slot via the explicit
`OverwriteByExpression`
- // recursion (parameters.scala) and that the `getDefaultTreePatternBits`
override on
- // `OverwriteByExpression` 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
OverwriteByExpression.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 overwrite = parsedPlan.collectFirst { case o: OverwriteByExpression =>
o }.getOrElse(
- fail(s"Expected OverwriteByExpression in parsed plan:\n$parsedPlan"))
- // Pruning prerequisite: the PARAMETER bit must be visible at the
OverwriteByExpression
- // level (it lives inside `table`, which is not a child); this exercises
the
- // `getDefaultTreePatternBits` override.
- assert(overwrite.containsPattern(PARAMETER),
- "OverwriteByExpression.getDefaultTreePatternBits must propagate
`table`'s PARAMETER bit")
+ val insert = parsedPlan.collectFirst { case i: UnresolvedInsert => i
}.getOrElse(
+ fail(s"Expected UnresolvedInsert in parsed plan:\n$parsedPlan"))
+ assert(insert.containsPattern(PARAMETER),
+ "UnresolvedInsert must propagate the table child's PARAMETER bit")
val bound = BindParameters.apply(
NameParameterizedQuery(parsedPlan, Seq("tname"),
Seq(Literal("foo_table"))))
- val boundOverwrite = bound.collectFirst { case o: OverwriteByExpression =>
o }.getOrElse(
- fail(s"Expected OverwriteByExpression in bound plan:\n$bound"))
- assert(!boundOverwrite.table.containsPattern(PARAMETER),
- s"Expected :tname inside OverwriteByExpression.table to be bound,
got:\n$boundOverwrite")
- }
-
- // 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.
+ 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 UnresolvedInsert.table to be bound,
got:\n$boundInsert")
+ }
+
+ 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")
@@ -2618,13 +2626,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/connector/AppendDataTransactionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/AppendDataTransactionSuite.scala
index aef9c65550fc..c37f84fc60a7 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
@@ -98,6 +98,23 @@ 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 (txn, txnTables) = executeTransaction {
+ sql(s"INSERT INTO IDENTIFIER(lower('$tableNameAsString')) " +
+ "VALUES (2, 200, 'software')")
+ }
+
+ assert(txn.currentState === Committed)
+ assert(txn.isClosed)
+ assert(txnTables.isEmpty)
+ 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 4dfd09693578..eb2a8fa0ce13 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
@@ -20,9 +20,9 @@ import scala.collection.mutable
import org.apache.spark.{SparkException, SparkRuntimeException}
import org.apache.spark.sql.{AnalysisException, SparkSession}
-import org.apache.spark.sql.catalyst.QueryPlanningTracker
+import org.apache.spark.sql.catalyst.{QueryPlanningTracker, TableIdentifier}
import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
-import org.apache.spark.sql.catalyst.plans.logical.{CreateFlowCommand,
CreateMaterializedViewAsSelect, CreateStreamingTable,
CreateStreamingTableAsSelect, CreateView, InsertIntoStatement, LogicalPlan}
+import org.apache.spark.sql.catalyst.plans.logical.{CreateFlowCommand,
CreateMaterializedViewAsSelect, CreateStreamingTable,
CreateStreamingTableAsSelect, CreateView, InsertIntoStatement, LogicalPlan,
UnresolvedInsert}
import org.apache.spark.sql.catalyst.util.StringUtils
import org.apache.spark.sql.execution.command.{CreateViewCommand,
SetCatalogCommand, SetCommand, SetNamespaceCommand}
import org.apache.spark.sql.pipelines.Language
@@ -417,24 +417,9 @@ class SqlGraphRegistrationContext(
val (flowTargetDatasetIdentifier, flowQueryLogicalPlan) =
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
- )
- }
- val qualifiedFlowTargetDatasetName = GraphIdentifierManager
- .parseAndQualifyTableIdentifier(
- rawTableIdentifier = flowTargetDatasetName,
- currentCatalog = context.getCurrentCatalogOpt,
- currentDatabase = context.getCurrentDatabaseOpt
- )
- .identifier
- (qualifiedFlowTargetDatasetName, i.query)
+ extractInsertFlow(i, queryOrigin)
+ case i: UnresolvedInsert =>
+ extractInsertFlow(i.toInsertIntoStatement(i.table), queryOrigin)
case _ =>
throw SqlGraphElementRegistrationException(
msg = "Unable flow type. Only INSERT INTO flows are supported.",
@@ -466,6 +451,27 @@ class SqlGraphRegistrationContext(
)
}
+ private def extractInsertFlow(
+ insert: InsertIntoStatement,
+ queryOrigin: QueryOrigin): (TableIdentifier, LogicalPlan) = {
+ 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)
+ }
+ val qualifiedFlowTargetDatasetName = GraphIdentifierManager
+ .parseAndQualifyTableIdentifier(
+ rawTableIdentifier = flowTargetDatasetName,
+ currentCatalog = context.getCurrentCatalogOpt,
+ currentDatabase = context.getCurrentDatabaseOpt)
+ .identifier
+ (qualifiedFlowTargetDatasetName, insert.query)
+ }
+
private def validateInsertIntoFlow(
insertIntoStatement: InsertIntoStatement,
queryOrigin: QueryOrigin
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
