Spark: copy-on-write `UPDATE ... WHERE <subquery>` can silently duplicate rows — race in `SparkCopyOnWriteScan.filter()` when both `Union` branches apply the runtime file filter concurrently (Spark 4.0.x / 4.1.x)

### Apache Iceberg version

1.11.0 (latest release)

### Query engine

Spark

### Please describe the bug 🐞

**Symptom.** A copy-on-write `UPDATE t ... WHERE EXISTS ()` intermittently commits a snapshot that **adds a rewritten copy of every row in the table while deleting only the affected data file(s)**. All rows outside the affected files end up duplicated (byte-identical old + new copy). No error, no warning. In production a 71M-row / 922-file table was doubled twice on consecutive days: the commit added 577 data files and deleted 2 (`added-records` 71,428,702, `deleted-records` 154,479). The next MERGE into the table then fails with `MERGE_CARDINALITY_VIOLATION` because every id now matches two target rows.

**Minimal reproducer** (plain Spark SQL, ~35 lines). `t` is a COW table with exactly one row per data file (so a correct UPDATE touches 1 file and a corrupt one rewrites all of them); `m` holds the id to update — any subquery works, this is just the smallest shape that races (see "Sensitivity").

```python
# ICEBERG_PKG=org.apache.iceberg:iceberg-spark-runtime-4.0_2.13:1.11.0 python repro.py [files=1000] [runs=10]
# 4.1: iceberg-spark-runtime-4.1_2.13:1.11.0 + pyspark 4.1.3 3.5: iceberg-spark-runtime-3.5_2.12:1.8.1 + pyspark 3.5.5
import os, sys, tempfile
from pyspark.sql import SparkSession, functions as F

FILES = int(sys.argv[1]) if len(sys.argv) > 1 else 1000
RUNS = int(sys.argv[2]) if len(sys.argv) > 2 else 10
spark = (SparkSession.builder.master("local[4]")
.config("spark.jars.packages", os.environ.get("ICEBERG_PKG", "org.apache.iceberg:iceberg-spark-runtime-4.0_2.13:1.11.0"))
.config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
.config("spark.sql.catalog.local", "org.apache.iceberg.spark.SparkCatalog")
.config("spark.sql.catalog.local.type", "hadoop")
.config("spark.sql.catalog.local.warehouse", tempfile.mkdtemp())
.getOrCreate())
spark.sparkContext.setLogLevel("ERROR")
spark.sql("CREATE NAMESPACE IF NOT EXISTS local.db")
spark.createDataFrame([(0,)], "id bigint").createOrReplaceTempView("m") # the id(s) to update

dups = 0
for i in range(1, RUNS + 1):
spark.sql("DROP TABLE IF EXISTS local.db.t")
spark.sql("CREATE TABLE local.db.t (id BIGINT, v STRING) USING iceberg TBLPROPERTIES ('write.update.mode'='copy-on-write')")
spark.range(FILES).withColumn("v", F.lit("old")).repartitionByRange(FILES, "id").writeTo("local.db.t").append() # 1 row per file
spark.sql("UPDATE local.db.t t SET v = 'new' WHERE EXISTS (SELECT 1 FROM m WHERE m.id = t.id)")
rows = spark.table("local.db.t").count()
c = spark.sql("SELECT summary['added-data-files'] a, summary['deleted-data-files'] d, summary['added-records'] r "
"FROM local.db.t.snapshots ORDER BY committed_at DESC LIMIT 1").first()
dups += rows != FILES
print(f"run {i:2d}: rows={rows} {'DUPLICATED' if rows != FILES else 'ok'} commit: added_files={c.a} deleted_files={c.d} added_rows={c.r}")
print(f"{dups}/{RUNS} runs produced duplicate rows")
```

Output (all Spark configs default, AQE on):
```
run 1: rows=1999 DUPLICATED commit: added_files=1 deleted_files=1 added_rows=1000
run 2: rows=1000 ok commit: added_files=1 deleted_files=1 added_rows=1
...
```
| stack | duplicated runs |
|---|---|
| Spark 4.0.4 + Iceberg 1.11.0 | 3/10 (9/23 across batches) |
| Spark 4.1.3 + Iceberg 1.11.0 | 4/10 (12/30 across batches) |
| Spark 3.5.5 + Iceberg 1.8.1 | 0/10 (0/20) |

A corrupt run writes `FILES` rows (`added_rows=1000`) while deleting only the 1 affected file; a correct run writes 1.

**Mechanism** (from source at tag `apache-iceberg-1.11.0`, `spark/v4.0`, plus Spark 4.0.4 sources, event logs and stage metrics of the production runs):

