dongjoon-hyun opened a new pull request, #60096: URL: https://github.com/apache/spark/pull/60096
### What changes were proposed in this pull request? This PR aims to fix a correctness issue of group-based row-level UPDATE. `DataSourceV2Strategy` no longer puts the `DynamicPruningSubquery` filters of `PartitionPruning` into the runtime filters of a read of a group-based row-level operation, that is, a read of a `RowLevelOperationTable` whose operation does not implement `SupportsDelta`. Such a read keeps only the group filter of `RowLevelOperationRuntimeGroupFiltering`. Delta-based operations and other reads keep dynamic partition pruning. ### Why are the changes needed? This is a correctness issue. When `spark.sql.optimizer.runtime.rowLevelOperationGroupFilter.enabled` is `false`, a group-based UPDATE writes some unchanged rows twice, without any error, if its condition has an `IN` or `EXISTS` subquery on a runtime filter column, such as an identity partition column. Since [SPARK-43975](https://issues.apache.org/jira/browse/SPARK-43975) ([v3.5.0](https://github.com/apache/spark/releases/tag/v3.5.0), 2023-09-09), such an UPDATE is rewritten as a union of two reads of the table, one for the updated rows and one for the rows to copy over. Both reads share the configured scan of the operation, and the operation replaces the groups that this scan returns. The read of the updated rows becomes a left semi join, and `PartitionPruning` adds a `DynamicPruningSubquery` to that read only. At runtime, this filter prunes the shared scan, so the write does not replace the pruned groups. However, the read of the copied rows has no such filter and still copies their rows over, so those rows are written twice. ```scala // spark.sql.catalog.cat=org.apache.spark.sql.connector.catalog.InMemoryRowLevelOperationTableCatalog spark.conf.set("spark.sql.optimizer.runtime.rowLevelOperationGroupFilter.enabled", "false") sql("CREATE TABLE cat.ns1.t (pk INT NOT NULL, id INT, salary INT, dep STRING) PARTITIONED BY (dep)") sql("INSERT INTO cat.ns1.t VALUES (1, 1, 300, 'hr'), (2, 2, 150, 'software'), (3, 3, 120, 'hr')") Seq("hr").toDF("d").createOrReplaceTempView("src") // Issue: It should update the 2 `hr` rows, but the table also ends up with the `software` row twice. sql("UPDATE cat.ns1.t SET salary = -1 WHERE dep IN (SELECT d FROM src)") ``` - `WHERE EXISTS (SELECT 1 FROM src WHERE d = dep)` gives the same result. - With the group filter enabled (the default), the result is correct, because `RowLevelOperationRuntimeGroupFiltering` adds the same group filter to every read of the scan. The groups that this filter keeps are a subset of the groups that the dynamic partition pruning filter keeps. - DELETE and MERGE are not affected, since dynamic partition pruning does not apply to their reads. This issue was reported in [the review](https://github.com/apache/spark/pull/60079#discussion_r4465170254) of [SPARK-60863](https://issues.apache.org/jira/browse/SPARK-60863), which fixed the same kind of issue for scalar subquery runtime filters. ### Does this PR introduce _any_ user-facing change? Yes. When the group filter is disabled, a group-based UPDATE whose condition has an `IN` or `EXISTS` subquery on a runtime filter column now writes each row once. Dynamic partition pruning no longer prunes the scan of a group-based row-level operation. With the group filter enabled (the default), it did not prune more groups than the group filter. ### How was this patch tested? Pass the CIs with a newly added test case. The test case in `GroupBasedUpdateTableSuite` fails without this PR, because the table ends up with the `software` row twice. ``` $ build/sbt 'sql/testOnly *GroupBased* *DeltaBased* *MergeIntoDataFrameSuite *TransactionSuite *DataSourceV2CatalystRuntimeFilterSuite *DataSourceV2EnhancedRuntimePartitionFilterSuite *DataSourceV2EnhancedDeleteFilterSuite *KeyGroupedPartitioningRuntimeFilterSuite *KeyGroupedPartitioningCatalystRuntimeFilterSuite *DynamicPartitionPruning*' ``` ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Opus 5.5 -- 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]
