edgarRd opened a new issue, #18901:
URL: https://github.com/apache/iceberg/issues/18901
### Apache Iceberg version
1.11.0
### Query engine
Spark
### Please describe the bug 🐞
### Problem
`SparkBatch.planInputPartitions()` creates a new table broadcast and a new
`FileIO` broadcast every time it is called:
```java
Broadcast<Table> tableBroadcast =
sparkContext.broadcast(SerializableTableWithSize.copyOf(table));
Broadcast<FileIO> fileIOBroadcast =
sparkContext.broadcast(SerializableFileIOWithSize.wrap(fileIO.get()));
```
`SparkScan.toBatch()` returns a new `SparkBatch` on each call, so nothing is
reused between calls, even though the table and the `FileIO` do not change
during the lifetime of a scan.
Spark calls `planInputPartitions()` once per copy of a scan's physical node,
while it is still planning the query. When dynamic partition pruning filters
nest over a chain of joins on a partition column - something similar to
([SPARK-48486](https://issues.apache.org/jira/browse/SPARK-48486)) - Spark
plans an exponential number of copies, and each one adds two broadcasts to the
driver. Iceberg never destroys these broadcasts, so they stay in the driver's
block manager until the plans that reference them are garbage collected.
### What we saw
A production query on Spark 4.1.1 with Iceberg 1.11.0, with 17 left joins of
which 10 are on the partition column:
- the driver created 264,382 broadcasts, exactly half from each of the two
`sparkContext.broadcast` calls above and none from anywhere else
- the driver's `MemoryStore` went from 14.2 GiB free to about 1 MiB in
roughly 6.5 minutes and never recovered
- the driver failed without starting a job or a stage
The same query succeeds with
`spark.sql.optimizer.dynamicPartitionPruning.enabled=false`.
The Spark 3.5, 4.0, 4.1 and 4.2 modules on `main` have the same code.
### Expected
Planning input partitions again for the same scan should reuse the
broadcasts it already created. The number of broadcasts should grow with the
number of scans in the query, not with the number of times Spark plans them.
### How to reproduce
Create seven small partitioned tables and join them on the partition column,
with a selective filter on the first one:
```sql
CREATE TABLE t0 (id BIGINT, ds STRING) USING iceberg PARTITIONED BY (ds);
INSERT INTO t0 VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd');
-- repeat for t1 .. t6
SELECT t0.id
FROM t0
LEFT JOIN t1 ON t0.ds = t1.ds AND t0.id = t1.id
LEFT JOIN t2 ON t0.ds = t2.ds AND t0.id = t2.id
LEFT JOIN t3 ON t0.ds = t3.ds AND t0.id = t3.id
LEFT JOIN t4 ON t0.ds = t4.ds AND t0.id = t4.id
LEFT JOIN t5 ON t0.ds = t5.ds AND t0.id = t5.id
LEFT JOIN t6 ON t0.ds = t6.ds AND t0.id = t6.id
WHERE t0.id < 3;
```
With default settings this query creates 415 to 429 broadcasts on `main`,
depending on the Spark version, for seven tables of four rows each. The count
can be read from the `Created broadcast N from broadcast at SparkBatch.java`
lines in the driver log, or from the ID of a broadcast created before and after
the query. Each additional join on `ds` roughly doubles it.
Calling `scan.toBatch().planInputPartitions()` repeatedly on a single scan
shows the same thing in isolation: every call creates two new broadcasts.
### Scope
The exponential number of DPP subqueries is a Spark planning problem and is
tracked in SPARK-48486 (it is unverified if this spark issue is exactly the
nesting root cause) . This issue is only about Iceberg creating new broadcasts
for every one of those planning calls, which turns slow planning into a driver
out-of-memory failure.
### Willingness to contribute
- [x] 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
- [ ] 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]
