ranflarion opened a new issue, #18556:
URL: https://github.com/apache/iceberg/issues/18556
### Apache Iceberg version
1.12.0 (latest release)
### Query engine
Spark
### Please describe the bug 🐞
`SparkBatch.equals` compares the table name and the hash code of the scan
that created the batch, so the batches of two scans whose hash codes collide
are equal even when the scans read different data. Spark's
`BatchScanExec.equals` compares batches and exchange reuse relies on it, so
Spark reuses one scan's shuffle for the other and returns its rows for both.
Reproduction in plain Spark SQL with default settings:
```sql
CREATE TABLE t (id BIGINT, data STRING) USING iceberg PARTITIONED BY (data);
INSERT INTO t VALUES (1, 'Aa'), (2, 'BB');
SELECT sum(id) FROM t WHERE data = 'Aa'
UNION ALL
SELECT sum(id) FROM t WHERE data = 'BB';
```
This returns `1` and `1` instead of `1` and `2`. `'Aa'` and `'BB'` have the
same `String.hashCode()`, so the two scans' filter descriptions, and with them
the scans, hash alike. The filters select whole identity partitions, so Iceberg
evaluates them completely and no Spark `Filter` above the scans tells the
branches apart. The final plan (Spark 4.2, main) reuses the first branch's
exchange for the second:
```
Union
:- HashAggregate(keys=[], functions=[sum(id#12L)], output=[sum(id)#18L])
: +- ShuffleQueryStage 0
: +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [plan_id=91]
: +- *(1) HashAggregate(keys=[], functions=[partial_sum(id#12L)],
output=[sum#38L])
: +- *(1) Project [id#12L]
: +- *(1) ColumnarToRow
: +- BatchScan testhadoop.default.table[id#12L, data#13]
IcebergScan(table=testhadoop.default.table, schemaId=0,
snapshotId=369853115188610976, branch=null, filters=data IS NOT NULL, data =
'Aa', runtimeFilters=, groupedBy=) RuntimeFilters: []
+- HashAggregate(keys=[], functions=[sum(id#14L)], output=[sum(id)#19L])
+- ShuffleQueryStage 2
+- ReusedExchange [sum#39L], Exchange SinglePartition,
ENSURE_REQUIREMENTS, [plan_id=91]
```
The scans' own `equals` methods compare their table, snapshot, read schema
and filters; `SparkScan.toBatch()` and `SparkChangelogScan.toBatch()` pass only
`hashCode()` to `SparkBatch` (since #5934), in every Spark module. Any two
scans of the same table whose 32-bit hash codes collide are affected when the
plans above them match, and string literals with equal hash codes, like `'Aa'`
and `'BB'`, make the collision deterministic. #13437 proposed a stronger digest
for this but was closed as stale.
### 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]
