[SPARK-58065] Support column lists for INSERT REPLACE WHERE

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.",

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

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",

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)
           }
         }

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) {

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))

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

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)
+}

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 " +