dongjoon-hyun opened a new pull request, #60079: URL: https://github.com/apache/spark/pull/60079
### What changes were proposed in this pull request? This PR aims to fix a correctness issue of group-based row-level DELETE and UPDATE. `DataSourceV2Strategy` no longer routes scalar subquery filters 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`. Only the group filter of `RowLevelOperationRuntimeGroupFiltering` prunes that scan at runtime, as before [SPARK-56467](https://issues.apache.org/jira/browse/SPARK-56467). Delta-based operations keep the routing. ### Why are the changes needed? This is a correctness issue. A group-based DELETE leaves rows that it should delete, and a group-based UPDATE duplicates rows, without any error. It happens with the default configurations when the condition compares a runtime filter column, such as an identity partition column, with a scalar subquery. Since SPARK-56467 ([v4.2.0](https://github.com/apache/spark/releases/tag/v4.2.0), 2026-07-11), `DataSourceV2Strategy` routes a deterministic filter that has a scalar subquery and references only runtime filter columns into `BatchScanExec.runtimeFilters`. However, a group-based operation replaces the groups that its configured scan returns, and all reads of the operation share that scan. The read of a DELETE, and the copy branch of an UPDATE whose condition has a subquery, filter by the negated condition `NOT (cond <=> TRUE)`, which selects the rows to copy over. Pushing this negated filter into the shared scan drops the groups where every row matches, so the write does not replace them. ```scala // spark.sql.catalog.cat=org.apache.spark.sql.connector.catalog.InMemoryRowLevelOperationTableCatalog sql("CREATE TABLE cat.ns1.t (pk INT NOT NULL, id INT, salary INT, dep STRING) " + "PARTITIONED BY (dep) TBLPROPERTIES ('use-catalyst-runtime-filtering' = 'true')") 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 1: It should delete the 2 `hr` rows, but it deletes nothing. sql("DELETE FROM cat.ns1.t WHERE dep = (SELECT max(d) FROM src)") // Issue 2: On the same initial table, it should update the 2 `hr` rows, but the table ends up // with both the old and the updated `hr` rows. sql("UPDATE cat.ns1.t SET salary = -1 WHERE dep = (SELECT max(d) FROM src)") ``` A scan is affected when it prunes with the negated filter: - `SupportsRuntimeV2Filtering` scans since v4.2.0, e.g. a scan with `supportsIterativePushdown()` that prunes with the `PartitionPredicate` Spark derives from the filter. - `SupportsRuntimeCatalystFiltering` scans, added by [SPARK-58523](https://issues.apache.org/jira/browse/SPARK-58523) ([v4.3.0](https://github.com/apache/spark/releases/tag/v4.3.0), 2026-10-27). Disabling `spark.sql.optimizer.runtime.rowLevelOperationGroupFilter.enabled` does not avoid the issue, because the negated filter prunes the scan by itself. ### Does this PR introduce _any_ user-facing change? Yes. A group-based DELETE or UPDATE whose condition compares a runtime filter column with a scalar subquery now deletes or updates the matching rows. As in Apache Spark 4.1, only the group filter prunes its scan, and the scan reads every group when the group filter is disabled. ### How was this patch tested? Pass the CIs with newly added test cases. The two test cases in `GroupBasedRowLevelOperationCatalystRuntimeFilterSuite` fail without this PR, because the scans get the condition or its negation as a runtime filter besides the group filter. The test case in `DeltaBasedRowLevelOperationCatalystRuntimeFilterSuite` checks that a delta-based DELETE still prunes its scan with the condition. ``` $ build/sbt 'sql/testOnly *GroupBased* *DeltaBased* *MergeIntoDataFrameSuite *TransactionSuite *SchemaAlignmentConfigSuite *DataSourceV2CatalystRuntimeFilterSuite *DataSourceV2EnhancedRuntimePartitionFilterSuite *DataSourceV2EnhancedDeleteFilterSuite *KeyGroupedPartitioningRuntimeFilterSuite *KeyGroupedPartitioningCatalystRuntimeFilterSuite' $ build/sbt 'sql/testOnly *DataSourceV2SQLSuiteV2Filter -- -z SPARK-56467 -z SPARK-58207' ``` ### 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]
