This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 9686bfff3ea0 [SPARK-60673][SQL] Read all referenced metadata columns
in DSv2 DELETE, UPDATE and MERGE
9686bfff3ea0 is described below
commit 9686bfff3ea06ad00c0416e5bee48fa5fbfd0d99
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Thu Oct 29 06:44:26 2026 -0700
[SPARK-60673][SQL] Read all referenced metadata columns in DSv2 DELETE,
UPDATE and MERGE
### What changes were proposed in this pull request?
This PR makes the rewrite of DSv2 `DELETE`, `UPDATE` and `MERGE` read every
target metadata column that the command references, not only the ones that the
row-level operation requests through `requiredMetadataAttributes()` and, for
delta-based operations, `rowId()`.
`RewriteRowLevelCommand.buildRelationWithAttrs` builds the read relation
from the data columns, the row ID columns and the required metadata columns.
This relation reads `RowLevelOperationTable`, which exposes no metadata
columns, so `AddMetadataColumns` cannot add a referenced metadata column after
the rewrite. A condition or an assignment that references any other metadata
column of the target then refers to an attribute that the read relation does
not produce.
- `RewriteRowLevelCommand` has a new helper,
`resolveReferencedMetadataAttrs`, that returns the target metadata columns
(`relation.metadataOutput`) that the given expressions reference and that the
relation does not output yet.
- `RewriteDeleteFromTable`, `RewriteUpdateTable` and
`RewriteMergeIntoTable` add these columns to the read relation. They take them
from the condition for `DELETE`, from the condition and the assignments for
`UPDATE`, and from the `ON` condition and all clauses for `MERGE`.
- A group-based `MERGE` copies carried-over target rows with all read
columns (`Keep(Copy, ...)` over `readRelation.output`), so every `MergeRows`
instruction now produces all read metadata columns instead of only the required
ones.
The write does not change. `ReplaceDataProjections` and
`WriteDeltaProjections` still project only the required metadata columns, so
`LogicalWriteInfo` is the same as before and the extra columns are only read.
Spark requests them from the row-level operation scan through `pruneColumns`,
in the same way as the required metadata columns and the row ID columns, and as
it already requests all metadata columns when `AddMetadataColumns` widens the
target in a later analyzer iteration. `Sup [...]
This PR does not change the runtime group filters. If a target metadata
column appears in the `ON` condition of a `MERGE`, or in the condition of a
group-based `UPDATE` with a subquery, the runtime group filter still fails in
the same way as when the operation requests that column. apache/spark#59896
(SPARK-60668) and apache/spark#59893 (SPARK-60664) fix those cases.
### Why are the changes needed?
A DSv2 row-level command fails during analysis with an internal error
(SQLSTATE `XX000`) when it references a target metadata column that the
operation does not request. For example, with the in-memory test table whose
operations request no metadata columns (`no-metadata=true`, columns `pk INT NOT
NULL, id INT, dep STRING`, partitioned by `dep`):
```sql
DELETE FROM cat.ns1.test_table WHERE _partition = 'hr' AND id < 2
```
```
org.apache.spark.sql.catalyst.ExtendedAnalysisException:
[MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_MISSING_FROM_INPUT] Resolved
attribute(s) "_partition" missing from "pk", "id", "dep" in operator !Filter
((_partition#89 = hr) AND (id#86 < 2)). SQLSTATE: XX000
```
`UPDATE` and `MERGE` fail the same way. The same statements work when the
operation requests the referenced columns, so whether a valid statement runs
depends on which metadata columns the connector needs for its write. Connectors
usually request only the metadata columns that they need to write the data
back, for example a file path, while users can reference any metadata column
that the table exposes.
The read relation has included only the requested metadata columns since
these rewrites were added (`DELETE` in Apache Spark 3.4.0, `UPDATE` and `MERGE`
in 3.5.0).
This fix is also a prerequisite for apache/spark#59895 (SPARK-60667), which
stops `AddMetadataColumns` from adding all metadata columns to the target of
these commands when they are resolved in a later analyzer iteration, for
example because of type coercion. On master, `DELETE ... WHERE _partition =
'hr' AND id < 1.5` succeeds only through that widening, and the group-based
variant then writes `index` and `_partition` as data columns, while `... AND id
< 2` fails as shown above. apac [...]
### Does this PR introduce _any_ user-facing change?
Yes. DSv2 `DELETE`, `UPDATE` and `MERGE` statements that reference a target
metadata column that the row-level operation does not request now run instead
of failing with `MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_MISSING_FROM_INPUT`.
Statements that reference only requested metadata columns, or none, build the
same plans as before.
### How was this patch tested?
Pass the CIs with the newly added test cases.
- `DeleteFromTableSuiteBase`, `UpdateTableSuiteBase` and
`MergeIntoTableSuiteBase` reference `_partition` and `index` in a `DELETE`
condition, in an `UPDATE` condition and assignment, and in the `ON` condition,
clause conditions and an assignment of a `MERGE` whose `NOT MATCHED BY SOURCE`
clause disables the runtime group filter. They exercise this change in the
`*NoMetadata*` suites, whose operations request no metadata columns. In the
other suites, the operations request both column [...]
- The `*NoMetadata*` suites check that such columns are read but not
written, using the scan schema for `DELETE`, the `LogicalWriteInfo` and the
write log. This includes `DELETE ... WHERE _partition = 'hr' AND id < 2` from
the review of apache/spark#59895, a `MERGE` that references `_partition` only
in the `ON` condition and `index` only in the condition of a `NOT MATCHED BY
SOURCE` clause, and a group-based `UPDATE` whose subquery condition makes the
rewrite read the table twice, wit [...]
Without this fix, the new test cases fail in the `*NoMetadata*` suites with
`MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_MISSING_FROM_INPUT`. They pass in the
other suites, whose operations request both metadata columns.
```
$ build/sbt "sql/testOnly
org.apache.spark.sql.connector.*DeleteFromTable*Suite*
org.apache.spark.sql.connector.*UpdateTable*Suite*
org.apache.spark.sql.connector.*UpdateAsDeleteAndInsertTableSuite
org.apache.spark.sql.connector.*MergeIntoTable*Suite* -- -z SPARK-60673"
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59900 from dongjoon-hyun/SPARK-60673.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit ded9b617ef96dd6dcdced9533fa01f8c86881ca6)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../catalyst/analysis/RewriteDeleteFromTable.scala | 10 ++-
.../catalyst/analysis/RewriteMergeIntoTable.scala | 15 +++-
.../catalyst/analysis/RewriteRowLevelCommand.scala | 11 +++
.../sql/catalyst/analysis/RewriteUpdateTable.scala | 15 +++-
.../sql/connector/DeleteFromTableSuiteBase.scala | 21 +++++
.../DeltaBasedNoMetadataDeleteFromTableSuite.scala | 24 ++++++
.../DeltaBasedNoMetadataMergeIntoTableSuite.scala | 89 ++++++++++++++++++++++
.../DeltaBasedNoMetadataUpdateTableSuite.scala | 25 ++++++
.../GroupBasedNoMetadataDeleteFromTableSuite.scala | 23 ++++++
.../GroupBasedNoMetadataMergeIntoTableSuite.scala | 78 +++++++++++++++++++
.../GroupBasedNoMetadataUpdateTableSuite.scala | 51 +++++++++++++
.../sql/connector/MergeIntoTableSuiteBase.scala | 35 +++++++++
.../spark/sql/connector/UpdateTableSuiteBase.scala | 23 ++++++
13 files changed, 412 insertions(+), 8 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala
index 8bb55b1822b1..ee8e2c3388c7 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala
@@ -74,7 +74,10 @@ object RewriteDeleteFromTable extends RewriteRowLevelCommand
{
val metadataAttrs = resolveRequiredMetadataAttrs(relation,
operationTable.operation)
// construct a read relation and include all required metadata columns
- val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs)
+ // as well as metadata columns referenced in the condition
+ val referencedMetadataAttrs = resolveReferencedMetadataAttrs(relation,
Seq(cond))
+ val readRelation = buildRelationWithAttrs(
+ relation, operationTable, metadataAttrs ++ referencedMetadataAttrs)
// construct a plan that contains unmatched rows in matched groups that
must be carried over
// such rows do not match the condition but have to be copied over as the
source can replace
@@ -105,7 +108,10 @@ object RewriteDeleteFromTable extends
RewriteRowLevelCommand {
val metadataAttrs = resolveRequiredMetadataAttrs(relation, operation)
// construct a read relation and include all required metadata columns
- val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs, rowIdAttrs)
+ // as well as metadata columns referenced in the condition
+ val referencedMetadataAttrs = resolveReferencedMetadataAttrs(relation,
Seq(cond))
+ val readRelation = buildRelationWithAttrs(
+ relation, operationTable, metadataAttrs ++ referencedMetadataAttrs,
rowIdAttrs)
// construct a plan that only contains records to delete
val deletedRowsPlan = Filter(cond, readRelation)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
index 496514a839db..fd348e92a6dc 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
@@ -161,7 +161,12 @@ object RewriteMergeIntoTable extends
RewriteRowLevelCommand with PredicateHelper
val metadataAttrs = resolveRequiredMetadataAttrs(relation,
operationTable.operation)
// construct a read relation and include all required metadata columns
- val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs)
+ // as well as metadata columns referenced in the condition or actions
+ // carried over rows contain all read metadata columns, so every
instruction must produce them
+ val actions = matchedActions ++ notMatchedActions ++
notMatchedBySourceActions
+ val referencedMetadataAttrs = resolveReferencedMetadataAttrs(relation,
cond +: actions)
+ val readMetadataAttrs = dedupAttrs(metadataAttrs ++
referencedMetadataAttrs)
+ val readRelation = buildRelationWithAttrs(relation, operationTable,
readMetadataAttrs)
val checkCardinality = shouldCheckCardinality(matchedActions)
@@ -172,7 +177,7 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand
with PredicateHelper
val mergeRowsPlan = buildReplaceDataMergeRowsPlan(
readRelation, joinPlan, matchedActions, notMatchedActions,
- notMatchedBySourceActions, metadataAttrs, checkCardinality)
+ notMatchedBySourceActions, readMetadataAttrs, checkCardinality)
// predicates of the ON condition can be used to filter the target table
(planning & runtime)
// only if there is no NOT MATCHED BY SOURCE clause
@@ -278,7 +283,11 @@ object RewriteMergeIntoTable extends
RewriteRowLevelCommand with PredicateHelper
val metadataAttrs = resolveRequiredMetadataAttrs(relation, operation)
// construct a read relation and include all required metadata columns
- val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs, rowIdAttrs)
+ // as well as metadata columns referenced in the condition or actions
+ val actions = matchedActions ++ notMatchedActions ++
notMatchedBySourceActions
+ val referencedMetadataAttrs = resolveReferencedMetadataAttrs(relation,
cond +: actions)
+ val readRelation = buildRelationWithAttrs(
+ relation, operationTable, metadataAttrs ++ referencedMetadataAttrs,
rowIdAttrs)
// if there is no NOT MATCHED BY SOURCE clause, predicates of the ON
condition that
// reference only the target table can be pushed down
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala
index a64bdb6cbf58..3acd090b6d06 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala
@@ -109,6 +109,17 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
relation)
}
+ // metadata attrs referenced in the command (e.g. in the condition) must be
read even if they are
+ // not required by the operation as the row-level operation table exposes no
metadata columns
+ // such attrs are only read, the write projections include only required
metadata attrs
+ protected def resolveReferencedMetadataAttrs(
+ relation: DataSourceV2Relation,
+ exprs: Seq[Expression]): Seq[AttributeReference] = {
+
+ val references = AttributeSet(exprs.flatMap(_.references)) --
relation.outputSet
+ relation.metadataOutput.filter(references.contains)
+ }
+
protected def resolveRowIdAttrs(
relation: DataSourceV2Relation,
operation: SupportsDelta): Seq[AttributeReference] = {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
index 352ddc0b0acb..03c902d8cb91 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
@@ -71,7 +71,10 @@ object RewriteUpdateTable extends RewriteRowLevelCommand {
val metadataAttrs = resolveRequiredMetadataAttrs(relation,
operationTable.operation)
// construct a read relation and include all required metadata columns
- val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs)
+ // as well as metadata columns referenced in the condition or assignments
+ val referencedMetadataAttrs = resolveReferencedMetadataAttrs(relation,
cond +: assignments)
+ val readRelation = buildRelationWithAttrs(
+ relation, operationTable, metadataAttrs ++ referencedMetadataAttrs)
// build a plan with updated and copied over records
val query = buildReplaceDataUpdateProjection(readRelation, assignments,
cond)
@@ -95,9 +98,12 @@ object RewriteUpdateTable extends RewriteRowLevelCommand {
val metadataAttrs = resolveRequiredMetadataAttrs(relation,
operationTable.operation)
// construct a read relation and include all required metadata columns
+ // as well as metadata columns referenced in the condition or assignments
// the same read relation will be used to read records that must be
updated and copied over
// the analyzer will take care of duplicated attr IDs
- val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs)
+ val referencedMetadataAttrs = resolveReferencedMetadataAttrs(relation,
cond +: assignments)
+ val readRelation = buildRelationWithAttrs(
+ relation, operationTable, metadataAttrs ++ referencedMetadataAttrs)
// build a plan for updated records that match the condition
val matchedRowsPlan = Filter(cond, readRelation)
@@ -163,7 +169,10 @@ object RewriteUpdateTable extends RewriteRowLevelCommand {
val metadataAttrs = resolveRequiredMetadataAttrs(relation, operation)
// construct a read relation and include all required metadata columns
- val readRelation = buildRelationWithAttrs(relation, operationTable,
metadataAttrs, rowIdAttrs)
+ // as well as metadata columns referenced in the condition or assignments
+ val referencedMetadataAttrs = resolveReferencedMetadataAttrs(relation,
cond +: assignments)
+ val readRelation = buildRelationWithAttrs(
+ relation, operationTable, metadataAttrs ++ referencedMetadataAttrs,
rowIdAttrs)
// build a plan for updated records that match the condition
val matchedRowsPlan = Filter(cond, readRelation)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala
index 22ac338fbd6e..92e427d3de0c 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala
@@ -360,6 +360,27 @@ abstract class DeleteFromTableSuiteBase extends
RowLevelOperationSuiteBase {
checkDeleteMetrics(numDeletedRows = 1, numCopiedRows = 0)
}
+ test("SPARK-60673: delete with metadata columns in the condition") {
+ // the metadata columns must be read even if the operation does not
require them
+ createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
+ """{ "pk": 1, "id": 1, "dep": "hr" }
+ |{ "pk": 2, "id": 2, "dep": "hr" }
+ |{ "pk": 3, "id": 1, "dep": "software" }
+ |""".stripMargin)
+
+ sql(s"DELETE FROM $tableNameAsString WHERE _partition = 'hr' AND id < 2")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(2, 2, "hr") :: Row(3, 1, "software") :: Nil)
+ checkDeleteMetrics(numDeletedRows = 1, numCopiedRows = 1)
+
+ sql(s"DELETE FROM $tableNameAsString t WHERE t.index = 0 AND t.id > 1")
+
+ checkAnswer(sql(s"SELECT * FROM $tableNameAsString"), Row(3, 1,
"software") :: Nil)
+ checkDeleteMetrics(numDeletedRows = 1, numCopiedRows = 0)
+ }
+
test("delete with IN subqueries") {
withTempView("deleted_id", "deleted_dep") {
createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNoMetadataDeleteFromTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNoMetadataDeleteFromTableSuite.scala
index b33cf87402b3..086a696aa8a4 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNoMetadataDeleteFromTableSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNoMetadataDeleteFromTableSuite.scala
@@ -17,6 +17,9 @@
package org.apache.spark.sql.connector
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
class DeltaBasedNoMetadataDeleteFromTableSuite extends
DeleteFromTableSuiteBase {
override protected def deltaDelete: Boolean = true
@@ -29,4 +32,25 @@ class DeltaBasedNoMetadataDeleteFromTableSuite extends
DeleteFromTableSuiteBase
}
override def enforceCheckConstraintOnDelete: Boolean = false
+
+ test("SPARK-60673: delete reads non-required metadata columns without
writing them") {
+ createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
+ """{ "pk": 1, "id": 1, "dep": "hr" }
+ |{ "pk": 2, "id": 2, "dep": "hr" }
+ |{ "pk": 3, "id": 1, "dep": "software" }
+ |""".stripMargin)
+
+ executeAndCheckScan(
+ s"DELETE FROM $tableNameAsString WHERE _partition = 'hr' AND id < 2",
+ // `pk` is used to encode deletes
+ // `id` and `_partition` are used in the condition
+ expectedScanSchema = "pk INT, id INT, _partition STRING")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(2, 2, "hr") :: Row(3, 1, "software") :: Nil)
+
+ checkLastWriteInfo(expectedRowIdSchema = Some(StructType(Array(PK_FIELD))))
+ checkLastWriteLog(deleteWriteLogEntry(id = 1, metadata = null))
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNoMetadataMergeIntoTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNoMetadataMergeIntoTableSuite.scala
index d6e148425313..379c45a6d8e8 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNoMetadataMergeIntoTableSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNoMetadataMergeIntoTableSuite.scala
@@ -17,8 +17,13 @@
package org.apache.spark.sql.connector
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
class DeltaBasedNoMetadataMergeIntoTableSuite extends MergeIntoTableSuiteBase {
+ import testImplicits._
+
override protected def deltaMerge = true
override protected def extraTableProps: java.util.Map[String, String] = {
@@ -27,4 +32,88 @@ class DeltaBasedNoMetadataMergeIntoTableSuite extends
MergeIntoTableSuiteBase {
props.put("no-metadata", "true")
props
}
+
+ test("SPARK-60673: merge reads non-required metadata columns without writing
them") {
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |{ "pk": 3, "salary": 300, "dep": "hr" }
+ |""".stripMargin)
+
+ val sourceDF = Seq(1, 2, 5).toDF("pk")
+ sourceDF.createOrReplaceTempView("source")
+
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN MATCHED AND t._partition = 'hr' THEN
+ | UPDATE SET t.salary = t.index + 10
+ |WHEN NOT MATCHED THEN
+ | INSERT (pk, salary, dep) VALUES (s.pk, 0, 'hr')
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Seq(
+ Row(1, 10, "hr"), // update
+ Row(2, 200, "software"), // unchanged
+ Row(3, 300, "hr"), // unchanged
+ Row(5, 0, "hr"))) // insert
+
+ checkLastWriteInfo(
+ expectedRowSchema = StructType(table.schema.map {
+ case attr if attr.name == "salary" => attr.copy(nullable = false) //
input is not nullable
+ case attr => attr
+ }),
+ expectedRowIdSchema = Some(StructType(Array(PK_FIELD))))
+
+ checkLastWriteLog(
+ updateWriteLogEntry(id = 1, metadata = null, data = Row(1, 10, "hr")),
+ insertWriteLogEntry(data = Row(5, 0, "hr")))
+ }
+ }
+
+ test("SPARK-60673: merge reads non-required metadata columns only used in
conditions") {
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |{ "pk": 3, "salary": 300, "dep": "hr" }
+ |""".stripMargin)
+
+ val sourceDF = Seq(1, 2).toDF("pk")
+ sourceDF.createOrReplaceTempView("source")
+
+ // `_partition` is only in the ON condition and `index` is only in the
condition of
+ // the NOT MATCHED BY SOURCE clause, which also disables the runtime
group filter
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk AND t._partition = 'hr'
+ |WHEN MATCHED THEN
+ | UPDATE SET t.salary = -1
+ |WHEN NOT MATCHED BY SOURCE AND t.index = 1 THEN
+ | DELETE
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Seq(
+ Row(1, -1, "hr"), // update
+ Row(2, 200, "software"))) // unchanged
+
+ checkLastWriteInfo(
+ expectedRowSchema = StructType(table.schema.map {
+ case attr if attr.name == "salary" => attr.copy(nullable = false) //
input is a constant
+ case attr => attr
+ }),
+ expectedRowIdSchema = Some(StructType(Array(PK_FIELD))))
+
+ checkLastWriteLog(
+ updateWriteLogEntry(id = 1, metadata = null, data = Row(1, -1, "hr")),
+ deleteWriteLogEntry(id = 3, metadata = null))
+ }
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNoMetadataUpdateTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNoMetadataUpdateTableSuite.scala
index bf657806a30d..4c738ac13a96 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNoMetadataUpdateTableSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNoMetadataUpdateTableSuite.scala
@@ -17,6 +17,9 @@
package org.apache.spark.sql.connector
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
class DeltaBasedNoMetadataUpdateTableSuite extends UpdateTableSuiteBase {
override protected def deltaUpdate: Boolean = true
@@ -27,4 +30,26 @@ class DeltaBasedNoMetadataUpdateTableSuite extends
UpdateTableSuiteBase {
props.put("no-metadata", "true")
props
}
+
+ test("SPARK-60673: update reads non-required metadata columns without
writing them") {
+ createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
+ """{ "pk": 1, "id": 1, "dep": "hr" }
+ |{ "pk": 2, "id": 2, "dep": "hr" }
+ |{ "pk": 3, "id": 1, "dep": "software" }
+ |""".stripMargin)
+
+ sql(s"UPDATE $tableNameAsString SET id = index + 10 WHERE _partition =
'hr' AND id < 2")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 10, "hr") :: Row(2, 2, "hr") :: Row(3, 1, "software") :: Nil)
+
+ checkLastWriteInfo(
+ expectedRowSchema = StructType(table.schema.map {
+ case attr if attr.name == "id" => attr.copy(nullable = false) // input
is not nullable
+ case attr => attr
+ }),
+ expectedRowIdSchema = Some(StructType(Array(PK_FIELD))))
+ checkLastWriteLog(updateWriteLogEntry(id = 1, metadata = null, data =
Row(1, 10, "hr")))
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedNoMetadataDeleteFromTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedNoMetadataDeleteFromTableSuite.scala
index 6ac4f6e32fb1..219305294ce2 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedNoMetadataDeleteFromTableSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedNoMetadataDeleteFromTableSuite.scala
@@ -17,6 +17,8 @@
package org.apache.spark.sql.connector
+import org.apache.spark.sql.Row
+
class GroupBasedNoMetadataDeleteFromTableSuite extends
DeleteFromTableSuiteBase {
override protected def extraTableProps: java.util.Map[String, String] = {
@@ -24,4 +26,25 @@ class GroupBasedNoMetadataDeleteFromTableSuite extends
DeleteFromTableSuiteBase
props.put("no-metadata", "true")
props
}
+
+ test("SPARK-60673: delete reads non-required metadata columns without
writing them") {
+ createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
+ """{ "pk": 1, "id": 1, "dep": "hr" }
+ |{ "pk": 2, "id": 2, "dep": "hr" }
+ |{ "pk": 3, "id": 1, "dep": "software" }
+ |""".stripMargin)
+
+ executeAndCheckScans(
+ s"DELETE FROM $tableNameAsString WHERE _partition = 'hr' AND id < 2",
+ primaryScanSchema = "pk INT, id INT, dep STRING, _partition STRING",
+ groupFilterScanSchema = Some("id INT, dep STRING, _partition STRING"))
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(2, 2, "hr") :: Row(3, 1, "software") :: Nil)
+
+ checkReplacedPartitions(Seq("hr"))
+ checkLastWriteInfo(expectedRowSchema = table.schema)
+ checkLastWriteLog(writeLogEntry(data = Row(2, 2, "hr")))
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedNoMetadataMergeIntoTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedNoMetadataMergeIntoTableSuite.scala
index 5feadcdec23e..9fca5542c691 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedNoMetadataMergeIntoTableSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedNoMetadataMergeIntoTableSuite.scala
@@ -17,11 +17,89 @@
package org.apache.spark.sql.connector
+import org.apache.spark.sql.Row
+
class GroupBasedNoMetadataMergeIntoTableSuite extends MergeIntoTableSuiteBase {
+ import testImplicits._
+
override protected def extraTableProps: java.util.Map[String, String] = {
val props = new java.util.HashMap[String, String]()
props.put("no-metadata", "true")
props
}
+
+ test("SPARK-60673: merge reads non-required metadata columns without writing
them") {
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |{ "pk": 3, "salary": 300, "dep": "hr" }
+ |""".stripMargin)
+
+ val sourceDF = Seq(1, 2, 5).toDF("pk")
+ sourceDF.createOrReplaceTempView("source")
+
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN MATCHED AND t._partition = 'hr' THEN
+ | UPDATE SET t.salary = t.index + 10
+ |WHEN NOT MATCHED THEN
+ | INSERT (pk, salary, dep) VALUES (s.pk, 0, 'hr')
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Seq(
+ Row(1, 10, "hr"), // update
+ Row(2, 200, "software"), // unchanged
+ Row(3, 300, "hr"), // unchanged
+ Row(5, 0, "hr"))) // insert
+
+ checkLastWriteInfo(expectedRowSchema = table.schema)
+ checkLastWriteLog(
+ writeLogEntry(data = Row(1, 10, "hr")),
+ writeLogEntry(data = Row(2, 200, "software")),
+ writeLogEntry(data = Row(3, 300, "hr")),
+ writeLogEntry(data = Row(5, 0, "hr")))
+ }
+ }
+
+ test("SPARK-60673: merge reads non-required metadata columns only used in
conditions") {
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |{ "pk": 3, "salary": 300, "dep": "hr" }
+ |""".stripMargin)
+
+ val sourceDF = Seq(1, 2).toDF("pk")
+ sourceDF.createOrReplaceTempView("source")
+
+ // `_partition` is only in the ON condition and `index` is only in the
condition of
+ // the NOT MATCHED BY SOURCE clause, which also disables the runtime
group filter
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk AND t._partition = 'hr'
+ |WHEN MATCHED THEN
+ | UPDATE SET t.salary = -1
+ |WHEN NOT MATCHED BY SOURCE AND t.index = 1 THEN
+ | DELETE
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Seq(
+ Row(1, -1, "hr"), // update
+ Row(2, 200, "software"))) // unchanged
+
+ checkLastWriteInfo(expectedRowSchema = table.schema)
+ checkLastWriteLog(
+ writeLogEntry(data = Row(1, -1, "hr")),
+ writeLogEntry(data = Row(2, 200, "software")))
+ }
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedNoMetadataUpdateTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedNoMetadataUpdateTableSuite.scala
index 31db56b8d559..da2955827329 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedNoMetadataUpdateTableSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedNoMetadataUpdateTableSuite.scala
@@ -17,11 +17,62 @@
package org.apache.spark.sql.connector
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.internal.SQLConf
+
class GroupBasedNoMetadataUpdateTableSuite extends UpdateTableSuiteBase {
+ import testImplicits._
+
override protected def extraTableProps: java.util.Map[String, String] = {
val props = new java.util.HashMap[String, String]()
props.put("no-metadata", "true")
props
}
+
+ test("SPARK-60673: update reads non-required metadata columns without
writing them") {
+ withTempView("updated_pk") {
+ createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
+ """{ "pk": 1, "id": 1, "dep": "hr" }
+ |{ "pk": 2, "id": 2, "dep": "hr" }
+ |{ "pk": 3, "id": 1, "dep": "software" }
+ |""".stripMargin)
+
+ sql(s"UPDATE $tableNameAsString SET id = index + 10 WHERE _partition =
'hr' AND id < 2")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 10, "hr") :: Row(2, 2, "hr") :: Row(3, 1, "software") :: Nil)
+
+ checkReplacedPartitions(Seq("hr"))
+ checkLastWriteInfo(expectedRowSchema = table.schema)
+ checkLastWriteLog(
+ writeLogEntry(data = Row(1, 10, "hr")),
+ writeLogEntry(data = Row(2, 2, "hr")))
+
+ // UPDATEs with subqueries are rewritten as a UNION of two reads
+ // the runtime group filter is disabled to test only the reads
+ val updatedPkDF = Seq(3).toDF()
+ updatedPkDF.createOrReplaceTempView("updated_pk")
+
+ withSQLConf(SQLConf.RUNTIME_ROW_LEVEL_OPERATION_GROUP_FILTER_ENABLED.key
-> "false") {
+ sql(
+ s"""UPDATE $tableNameAsString
+ |SET id = index + 20
+ |WHERE _partition = 'software' AND pk IN (SELECT * FROM
updated_pk)
+ |""".stripMargin)
+ }
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 10, "hr") :: Row(2, 2, "hr") :: Row(3, 20, "software") :: Nil)
+
+ checkReplacedPartitions(Seq("hr", "software"))
+ checkLastWriteInfo(expectedRowSchema = table.schema)
+ checkLastWriteLog(
+ writeLogEntry(data = Row(1, 10, "hr")),
+ writeLogEntry(data = Row(2, 2, "hr")),
+ writeLogEntry(data = Row(3, 20, "software")))
+ }
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
index 93338ba043d9..37983b75faf4 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
@@ -893,6 +893,41 @@ abstract class MergeIntoTableSuiteBase extends
RowLevelOperationSuiteBase
}
}
+ test("SPARK-60673: merge with metadata columns in conditions and
assignments") {
+ // the metadata columns must be read even if the operation does not
require them
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |{ "pk": 3, "salary": 300, "dep": "hr" }
+ |{ "pk": 4, "salary": 400, "dep": "software" }
+ |""".stripMargin)
+
+ val sourceDF = Seq((1, "hr"), (2, "hr"), (5, "hr")).toDF("pk", "dep")
+ sourceDF.createOrReplaceTempView("source")
+
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk AND t._partition = s.dep
+ |WHEN MATCHED AND t.index = 0 THEN
+ | UPDATE SET t.salary = t.index + 10
+ |WHEN NOT MATCHED THEN
+ | INSERT (pk, salary, dep) VALUES (s.pk, 0, s.dep)
+ |WHEN NOT MATCHED BY SOURCE AND t._partition = 'software' THEN
+ | DELETE
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Seq(
+ Row(1, 10, "hr"), // update
+ Row(2, 0, "hr"), // insert
+ Row(3, 300, "hr"), // unchanged
+ Row(5, 0, "hr"))) // insert
+ }
+ }
+
test("merge with one NOT MATCHED BY SOURCE clause") {
withTempView("source") {
createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
index c0506ca03dcc..c2b6c1f7ec94 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
@@ -523,6 +523,29 @@ abstract class UpdateTableSuiteBase extends
RowLevelOperationSuiteBase {
checkUpdateMetrics(numUpdatedRows = 1, numCopiedRows = 0)
}
+ test("SPARK-60673: update with metadata columns in the condition and
assignments") {
+ // the metadata columns must be read even if the operation does not
require them
+ createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
+ """{ "pk": 1, "id": 1, "dep": "hr" }
+ |{ "pk": 2, "id": 2, "dep": "hr" }
+ |{ "pk": 3, "id": 1, "dep": "software" }
+ |""".stripMargin)
+
+ sql(s"UPDATE $tableNameAsString SET id = -1 WHERE _partition = 'hr' AND id
< 2")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, -1, "hr") :: Row(2, 2, "hr") :: Row(3, 1, "software") :: Nil)
+ checkUpdateMetrics(numUpdatedRows = 1, numCopiedRows = 1)
+
+ sql(s"UPDATE $tableNameAsString t SET t.id = t.index + 10 WHERE t.pk = 3")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, -1, "hr") :: Row(2, 2, "hr") :: Row(3, 10, "software") :: Nil)
+ checkUpdateMetrics(numUpdatedRows = 1, numCopiedRows = 0)
+ }
+
test("update with IN subqueries") {
withTempView("updated_id", "updated_dep") {
createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
