dongjoon-hyun commented on code in PR #60096:
URL: https://github.com/apache/spark/pull/60096#discussion_r4470588633
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -233,7 +233,15 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
// dynamicFilters need no such check: a DynamicPruningSubquery over a
non-deterministic
// filtering plan is itself non-deterministic, so
CleanupDynamicPruningFilters has already
// rewritten it to TrueLiteral by the time we get here.
- val runtimeFilters = dynamicFilters ++ scalarSubqueryFilters
+ // A group-based read keeps only the group filter, for the reason above.
PartitionPruning
+ // plans a DynamicPruningSubquery on one read only, such as the read of
the updated rows of
+ // an UPDATE with a subquery, so it would drop groups from the shared
scan whose rows the
+ // read of the copied rows still writes.
+ val runtimeFilters = if (groupBasedRead) {
Review Comment:
Thank you, @peter-toth. You're right. I confirmed both shapes on master:
with the default configs, the `NOT EXISTS` DELETE deletes nothing and the `NOT
EXISTS` UPDATE keeps `(2, 150, software)` next to the updated row, with AQE on
and off. Both are also wrong with the group filter disabled, and this PR gives
the right rows in all of these cases. I updated the PR title and the
description to cover DELETE and UPDATE with any group filter setting, and I'll
update SPARK-60878 too. For the versions, I couldn't run it on 3.4 either, so
the description only says that the DELETE case needs SPARK-38085 and
SPARK-38959, which are both in v3.4.0.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedUpdateTableSuite.scala:
##########
@@ -222,6 +222,30 @@ class GroupBasedUpdateTableSuite extends
UpdateTableSuiteBase {
}
}
+ test("SPARK-60878: update with a subquery on the group key and the group
filter disabled") {
Review Comment:
Thank you. I added both test cases as you suggested in a962bac4402. They
fail on master and pass on this PR.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -233,7 +233,15 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
// dynamicFilters need no such check: a DynamicPruningSubquery over a
non-deterministic
// filtering plan is itself non-deterministic, so
CleanupDynamicPruningFilters has already
// rewritten it to TrueLiteral by the time we get here.
- val runtimeFilters = dynamicFilters ++ scalarSubqueryFilters
+ // A group-based read keeps only the group filter, for the reason above.
PartitionPruning
+ // plans a DynamicPruningSubquery on one read only, such as the read of
the updated rows of
+ // an UPDATE with a subquery, so it would drop groups from the shared
scan whose rows the
+ // read of the copied rows still writes.
Review Comment:
Thank you. I applied your suggestion as it is in a962bac4402.
--
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]
