longvu-db commented on code in PR #56705:
URL: https://github.com/apache/spark/pull/56705#discussion_r3511569879
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/write/RowLevelOperation.java:
##########
@@ -39,7 +39,7 @@ public interface RowLevelOperation {
* @since 3.3.0
*/
enum Command {
- DELETE, UPDATE, MERGE
+ DELETE, UPDATE, MERGE, REPLACE
Review Comment:
```suggestion
DELETE, UPDATE, MERGE, INSERT_INTO_REPLACE
```
To better distinguish against the SQL REPLACE TABLE command
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -3198,6 +3198,18 @@
],
"sqlState" : "0A000"
},
+ "INSERT_REPLACE_USING_DUPLICATE_SCOPE_COLUMN" : {
Review Comment:
I think let's not block duplicate columns, users would expect `INSERT
REPLACE USING` to have the same matching semantics as `JOIN USING`, and `JOIN
USING` allows duplicate columns.
Also the term "scope column" is likely to be unfamiliar to users.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -1283,10 +1284,24 @@ class Analyzer(
object ResolveInsertInto extends ResolveInsertionBase {
override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsWithPruning(
AlwaysProcess.fn, ruleId) {
+ case i: InsertIntoStatement
+ if i.table.isInstanceOf[DataSourceV2Relation] &&
+ i.query.resolved &&
+ i.replaceCriteriaOpt.exists(_.isInstanceOf[InsertReplaceUsing]) =>
+ val r = i.table.asInstanceOf[DataSourceV2Relation]
+ val scopeColumns =
i.replaceCriteriaOpt.get.asInstanceOf[InsertReplaceUsing].cols
+ // Align the source to the target with standard INSERT resolution so
RewriteReplaceUsing can
+ // compare scope columns and build the insert payload positionally by
ordinal.
+ val alignedQuery = TableOutputResolver.resolveOutputColumns(
+ r.table.name, r.output, i.query, byName = i.byName, conf)
+ ReplaceUsingTable(r, scopeColumns, alignedQuery)
+
case i: InsertIntoStatement
if i.table.isInstanceOf[DataSourceV2Relation] &&
i.query.resolved &&
i.replaceCriteriaOpt.isDefined =>
Review Comment:
```suggestion
i.replaceCriteriaOpt.exists(_.isInstanceOf[InsertReplaceOn])=>
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -1283,10 +1284,24 @@ class Analyzer(
object ResolveInsertInto extends ResolveInsertionBase {
override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsWithPruning(
AlwaysProcess.fn, ruleId) {
+ case i: InsertIntoStatement
+ if i.table.isInstanceOf[DataSourceV2Relation] &&
+ i.query.resolved &&
+ i.replaceCriteriaOpt.exists(_.isInstanceOf[InsertReplaceUsing]) =>
+ val r = i.table.asInstanceOf[DataSourceV2Relation]
+ val scopeColumns =
i.replaceCriteriaOpt.get.asInstanceOf[InsertReplaceUsing].cols
+ // Align the source to the target with standard INSERT resolution so
RewriteReplaceUsing can
+ // compare scope columns and build the insert payload positionally by
ordinal.
+ val alignedQuery = TableOutputResolver.resolveOutputColumns(
+ r.table.name, r.output, i.query, byName = i.byName, conf)
+ ReplaceUsingTable(r, scopeColumns, alignedQuery)
+
case i: InsertIntoStatement
if i.table.isInstanceOf[DataSourceV2Relation] &&
i.query.resolved &&
i.replaceCriteriaOpt.isDefined =>
+ // REPLACE USING has a scope-based row-level rewrite path. REPLACE ON
remains unsupported
Review Comment:
```suggestion
// INSERT REPLACE ON remains unsupported
```
This case match focuses on blocking INSERT REPLACE ON
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -1283,10 +1284,24 @@ class Analyzer(
object ResolveInsertInto extends ResolveInsertionBase {
override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsWithPruning(
AlwaysProcess.fn, ruleId) {
+ case i: InsertIntoStatement
+ if i.table.isInstanceOf[DataSourceV2Relation] &&
+ i.query.resolved &&
+ i.replaceCriteriaOpt.exists(_.isInstanceOf[InsertReplaceUsing]) =>
+ val r = i.table.asInstanceOf[DataSourceV2Relation]
+ val scopeColumns =
i.replaceCriteriaOpt.get.asInstanceOf[InsertReplaceUsing].cols
Review Comment:
```suggestion
val replaceUsingCols =
i.replaceCriteriaOpt.get.asInstanceOf[InsertReplaceUsing].cols
```
I think "`replaceUsingCols`" would be more familiar than "`scopeColumns`"
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -49,10 +49,11 @@ case class BatchScanExec(
override protected lazy val sparkMetrics: Map[String, SQLMetric] = {
val name = "number of output rows"
val metric = table match {
- // Use SLAM for the scan-output count when this scan reads on behalf of
a row-level DELETE,
- // so that the driver-side derivation `numDeletedRows = numScannedRows -
numCopiedRows` in
- // `ReplaceDataExec.getWriteSummary` stays correct under stage retries.
- case rlot: RowLevelOperationTable if rlot.operation.command() == DELETE
=>
+ // Use SLAM for the scan-output count when this scan reads on behalf of
a row-level DELETE or
+ // REPLACE, so that the driver-side derivation `numDeletedRows =
numScannedRows -
+ // numCopiedRows` in `ReplaceDataExec` stays correct under stage retries.
+ case rlot: RowLevelOperationTable
+ if rlot.operation.command() == DELETE || rlot.operation.command() ==
REPLACE =>
Review Comment:
```suggestion
if Set(DELETE,
INSERT_REPLACE_USING).contains(rlot.operation.command()) =>
```
##########
docs/sql-v2-data-sources.md:
##########
@@ -216,6 +216,10 @@ A `Table` gains read and write abilities by implementing
mix-in interfaces:
batch or streaming writes. See [Write Path](#write-path).
- **`SupportsRowLevelOperations`** enables `DELETE`, `UPDATE`, and `MERGE
INTO` through
a read-and-rewrite cycle. See [Row-Level DML](#row-level-dml).
+- **`SupportsRowLevelReplace`** additionally enables `INSERT INTO ... REPLACE
USING (cols)`,
Review Comment:
Let's have a dedicated PR for docs change (it will need a dedicated Spark
Jira Doc ticket as well), since the current PR is already quite big.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -3198,6 +3198,18 @@
],
"sqlState" : "0A000"
},
+ "INSERT_REPLACE_USING_DUPLICATE_SCOPE_COLUMN" : {
+ "message" : [
+ "INSERT INTO ... REPLACE USING references the scope column <columnName>
more than once. Each scope column must be listed at most once."
+ ],
+ "sqlState" : "42701"
+ },
+ "INSERT_REPLACE_USING_NON_DETERMINISTIC_SOURCE" : {
Review Comment:
```suggestion
"INSERT_REPLACE_USING_NON_DETERMINISTIC_SOURCE_QUERY_NOT_SUPPORTED" : {
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteReplaceUsing.scala:
##########
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute,
EqualNullSafe, Exists, Expression, Literal, MetadataAttribute, NamedExpression,
OuterReference}
+import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
+import org.apache.spark.sql.catalyst.plans.{LeftAnti, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Assignment,
CTERelationDef, CTERelationRef, Filter, Join, JoinHint, LogicalPlan, Project,
ReplaceData, ReplaceUsingTable, Union, WithCTE, WriteDelta}
+import org.apache.spark.sql.catalyst.util.RowDeltaUtils.{COPY_OPERATION,
INSERT_OPERATION, OPERATION_COLUMN}
+import org.apache.spark.sql.connector.write.{RowLevelOperationTable,
SupportsDelta}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.types.IntegerType
+
+/**
+ * Rewrites `INSERT INTO ... REPLACE USING (cols) <query>` (a
[[ReplaceUsingTable]]) onto the
+ * row-level write machinery via a [[RowLevelOperation.Command.REPLACE]]
operation.
+ *
+ * Scoped replace deletes every target row whose scope-column tuple appears in
the source and
+ * appends all source rows, including duplicates that share a scope tuple. The
replacement state
+ * is computed from independent target and source branches so that source rows
are inserted
+ * regardless of whether their scope tuple already exists in the target.
+ *
+ * The source is shared through a CTE so the delete/carryover join, the insert
branch, and the
+ * runtime group filter all refer to the same analyzed source plan. A
non-deterministic source is
+ * rejected because it could otherwise delete based on a different source
result than it inserts.
+ */
+object RewriteReplaceUsing extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case r @ ReplaceUsingTable(aliasedTable, scopeColumns, source)
+ if r.resolved && source.resolved =>
+
+ // The source feeds the delete decision, insert payload, and runtime
group filter.
+ // Evaluating a non-deterministic source in those independent plan
locations could produce
+ // mutually inconsistent results.
+ if (!source.deterministic) {
+ throw QueryCompilationErrors.insertReplaceUsingNonDeterministicSource()
+ }
+
+ val (rel, operationTable) = buildReplaceOperationTable(aliasedTable)
+ val scopeOrdinals = resolveScopeOrdinals(rel, scopeColumns)
+ operationTable.operation match {
+ case _: SupportsDelta =>
+ buildWriteDeltaPlan(rel, operationTable, scopeOrdinals, source)
+ case _ =>
+ buildReplaceDataPlan(rel, operationTable, scopeOrdinals, source)
+ }
+ }
+
+ // Build a copy-on-write replacement from retained target rows and inserted
source rows.
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ scopeOrdinals: Seq[Int],
+ source: LogicalPlan): LogicalPlan = {
+
+ val rowAttrs = relation.output
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation,
operationTable.operation)
+ val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs)
+
+ val sourceCte = CTERelationDef(source)
Review Comment:
Seems odd that we have to directly mention CTE inside the INSERT REPLACE
USING abstraction, they are orthogonal features that should be abstracted from
each other, `RewriteMergeIntoTable`/`RewriteDelete`/... also don't seem to
reference them.
If possible, could we defer the support/testing of CTE to another PR, in
order to not make this PR too big?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -56,6 +57,33 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
RowLevelOperationTable(table, operation)
}
+ /**
+ * Resolves the target of an `INSERT INTO ... REPLACE` command and builds its
+ * [[RowLevelOperation.Command.REPLACE]] operation table, gating on the
connector opt-in.
+ *
+ * REPLACE is dispatched only to connectors that explicitly opt in via
+ * [[SupportsRowLevelReplace]]. A table that supports row-level operations
but not REPLACE takes
+ * the same unsupported-operation path as a non-row-level table, so its
operation builder is
+ * never invoked with a command it did not expect. Keeping the gate here
ensures rewrites that
+ * dispatch [[RowLevelOperation.Command.REPLACE]] reject unsupported tables
identically.
+ */
+ protected def buildReplaceOperationTable(
Review Comment:
Should we do all these specific handling `case _ =>
throw
QueryCompilationErrors.unsupportedInsertReplaceOnOrUsing(rel.table.name())`,
`EliminateSubqueryAliases(aliasedTable) match {`, `case other =>
throw QueryCompilationErrors.unsupportedInsertReplaceOnOrUsing(
other.simpleString(maxFields = 2))` in
`RewriteInsertReplaceUsing.scala` like `RewriteMergeIntoTable.scala`?
`buildReplaceOperationTable` and `buildOperationTable` are quite similar, we
should avoid duplications, also readers won't expect
`buildReplaceOperationTable` to do some additional analysis handling, from the
func name.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/write/ReplaceSummaryImpl.scala:
##########
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.write
+
+/**
+ * Implementation of [[ReplaceSummary]] that provides REPLACE operation
summary.
+ */
+private[sql] case class ReplaceSummaryImpl(
+ numInsertedRows: Long,
+ numDeletedRows: Long,
+ numCopiedRows: Long)
+ extends ReplaceSummary {
Review Comment:
Instead of defining a new Summary, would it be possible to just extend
`DeleteSummary` and `InsertSummary`?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala:
##########
@@ -1166,6 +1170,37 @@ case class MergeIntoTable(
}
}
+/**
+ * The logical plan of the `INSERT INTO t REPLACE USING (cols) <query>`
command, resolved from
+ * [[InsertIntoStatement]] when its replace criteria is an
[[InsertReplaceUsing]].
+ *
+ * Scoped replace deletes every target row whose scope-column tuple appears in
the source and
+ * appends all source rows, including duplicates that share the same scope
tuple. The command uses
+ * [[RowLevelOperation.Command.REPLACE]] so connectors can distinguish this
scope-level replacement
+ * from per-row MERGE semantics.
+ *
+ * @param targetTable the target relation to replace rows in
+ * @param scopeColumns unqualified target column names whose values define a
replace scope tuple
+ * @param query the source query, already aligned to the target table
output
+ */
+case class ReplaceUsingTable(
Review Comment:
Since InsertReplaceUsingTable is also an INSERT, should we extend
https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala#L99?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -1283,10 +1284,24 @@ class Analyzer(
object ResolveInsertInto extends ResolveInsertionBase {
override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsWithPruning(
AlwaysProcess.fn, ruleId) {
+ case i: InsertIntoStatement
+ if i.table.isInstanceOf[DataSourceV2Relation] &&
+ i.query.resolved &&
+ i.replaceCriteriaOpt.exists(_.isInstanceOf[InsertReplaceUsing]) =>
+ val r = i.table.asInstanceOf[DataSourceV2Relation]
+ val scopeColumns =
i.replaceCriteriaOpt.get.asInstanceOf[InsertReplaceUsing].cols
+ // Align the source to the target with standard INSERT resolution so
RewriteReplaceUsing can
+ // compare scope columns and build the insert payload positionally by
ordinal.
+ val alignedQuery = TableOutputResolver.resolveOutputColumns(
+ r.table.name, r.output, i.query, byName = i.byName, conf)
Review Comment:
I wonder if we should combine this case match's implementation
`i.replaceCriteriaOpt.exists(_.isInstanceOf[InsertReplaceUsing])`, with the
case match's below `i.replaceCriteriaOpt.isEmpty`, because we would also have
to integrate BY NAME, with schema evolution, column list, into INSERT REPLACE
USING, in the future.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SupportsRowLevelReplace.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.catalog;
+
+import org.apache.spark.annotation.Experimental;
+import org.apache.spark.sql.connector.write.RowLevelOperation;
+
+/**
+ * A mix-in interface for {@link SupportsRowLevelOperations} that signals a
table can handle the
+ * {@link RowLevelOperation.Command#REPLACE} command.
+ * <p>
+ * Spark dispatches {@code INSERT INTO ... REPLACE USING (cols)} through this
command.
+ * {@link RowLevelOperation.Command#REPLACE} is dispatched to a connector's
+ * {@link SupportsRowLevelOperations#newRowLevelOperationBuilder} only when
the table implements
Review Comment:
Forgive me if this is a dumb question, but why do we need a new trait for a
new DML command? What makes INSERT REPLACE USING different from
DELETE/UPDATE/MERGE? At the end of the day, it simply deletes some rows and
inserts some rows.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteReplaceUsing.scala:
##########
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute,
EqualNullSafe, Exists, Expression, Literal, MetadataAttribute, NamedExpression,
OuterReference}
+import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
+import org.apache.spark.sql.catalyst.plans.{LeftAnti, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Assignment,
CTERelationDef, CTERelationRef, Filter, Join, JoinHint, LogicalPlan, Project,
ReplaceData, ReplaceUsingTable, Union, WithCTE, WriteDelta}
+import org.apache.spark.sql.catalyst.util.RowDeltaUtils.{COPY_OPERATION,
INSERT_OPERATION, OPERATION_COLUMN}
+import org.apache.spark.sql.connector.write.{RowLevelOperationTable,
SupportsDelta}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.types.IntegerType
+
+/**
+ * Rewrites `INSERT INTO ... REPLACE USING (cols) <query>` (a
[[ReplaceUsingTable]]) onto the
+ * row-level write machinery via a [[RowLevelOperation.Command.REPLACE]]
operation.
+ *
+ * Scoped replace deletes every target row whose scope-column tuple appears in
the source and
+ * appends all source rows, including duplicates that share a scope tuple. The
replacement state
+ * is computed from independent target and source branches so that source rows
are inserted
+ * regardless of whether their scope tuple already exists in the target.
+ *
+ * The source is shared through a CTE so the delete/carryover join, the insert
branch, and the
+ * runtime group filter all refer to the same analyzed source plan. A
non-deterministic source is
+ * rejected because it could otherwise delete based on a different source
result than it inserts.
+ */
+object RewriteReplaceUsing extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case r @ ReplaceUsingTable(aliasedTable, scopeColumns, source)
+ if r.resolved && source.resolved =>
+
+ // The source feeds the delete decision, insert payload, and runtime
group filter.
+ // Evaluating a non-deterministic source in those independent plan
locations could produce
+ // mutually inconsistent results.
+ if (!source.deterministic) {
+ throw QueryCompilationErrors.insertReplaceUsingNonDeterministicSource()
+ }
+
+ val (rel, operationTable) = buildReplaceOperationTable(aliasedTable)
+ val scopeOrdinals = resolveScopeOrdinals(rel, scopeColumns)
+ operationTable.operation match {
+ case _: SupportsDelta =>
+ buildWriteDeltaPlan(rel, operationTable, scopeOrdinals, source)
+ case _ =>
+ buildReplaceDataPlan(rel, operationTable, scopeOrdinals, source)
+ }
+ }
+
+ // Build a copy-on-write replacement from retained target rows and inserted
source rows.
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ scopeOrdinals: Seq[Int],
+ source: LogicalPlan): LogicalPlan = {
+
+ val rowAttrs = relation.output
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation,
operationTable.operation)
+ val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs)
+
+ val sourceCte = CTERelationDef(source)
+ val scopeRef = newSourceRef(sourceCte)
+ val insertRef = newSourceRef(sourceCte)
+ val filterRef = newSourceRef(sourceCte)
+
+ // COW rewrites whole groups, so target rows whose scope tuple is absent
from the source must
+ // be re-emitted as COPY to preserve them (and their metadata).
+ val antiJoinCond = scopeEquality(readRelation.output, scopeRef.output,
scopeOrdinals)
+ val carryoverJoin = Join(readRelation, scopeRef, LeftAnti,
Some(antiJoinCond), JoinHint.NONE)
+ val carryover = addOperationColumn(COPY_OPERATION, carryoverJoin)
+
+ // All source rows are inserted. Carryover supplies target metadata;
insert rows null it out.
+ val insertData = rowAttrs.indices.map { i =>
+ Alias(insertRef.output(i), rowAttrs(i).name)()
+ }
+ val insertMetadata = metadataAttrs.map(attr => Alias(Literal(null,
attr.dataType), attr.name)())
+ val insertOutput = operationAlias(INSERT_OPERATION) +: (insertData ++
insertMetadata)
+ val inserts = Project(insertOutput, insertRef)
+
+ val replacementQuery = Union(carryover :: inserts :: Nil)
Review Comment:
For the logic of building the replacementQuery, can we make it a separate
function like `buildReplaceDataUpdateProjection` in
https://github.com/apache/spark/blob/3f2ac2cc4c2c1932ace3f286c04226bdb38100a6/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala#L69?
Right now the `buildReplaceDataPlan` is a bit long, which makes it hard to
distinguish the steps within
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/ReplaceUsingTableSuiteBase.scala:
##########
@@ -0,0 +1,545 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector
+
+import org.apache.spark.internal.config
+import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.connector.catalog.{InMemoryRowLevelOperationTable,
InMemoryTable}
+import org.apache.spark.sql.connector.write.{BatchWrite, ReplaceSummary,
ReplaceSummaryImpl, WriterCommitMessage, WriteSummary}
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.datasources.v2.{ReplaceDataExec,
WriteDeltaExec}
+import org.apache.spark.sql.execution.metric.SQLMetric
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
+
+/**
+ * Shared end-to-end coverage for `INSERT INTO ... REPLACE USING` execution.
+ *
+ * Scoped replace deletes every target row whose scope-column tuple appears in
the source and
+ * appends all source rows. Concrete subclasses fix the table layout to a
group-based
+ * (copy-on-write) or delta-based (merge-on-read) row-level operation.
Assertions are written
+ * against final table contents so they hold for both backends.
+ */
+abstract class ReplaceUsingTableSuiteBase extends RowLevelOperationSuiteBase {
+
+ import testImplicits._
+
+ private val schemaString = "pk INT NOT NULL, id INT, dep STRING"
+
+ protected def isDeltaBasedReplace: Boolean = false
+
+ // Select the in-memory table subclass that implements
SupportsRowLevelReplace via props.
+ override protected def extraTableProps: java.util.Map[String, String] = {
+ val props = new java.util.HashMap[String, String]()
+ props.put(InMemoryRowLevelOperationTable.SUPPORTS_ROW_LEVEL_REPLACE,
"true")
+ props
+ }
+
+ test("replace using deletes matching scopes and inserts all source rows") {
Review Comment:
We should add the following tests:
1. Self-INSERTing: INSERT INTO t REPLACE USING ... SELECT ... FROM t
2. Non-Deterministic formats such as Parquet, CSV,...
3. INSERT REPLACE USING analysis Exceptions: Not enough source query columns
compared to table columns, too many source query columns than table columns,
empty REPLACE USING clause,...
4. INSERT adjacent features: WITH SCHEMA EVOLUTION, constraint check, BY
NAME (matching on a column in a different position in the table and the source
query
5. Matching on multiple columns, where there exists one that carries the
NULL value, if a tuple has a NULL, then it shouldn't match.
6. INSERT REPLACE USING on a partitioned table, behavior should be similar
to DPO, except not matching on NULL partitions
7. Palindrome on REPLACE USING clauses, i.e. (col1, col2) and (col2, col1)
8. Different source query shape (SELECT * FROM table)
9. Empty target table
10. REPLACE USING column exists in table but not in source query, exists in
source query but not in table, exist in neither,...
11. INSERT REPLACE USING with `optionsClause`, should still respect
`optionsClause`
https://github.com/apache/spark/blob/master/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4#L594C82-L594C96
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteReplaceUsing.scala:
##########
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute,
EqualNullSafe, Exists, Expression, Literal, MetadataAttribute, NamedExpression,
OuterReference}
+import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
+import org.apache.spark.sql.catalyst.plans.{LeftAnti, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Assignment,
CTERelationDef, CTERelationRef, Filter, Join, JoinHint, LogicalPlan, Project,
ReplaceData, ReplaceUsingTable, Union, WithCTE, WriteDelta}
+import org.apache.spark.sql.catalyst.util.RowDeltaUtils.{COPY_OPERATION,
INSERT_OPERATION, OPERATION_COLUMN}
+import org.apache.spark.sql.connector.write.{RowLevelOperationTable,
SupportsDelta}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.types.IntegerType
+
+/**
+ * Rewrites `INSERT INTO ... REPLACE USING (cols) <query>` (a
[[ReplaceUsingTable]]) onto the
+ * row-level write machinery via a [[RowLevelOperation.Command.REPLACE]]
operation.
+ *
+ * Scoped replace deletes every target row whose scope-column tuple appears in
the source and
+ * appends all source rows, including duplicates that share a scope tuple. The
replacement state
+ * is computed from independent target and source branches so that source rows
are inserted
+ * regardless of whether their scope tuple already exists in the target.
+ *
+ * The source is shared through a CTE so the delete/carryover join, the insert
branch, and the
+ * runtime group filter all refer to the same analyzed source plan. A
non-deterministic source is
+ * rejected because it could otherwise delete based on a different source
result than it inserts.
+ */
+object RewriteReplaceUsing extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case r @ ReplaceUsingTable(aliasedTable, scopeColumns, source)
+ if r.resolved && source.resolved =>
+
+ // The source feeds the delete decision, insert payload, and runtime
group filter.
+ // Evaluating a non-deterministic source in those independent plan
locations could produce
+ // mutually inconsistent results.
+ if (!source.deterministic) {
+ throw QueryCompilationErrors.insertReplaceUsingNonDeterministicSource()
+ }
+
+ val (rel, operationTable) = buildReplaceOperationTable(aliasedTable)
+ val scopeOrdinals = resolveScopeOrdinals(rel, scopeColumns)
+ operationTable.operation match {
+ case _: SupportsDelta =>
+ buildWriteDeltaPlan(rel, operationTable, scopeOrdinals, source)
+ case _ =>
+ buildReplaceDataPlan(rel, operationTable, scopeOrdinals, source)
+ }
+ }
+
+ // Build a copy-on-write replacement from retained target rows and inserted
source rows.
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ scopeOrdinals: Seq[Int],
+ source: LogicalPlan): LogicalPlan = {
+
+ val rowAttrs = relation.output
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation,
operationTable.operation)
+ val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs)
+
+ val sourceCte = CTERelationDef(source)
+ val scopeRef = newSourceRef(sourceCte)
+ val insertRef = newSourceRef(sourceCte)
+ val filterRef = newSourceRef(sourceCte)
+
+ // COW rewrites whole groups, so target rows whose scope tuple is absent
from the source must
+ // be re-emitted as COPY to preserve them (and their metadata).
+ val antiJoinCond = scopeEquality(readRelation.output, scopeRef.output,
scopeOrdinals)
+ val carryoverJoin = Join(readRelation, scopeRef, LeftAnti,
Some(antiJoinCond), JoinHint.NONE)
+ val carryover = addOperationColumn(COPY_OPERATION, carryoverJoin)
+
+ // All source rows are inserted. Carryover supplies target metadata;
insert rows null it out.
+ val insertData = rowAttrs.indices.map { i =>
+ Alias(insertRef.output(i), rowAttrs(i).name)()
+ }
+ val insertMetadata = metadataAttrs.map(attr => Alias(Literal(null,
attr.dataType), attr.name)())
+ val insertOutput = operationAlias(INSERT_OPERATION) +: (insertData ++
insertMetadata)
+ val inserts = Project(insertOutput, insertRef)
+
+ val replacementQuery = Union(carryover :: inserts :: Nil)
+ val projections = buildReplaceDataProjections(replacementQuery, rowAttrs,
metadataAttrs)
+
+ val groupFilterCond = if (groupFilterEnabled) {
+ Some(scopeExists(relation, filterRef, scopeOrdinals))
+ } else {
+ None
+ }
+
+ // The replaced scope is source-defined, so there is no static target-only
predicate to push
+ // down during planning. File pruning flows through groupFilterCond, which
+ // RowLevelOperationRuntimeGroupFiltering turns into a dynamic subquery.
+ val writeRelation = relation.copy(table = operationTable)
+ val replaceData =
+ ReplaceData(writeRelation, TrueLiteral, replacementQuery, relation,
projections,
+ groupFilterCond)
+ WithCTE(replaceData, sourceCte :: Nil)
+ }
+
+ // Build a merge-on-read delta from matched-row deletes and inserted source
rows.
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ scopeOrdinals: Seq[Int],
+ source: LogicalPlan): LogicalPlan = {
+
+ val operation = operationTable.operation.asInstanceOf[SupportsDelta]
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operation)
+ val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs, rowIdAttrs)
+
+ // Keep row columns before metadata columns, matching the schema shape
expected by
+ // buildWriteDeltaProjections. Row ID attributes are metadata-classified
and land in metaPart.
+ val (metaPart, rowPart) = readRelation.output.partition { attr =>
+ MetadataAttribute.isValid(attr.metadata)
+ }
+ val branchNames = OPERATION_COLUMN +: (rowPart ++ metaPart).map(_.name)
+
+ val sourceCte = CTERelationDef(source)
+ val scopeRef = newSourceRef(sourceCte)
+ val insertRef = newSourceRef(sourceCte)
+ val filterRef = newSourceRef(sourceCte)
+
+ // Matched target rows are encoded as deletes: row ID and metadata are
retained, row columns
+ // are nulled out.
+ val semiJoinCond = scopeEquality(readRelation.output, scopeRef.output,
scopeOrdinals)
+ val matchingRows = Join(readRelation, scopeRef, LeftSemi,
Some(semiJoinCond), JoinHint.NONE)
+ val deleteOutput = deltaDeleteOutput(rowPart, rowIdAttrs, metaPart)
+ val deletes = Project(named(deleteOutput, branchNames), matchingRows)
+
+ // Every source row is inserted; row IDs and metadata are null for new
rows.
+ val relationValueByExprId =
+ relation.output.zipWithIndex.map {
+ case (attr, i) => attr.exprId -> insertRef.output(i)
+ }.toMap
+ val insertAssignments = rowPart.map { attr =>
+ val value = relationValueByExprId.getOrElse(attr.exprId, Literal(null,
attr.dataType))
+ Assignment(attr, value)
+ }
+ val insertOutput = deltaInsertOutput(insertAssignments, metaPart)
+ val inserts = Project(named(insertOutput, branchNames), insertRef)
+
+ val deltaQuery = Union(deletes :: inserts :: Nil)
+ val projections = buildWriteDeltaProjections(deltaQuery, rowAttrs,
rowIdAttrs, metadataAttrs)
+
+ val groupFilterCond = if (groupFilterEnabled) {
+ Some(scopeExists(relation, filterRef, scopeOrdinals))
+ } else {
+ None
+ }
+
+ val writeRelation = relation.copy(table = operationTable)
+ val writeDelta =
+ WriteDelta(writeRelation, TrueLiteral, deltaQuery, relation,
projections, groupFilterCond)
+ WithCTE(writeDelta, sourceCte :: Nil)
+ }
+
+ private def operationAlias(operation: Int): NamedExpression = {
+ Alias(Literal(operation, IntegerType), OPERATION_COLUMN)()
+ }
+
+ private def named(exprs: Seq[Expression], names: Seq[String]):
Seq[NamedExpression] = {
+ exprs.zip(names).map {
+ case (ne: NamedExpression, name) if ne.name == name => ne
+ case (e, name) => Alias(e, name)()
+ }
+ }
+
+ private def resolveScopeOrdinals(
+ relation: DataSourceV2Relation,
+ scopeColumns: Seq[String]): Seq[Int] = {
+
+ val seen = scala.collection.mutable.HashSet.empty[Int]
+ scopeColumns.map { name =>
+ val resolved = relation.resolve(Seq(name), conf.resolver).getOrElse {
+ throw QueryCompilationErrors.unresolvedColumnError(name,
relation.output.map(_.name))
+ }
+ val ordinal = relation.output.indexWhere(_.exprId ==
resolved.toAttribute.exprId)
+ if (ordinal < 0) {
+ throw QueryCompilationErrors.unresolvedColumnError(name,
relation.output.map(_.name))
+ }
+ if (!seen.add(ordinal)) {
+ throw
QueryCompilationErrors.insertReplaceUsingDuplicateScopeColumn(name)
+ }
+ ordinal
+ }
+ }
+
+ private def newSourceRef(sourceDef: CTERelationDef): CTERelationRef = {
+ val ref = CTERelationRef(
+ sourceDef.id,
+ _resolved = true,
+ sourceDef.child.output,
+ sourceDef.child.isStreaming)
+ ref.newInstance().asInstanceOf[CTERelationRef]
+ }
+
+ private def scopeEquality(
+ targetOutput: Seq[Attribute],
+ sourceOutput: Seq[Attribute],
+ scopeOrdinals: Seq[Int]): Expression = {
+ scopeOrdinals
+ .map(ordinal => EqualNullSafe(targetOutput(ordinal),
sourceOutput(ordinal)): Expression)
+ .reduce(And)
+ }
+
+ // A correlated EXISTS over the source on scope equality, used as the
runtime group filter.
+ private def scopeExists(
+ relation: DataSourceV2Relation,
+ sourceRef: CTERelationRef,
+ scopeOrdinals: Seq[Int]): Expression = {
+ val cond = scopeOrdinals
+ .map { ordinal =>
Review Comment:
I think we should be able to reuse `scopeEquality` here, for building the
`cond`
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala:
##########
@@ -49,10 +49,11 @@ case class BatchScanExec(
override protected lazy val sparkMetrics: Map[String, SQLMetric] = {
val name = "number of output rows"
val metric = table match {
- // Use SLAM for the scan-output count when this scan reads on behalf of
a row-level DELETE,
- // so that the driver-side derivation `numDeletedRows = numScannedRows -
numCopiedRows` in
- // `ReplaceDataExec.getWriteSummary` stays correct under stage retries.
- case rlot: RowLevelOperationTable if rlot.operation.command() == DELETE
=>
+ // Use SLAM for the scan-output count when this scan reads on behalf of
a row-level DELETE or
+ // REPLACE, so that the driver-side derivation `numDeletedRows =
numScannedRows -
Review Comment:
```suggestion
// INSERT REPLACE USING, so that the driver-side derivation
`numDeletedRows = numScannedRows -
```
Let's change all references of REPLACE in this file, to be INSERT REPLACE
USING, to distinguish against the SQL Command `REPLACE` (CTAS)
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteReplaceUsing.scala:
##########
@@ -0,0 +1,237 @@
+/*
Review Comment:
RewriteInsertReplaceUsing.scala
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -3198,6 +3198,18 @@
],
"sqlState" : "0A000"
},
+ "INSERT_REPLACE_USING_DUPLICATE_SCOPE_COLUMN" : {
+ "message" : [
+ "INSERT INTO ... REPLACE USING references the scope column <columnName>
more than once. Each scope column must be listed at most once."
+ ],
+ "sqlState" : "42701"
+ },
+ "INSERT_REPLACE_USING_NON_DETERMINISTIC_SOURCE" : {
+ "message" : [
+ "INSERT INTO ... REPLACE USING requires a deterministic source query
because the source determines both which target rows are deleted and which rows
are inserted. Remove non-deterministic expressions (e.g. rand()) from the
source."
Review Comment:
```suggestion
"IINSERT INTO ... REPLACE USING does not support a non-deterministic
source query."
```
INSERT REPLACE USING can support Source Materialization in the future:
#55858, also an non-deterministic source can come from non-deterministic
formats like Parquet/CSV, not just functions.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -3198,6 +3198,18 @@
],
"sqlState" : "0A000"
},
+ "INSERT_REPLACE_USING_DUPLICATE_SCOPE_COLUMN" : {
+ "message" : [
+ "INSERT INTO ... REPLACE USING references the scope column <columnName>
more than once. Each scope column must be listed at most once."
+ ],
+ "sqlState" : "42701"
+ },
+ "INSERT_REPLACE_USING_NON_DETERMINISTIC_SOURCE" : {
+ "message" : [
+ "INSERT INTO ... REPLACE USING requires a deterministic source query
because the source determines both which target rows are deleted and which rows
are inserted. Remove non-deterministic expressions (e.g. rand()) from the
source."
+ ],
+ "sqlState" : "42K0E"
Review Comment:
```suggestion
"sqlState" : "0A000"
```
Feature not supported sqlState
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteReplaceUsing.scala:
##########
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute,
EqualNullSafe, Exists, Expression, Literal, MetadataAttribute, NamedExpression,
OuterReference}
+import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
+import org.apache.spark.sql.catalyst.plans.{LeftAnti, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Assignment,
CTERelationDef, CTERelationRef, Filter, Join, JoinHint, LogicalPlan, Project,
ReplaceData, ReplaceUsingTable, Union, WithCTE, WriteDelta}
+import org.apache.spark.sql.catalyst.util.RowDeltaUtils.{COPY_OPERATION,
INSERT_OPERATION, OPERATION_COLUMN}
+import org.apache.spark.sql.connector.write.{RowLevelOperationTable,
SupportsDelta}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.types.IntegerType
+
+/**
+ * Rewrites `INSERT INTO ... REPLACE USING (cols) <query>` (a
[[ReplaceUsingTable]]) onto the
+ * row-level write machinery via a [[RowLevelOperation.Command.REPLACE]]
operation.
+ *
+ * Scoped replace deletes every target row whose scope-column tuple appears in
the source and
+ * appends all source rows, including duplicates that share a scope tuple. The
replacement state
+ * is computed from independent target and source branches so that source rows
are inserted
+ * regardless of whether their scope tuple already exists in the target.
+ *
+ * The source is shared through a CTE so the delete/carryover join, the insert
branch, and the
+ * runtime group filter all refer to the same analyzed source plan. A
non-deterministic source is
+ * rejected because it could otherwise delete based on a different source
result than it inserts.
+ */
+object RewriteReplaceUsing extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case r @ ReplaceUsingTable(aliasedTable, scopeColumns, source)
+ if r.resolved && source.resolved =>
+
+ // The source feeds the delete decision, insert payload, and runtime
group filter.
+ // Evaluating a non-deterministic source in those independent plan
locations could produce
+ // mutually inconsistent results.
+ if (!source.deterministic) {
+ throw QueryCompilationErrors.insertReplaceUsingNonDeterministicSource()
+ }
+
+ val (rel, operationTable) = buildReplaceOperationTable(aliasedTable)
+ val scopeOrdinals = resolveScopeOrdinals(rel, scopeColumns)
+ operationTable.operation match {
+ case _: SupportsDelta =>
+ buildWriteDeltaPlan(rel, operationTable, scopeOrdinals, source)
+ case _ =>
+ buildReplaceDataPlan(rel, operationTable, scopeOrdinals, source)
+ }
+ }
+
+ // Build a copy-on-write replacement from retained target rows and inserted
source rows.
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ scopeOrdinals: Seq[Int],
+ source: LogicalPlan): LogicalPlan = {
+
+ val rowAttrs = relation.output
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation,
operationTable.operation)
+ val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs)
+
+ val sourceCte = CTERelationDef(source)
+ val scopeRef = newSourceRef(sourceCte)
+ val insertRef = newSourceRef(sourceCte)
+ val filterRef = newSourceRef(sourceCte)
+
+ // COW rewrites whole groups, so target rows whose scope tuple is absent
from the source must
+ // be re-emitted as COPY to preserve them (and their metadata).
+ val antiJoinCond = scopeEquality(readRelation.output, scopeRef.output,
scopeOrdinals)
+ val carryoverJoin = Join(readRelation, scopeRef, LeftAnti,
Some(antiJoinCond), JoinHint.NONE)
+ val carryover = addOperationColumn(COPY_OPERATION, carryoverJoin)
+
+ // All source rows are inserted. Carryover supplies target metadata;
insert rows null it out.
+ val insertData = rowAttrs.indices.map { i =>
+ Alias(insertRef.output(i), rowAttrs(i).name)()
+ }
+ val insertMetadata = metadataAttrs.map(attr => Alias(Literal(null,
attr.dataType), attr.name)())
+ val insertOutput = operationAlias(INSERT_OPERATION) +: (insertData ++
insertMetadata)
+ val inserts = Project(insertOutput, insertRef)
+
+ val replacementQuery = Union(carryover :: inserts :: Nil)
+ val projections = buildReplaceDataProjections(replacementQuery, rowAttrs,
metadataAttrs)
+
+ val groupFilterCond = if (groupFilterEnabled) {
+ Some(scopeExists(relation, filterRef, scopeOrdinals))
+ } else {
+ None
+ }
+
+ // The replaced scope is source-defined, so there is no static target-only
predicate to push
+ // down during planning. File pruning flows through groupFilterCond, which
+ // RowLevelOperationRuntimeGroupFiltering turns into a dynamic subquery.
+ val writeRelation = relation.copy(table = operationTable)
+ val replaceData =
+ ReplaceData(writeRelation, TrueLiteral, replacementQuery, relation,
projections,
+ groupFilterCond)
+ WithCTE(replaceData, sourceCte :: Nil)
+ }
+
+ // Build a merge-on-read delta from matched-row deletes and inserted source
rows.
+ private def buildWriteDeltaPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ scopeOrdinals: Seq[Int],
+ source: LogicalPlan): LogicalPlan = {
+
+ val operation = operationTable.operation.asInstanceOf[SupportsDelta]
+ val rowAttrs = relation.output
+ val rowIdAttrs = resolveRowIdAttrs(relation, operation)
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation, operation)
+ val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs, rowIdAttrs)
+
+ // Keep row columns before metadata columns, matching the schema shape
expected by
+ // buildWriteDeltaProjections. Row ID attributes are metadata-classified
and land in metaPart.
+ val (metaPart, rowPart) = readRelation.output.partition { attr =>
+ MetadataAttribute.isValid(attr.metadata)
+ }
+ val branchNames = OPERATION_COLUMN +: (rowPart ++ metaPart).map(_.name)
+
+ val sourceCte = CTERelationDef(source)
+ val scopeRef = newSourceRef(sourceCte)
+ val insertRef = newSourceRef(sourceCte)
+ val filterRef = newSourceRef(sourceCte)
+
+ // Matched target rows are encoded as deletes: row ID and metadata are
retained, row columns
+ // are nulled out.
+ val semiJoinCond = scopeEquality(readRelation.output, scopeRef.output,
scopeOrdinals)
+ val matchingRows = Join(readRelation, scopeRef, LeftSemi,
Some(semiJoinCond), JoinHint.NONE)
+ val deleteOutput = deltaDeleteOutput(rowPart, rowIdAttrs, metaPart)
+ val deletes = Project(named(deleteOutput, branchNames), matchingRows)
+
+ // Every source row is inserted; row IDs and metadata are null for new
rows.
+ val relationValueByExprId =
+ relation.output.zipWithIndex.map {
+ case (attr, i) => attr.exprId -> insertRef.output(i)
+ }.toMap
+ val insertAssignments = rowPart.map { attr =>
+ val value = relationValueByExprId.getOrElse(attr.exprId, Literal(null,
attr.dataType))
+ Assignment(attr, value)
+ }
+ val insertOutput = deltaInsertOutput(insertAssignments, metaPart)
+ val inserts = Project(named(insertOutput, branchNames), insertRef)
+
+ val deltaQuery = Union(deletes :: inserts :: Nil)
+ val projections = buildWriteDeltaProjections(deltaQuery, rowAttrs,
rowIdAttrs, metadataAttrs)
+
+ val groupFilterCond = if (groupFilterEnabled) {
+ Some(scopeExists(relation, filterRef, scopeOrdinals))
+ } else {
+ None
+ }
+
+ val writeRelation = relation.copy(table = operationTable)
+ val writeDelta =
+ WriteDelta(writeRelation, TrueLiteral, deltaQuery, relation,
projections, groupFilterCond)
+ WithCTE(writeDelta, sourceCte :: Nil)
+ }
+
+ private def operationAlias(operation: Int): NamedExpression = {
+ Alias(Literal(operation, IntegerType), OPERATION_COLUMN)()
+ }
+
+ private def named(exprs: Seq[Expression], names: Seq[String]):
Seq[NamedExpression] = {
+ exprs.zip(names).map {
+ case (ne: NamedExpression, name) if ne.name == name => ne
+ case (e, name) => Alias(e, name)()
+ }
+ }
+
+ private def resolveScopeOrdinals(
+ relation: DataSourceV2Relation,
+ scopeColumns: Seq[String]): Seq[Int] = {
+
+ val seen = scala.collection.mutable.HashSet.empty[Int]
+ scopeColumns.map { name =>
+ val resolved = relation.resolve(Seq(name), conf.resolver).getOrElse {
+ throw QueryCompilationErrors.unresolvedColumnError(name,
relation.output.map(_.name))
+ }
+ val ordinal = relation.output.indexWhere(_.exprId ==
resolved.toAttribute.exprId)
+ if (ordinal < 0) {
+ throw QueryCompilationErrors.unresolvedColumnError(name,
relation.output.map(_.name))
+ }
+ if (!seen.add(ordinal)) {
+ throw
QueryCompilationErrors.insertReplaceUsingDuplicateScopeColumn(name)
+ }
+ ordinal
+ }
+ }
+
+ private def newSourceRef(sourceDef: CTERelationDef): CTERelationRef = {
+ val ref = CTERelationRef(
+ sourceDef.id,
+ _resolved = true,
+ sourceDef.child.output,
+ sourceDef.child.isStreaming)
+ ref.newInstance().asInstanceOf[CTERelationRef]
+ }
+
+ private def scopeEquality(
+ targetOutput: Seq[Attribute],
+ sourceOutput: Seq[Attribute],
+ scopeOrdinals: Seq[Int]): Expression = {
+ scopeOrdinals
+ .map(ordinal => EqualNullSafe(targetOutput(ordinal),
sourceOutput(ordinal)): Expression)
Review Comment:
```suggestion
.map(ordinal => EqualNullSafe(targetOutput(ordinal),
sourceOutput(ordinal)))
```
Is type enforcement necessary? If not let's remove it
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -1283,10 +1284,24 @@ class Analyzer(
object ResolveInsertInto extends ResolveInsertionBase {
override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsWithPruning(
AlwaysProcess.fn, ruleId) {
+ case i: InsertIntoStatement
+ if i.table.isInstanceOf[DataSourceV2Relation] &&
+ i.query.resolved &&
+ i.replaceCriteriaOpt.exists(_.isInstanceOf[InsertReplaceUsing]) =>
+ val r = i.table.asInstanceOf[DataSourceV2Relation]
+ val scopeColumns =
i.replaceCriteriaOpt.get.asInstanceOf[InsertReplaceUsing].cols
+ // Align the source to the target with standard INSERT resolution so
RewriteReplaceUsing can
+ // compare scope columns and build the insert payload positionally by
ordinal.
+ val alignedQuery = TableOutputResolver.resolveOutputColumns(
Review Comment:
From https://github.com/apache/spark/pull/56705/changes#r3525640699, I'm not
sure if we should call manually `TableOutputResolver.resolveOutputColumns(`,
because we have a Spark Rule
https://github.com/apache/spark/blob/3f2ac2cc4c2c1932ace3f286c04226bdb38100a6/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala#L3898
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala:
##########
@@ -1166,6 +1170,37 @@ case class MergeIntoTable(
}
}
+/**
+ * The logical plan of the `INSERT INTO t REPLACE USING (cols) <query>`
command, resolved from
+ * [[InsertIntoStatement]] when its replace criteria is an
[[InsertReplaceUsing]].
+ *
+ * Scoped replace deletes every target row whose scope-column tuple appears in
the source and
+ * appends all source rows, including duplicates that share the same scope
tuple. The command uses
+ * [[RowLevelOperation.Command.REPLACE]] so connectors can distinguish this
scope-level replacement
+ * from per-row MERGE semantics.
+ *
+ * @param targetTable the target relation to replace rows in
+ * @param scopeColumns unqualified target column names whose values define a
replace scope tuple
+ * @param query the source query, already aligned to the target table
output
+ */
+case class ReplaceUsingTable(
+ targetTable: LogicalPlan,
+ scopeColumns: Seq[String],
+ query: LogicalPlan)
+ extends BinaryCommand with SupportsSubquery with TransactionalWrite {
+
+ override def table: LogicalPlan = EliminateSubqueryAliases(targetTable)
Review Comment:
Do we really need this? Also why `EliminateSubqueryAliases(targetTable)`?
Just seems odd compared to other V2Commands
https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala:
##########
@@ -114,6 +114,12 @@ class
RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla
// rewrite the group filter subquery as joins
val filter = Filter(cond, relation)
RewritePredicateSubquery(filter)
+
+ case REPLACE =>
Review Comment:
We can group this case, with the MERGE case
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteReplaceUsing.scala:
##########
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute,
EqualNullSafe, Exists, Expression, Literal, MetadataAttribute, NamedExpression,
OuterReference}
+import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
+import org.apache.spark.sql.catalyst.plans.{LeftAnti, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.logical.{Assignment,
CTERelationDef, CTERelationRef, Filter, Join, JoinHint, LogicalPlan, Project,
ReplaceData, ReplaceUsingTable, Union, WithCTE, WriteDelta}
+import org.apache.spark.sql.catalyst.util.RowDeltaUtils.{COPY_OPERATION,
INSERT_OPERATION, OPERATION_COLUMN}
+import org.apache.spark.sql.connector.write.{RowLevelOperationTable,
SupportsDelta}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.types.IntegerType
+
+/**
+ * Rewrites `INSERT INTO ... REPLACE USING (cols) <query>` (a
[[ReplaceUsingTable]]) onto the
+ * row-level write machinery via a [[RowLevelOperation.Command.REPLACE]]
operation.
+ *
+ * Scoped replace deletes every target row whose scope-column tuple appears in
the source and
+ * appends all source rows, including duplicates that share a scope tuple. The
replacement state
+ * is computed from independent target and source branches so that source rows
are inserted
+ * regardless of whether their scope tuple already exists in the target.
+ *
+ * The source is shared through a CTE so the delete/carryover join, the insert
branch, and the
+ * runtime group filter all refer to the same analyzed source plan. A
non-deterministic source is
+ * rejected because it could otherwise delete based on a different source
result than it inserts.
+ */
+object RewriteReplaceUsing extends RewriteRowLevelCommand {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+ case r @ ReplaceUsingTable(aliasedTable, scopeColumns, source)
+ if r.resolved && source.resolved =>
+
+ // The source feeds the delete decision, insert payload, and runtime
group filter.
+ // Evaluating a non-deterministic source in those independent plan
locations could produce
+ // mutually inconsistent results.
+ if (!source.deterministic) {
+ throw QueryCompilationErrors.insertReplaceUsingNonDeterministicSource()
+ }
+
+ val (rel, operationTable) = buildReplaceOperationTable(aliasedTable)
+ val scopeOrdinals = resolveScopeOrdinals(rel, scopeColumns)
+ operationTable.operation match {
+ case _: SupportsDelta =>
+ buildWriteDeltaPlan(rel, operationTable, scopeOrdinals, source)
+ case _ =>
+ buildReplaceDataPlan(rel, operationTable, scopeOrdinals, source)
+ }
+ }
+
+ // Build a copy-on-write replacement from retained target rows and inserted
source rows.
+ private def buildReplaceDataPlan(
+ relation: DataSourceV2Relation,
+ operationTable: RowLevelOperationTable,
+ scopeOrdinals: Seq[Int],
+ source: LogicalPlan): LogicalPlan = {
+
+ val rowAttrs = relation.output
+ val metadataAttrs = resolveRequiredMetadataAttrs(relation,
operationTable.operation)
+ val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs)
+
+ val sourceCte = CTERelationDef(source)
+ val scopeRef = newSourceRef(sourceCte)
+ val insertRef = newSourceRef(sourceCte)
+ val filterRef = newSourceRef(sourceCte)
+
+ // COW rewrites whole groups, so target rows whose scope tuple is absent
from the source must
+ // be re-emitted as COPY to preserve them (and their metadata).
+ val antiJoinCond = scopeEquality(readRelation.output, scopeRef.output,
scopeOrdinals)
+ val carryoverJoin = Join(readRelation, scopeRef, LeftAnti,
Some(antiJoinCond), JoinHint.NONE)
+ val carryover = addOperationColumn(COPY_OPERATION, carryoverJoin)
+
+ // All source rows are inserted. Carryover supplies target metadata;
insert rows null it out.
+ val insertData = rowAttrs.indices.map { i =>
+ Alias(insertRef.output(i), rowAttrs(i).name)()
Review Comment:
I'm not sure if we need all these `Alias`-ing in the INSERT steps, I think
it should already be handled by `TableOutputResolver.resolveOutputColumns`, or
something in the `Analyzer`, INSERT SQLs in DSv1 already has a way to allow
INSERT source query with different names into the table via aliasing (or
projecting)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
