From fead749eb500f28c80b561591c55211a93eafb44 Mon Sep 17 00:00:00 2001 From: Wenchen Fan Date: Fri, 28 Aug 2026 10:27:06 +0800 Subject: [PATCH] [SPARK-58933][SQL] Resolve expressions in INSERT target IDENTIFIER clauses ### What changes were proposed in this pull request? This PR introduces a temporary binary `UnresolvedInsert` plan for INSERT statements whose target uses a dynamic `IDENTIFIER(...)` expression. Its target and input query are ordinary children, so analyzer rules can resolve parameters, functions, and SQL variables in the target expression through normal tree traversal. After the expression is evaluated, its builder creates an `UnresolvedInsertTarget` containing the raw multipart identifier, target options, and write privileges. The analyzer then converts this marker to an `UnresolvedRelation` and lowers `UnresolvedInsert` to the existing unary `InsertIntoStatement`. The target is subsequently qualified and resolved by the existing INSERT analysis path. `UnresolvedInsertTarget` is deliberately not a `NamedRelation`, so CTE substitution cannot treat an INSERT target as a readable relation. Static table names and literal `IDENTIFIER` clauses still parse directly to `InsertIntoStatement` and retain the existing behavior. `QueryExecution` resolves only the dynamic target identifier before transaction detection. This allows transactional INSERTs to be recognized without adding a separate target-resolution path or changing the long-lived child contract of `InsertIntoStatement`. `ResolveUnresolvedInsert` uses a dedicated tree pattern and rule ID, so the fixed-point Resolution batch skips plans that cannot contain the temporary carrier. ### Why are the changes needed? `InsertIntoStatement.table` is not a logical-plan child, so normal analyzer rules do not visit a dynamic `IDENTIFIER` expression stored in that slot. Nested functions, parameters, or SQL variables can therefore remain unresolved. Making `InsertIntoStatement` permanently binary would affect analyzer, planner, lineage, and command consumers that rely on its input query being its only child. The temporary `UnresolvedInsert` node limits the binary shape to identifier-expression resolution, while `UnresolvedInsertTarget` preserves the raw name until normal relation resolution. This also avoids qualifying an already-resolved identifier a second time. ### Does this PR introduce _any_ user-facing change? Yes. INSERT statements can now resolve expressions inside a target `IDENTIFIER` clause. For example, the following statement now resolves and executes correctly: ```sql INSERT INTO IDENTIFIER( lower(regexp_replace(:table_name, 'PLACEHOLDER', 'TBL'))) REPLACE WHERE id = 1 VALUES (1, 'updated') ``` ### How was this patch tested? Added parser, analyzer, end-to-end, parse-metadata lineage and classification, and transactional-catalog coverage for dynamic INSERT targets, including parameters, nested functions, SQL variables, qualified multipart names, query-source and select-list extraction, and a target sharing its name with a CTE. The following compile and test checks passed: ```bash build/sbt catalyst/Test/compile build/sbt 'sql/testOnly org.apache.spark.sql.ParametersSuite' build/sbt 'sql/testOnly org.apache.spark.sql.catalyst.parser.ParseSqlResultSuite' build/sbt 'sql/testOnly org.apache.spark.sql.connector.AppendDataTransactionSuite' build/sbt pipelines/compile build/sbt connect/compile SPARK_GENERATE_GOLDEN_FILES=1 build/sbt \ 'sql/testOnly org.apache.spark.sql.SQLQueryTestSuite -- -z explain.sql' SPARK_GENERATE_GOLDEN_FILES=1 build/sbt \ 'sql/testOnly org.apache.spark.sql.SQLQueryTestSuite -- -z explain-aqe.sql' ``` ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex (GPT-5) Closes #58204 from cloud-fan/fix-identifier-dml-table-resolution. Authored-by: Wenchen Fan Signed-off-by: Wenchen Fan --- .../sql/catalyst/analysis/Analyzer.scala | 65 +++++--- .../analysis/ResolveIdentifierClause.scala | 18 +-- .../analysis/ResolveInlineTables.scala | 15 +- .../sql/catalyst/analysis/parameters.scala | 13 +- .../sql/catalyst/analysis/unresolved.scala | 22 ++- .../sql/catalyst/parser/AstBuilder.scala | 83 +++++----- .../catalyst/plans/logical/statements.scala | 92 +++++++++-- .../sql/catalyst/rules/RuleIdCollection.scala | 1 + .../sql/catalyst/trees/TreePatterns.scala | 1 + .../connect/pipelines/PipelinesHandler.scala | 3 +- .../spark/sql/execution/QueryExecution.scala | 22 ++- .../apache/spark/sql/ParametersSuite.scala | 145 +++++++++--------- .../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 de0e8a171962f..03434579bc302 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 cfa6f33588062..320cfc0a77c37 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 1c9bf9dd22bc3..800b346273503 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 31c835986e20d..1dac9a7b9f7d6 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 a5b467d0f0816..fdab9bec5e665 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 ... ` 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 e6de6a1c9bc6c..f7c4d6c80f06b 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 4fbe71ed7d3e5..99a8548ab4430 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 a890d43f0672c..f5cdba11b7f33 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 a18adc7ad8330..4427530a778c9 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 a7249ef70e49c..7623a81415c6d 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 d843af844d94a..adaae69119cdd 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 c32cecbb9a826..aa7e90491fa64 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 ... 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() ...` 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() 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()` 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() 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 aef9c65550fc4..c37f84fc60a72 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 4dfd096935781..eb2a8fa0ce134 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