1. Spark rewrites an UPDATE whose WHERE contains a subquery as `ReplaceData(Union(Filter(cond, S), Filter(NOT cond, S)))` (`RewriteUpdateTable.buildReplaceDataWithUnionPlan`): an "updated rows" branch and a "carry-over" branch (unchanged rows of the affected files) that read **the same** `SparkCopyOnWriteScan` `S` (`GroupBasedRowLevelOperationScanPlanning` replaces all occurrences with one scan relation; `RowLevelOperationRuntimeGroupFiltering` attaches the runtime file filter to both occurrences via `r.scan eq scan`). Both `BatchScanExec`s call `S.filter(...)` then `S.toBatch().planInputPartitions()` (`BatchScanExec.filteredPartitions`, lines 63–117 in 4.0.4).
2. `SparkCopyOnWriteScan.filter()` (`SparkCopyOnWriteScan.java` 124–138) is not synchronized and is not atomic: it **publishes `this.filteredLocations = fileLocations` (line 125) first**, then streams `tasks()` to build `filteredTasks`, logs, and only then calls `resetTasks(filteredTasks)` (line 138), which is what invalidates the memoized task groups (`SparkPartitioningAwareScan.resetTasks`, 246–249: `taskGroups = null; tasks = filteredTasks`). `tasks()`/`taskGroups()` are individually `synchronized` (179, 204) but the check-then-act spans the whole method. The guard on line 124 (`filteredLocations == null || fileLocations.size() < filteredLocations.size()`) exists precisely because "Spark may call this multiple times for UPDATEs with subqueries ... the same scan on both sides" — it anticipates multiple calls, but not concurrent ones.
3. `taskGroups()` is memoized **at planning time**, before any runtime filter runs (`outputPartitioning()` calls `taskGroups().size()`, lines 107–122; Spark's `EnsureRequirements` invokes it), and `SparkBatch` captures the task-group list at construction.
4. When AQE prepares the two branch stages concurrently, the second caller sees `filteredLocations` already set → takes the no-op path → `toBatch()` captures the still-memoized **pre-narrowing** task groups → that branch reads **all** files. Exactly one `"N of M task(s) ... matched runtime file filter"` INFO line is logged per run (the winner). If the losing branch is the carry-over branch, every row is rewritten while the commit deletes only `configuredScan.tasks()` = the narrowed files → duplicates. If it is the updated-rows branch the result is correct but the whole table is scanned. Hence ~50% corruption per run once the calls overlap.

Production evidence (Spark 4.0.4, 922 files): both branches carry `RuntimeFilters: [dynamicpruningexpression(_file IN subquery#...)]` in all 9 AQE plan versions; one narrowing log (`2 of 922 task(s) matched ... 2 location(s)`); stage metrics show one branch 2 tasks / 154,479 rows and the other 1,086 tasks / 71,428,702 rows in **both** a corrupt and a clean run — the corrupt one had the full scan feeding the carry-over branch (SortMergeJoin output 71,428,700 → write of 577 files), the clean one had it feeding the updated-rows branch (2 rows out). The two branch jobs were submitted 4 ms apart.

**Sensitivity (controls, Spark 4.0.4).** The hazard is constant; corruption requires AQE to prepare the two branch stages concurrently so the two `filter()` calls overlap. In the minimal repro that happens for a local-relation subquery side (branch jobs submitted 1–2 ms apart → 3/10); with `m` as a physical table, a temp view over a table scan, a cached read, or a derived aggregate, the branches were prepared sequentially (~25–65 ms apart, unchanged at 5,000 files) → 0/N. Production reached the concurrent case with a derived temp view over an already-materialized cached DataFrame + a table scan (4 ms apart). Spark 3.5.5 never overlapped in our runs, although the Iceberg code is identical.

**Workarounds verified with the same harness:** expressing the operation as `MERGE INTO local.db.t t USING m ON t.id = m.id WHEN MATCHED THEN UPDATE SET v = 'new'` (single read of the target, one `filter()` call) → 0/20 on Spark 4.0.4 and 0/20 on 4.1.3; `spark.sql.adaptive.enabled=false` for the statement → 0/10. Merge-on-read avoids the carry-over rewrite entirely. Making `m` a physical table only changes the timing.

**Prior art searched (none found):** issues/PRs mentioning `SparkCopyOnWriteScan` (#16295/#16303 migration to `SupportsRuntimeV2Filtering`, #6209, #9215 — none about concurrency), `resetTasks`, `filteredLocations`, duplicate-row issues (#14667, #7005 are different mechanisms), PR #2206 (original UPDATE-with-subquery design), dev list, Spark JIRA `SupportsRuntimeV2Filtering` (SPARK-45866 is context only).

**Suggested fix.**
- Make `filter()` atomic with respect to the task groups: synchronize `filter()`/`resetTasks()` on the scan (the getters already are), or compute `filteredTasks` and call `resetTasks()` **before** publishing `filteredLocations` (and make it `volatile`). Either closes the window.
- Defense in depth: in the copy-on-write commit (`SparkWrite.CopyOnWriteOperation`), fail if the rewritten data cannot have come only from `configuredScan.tasks()` (e.g. added records exceed the rows in the replaced files for an UPDATE/DELETE), instead of committing duplicates silently.
- Spark side (I can cross-file): `SupportsRuntimeV2Filtering.filter()` may be invoked concurrently on one shared scan from the two branches of the UPDATE rewrite; the API does not document thread-safety expectations, and 4.x prepares the branches concurrently where 3.5 did not.

### Willingness to contribute

- [ ] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [x] I cannot contribute a fix for this bug at this time

Start by running the minimal Python reproducer across the listed Spark and Iceberg versions. Read SparkCopyOnWriteScan.filter(), resetTasks(), and taskGroups(), then inspect RewriteUpdateTable.buildReplaceDataWithUnionPlan and the shared scan behavior. Done means concurrent runtime filtering no longer causes UPDATE ... WHERE EXISTS to duplicate unaffected rows.

Written by the indexing model from the issue text.