peter-toth commented on code in PR #60079:
URL: https://github.com/apache/spark/pull/60079#discussion_r4465170254
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -204,7 +204,16 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
// pruning while the FilterExec above the scan re-evaluates them, so the
two evaluations
// may disagree and rows the source pruned away could not be recovered.
This is the
// runtime counterpart of the pushFilters guard in PushDownUtils
(SPARK-58207).
- val scalarSubqueryFilters = if (relation.runtimeFilterAttrs.nonEmpty) {
+ // Nothing is routed for a read of a group-based row-level operation
either. The operation
+ // replaces the groups its scan returns, and its reads share that scan,
so a filter of one
+ // read must not prune it: the negated condition that selects the rows
to copy over would
+ // drop the groups where every row matches. Only the group filter of
+ // RowLevelOperationRuntimeGroupFiltering prunes that scan.
Review Comment:
**Finding 1.** DPP can also prune a group-based read, so this sentence does
not hold yet. An `IN` or `EXISTS` subquery makes the UPDATE a union of two
reads of the shared scan. The read of the updated rows becomes a left semi
join, and `PartitionPruning` puts a `DynamicPruningSubquery` on that read only.
With the group filter off, that filter prunes the shared scan to `hr`, but the
copy read still writes the `software` row:
```scala
// the table and `src` of the new UPDATE test
withSQLConf(SQLConf.RUNTIME_ROW_LEVEL_OPERATION_GROUP_FILTER_ENABLED.key ->
"false") {
sql(s"UPDATE $tableNameAsString SET salary = -1 WHERE dep IN (SELECT d
FROM src)")
}
// expected: (1, 1, -1, hr), (2, 2, 150, software), (3, 3, -1, hr)
// on master and on this PR: the same, plus (2, 2, 150, software) a second
time
```
- `WHERE EXISTS (SELECT 1 FROM src WHERE d = dep)` gives the same rows.
- With `spark.sql.optimizer.dynamicPartitionPruning.enabled=false`, the rows
are right.
- With the group filter on (the default), the rows are right, since both
reads get the group filter.
- DELETE and MERGE get no DPP on their reads, and their rows are right with
the group filter off.
This is pre-existing, and I found no ticket for it. Keeping only the group
filter among the dynamic filters of a group-based read fixes it:
```scala
val runtimeFilters = if (groupBasedRead) {
// only the group filter may prune the shared scan, see above
dynamicFilters.filterNot(_.isInstanceOf[DynamicPruningSubquery])
} else {
dynamicFilters ++ scalarSubqueryFilters
}
```
With it, the shapes above give the right rows with the group filter on and
off. With the group filter on, each read keeps exactly the group filter.
`*GroupBased* *DeltaBased* *DataSourceV2CatalystRuntimeFilterSuite
*DataSourceV2EnhancedRuntimePartitionFilterSuite
*DataSourceV2EnhancedDeleteFilterSuite` pass with it (2509 tests). A separate
ticket works too, but then this comment and the description's "the scan reads
every group when the group filter is disabled" need rewording.
--
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]
