This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 442b2ab64c67 [SPARK-58065][SQL] Support column lists for `INSERT
REPLACE WHERE`
442b2ab64c67 is described below
commit 442b2ab64c671fa05aca16e5eed296a6c62c2a6a
Author: Leon Windheuser <[email protected]>
AuthorDate: Wed Jul 22 22:40:07 2026 +0800
[SPARK-58065][SQL] Support column lists for `INSERT REPLACE WHERE`
### What changes were proposed in this pull request?
* Refactor the `REPLACE WHERE` `OverwriteByExpression` into the
`InsertIntoStatement` abstraction.
* Add support for column lists for `INSERT REPLACE WHERE` which was lacking.
### Why are the changes needed?
* The `InsertIntoStatement` gives us the `userSpecifiedCols` field to
express column lists more easily. The `OverwriteByExpression` did not have that.
* Other `INSERT INTO` statements support a column list.
### Does this PR introduce _any_ user-facing change?
* Allows providing column lists in an `INSERT INTO ... REPLACE WHERE`
statement.
### How was this patch tested?
New unit tests.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code with Opus 4.8
Closes #57142 from leonwind/lw/replace-where-col-list.
Authored-by: Leon Windheuser <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit aed53d9d445c28e2e94be10371410c377ec354f6)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 12 ++
.../spark/sql/catalyst/parser/SqlBaseParser.g4 | 2 +-
.../spark/sql/errors/QueryParsingErrors.scala | 14 ++
.../spark/sql/catalyst/analysis/Analyzer.scala | 28 +++-
.../analysis/ResolveIdentifierClause.scala | 6 +-
.../spark/sql/catalyst/analysis/parameters.scala | 6 +-
.../spark/sql/catalyst/parser/AstBuilder.scala | 96 +++++++------
.../sql/catalyst/plans/logical/statements.scala | 42 +++++-
.../spark/sql/errors/QueryCompilationErrors.scala | 4 +
.../org/apache/spark/sql/internal/SQLConf.scala | 10 ++
.../spark/sql/catalyst/parser/DDLParserSuite.scala | 151 +++++++++++++++++---
.../spark/sql/execution/datasources/rules.scala | 8 +-
.../identifier-clause-legacy.sql.out | 12 +-
.../postgreSQL/window_part3.sql.out | 2 +-
.../results/identifier-clause-legacy.sql.out | 12 +-
.../results/postgreSQL/window_part3.sql.out | 2 +-
.../org/apache/spark/sql/ParametersSuite.scala | 71 ++++++----
.../spark/sql/connector/DataSourceV2SQLSuite.scala | 155 +++++++++++++++++++++
.../execution/command/PlanResolutionSuite.scala | 17 +++
.../org/apache/spark/sql/sources/InsertSuite.scala | 9 +-
20 files changed, 537 insertions(+), 122 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 9f6fb5ee99b2..854923c02eb9 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -3252,6 +3252,12 @@
],
"sqlState" : "0A000"
},
+ "INSERT_REPLACE_ON_COLUMN_LIST_NOT_ALLOWED" : {
+ "message" : [
+ "A column list is not allowed with INSERT INTO ... REPLACE ON."
+ ],
+ "sqlState" : "42000"
+ },
"INSERT_REPLACE_ON_NOT_ENABLED" : {
"message" : [
"INSERT INTO ... REPLACE ON is not enabled. Set
'spark.sql.insertIntoReplaceOn.enabled' to true to enable it."
@@ -3270,6 +3276,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 ca5de63ecb0b..29094679e0fb 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
@@ -597,7 +597,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 cb1fd7ad066d..7cf205e75ac4 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
@@ -58,6 +58,13 @@ private[sql] object QueryParsingErrors extends
DataTypeErrorsBase {
ctx)
}
+ def insertReplaceOnColumnListNotAllowed(ctx: ParserRuleContext): Throwable =
{
+ new ParseException(
+ errorClass = "INSERT_REPLACE_ON_COLUMN_LIST_NOT_ALLOWED",
+ messageParameters = Map.empty,
+ ctx)
+ }
+
def insertReplaceUsingNotEnabled(ctx: ParserRuleContext): Throwable = {
new ParseException(
errorClass = "INSERT_REPLACE_USING_NOT_ENABLED",
@@ -72,6 +79,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 e921d0e77c63..c629de82c6ff 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
@@ -1160,6 +1160,8 @@ 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 {
+ 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)
@@ -1351,14 +1353,14 @@ 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())
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) {
@@ -1392,7 +1394,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,
@@ -1405,17 +1411,29 @@ class Analyzer(
withSchemaEvolution = i.withSchemaEvolution)
}
} else {
+ val deleteExpr = i.replaceCriteriaOpt match {
+ case Some(InsertReplaceWhere(condition)) =>
+ if (staticPartitions.nonEmpty) {
+ throw SparkException.internalError(
+ 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 cfa6f3358806..2bccc99f31d8 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 31c835986e20..deb63a7d9789 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 2100241ce9db..6b60cb9ebd3a 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
@@ -939,11 +939,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)
@@ -973,36 +972,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)
@@ -1109,9 +1087,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))
)
@@ -1129,27 +1108,62 @@ class AstBuilder extends DataTypeAstBuilder
if (byName &&
!SQLConf.get.getConf(SQLConf.INSERT_INTO_REPLACE_ON_BY_NAME_ENABLED)) {
throw QueryParsingErrors.insertReplaceOnByNameNotEnabled(ctx)
}
+ // Only REPLACE WHERE supports a column list, but the grammar rule is
shared with REPLACE ON,
+ // so reject the column list here.
+ if (ctx.identifierList() != null) {
+ throw
QueryParsingErrors.insertReplaceOnColumnListNotAllowed(ctx.identifierList())
+ }
val replaceOnCond = expression(ctx.replaceCondition)
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 grammar rule shared with REPLACE ON accepts a table alias, but
REPLACE WHERE has no
+ // use for it. Reject it explicitly for a clear parse error.
+ 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,
@@ -1181,9 +1195,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 4fbe71ed7d3e..e10fa680ddda 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/errors/QueryCompilationErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index 13fdb7b8756f..5f0f23b6de55 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -682,6 +682,10 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
unsupportedTableOperationError(tableName, "INSERT INTO ... REPLACE
ON/USING")
}
+ def unsupportedInsertReplaceWhere(tableName: String): Throwable = {
+ unsupportedTableOperationError(tableName, "INSERT INTO ... REPLACE WHERE")
+ }
+
def writeIntoViewNotAllowedError(identifier: TableIdentifier, t:
TreeNode[_]): Throwable = {
new AnalysisException(
errorClass = "VIEW_WRITE_NOT_ALLOWED",
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 f71f27de5c6a..5cb317a52b33 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
@@ -5467,6 +5467,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 f33c11c8a2d4..5d8ff7d2155b 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) {
@@ -2082,6 +2182,18 @@ class DDLParserSuite extends AnalysisTest {
)
}
+ test("insert table: REPLACE ON rejects a column list") {
+ checkError(
+ exception = parseException(
+ "INSERT INTO testcat.ns1.ns2.tbl (a, b) REPLACE ON col1 = col2 SELECT
* FROM source"),
+ condition = "INSERT_REPLACE_ON_COLUMN_LIST_NOT_ALLOWED",
+ parameters = Map.empty,
+ context = ExpectedContext(
+ fragment = "INSERT INTO testcat.ns1.ns2.tbl (a, b) REPLACE ON col1 =
col2",
+ start = 0,
+ stop = 60))
+ }
+
test("INSERT INTO REPLACE ON with source query alias") {
val table = "testcat.ns1.ns2.tbl"
parseCompare(
@@ -2136,12 +2248,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/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
index 3e20900da55b..040a067ec820 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
@@ -468,8 +468,12 @@ object PreprocessTableInsertion extends
ResolveInsertionBase {
throw QueryCompilationErrors.unsupportedInsertWithSchemaEvolution()
}
- if (insert.replaceCriteriaOpt.isDefined) {
- throw QueryCompilationErrors.unsupportedInsertReplaceOnOrUsing(tblName)
+ insert.replaceCriteriaOpt match {
+ case Some(_: InsertReplaceWhere) =>
+ throw QueryCompilationErrors.unsupportedInsertReplaceWhere(tblName)
+ case Some(_) =>
+ throw QueryCompilationErrors.unsupportedInsertReplaceOnOrUsing(tblName)
+ case None =>
}
val normalizedPartSpec = normalizePartitionSpec(
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/identifier-clause-legacy.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/identifier-clause-legacy.sql.out
index 4cecec03578a..6164ff65dabc 100644
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/identifier-clause-legacy.sql.out
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/identifier-clause-legacy.sql.out
@@ -1358,7 +1358,7 @@ org.apache.spark.sql.catalyst.parser.ParseException
"errorClass" : "PARSE_SYNTAX_ERROR",
"sqlState" : "42601",
"messageParameters" : {
- "error" : "''my_table''",
+ "error" : "'('",
"hint" : ""
}
}
@@ -1630,7 +1630,7 @@ org.apache.spark.sql.catalyst.parser.ParseException
"errorClass" : "PARSE_SYNTAX_ERROR",
"sqlState" : "42601",
"messageParameters" : {
- "error" : "''schema.table''",
+ "error" : "'('",
"hint" : ""
}
}
@@ -2124,8 +2124,8 @@ org.apache.spark.sql.catalyst.parser.ParseException
"errorClass" : "PARSE_SYNTAX_ERROR",
"sqlState" : "42601",
"messageParameters" : {
- "error" : "':'",
- "hint" : ": extra input ':'"
+ "error" : "'('",
+ "hint" : ""
},
"queryContext" : [ {
"objectType" : "EXECUTE IMMEDIATE",
@@ -2292,8 +2292,8 @@ org.apache.spark.sql.catalyst.parser.ParseException
"errorClass" : "PARSE_SYNTAX_ERROR",
"sqlState" : "42601",
"messageParameters" : {
- "error" : "':'",
- "hint" : ": extra input ':'"
+ "error" : "'('",
+ "hint" : ""
},
"queryContext" : [ {
"objectType" : "EXECUTE IMMEDIATE",
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/postgreSQL/window_part3.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/postgreSQL/window_part3.sql.out
index ea75c4414243..ab9b354ea7bd 100644
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/postgreSQL/window_part3.sql.out
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/postgreSQL/window_part3.sql.out
@@ -364,7 +364,7 @@ org.apache.spark.sql.catalyst.parser.ParseException
"errorClass" : "PARSE_SYNTAX_ERROR",
"sqlState" : "42601",
"messageParameters" : {
- "error" : "'BY'",
+ "error" : "'('",
"hint" : ""
}
}
diff --git
a/sql/core/src/test/resources/sql-tests/results/identifier-clause-legacy.sql.out
b/sql/core/src/test/resources/sql-tests/results/identifier-clause-legacy.sql.out
index cc8bd806365b..3b3bc11999c7 100644
---
a/sql/core/src/test/resources/sql-tests/results/identifier-clause-legacy.sql.out
+++
b/sql/core/src/test/resources/sql-tests/results/identifier-clause-legacy.sql.out
@@ -1502,7 +1502,7 @@ org.apache.spark.sql.catalyst.parser.ParseException
"errorClass" : "PARSE_SYNTAX_ERROR",
"sqlState" : "42601",
"messageParameters" : {
- "error" : "''my_table''",
+ "error" : "'('",
"hint" : ""
}
}
@@ -1809,7 +1809,7 @@ org.apache.spark.sql.catalyst.parser.ParseException
"errorClass" : "PARSE_SYNTAX_ERROR",
"sqlState" : "42601",
"messageParameters" : {
- "error" : "''schema.table''",
+ "error" : "'('",
"hint" : ""
}
}
@@ -2390,8 +2390,8 @@ org.apache.spark.sql.catalyst.parser.ParseException
"errorClass" : "PARSE_SYNTAX_ERROR",
"sqlState" : "42601",
"messageParameters" : {
- "error" : "':'",
- "hint" : ": extra input ':'"
+ "error" : "'('",
+ "hint" : ""
},
"queryContext" : [ {
"objectType" : "EXECUTE IMMEDIATE",
@@ -2574,8 +2574,8 @@ org.apache.spark.sql.catalyst.parser.ParseException
"errorClass" : "PARSE_SYNTAX_ERROR",
"sqlState" : "42601",
"messageParameters" : {
- "error" : "':'",
- "hint" : ": extra input ':'"
+ "error" : "'('",
+ "hint" : ""
},
"queryContext" : [ {
"objectType" : "EXECUTE IMMEDIATE",
diff --git
a/sql/core/src/test/resources/sql-tests/results/postgreSQL/window_part3.sql.out
b/sql/core/src/test/resources/sql-tests/results/postgreSQL/window_part3.sql.out
index 893528803096..95e8101801e5 100644
---
a/sql/core/src/test/resources/sql-tests/results/postgreSQL/window_part3.sql.out
+++
b/sql/core/src/test/resources/sql-tests/results/postgreSQL/window_part3.sql.out
@@ -383,7 +383,7 @@ org.apache.spark.sql.catalyst.parser.ParseException
"errorClass" : "PARSE_SYNTAX_ERROR",
"sqlState" : "42601",
"messageParameters" : {
- "error" : "'BY'",
+ "error" : "'('",
"hint" : ""
}
}
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 ca7732772b58..91a999ad663f 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
@@ -23,7 +23,7 @@ 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.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, 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(<sql-variable>) ...` places
a
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
index 0896df430a42..78baca76aa75 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
@@ -4566,6 +4566,161 @@ class DataSourceV2SQLSuiteV1Filter
}
}
+ test("Selective Overwrite: REPLACE WHERE with column list - maps source
columns by position") {
+ val df = spark.createDataFrame(Seq((1L, "a"), (2L, "b"), (3L,
"c"))).toDF("id", "data")
+ df.createOrReplaceTempView("source")
+ val df2 = spark.createDataFrame(Seq(("d", 4L), ("e", 5L), ("f",
6L))).toDF("data", "id")
+ df2.createOrReplaceTempView("source2")
+
+ val t = "testcat.tbl"
+ withTable(t) {
+ spark.sql(
+ s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY
(id)")
+ spark.sql(s"INSERT INTO TABLE $t SELECT * FROM source")
+
+ spark.sql(s"INSERT INTO $t (data, id) REPLACE WHERE id = 3 SELECT * FROM
source2")
+ checkAnswer(
+ spark.table(s"$t"),
+ Seq(Row(1L, "a"), Row(2L, "b"), Row(4L, "d"), Row(5L, "e"), Row(6L,
"f")))
+ }
+ }
+
+ test("Selective Overwrite: REPLACE WHERE with column list - omitted column
becomes null") {
+ val df = spark.createDataFrame(Seq((1L, "a"), (2L, "b"), (3L,
"c"))).toDF("id", "data")
+ df.createOrReplaceTempView("source")
+ val df2 = spark.createDataFrame(Seq(Tuple1(4L), Tuple1(5L))).toDF("id")
+ df2.createOrReplaceTempView("source2_ids")
+
+ val t = "testcat.tbl"
+ withTable(t) {
+ spark.sql(
+ s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY
(id)")
+ spark.sql(s"INSERT INTO TABLE $t SELECT * FROM source")
+
+ spark.sql(s"INSERT INTO $t (id) REPLACE WHERE id = 3 SELECT * FROM
source2_ids")
+ checkAnswer(
+ spark.table(s"$t"),
+ Seq(Row(1L, "a"), Row(2L, "b"), Row(4L, null), Row(5L, null)))
+ }
+ }
+
+ test("Selective Overwrite: REPLACE WHERE with column list - too few data
columns") {
+ val t = "testcat.tbl"
+ withTable(t) {
+ spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo
PARTITIONED BY (id)")
+ checkError(
+ exception = analysisException(
+ s"INSERT INTO $t (id, data) REPLACE WHERE id = 1 SELECT 1"),
+ condition = "INSERT_COLUMN_ARITY_MISMATCH.NOT_ENOUGH_DATA_COLUMNS",
+ parameters = Map(
+ "tableName" -> "`testcat`.`tbl`",
+ "tableColumns" -> "`id`, `data`",
+ "dataColumns" -> "`1`"))
+ }
+ }
+
+ test("Selective Overwrite: REPLACE WHERE with column list - too many data
columns") {
+ val t = "testcat.tbl"
+ withTable(t) {
+ spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo
PARTITIONED BY (id)")
+ checkError(
+ exception = analysisException(
+ s"INSERT INTO $t (id) REPLACE WHERE id = 1 SELECT 1, 'a'"),
+ condition = "INSERT_COLUMN_ARITY_MISMATCH.TOO_MANY_DATA_COLUMNS",
+ parameters = Map(
+ "tableName" -> "`testcat`.`tbl`",
+ "tableColumns" -> "`id`",
+ "dataColumns" -> "`1`, `a`"))
+ }
+ }
+
+ test("Selective Overwrite: REPLACE WHERE with column list - unknown target
column") {
+ val t = "testcat.tbl"
+ withTable(t) {
+ spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo
PARTITIONED BY (id)")
+ checkError(
+ exception = analysisException(
+ s"INSERT INTO $t (id, nonexistent) REPLACE WHERE id = 1 SELECT 1,
'a'"),
+ condition = "UNRESOLVED_COLUMN.WITH_SUGGESTION",
+ parameters = Map(
+ "objectName" -> "`nonexistent`",
+ "proposal" -> "`id`, `data`"),
+ context = ExpectedContext(
+ fragment = s"INSERT INTO $t (id, nonexistent) REPLACE WHERE id = 1",
+ start = 0,
+ stop = 50 + t.length))
+ }
+ }
+
+ test("Selective Overwrite: REPLACE WHERE with column list - duplicate
columns") {
+ val t = "testcat.tbl"
+ withTable(t) {
+ spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo
PARTITIONED BY (id)")
+ checkError(
+ exception = analysisException(
+ s"INSERT INTO $t (id, id) REPLACE WHERE id = 1 SELECT 1, 2"),
+ condition = "COLUMN_ALREADY_EXISTS",
+ parameters = Map("columnName" -> "`id`"))
+ }
+ }
+
+ test("Selective Overwrite: REPLACE WHERE with column list passes write
options to the write") {
+ val t = "testcat.tbl"
+ withTable(t) {
+ spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo
PARTITIONED BY (id)")
+ spark.sql(s"INSERT INTO $t VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+
+ // The in-memory table fails the write only if the option reaches the
write path.
+ intercept[IllegalStateException] {
+ spark.sql(
+ s"INSERT INTO $t WITH
(`${InMemoryBaseTable.SIMULATE_FAILED_WRITE_OPTION}` = true) " +
+ "(data, id) REPLACE WHERE id = 3 SELECT 'z', 3")
+ }
+ checkAnswer(
+ spark.table(t),
+ Seq(Row(1L, "a"), Row(2L, "b"), Row(3L, "c")))
+ }
+ }
+
+ test("Selective Overwrite: REPLACE WHERE with column list maps struct fields
by position") {
+ val t = "testcat.tbl"
+ withTable(t) {
+ spark.sql(
+ s"CREATE TABLE $t (id bigint, info struct<x: int, y: int>) USING foo
PARTITIONED BY (id)")
+ spark.sql(
+ s"INSERT INTO $t VALUES (1, named_struct('x', 10, 'y', 20)), " +
+ "(2, named_struct('x', 30, 'y', 40))")
+
+ // Column list reorders top-level columns; struct fields still map by
position, so the
+ // source's ('y', 'x') order lands in the target's (x, y).
+ spark.sql(
+ s"INSERT INTO $t (info, id) REPLACE WHERE id = 1 " +
+ "SELECT named_struct('y', 100, 'x', 200), 1")
+ checkAnswer(
+ spark.sql(s"SELECT id, info.x, info.y FROM $t ORDER BY id"),
+ Seq(Row(1L, 100, 200), Row(2L, 30, 40)))
+ }
+ }
+
+ test("Selective Overwrite: WITH SCHEMA EVOLUTION with column list widens the
reordered column") {
+ val t = "testcat.tbl"
+ withTable(t) {
+ spark.sql(s"CREATE TABLE $t (id int, data string) USING foo PARTITIONED
BY (id)")
+ spark.sql(s"INSERT INTO $t VALUES (1, 'a'), (2, 'b')")
+
+ // The column list reorders to (data, id) and supplies a BIGINT for id
at position 1.
+ // Schema evolution must widen `id` (not `data`) to BIGINT.
+ spark.sql(
+ s"INSERT WITH SCHEMA EVOLUTION INTO $t (data, id) " +
+ "REPLACE WHERE id = 1 SELECT 'a', CAST(100 AS BIGINT)")
+
+ assert(spark.table(t).schema("id").dataType === LongType)
+ checkAnswer(
+ spark.table(t).orderBy("data"),
+ Seq(Row(100L, "a"), Row(2L, "b")))
+ }
+ }
+
test("Overwrite: REPLACE WHERE without BY NAME - positional matching") {
val df = spark.createDataFrame(Seq((1L, "a"), (2L, "b"), (3L,
"c"))).toDF("id", "data")
df.createOrReplaceTempView("source")
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 cdba60d9960d..a924b637a79f 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 c790dc1d04c9..f2645e32e8a9 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
@@ -997,9 +997,10 @@ class InsertSuite extends DataSourceTest with
SharedSparkSession {
}
}
- for ((label, insertClause) <- Seq(
- ("REPLACE ON", "AS t REPLACE ON t.i = 1"),
- ("REPLACE USING", "AS t REPLACE USING (i)"))) {
+ for ((label, insertClause, operation) <- Seq(
+ ("REPLACE WHERE", "REPLACE WHERE i = 1", "INSERT INTO ... REPLACE
WHERE"),
+ ("REPLACE ON", "AS t REPLACE ON t.i = 1", "INSERT INTO ... REPLACE
ON/USING"),
+ ("REPLACE USING", "AS t REPLACE USING (i)", "INSERT INTO ... REPLACE
ON/USING"))) {
test(s"INSERT INTO ... $label is unsupported for V1 tables") {
withTable("test_table") {
val schema = new StructType().add("i", "int").add("j", "string")
@@ -1027,7 +1028,7 @@ class InsertSuite extends DataSourceTest with
SharedSparkSession {
sqlState = "0A000",
parameters = Map(
"tableName" -> "`spark_catalog`.`default`.`test_table`",
- "operation" -> "INSERT INTO ... REPLACE ON/USING")
+ "operation" -> operation)
)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
