Spark 4.1: Add rewrite option to enable executor cache for delete files

docs/docs/spark-procedures.md

@@ -415,6 +415,7 @@ Iceberg can compact data files in parallel using Spark with the `rewriteDataFile
 | `output-spec-id` | current partition spec id | Identifier of the output partition spec. Data will be reorganized during the rewrite to align with the output partitioning. |
 | `remove-dangling-deletes` | false | Remove dangling position and equality deletes after rewriting. A delete file is considered dangling if it does not apply to any live data files. Enabling this will generate an additional commit for the removal. |
 | `max-files-to-rewrite` | null | This option sets an upper limit on the number of eligible files that will be rewritten. If this option is not specified, all eligible files will be rewritten. |
+| `cache-delete-files` | false | Use the executor cache for delete files while rewriting. Enable this when the same delete file applies to many data files, which is common with equality deletes |
 
 !!! info
     Dangling delete files are removed based solely on data sequence numbers. This action does not apply to global

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java

@@ -70,6 +70,22 @@ public class RewriteDataFilesSparkAction
     extends BaseSnapshotUpdateSparkAction<RewriteDataFilesSparkAction> implements RewriteDataFiles {
 
   private static final Logger LOG = LoggerFactory.getLogger(RewriteDataFilesSparkAction.class);
+
+  /**
+   * Use the executor cache for delete files while rewriting.
+   *
+   * <p>Enable this when the same delete file applies to many data files, which is common with
+   * equality deletes.
+   *
+   * <p>This option sets {@link SparkSQLProperties#EXECUTOR_CACHE_DELETE_FILES_ENABLED} for the
+   * rewrite, so any value configured for that property in the session is ignored.
+   *
+   * <p>Defaults to false.
+   */
+  public static final String CACHE_DELETE_FILES = "cache-delete-files";
+
+  public static final boolean CACHE_DELETE_FILES_DEFAULT = false;
+
   private static final Set<String> VALID_OPTIONS =
       ImmutableSet.of(
           MAX_CONCURRENT_FILE_GROUP_REWRITES,
@@ -82,6 +98,7 @@ public class RewriteDataFilesSparkAction
           REWRITE_JOB_ORDER,
           OUTPUT_SPEC_ID,
           REMOVE_DANGLING_DELETES,
+          CACHE_DELETE_FILES,
           BinPackRewriteFilePlanner.MAX_FILES_TO_REWRITE);
 
   private static final RewriteDataFilesSparkAction.Result EMPTY_RESULT =
@@ -105,10 +122,6 @@ public class RewriteDataFilesSparkAction
     super(((org.apache.spark.sql.classic.SparkSession) spark).cloneSession());
     // Disable Adaptive Query Execution as this may change the output partitioning of our write
     spark().conf().set(SQLConf.ADAPTIVE_EXECUTION_ENABLED().key(), false);
-    // Disable executor cache for delete files as each partition is rewritten separately.
-    // Note: when compacting to a different target spec, data from multiple partitions
-    // may be grouped together, but caching is still disabled to avoid connection pool issues.
-    spark().conf().set(SparkSQLProperties.EXECUTOR_CACHE_DELETE_FILES_ENABLED, "false");
     this.caseSensitive = SparkUtil.caseSensitive(spark);
     this.table = table;
   }
@@ -424,6 +437,14 @@ void validateAndInitOptions() {
         PropertyUtil.propertyAsBoolean(
             options(), REMOVE_DANGLING_DELETES, REMOVE_DANGLING_DELETES_DEFAULT);
 
+    boolean cacheDeleteFiles =
+        PropertyUtil.propertyAsBoolean(options(), CACHE_DELETE_FILES, CACHE_DELETE_FILES_DEFAULT);
+    spark()
+        .conf()
+        .set(
+            SparkSQLProperties.EXECUTOR_CACHE_DELETE_FILES_ENABLED,
+            String.valueOf(cacheDeleteFiles));
+
     Preconditions.checkArgument(
         maxConcurrentFileGroupRewrites >= 1,
         "Cannot set %s to %s, the value must be positive.",

spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkExecutorCache.java

@@ -49,6 +49,8 @@
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.actions.RewriteDataFiles;
+import org.apache.iceberg.actions.SizeBasedFileRewritePlanner;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.data.FileHelpers;
@@ -68,6 +70,8 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.spark.SparkExecutorCache.CacheValue;
 import org.apache.iceberg.spark.SparkExecutorCache.Conf;
+import org.apache.iceberg.spark.actions.RewriteDataFilesSparkAction;
+import org.apache.iceberg.spark.actions.SparkActions;
 import org.apache.iceberg.util.CharSequenceSet;
 import org.apache.iceberg.util.Pair;
 import org.apache.spark.SparkEnv;
@@ -214,6 +218,38 @@ public void testDeleteFilesCacheDisabledConfig() throws Exception {
         });
   }
 
+  @TestTemplate
+  void cacheDeleteFilesDisabledByDefault() throws Exception {
+    List<DeleteFile> deleteFiles = createAndInitTable(TableProperties.DELETE_MODE, MERGE_ON_READ);
+
+    RewriteDataFiles.Result result =
+        SparkActions.get(spark)
+            .rewriteDataFiles(Spark3Util.loadIcebergTable(spark, targetTableName))
+            .option(SizeBasedFileRewritePlanner.REWRITE_ALL, "true")
+            .execute();
+
+    assertThat(result.rewrittenDataFilesCount()).isEqualTo(2);
+
+    // both delete files apply to both data files and the cache is off, so each is opened per file
+    assertThat(deleteFiles).allMatch(deleteFile -> streamCount(deleteFile) == 2);
+  }
+
+  @TestTemplate
+  void cacheDeleteFilesEnabledByOption() throws Exception {
+    List<DeleteFile> deleteFiles = createAndInitTable(TableProperties.DELETE_MODE, MERGE_ON_READ);
+
+    RewriteDataFiles.Result result =
+        SparkActions.get(spark)
+            .rewriteDataFiles(Spark3Util.loadIcebergTable(spark, targetTableName))
+            .option(SizeBasedFileRewritePlanner.REWRITE_ALL, "true")
+            .option(RewriteDataFilesSparkAction.CACHE_DELETE_FILES, "true")
+            .execute();
+
+    assertThat(result.rewrittenDataFilesCount()).isEqualTo(2);
+
+    assertThat(deleteFiles).allMatch(deleteFile -> streamCount(deleteFile) == 1);
+  }
+
   @TestTemplate
   public void testConcurrentAccess() throws InterruptedException {
     SparkExecutorCache cache = SparkExecutorCache.getOrCreate();

spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java

@@ -111,7 +111,6 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Streams;
 import org.apache.iceberg.spark.FileRewriteCoordinator;
 import org.apache.iceberg.spark.ScanTaskSetManager;
-import org.apache.iceberg.spark.SparkReadConf;
 import org.apache.iceberg.spark.SparkReadOptions;
 import org.apache.iceberg.spark.SparkTableUtil;
 import org.apache.iceberg.spark.SparkWriteOptions;
@@ -2242,18 +2241,6 @@ public void testRewriteDataFilesPreservesLineage() throws NoSuchTableException {
     assertEquals("Rows must match", expectedRecords, actualRecordsWithLineage);
   }
 
-  @TestTemplate
-  public void testExecutorCacheForDeleteFilesDisabled() {
-    Table table = createTablePartitioned(1, 1);
-    RewriteDataFilesSparkAction action = SparkActions.get(spark).rewriteDataFiles(table);
-
-    // The constructor should have set the configuration to false
-    SparkReadConf readConf = new SparkReadConf(action.spark(), table);
-    assertThat(readConf.cacheDeleteFilesOnExecutors())
-        .as("Executor cache for delete files should be disabled in RewriteDataFilesSparkAction")
-        .isFalse();
-  }
-
   @TestTemplate
   public void testZOrderUDFWithDateType() {
     SparkZOrderUDF zorderUDF = new SparkZOrderUDF(1, 16, 1024);