SPJ Broken After Partition Evolution (Iceberg 1.10.0)

[andyguwc]

Apache Iceberg version

1.10.0

Query engine

Spark

Please describe the bug 🐞

SPJ Broken After Partition Evolution (Iceberg 1.10.0)

Summary

Storage-Partitioned Join (SPJ) stops working for MERGE INTO operations after a table undergoes partition evolution, even when all data has been rewritten into the new partition layout and old snapshots/specs have been fully cleaned up. The workaround is to recreate the table from scratch with the desired partition spec.

Environment

• Iceberg Spark Runtime: iceberg-spark-runtime-3.5_2.12-1.10.0
• Spark: 3.5 (AWS Glue 5.0)
• Catalog: AWS Glue Data Catalog

SPJ Configuration

The following Spark settings are configured and work correctly for tables that were created with their partition spec from the start (no evolution history):

spark.sql.sources.v2.bucketing.enabled = true
spark.sql.sources.v2.bucketing.pushPartValues.enabled = true
spark.sql.sources.v2.bucketing.pushPartKeys.enabled = true
spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled = true
spark.sql.requireAllClusterKeysForCoPartition = false
spark.sql.iceberg.planning.preserve-data-grouping = true

Table Definition

Tables are created with Iceberg format version 2, bucket partitioning on _ID, and hash distribution for writes and merges:

CREATE TABLE <catalog>.<db>.<table>
USING iceberg
PARTITIONED BY (bucket(64, _ID))
TBLPROPERTIES (
    'commit.manifest.min-count-to-merge' = '50',
    'format' = 'parquet',
    'format-version' = '2',
    'history.expire.max-snapshot-age-ms' = '604800000',
    'history.expire.min-snapshots-to-keep' = '10',
    'write.distribution-mode' = 'hash',
    'write.merge.distribution-mode' = 'hash',
    'write.metadata.delete-after-commit.enabled' = 'true',
    'write.metadata.previous-versions-max' = '10',
    'write.target-file-size-bytes' = '536870912'
)

Steps Performed

1. Table originally created with bucket(4, _ID)
2. Evolved partition via ALTER TABLE DROP PARTITION FIELD / ALTER TABLE ADD PARTITION FIELD bucket(64, _ID)
3. Ran rewrite_data_files with rewrite-all: true — all data files rewritten under new partition layout
4. Ran expire_snapshots with retain_last: 1 and older_than_minutes: 1 — old snapshots fully removed
5. Ran remove_orphan_files — old data files cleaned up from S3
6. Verified via DESCRIBE TABLE EXTENDED: only Part 0: bucket(64, _ID) remains (single clean spec)
7. Staging table created with matching PARTITIONED BY (bucket(64, _ID))

Expected

MERGE INTO should use SPJ (no Exchange/shuffle between staging and target table scans), since both tables share identical bucket(64, _ID) partitioning with a single spec.

Actual

Physical plan shows Exchange hashpartitioning(_ID, 32) and ShuffledHashJoin — full shuffle. Target table BatchScan shows groupedBy= (empty), while staging table correctly shows groupedBy=_ID_bucket. SPJ works for tables that were created with this partition spec originally (no evolution history).

Root Cause

Partition evolution metadata (even after cleanup) leaves residual state that prevents Spark from recognizing the table's bucket grouping during MERGE planning. Related upstream issue: apache/iceberg#13534 (closed without merge, Sep 2025).

Workaround

Recreate the table fresh via CREATE TABLE ... AS SELECT with the desired partition spec, then swap via ALTER TABLE RENAME. The new table has no evolution history and SPJ works immediately.

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

[moomindani]

I investigated this and found that SPJ works correctly after partition evolution when all data files use the new spec — SparkPartitioningAwareScan.specs() only considers specs with actual data, so the grouping key computation should work after a successful rewrite_data_files + expire_snapshots.

Could you check if any files still reference the old spec after your cleanup?

SELECT spec_id, count(*) FROM <catalog>.<db>.<table>.files GROUP BY spec_id;
DESCRIBE EXTENDED <catalog>.<db>.<table>;

If old-spec files remain, that would explain the empty grouping key.

[hemanthboyina]

tried reproducing the issue and as @moomindani mentioned I am also unable to reproduce it. Is anything am I missing here @andyguwc ?

`@TestTemplate
public void testMergeAfterBucketEvolutionAndRewrite() {
sql(
"CREATE TABLE %s (id INT, salary INT, dep STRING)"
+ "USING iceberg "
+ "PARTITIONED BY (bucket(4, id))"
+ "TBLPROPERTIES (%s)",
tableName, tablePropsAsString(COMMON_TABLE_PROPERTIES));

append(tableName, "{ \"id\": 1, \"salary\": 100, \"dep\": \"hr\" }");
append(tableName, "{ \"id\": 2, \"salary\": 200, \"dep\": \"hr\" }");
append(tableName, "{ \"id\": 3, \"salary\": 300, \"dep\": \"hardware\" }");

Table table = validationCatalog.loadTable(tableIdent);

// Evolve: drop bucket(4, id), add bucket(64, id)
table.updateSpec().removeField(Expressions.bucket("id", 4)).commit();
table.updateSpec().addField(Expressions.bucket("id", 64)).commit();

SparkActions.get(spark).rewriteDataFiles(table).option("rewrite-all", "true").execute();
SparkActions.get(spark).expireSnapshots(table).retainLast(1).execute();

// Create staging table with matching bucket(64, id)
sql(
    "CREATE TABLE %s (id INT, salary INT, dep STRING)"
        + "USING iceberg "
        + "PARTITIONED BY (bucket(64, id))"
        + "TBLPROPERTIES (%s)",
    tableName(OTHER_TABLE_NAME), tablePropsAsString(COMMON_TABLE_PROPERTIES));

append(tableName(OTHER_TABLE_NAME), "{ \"id\": 1, \"salary\": 150, \"dep\": \"hr\" }");
append(tableName(OTHER_TABLE_NAME), "{ \"id\": 4, \"salary\": 400, \"dep\": \"software\" }");

sql("REFRESH TABLE %s", tableName);

Map<String, String> mergeTableProps =
    ImmutableMap.of(
        TableProperties.MERGE_MODE,
        MERGE_ON_READ.modeName(),
        TableProperties.MERGE_DISTRIBUTION_MODE,
        "none");

sql("ALTER TABLE %s SET TBLPROPERTIES(%s)", tableName, tablePropsAsString(mergeTableProps));

withSQLConf(
    ENABLED_SPJ_SQL_CONF,
    () -> {
      SparkPlan plan =
          executeAndKeepPlan(
              "MERGE INTO %s AS t USING %s AS s "
                  + "ON t.id = s.id "
                  + "WHEN MATCHED THEN "
                  + "  UPDATE SET t.salary = s.salary "
                  + "WHEN NOT MATCHED THEN "
                  + "  INSERT *",
              tableName, tableName(OTHER_TABLE_NAME));
      String planAsString = plan.toString();
      // SPJ should work — no Exchange shuffles for MoR merge
      assertThat(planAsString)
          .as("MERGE INTO after bucket evolution + rewrite should use SPJ (no Exchange)")
          .doesNotContain("Exchange");
    });

}`

[RussellSpitzer]

I also was unable to reproduce this issue on Spark 3.5 or on Spark 4.1.

[jbewing]

Wasn't able to reproduce this issue, but still was able to reproduce #13530