diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index dd3220d10b284..8b3b96a8fefeb 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -3224,6 +3224,12 @@ ], "sqlState" : "0A000" }, + "INSERT_REPLACE_WHERE_COLUMN_LIST_NOT_ENABLED" : { + "message" : [ + "INSERT INTO ... (column_list) REPLACE WHERE is not enabled. Set 'spark.sql.insertIntoReplaceWhereColumnList.enabled' to true to enable it." + ], + "sqlState" : "0A000" + }, "INSERT_REPLACE_WHERE_TABLE_ALIAS_NOT_ALLOWED" : { "message" : [ "Table alias is not allowed with INSERT INTO ... REPLACE WHERE because the WHERE condition is evaluated against the target table directly.", diff --git a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 index 38d4460965fb1..00090f83de62f 100644 --- a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 +++ b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 @@ -594,7 +594,7 @@ query insertInto : INSERT (WITH SCHEMA EVOLUTION)? OVERWRITE TABLE? identifierReference optionsClause? (partitionSpec (IF errorCapturingNot EXISTS)?)? ((BY NAME) | identifierList)? #insertOverwriteTable | INSERT (WITH SCHEMA EVOLUTION)? INTO TABLE? identifierReference optionsClause? partitionSpec? (IF errorCapturingNot EXISTS)? ((BY NAME) | identifierList)? #insertIntoTable - | INSERT (WITH SCHEMA EVOLUTION)? INTO TABLE? identifierReference tableAlias optionsClause? (BY NAME)? + | INSERT (WITH SCHEMA EVOLUTION)? INTO TABLE? identifierReference tableAlias optionsClause? ((BY NAME) | identifierList)? REPLACE (WHERE | ON) replaceCondition=booleanExpression #insertIntoReplaceBooleanCond | INSERT (WITH SCHEMA EVOLUTION)? INTO TABLE? identifierReference tableAlias optionsClause? (BY NAME)? REPLACE USING identifierList #insertIntoReplaceUsing diff --git a/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala b/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala index 01ba078be7298..4ef7f4ef1318c 100644 --- a/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala +++ b/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala @@ -72,6 +72,13 @@ private[sql] object QueryParsingErrors extends DataTypeErrorsBase { ctx) } + def insertReplaceWhereColumnListNotEnabled(ctx: ParserRuleContext): Throwable = { + new ParseException( + errorClass = "INSERT_REPLACE_WHERE_COLUMN_LIST_NOT_ENABLED", + messageParameters = Map.empty, + ctx) + } + def insertReplaceWhereTableAliasNotAllowed(ctx: TableAliasContext): Throwable = { new ParseException( errorClass = "INSERT_REPLACE_WHERE_TABLE_ALIAS_NOT_ALLOWED", 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 49fdbd45b91c2..5bdb4d276d93c 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 @@ -1154,6 +1154,9 @@ class Analyzer( // Thus, we need to look at the raw plan if `relation` is a temporary view. // unwrapRelationPlan also resolves V2TableReference nodes in temp view plans. unwrapRelationPlan(relation) match { + // Only REPLACE WHERE is rejected here. REPLACE ON/USING is rejected separately. + case v: View if i.replaceCriteriaOpt.exists(_.isReplaceWhere) => + throw QueryCompilationErrors.writeIntoViewNotAllowedError(v.desc.identifier, i) case v: View => throw QueryCompilationErrors.insertIntoViewNotAllowedError(v.desc.identifier, table) case other => i.copy(table = other) @@ -1345,14 +1348,15 @@ class Analyzer( case i: InsertIntoStatement if i.table.isInstanceOf[DataSourceV2Relation] && i.query.resolved && - i.replaceCriteriaOpt.isDefined => + i.replaceCriteriaOpt.exists(_.isReplaceOnOrUsing) => throw QueryCompilationErrors.unsupportedInsertReplaceOnOrUsing( i.table.asInstanceOf[DataSourceV2Relation].table.name()) + // Handles all plain INSERTs and REPLACE WHERE. case i: InsertIntoStatement if i.table.isInstanceOf[DataSourceV2Relation] && i.query.resolved && - i.replaceCriteriaOpt.isEmpty => + (i.replaceCriteriaOpt.isEmpty || i.replaceCriteriaOpt.exists(_.isReplaceWhere)) => val r = i.table.asInstanceOf[DataSourceV2Relation] // ifPartitionNotExists is append with validation, but validation is not supported if (i.ifPartitionNotExists) { @@ -1386,7 +1390,11 @@ class Analyzer( query, withSchemaEvolution = i.withSchemaEvolution) } - } else if (conf.partitionOverwriteMode == PartitionOverwriteMode.DYNAMIC) { + // Dynamic partition overwrite applies only to plain INSERT OVERWRITE. REPLACE WHERE always + // deletes by its condition, so it falls through to the OverwriteByExpression branch below + // even when the session is in DYNAMIC partition-overwrite mode. + } else if (i.replaceCriteriaOpt.isEmpty && + conf.partitionOverwriteMode == PartitionOverwriteMode.DYNAMIC) { if (isByName) { OverwritePartitionsDynamic.byName( r, @@ -1399,17 +1407,27 @@ class Analyzer( withSchemaEvolution = i.withSchemaEvolution) } } else { + val deleteExpr = i.replaceCriteriaOpt match { + case Some(InsertReplaceWhere(condition)) => + assert(staticPartitions.isEmpty, + s"REPLACE WHERE must not carry static partitions, but got: $staticPartitions") + condition + case Some(other) => throw SparkException.internalError( + s"Replace criteria ${other.getClass.getSimpleName} must not reach " + + "ResolveInsertInto; REPLACE ON/USING are rejected earlier.") + case None => staticDeleteExpression(r, staticPartitions) + } if (isByName) { OverwriteByExpression.byName( table = r, df = query, - deleteExpr = staticDeleteExpression(r, staticPartitions), + deleteExpr = deleteExpr, withSchemaEvolution = i.withSchemaEvolution) } else { OverwriteByExpression.byPosition( table = r, query = query, - deleteExpr = staticDeleteExpression(r, staticPartitions), + deleteExpr = deleteExpr, withSchemaEvolution = i.withSchemaEvolution) } } 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..2bccc99f31d86 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 @@ -73,9 +73,9 @@ class ResolveIdentifierClause(earlyBatches: Seq[RuleExecutor[LogicalPlan]#Batch] 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. + // 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) { 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..deb63a7d97893 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 @@ -183,9 +183,9 @@ object BindParameters extends Rule[LogicalPlan] with QueryErrorsBase { // 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. + // 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. val withBoundTable = p1 match { case i: InsertIntoStatement if i.table.containsPattern(PARAMETER) => i.copy(table = bind(i.table)(f)) 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 8389e79331a96..bebd45869c1c0 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 @@ -938,11 +938,10 @@ 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). + // 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). case table: InsertIntoTableContext => val insertParams = visitInsertIntoTable(table) val privileges = Set(TableWritePrivilege.INSERT) @@ -972,36 +971,15 @@ class AstBuilder extends DataTypeAstBuilder // while REPLACE WHERE still can. val isInsertReplaceWhere = ctx.WHERE() != null if (isInsertReplaceWhere) { - // The unified grammar rule for REPLACE WHERE | ON accepts a table alias for - // symmetry with REPLACE ON (whose condition can reference the target via the - // alias, e.g. `t.col`). The REPLACE WHERE branch has no use for the alias - // because the WHERE condition is evaluated against the target table directly. - // Reject explicitly so users get a clear parse error instead of a confusing - // column-not-found at analysis time. - if (ctx.tableAlias() != null && ctx.tableAlias().strictIdentifier() != null) { - throw QueryParsingErrors.insertReplaceWhereTableAliasNotAllowed(ctx.tableAlias()) - } - val options = Option(ctx.optionsClause()) + val insertParams = visitInsertIntoReplaceWhere(ctx) 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) @@ -1108,9 +1086,10 @@ class AstBuilder extends DataTypeAstBuilder } val replaceUsingCols = visitIdentifierList(ctx.identifierList()) - createInsertIntoReplaceOnOrUsingParams( + createInsertIntoReplaceParams( relationCtx = ctx.identifierReference(), optionsCtx = ctx.optionsClause(), + userSpecifiedCols = Seq.empty, byName = byName, replaceCriteriaOpt = Some(InsertReplaceUsing(replaceUsingCols)) ) @@ -1132,23 +1111,56 @@ class AstBuilder extends DataTypeAstBuilder val tableAliasOpt = getTableAliasWithoutColumnAlias(ctx.tableAlias(), "INSERT REPLACE ON") - createInsertIntoReplaceOnOrUsingParams( + createInsertIntoReplaceParams( relationCtx = ctx.identifierReference(), optionsCtx = ctx.optionsClause(), + userSpecifiedCols = Seq.empty, byName = byName, replaceCriteriaOpt = Some(InsertReplaceOn(replaceOnCond, tableAliasOpt)) ) } - private def createInsertIntoReplaceOnOrUsingParams( + /** + * Add an INSERT INTO REPLACE WHERE operation to the logical plan. + */ + def visitInsertIntoReplaceWhere( + ctx: InsertIntoReplaceBooleanCondContext): InsertTableParams = withOrigin(ctx) { + // The unified grammar rule for REPLACE WHERE | ON accepts a table alias for symmetry with + // REPLACE ON (whose condition can reference the target via the alias, e.g. `t.col`). The + // REPLACE WHERE branch has no use for the alias because the WHERE condition is evaluated + // against the target table directly. Reject explicitly so users get a clear parse error + // instead of a confusing column-not-found at analysis time. + if (ctx.tableAlias() != null && ctx.tableAlias().strictIdentifier() != null) { + throw QueryParsingErrors.insertReplaceWhereTableAliasNotAllowed(ctx.tableAlias()) + } + // BY NAME and a column list are mutually exclusive in the grammar. + val userSpecifiedCols = Option(ctx.identifierList()).map(visitIdentifierList).getOrElse(Nil) + if (userSpecifiedCols.nonEmpty && + !SQLConf.get.getConf(SQLConf.INSERT_INTO_REPLACE_WHERE_COLUMN_LIST_ENABLED)) { + throw QueryParsingErrors.insertReplaceWhereColumnListNotEnabled(ctx.identifierList()) + } + + val deleteExpr = expression(ctx.replaceCondition) + + createInsertIntoReplaceParams( + relationCtx = ctx.identifierReference(), + optionsCtx = ctx.optionsClause(), + userSpecifiedCols = userSpecifiedCols, + byName = ctx.NAME() != null, + replaceCriteriaOpt = Some(InsertReplaceWhere(deleteExpr)) + ) + } + + private def createInsertIntoReplaceParams( relationCtx: IdentifierReferenceContext, optionsCtx: OptionsClauseContext, + userSpecifiedCols: Seq[String], byName: Boolean, replaceCriteriaOpt: Option[InsertReplaceCriteria]): InsertTableParams = { InsertTableParams( relationCtx = relationCtx, options = Option(optionsCtx), - userSpecifiedCols = Seq.empty, + userSpecifiedCols = userSpecifiedCols, partitionSpec = Map[String, Option[String]](), ifPartitionNotExists = false, byName = byName, @@ -1180,9 +1192,9 @@ class AstBuilder extends DataTypeAstBuilder * 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`. + * 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 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..e10fa680ddda7 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 @@ -174,7 +174,7 @@ case class QualifiedColType( * Only valid for static partitions. * @param byName If true, reorder the data columns to match the column names of the * target table. - * @param replaceCriteriaOpt If specified, indicates an INSERT REPLACE ON/USING operation, + * @param replaceCriteriaOpt If specified, indicates an INSERT REPLACE ON/USING/WHERE operation, * which atomically deletes existing rows that satisfy the replace * criteria and then inserts the query result rows into the table. * @param withSchemaEvolution If true, enables automatic schema evolution for the operation. @@ -201,12 +201,13 @@ case class InsertIntoStatement( "IF NOT EXISTS is only valid with static partitions") require(userSpecifiedCols.isEmpty || !byName, "BY NAME is only valid without specified cols") - require(replaceCriteriaOpt.isEmpty || userSpecifiedCols.isEmpty, - "userSpecifiedCols is not compatible with REPLACE USING/ON") + require( + !replaceCriteriaOpt.exists(_.isInstanceOf[InsertReplaceUsing]) || userSpecifiedCols.isEmpty, + "userSpecifiedCols is not compatible with REPLACE USING") require(replaceCriteriaOpt.isEmpty || partitionSpec.isEmpty, - "partitionSpec is not compatible with REPLACE USING/ON") + "partitionSpec is not compatible with REPLACE USING/ON/WHERE") require(replaceCriteriaOpt.isEmpty || overwrite, - "REPLACE USING/ON requires overwrite to be true") + "REPLACE USING/ON/WHERE requires overwrite to be true") override def child: LogicalPlan = query override protected def withNewChildInternal(newChild: LogicalPlan): InsertIntoStatement = @@ -215,7 +216,9 @@ case class InsertIntoStatement( // `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`). + // (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) @@ -226,6 +229,23 @@ case class InsertIntoStatement( sealed abstract class InsertReplaceCriteria extends Expression with Unevaluable { override def nullable: Boolean = false override def dataType: DataType = throw new UnresolvedException("dataType") + + /** + * True for INSERT ... REPLACE WHERE, which `ResolveInsertInto` rewrites into an + * [[OverwriteByExpression]]. + */ + def isReplaceWhere: Boolean = this match { + case _: InsertReplaceWhere => true + case _: InsertReplaceOn | _: InsertReplaceUsing => false + } + + /** + * True for INSERT ... REPLACE ON / REPLACE USING. + */ + def isReplaceOnOrUsing: Boolean = this match { + case _: InsertReplaceOn | _: InsertReplaceUsing => true + case _: InsertReplaceWhere => false + } } /** @@ -249,3 +269,13 @@ 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/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index 6fe7a957a1216..349a59f99ae60 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -5314,6 +5314,16 @@ object SQLConf { .booleanConf .createWithDefault(true) + val INSERT_INTO_REPLACE_WHERE_COLUMN_LIST_ENABLED = + buildConf("spark.sql.insertIntoReplaceWhereColumnList.enabled") + .doc("Enable the SQL syntax INSERT INTO ... (column_list) REPLACE WHERE (...). " + + "Allows using a column list with INSERT INTO REPLACE WHERE.") + .internal() + .version("4.3.0") + .withBindingPolicy(ConfigBindingPolicy.SESSION) + .booleanConf + .createWithDefault(true) + val BIN_BY_ENABLED = buildConf("spark.sql.binByRelationOperator.enabled") .doc("Enable the BIN BY relation operator for aligning range-typed rows to " + diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala index f8d16cb0097c1..ea73353f11ca2 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala @@ -1846,23 +1846,30 @@ class DDLParserSuite extends AnalysisTest { test("insert table: REPLACE WHERE with BY NAME") { parseCompare( "INSERT INTO testcat.ns1.ns2.tbl BY NAME REPLACE WHERE a > 5 SELECT * FROM source", - OverwriteByExpression.byName( - UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")), - Project(Seq(UnresolvedStar(None)), UnresolvedRelation(Seq("source"))), - GreaterThan( - UnresolvedAttribute("a"), - Literal(5)))) + InsertIntoStatement( + table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")), + partitionSpec = Map.empty, + userSpecifiedCols = Nil, + query = Project(Seq(UnresolvedStar(None)), UnresolvedRelation(Seq("source"))), + overwrite = true, + ifPartitionNotExists = false, + byName = true, + replaceCriteriaOpt = Some(InsertReplaceWhere( + GreaterThan(UnresolvedAttribute("a"), Literal(5)))))) } test("insert table: REPLACE WHERE without BY NAME") { parseCompare( "INSERT INTO testcat.ns1.ns2.tbl REPLACE WHERE a > 5 SELECT * FROM source", - OverwriteByExpression.byPosition( - UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")), - Project(Seq(UnresolvedStar(None)), UnresolvedRelation(Seq("source"))), - GreaterThan( - UnresolvedAttribute("a"), - Literal(5)))) + InsertIntoStatement( + table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")), + partitionSpec = Map.empty, + userSpecifiedCols = Nil, + query = Project(Seq(UnresolvedStar(None)), UnresolvedRelation(Seq("source"))), + overwrite = true, + ifPartitionNotExists = false, + replaceCriteriaOpt = Some(InsertReplaceWhere( + GreaterThan(UnresolvedAttribute("a"), Literal(5)))))) } test("insert table: REPLACE WHERE rejects tableAlias with BY NAME") { @@ -1889,6 +1896,99 @@ class DDLParserSuite extends AnalysisTest { start = 0, stop = 55)) } + test("insert table: REPLACE WHERE rejects tableAlias with column list") { + checkError( + exception = parseException( + "INSERT INTO testcat.ns1.ns2.tbl AS t (a, b) REPLACE WHERE a > 5 SELECT * FROM source"), + condition = "INSERT_REPLACE_WHERE_TABLE_ALIAS_NOT_ALLOWED", + parameters = Map.empty, + context = ExpectedContext( + fragment = "INSERT INTO testcat.ns1.ns2.tbl AS t (a, b) REPLACE WHERE a > 5", + start = 0, stop = 62)) + } + + test("insert table: REPLACE WHERE with column list") { + parseCompare( + "INSERT INTO testcat.ns1.ns2.tbl (a, b) REPLACE WHERE a > 5 SELECT * FROM source", + InsertIntoStatement( + table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")), + partitionSpec = Map.empty, + userSpecifiedCols = Seq("a", "b"), + query = Project(Seq(UnresolvedStar(None)), UnresolvedRelation(Seq("source"))), + overwrite = true, + ifPartitionNotExists = false, + replaceCriteriaOpt = Some(InsertReplaceWhere( + GreaterThan(UnresolvedAttribute("a"), Literal(5)))))) + } + + test("insert table: REPLACE WHERE rejects BY NAME with column list") { + checkError( + exception = parseException( + "INSERT INTO testcat.ns1.ns2.tbl BY NAME (a, b) REPLACE WHERE a > 5 SELECT * FROM source"), + condition = "PARSE_SYNTAX_ERROR", + parameters = Map("error" -> "'a'", "hint" -> "")) + + checkError( + exception = parseException( + "INSERT INTO testcat.ns1.ns2.tbl (a, b) BY NAME REPLACE WHERE a > 5 SELECT * FROM source"), + condition = "PARSE_SYNTAX_ERROR", + parameters = Map("error" -> "'BY'", "hint" -> "")) + } + + test("insert table: REPLACE WHERE rejects column list when the feature flag is disabled") { + withSQLConf(SQLConf.INSERT_INTO_REPLACE_WHERE_COLUMN_LIST_ENABLED.key -> "false") { + checkError( + exception = parseException( + "INSERT INTO testcat.ns1.ns2.tbl (a, b) REPLACE WHERE a > 5 SELECT * FROM source"), + condition = "INSERT_REPLACE_WHERE_COLUMN_LIST_NOT_ENABLED", + parameters = Map.empty, + context = ExpectedContext( + fragment = "INSERT INTO testcat.ns1.ns2.tbl (a, b) REPLACE WHERE a > 5", + start = 0, + stop = 57)) + } + } + + test("insert table: REPLACE WHERE rejects empty column list") { + checkError( + exception = parseException( + "INSERT INTO testcat.ns1.ns2.tbl () REPLACE WHERE a > 5 SELECT * FROM source"), + condition = "PARSE_SYNTAX_ERROR", + parameters = Map("error" -> "')'", "hint" -> "")) + } + + test("insert table: INSERT WITH SCHEMA EVOLUTION INTO ... (cols) REPLACE WHERE") { + parseCompare( + "INSERT WITH SCHEMA EVOLUTION INTO testcat.ns1.ns2.tbl (a, b) " + + "REPLACE WHERE a > 5 SELECT * FROM source", + InsertIntoStatement( + table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")), + partitionSpec = Map.empty, + userSpecifiedCols = Seq("a", "b"), + query = Project(Seq(UnresolvedStar(None)), UnresolvedRelation(Seq("source"))), + overwrite = true, + ifPartitionNotExists = false, + replaceCriteriaOpt = Some(InsertReplaceWhere( + GreaterThan(UnresolvedAttribute("a"), Literal(5)))), + withSchemaEvolution = true)) + } + + test("insert table: REPLACE WHERE with column list and options") { + val opts = new CaseInsensitiveStringMap(java.util.Map.of("key", "value")) + parseCompare( + "INSERT INTO testcat.ns1.ns2.tbl WITH (key = 'value') (a, b) " + + "REPLACE WHERE a > 5 SELECT * FROM source", + InsertIntoStatement( + table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl"), opts), + partitionSpec = Map.empty, + userSpecifiedCols = Seq("a", "b"), + query = Project(Seq(UnresolvedStar(None)), UnresolvedRelation(Seq("source"))), + overwrite = true, + ifPartitionNotExists = false, + replaceCriteriaOpt = Some(InsertReplaceWhere( + GreaterThan(UnresolvedAttribute("a"), Literal(5)))))) + } + for { isByName <- Seq(true, false) userSpecifiedCols <- if (!isByName) { @@ -2136,12 +2236,15 @@ class DDLParserSuite extends AnalysisTest { """INSERT INTO testcat.ns1.ns2.tbl |REPLACE WHERE a > 5 |(SELECT * FROM source) AS s""".stripMargin, - expected = OverwriteByExpression.byPosition( - UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")), - Project(Seq(UnresolvedStar(None)), UnresolvedRelation(Seq("source"))), - GreaterThan( - UnresolvedAttribute("a"), - Literal(5)))) + expected = InsertIntoStatement( + table = UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")), + partitionSpec = Map.empty, + userSpecifiedCols = Nil, + query = Project(Seq(UnresolvedStar(None)), UnresolvedRelation(Seq("source"))), + overwrite = true, + ifPartitionNotExists = false, + replaceCriteriaOpt = Some(InsertReplaceWhere( + GreaterThan(UnresolvedAttribute("a"), Literal(5)))))) } test("INSERT INTO REPLACE ON with compound condition") { 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 ca7732772b588..45ceea27ece60 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.expressions.Literal +import org.apache.spark.sql.catalyst.analysis.{BindParameters, CTESubstitution, ExpressionWithUnresolvedIdentifier, NameParameterizedQuery, PlanWithUnresolvedIdentifier, UnresolvedAttribute} +import org.apache.spark.sql.catalyst.expressions.{EqualTo, 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, InsertReplaceWhere, Limit, ReplaceTableAsSelect, 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 @@ -2533,11 +2533,10 @@ class ParametersSuite extends SharedSparkSession { } } - // 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. + // 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. 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`. @@ -2545,13 +2544,13 @@ 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") + val insert = parsedPlan.collectFirst { case i: InsertIntoStatement => i }.getOrElse( + fail(s"Expected InsertIntoStatement in parsed plan:\n$parsedPlan")) + assert(insert.table.isInstanceOf[PlanWithUnresolvedIdentifier], + s"Expected InsertIntoStatement.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 - // OverwriteByExpression is a CTEInChildren) -- never as `WithCTE(OverwriteByExpression, _)`. + // InsertIntoStatement is a CTEInChildren) -- never as `WithCTE(InsertIntoStatement, _)`. val substituted = CTESubstitution.apply(parsedPlan) substituted.foreach { case WithCTE(_: CTEInChildren, _) => @@ -2561,29 +2560,51 @@ 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` + // `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 - // `OverwriteByExpression` exposes the PARAMETER bit for pruning. Done at the rule level + // `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 OverwriteByExpression.table") { + test("SPARK-46625: BindParameters recurses into InsertIntoStatement.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 + 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. - assert(overwrite.containsPattern(PARAMETER), - "OverwriteByExpression.getDefaultTreePatternBits must propagate `table`'s PARAMETER bit") + assert(insert.containsPattern(PARAMETER), + "InsertIntoStatement.getDefaultTreePatternBits must propagate `table`'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") + val boundInsert = bound.collectFirst { case i: InsertIntoStatement => i }.getOrElse( + fail(s"Expected InsertIntoStatement in bound plan:\n$bound")) + assert(!boundInsert.table.containsPattern(PARAMETER), + s"Expected :tname inside InsertIntoStatement.table to be bound, got:\n$boundInsert") + } + + test("BindParameters binds IDENTIFIER(:p) in the table slot for REPLACE WHERE with a column " + + "list and preserves the column list") { + 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")) + assert(insert.containsPattern(PARAMETER), + "InsertIntoStatement must expose the table slot'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")) + assert(!boundInsert.table.containsPattern(PARAMETER), + s"Expected :tname inside InsertIntoStatement.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() ...` places a @@ -2706,4 +2727,26 @@ class ParametersSuite extends SharedSparkSession { case _ => } } + + test("BindParameters binds parameters inside the REPLACE WHERE condition") { + val parsedPlan = spark.sessionState.sqlParser.parsePlan( + """INSERT INTO some_table REPLACE WHERE a = :val + |SELECT 1 AS a""".stripMargin) + val insert = parsedPlan.collectFirst { case i: InsertIntoStatement => i }.getOrElse( + fail(s"Expected InsertIntoStatement in parsed plan:\n$parsedPlan")) + assert(insert.containsPattern(PARAMETER), + "InsertIntoStatement must expose the REPLACE WHERE condition's PARAMETER bit for pruning") + + val bound = BindParameters.apply( + NameParameterizedQuery(parsedPlan, Seq("val"), Seq(Literal(10)))) + + val boundInsert = bound.collectFirst { case i: InsertIntoStatement => i }.getOrElse( + fail(s"Expected InsertIntoStatement in bound plan:\n$bound")) + boundInsert.replaceCriteriaOpt match { + case Some(InsertReplaceWhere(cond)) => + assert(cond === EqualTo(UnresolvedAttribute("a"), Literal(10)), + s"Expected :val inside the REPLACE WHERE condition to be bound, got:\n$cond") + case other => fail(s"Expected InsertReplaceWhere criteria, got: $other") + } + } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala index 6968a75f5eee3..42d813b91cf53 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala @@ -1397,6 +1397,23 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { ) } + test("INSERT INTO REPLACE WHERE with column list") { + val parsed = parseAndResolve( + """INSERT INTO testcat.tab (i, s) + |REPLACE WHERE i = 1 + |SELECT * FROM v2Table""".stripMargin) + parsed match { + case overwriteByExpression: OverwriteByExpression => + // ResolveInsertInto resolves the column list into a Project and converts the statement + // into a by-name OverwriteByExpression whose delete expression is the REPLACE WHERE + // condition, resolved against the target table. + assert(overwriteByExpression.isByName) + assert(overwriteByExpression.deleteExpr.resolved) + case other => + fail(s"Expected OverwriteByExpression, but got: $other") + } + } + test("INSERT INTO REPLACE ON is blocked when feature flag is disabled") { withSQLConf(SQLConf.INSERT_INTO_REPLACE_ON_ENABLED.key -> "false") { val ex = intercept[ParseException] { diff --git a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala index c790dc1d04c95..8f611f37713c1 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala @@ -998,6 +998,7 @@ class InsertSuite extends DataSourceTest with SharedSparkSession { } for ((label, insertClause) <- Seq( + ("REPLACE WHERE", "REPLACE WHERE i = 1"), ("REPLACE ON", "AS t REPLACE ON t.i = 1"), ("REPLACE USING", "AS t REPLACE USING (i)"))) { test(s"INSERT INTO ... $label is unsupported for V1 tables") {