ignaski opened a new issue, #18004:
URL: https://github.com/apache/iceberg/issues/18004
### Apache Iceberg version
1.11.0 (latest release)
### Query engine
Spark
### Please describe the bug 🐞
**Symptom.** A copy-on-write `UPDATE t ... WHERE EXISTS (<subquery>)`
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
--
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]
