From f005aca4a9a4313588fe8514124b39f283119a1e Mon Sep 17 00:00:00 2001
From: raashed <mohammedraashed10@gmail.com>
Date: Fri, 16 Oct 2026 02:30:28 +0530
Subject: [PATCH] Spark 4.2: Add test for rewrite_table_path after expiring
 rewritten position deletes

Rewriting position delete files and then expiring the snapshot that added
the old ones leaves DELETED entries in the manifest whose files no longer
exist. rewrite_table_path used to try to rewrite those files and fail with
a NotFoundException. That was fixed by #15470, but nothing covered this
case, so add a regression test.
---
 .../actions/TestRewriteTablePathsAction.java  | 55 +++++++++++++++++++
 1 file changed, 55 insertions(+)

diff --git a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java
index d520f350d291..25d87d77d6b2 100644
--- a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java
+++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java
@@ -536,6 +536,61 @@ private void runPositionDeletesTest(String fileFormat) throws Exception {
         .hasSize(1);
   }
 
+  @TestTemplate
+  void positionDeletesRewrittenAndExpired() throws Exception {
+    assumeThat(formatVersion)
+        .as("Rewriting position delete files in place only applies to v2")
+        .isEqualTo(2);
+    Table tableWithPosDeletes = createTableWithSnapshots(newTableLocation(), 2);
+    String dataFile =
+        SnapshotChanges.builderFor(tableWithPosDeletes)
+            .build()
+            .addedDataFiles()
+            .iterator()
+            .next()
+            .location();
+    List<Pair<CharSequence, Long>> deletes = Lists.newArrayList(Pair.of(dataFile, 0L));
+
+    DeleteFile oldDeletes =
+        FileHelpers.writeDeleteFile(
+                tableWithPosDeletes,
+                tableWithPosDeletes
+                    .io()
+                    .newOutputFile(tableWithPosDeletes.location() + "/data/old-deletes.parquet"),
+                deletes,
+                formatVersion)
+            .first();
+    tableWithPosDeletes.newRowDelta().addDeletes(oldDeletes).commit();
+    long oldDeletesSnapshotId = tableWithPosDeletes.currentSnapshot().snapshotId();
+
+    DeleteFile newDeletes =
+        FileHelpers.writeDeleteFile(
+                tableWithPosDeletes,
+                tableWithPosDeletes
+                    .io()
+                    .newOutputFile(tableWithPosDeletes.location() + "/data/new-deletes.parquet"),
+                deletes,
+                formatVersion)
+            .first();
+    tableWithPosDeletes.newRewrite().deleteFile(oldDeletes).addFile(newDeletes).commit();
+
+    // the manifest keeps oldDeletes as a DELETED entry, but expiration removes the file itself
+    actions().expireSnapshots(tableWithPosDeletes).expireSnapshotId(oldDeletesSnapshotId).execute();
+    assertThat(new File(removePrefix(oldDeletes.location()))).doesNotExist();
+
+    RewriteTablePath.Result result =
+        actions()
+            .rewriteTablePath(tableWithPosDeletes)
+            .stagingLocation(stagingLocation())
+            .rewriteLocationPrefix(tableWithPosDeletes.location(), targetTableLocation())
+            .execute();
+    assertThat(result.rewrittenDeleteFilePathsCount()).isEqualTo(1);
+
+    copyTableFiles(result);
+    assertThat(spark.read().format("iceberg").load(targetTableLocation()).collectAsList())
+        .hasSize(1);
+  }
+
   @TestTemplate
   public void testPositionDeletesAcrossFiles() throws Exception {
     assumeThat(formatVersion)