This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new 4e111b4dadca [SPARK-58933][4.3][SQL] Resolve expressions in INSERT
target IDENTIFIER clauses
4e111b4dadca is described below
commit 4e111b4dadca06177122077eca80e84bde270952
Author: Wenchen Fan <[email protected]>
AuthorDate: Fri Aug 28 16:13:42 2026 +0800
[SPARK-58933][4.3][SQL] Resolve expressions in INSERT target IDENTIFIER
clauses
### What changes were proposed in this pull request?
This backports #58204 (`3bc29d9d0ae031dfe0e793886ba652ca98ae7f29`) to
branch-4.3.
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
`InsertIntoStatement`.
The parse-metadata changes from the original commit are omitted because the
corresponding
`ParseSqlResult` and `SqlStatementCodes` infrastructure is not present on
branch-4.3.
### 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.3 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. 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`: 162 tests passed. `AppendDataTransactionSuite`: 19 tests
passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
Closes #58372 from cloud-fan/SPARK-58933-branch-4.3.
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 | 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/execution/QueryExecution.scala | 22 +++-
.../org/apache/spark/sql/ParametersSuite.scala | 144 +++++++++++----------
.../sql/connector/AppendDataTransactionSuite.scala | 24 ++++
.../graph/SqlGraphRegistrationContext.scala | 59 +++++----
14 files changed, 305 insertions(+), 193 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 176980e1fafd..a1fc8012d0c3 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 ::
@@ -1352,6 +1383,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(
@@ -1630,29 +1670,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 534f87c1e7eb..6930a13a0b62 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 d2c1ab9a8bc2..907e7c6cc051 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 0679d0c332df..7bb61fea6658 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
@@ -203,6 +203,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/execution/QueryExecution.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
index 1fa70351f2d0..586cb5226173 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/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]
