jbewing opened a new issue, #13530:
URL: https://github.com/apache/iceberg/issues/13530
### Apache Iceberg version
1.9.0
### Query engine
Spark
### Please describe the bug 🐞
### What
If partitions are evolved for an Iceberg table, there is a potential to
disable Spark's newer "Storage Partitioned Join" feature if partitions are
re-ordered in a given table. Say we have two tables such as:
```sql
CREATE TABLE table1 (id BIGINT, col_a INT, col_b STRING)
USING ICEBERG
PARTITIONED BY (id, col_a, col_b);
CREATE TABLE table2 (id BIGINT, col_a INT, col_b STRING)
USING ICEBERG
PARTITIONED BY (id, col_b, col_a);
```
Now suppose the user has written a bunch of data to table1 but actually
inadvertently messed up the partitioning. They meant for the partitioning in
table1 to match that of table2 exactly.
If the user was to run
```sql
-- Drop existing partitions
ALTER TABLE table1 DROP PARTITION FIELD col_a;
ALTER TABLE table1 DROP PARTITION FIELD col_b;
-- Re-add them, but with col_b first to match table2
ALTER TABLE table1 ADD PARTITION FIELD col_b;
ALTER TABLE table1 ADD PARTITION FIELD col_a;
```
Now suppose the user writes a bunch of data into table1 and table2.
Since, table1 has the partition fields (id, col_b, col_a) which precisely
matches table2. You would expect that the Spark Storage Partitioned Joins
feature would kick in if you were to then join these tables via a command like
```sql
SELECT *
FROM table1
INNER JOIN table2
ON (
table1.id = table2.id
AND table1.col_b = table2.col_b
AND table1.col_a = table2.col_a
)
```
However, when reporting partition information for these tables to Spark with
relevant SPJ configs enabled, a grouping key of `(id, col_a, col_b)` if
reported for table1 whereas a grouping key of `(id, col_b, col_a)` is reported
for table2 which will ultimately results in Spark deciding that this join isn't
a candidate for Storage Partitioned Join optimization.
### Minimal Reproducible Example
I've added the following unit test to TestStoragePartitionedJoins.java for
Spark 3.5 which fails on the master branch
```java
@TestTemplate
public void testJoinsWhichReorderPartitionsViaSpecEvolution() {
sql(
"CREATE TABLE %s (id BIGINT, int_col INT, dep STRING)"
+ "USING iceberg "
+ "PARTITIONED BY (bucket(8, id), int_col, dep) "
+ "TBLPROPERTIES (%s)",
tableName, tablePropsAsString(TABLE_PROPERTIES));
Table table = validationCatalog.loadTable(tableIdent);
// evolve the spec in the first table by reordering the final two
partition columns by first
// removing them and re-adding them
table.updateSpec().removeField("int_col").commit();
table.updateSpec().removeField("dep").commit();
table.updateSpec().addField("dep").commit();
table.updateSpec().addField("int_col").commit();
sql("REFRESH TABLE %s", tableName);
sql("INSERT INTO %s VALUES (1L, 100, 'software')", tableName);
sql("INSERT INTO %s VALUES (2L, 200, 'hr')", tableName);
// create another table partitioned by `other_dep` partitioned
identically to the original
// partitioning of the first table but with the last two partition
columns reordered (e.g.
// the current state of the first table)
sql(
"CREATE TABLE %s (other_id BIGINT, other_int_col INT, other_dep
STRING)"
+ "USING iceberg "
+ "PARTITIONED BY (bucket(8, other_id), other_dep,
other_int_col)"
+ "TBLPROPERTIES (%s)",
tableName(OTHER_TABLE_NAME), tablePropsAsString(TABLE_PROPERTIES));
sql("INSERT INTO %s VALUES (1L, 100, 'software')",
tableName(OTHER_TABLE_NAME));
sql("INSERT INTO %s VALUES (2L, 200, 'hr')",
tableName(OTHER_TABLE_NAME));
assertPartitioningAwarePlan(
1, /* expected num of shuffles with SPJ */
3, /* expected num of shuffles without SPJ */
"SELECT * "
+ "FROM %s "
+ "INNER JOIN %s "
+ "ON id = other_id AND int_col = other_int_col AND dep =
other_dep "
+ "ORDER BY id, int_col, dep",
tableName,
tableName(OTHER_TABLE_NAME));
}
```
### Proposed Fix
I believe that table1 and table2 should either be reporting identical
grouping keys if table1's partition structure has evolved to completely match
that of table2. The current issue seems to be that grouping keys are ordered by
"fieldId" which means that if you re-order where a partition appears in the
spec over time, it will stay "sticky" to its original position in the spec
rather than reflecting it's latest position in the spec.
I'm willing to contribute this change as I've verified that it fixes the
above issue and causes the unit test in question to pass. I'd love feedback
from the community if there's a better way to do this or if this is an issue
that the Spark execution engine should handle.
### 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
- [ ] 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]
