You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ru...@apache.org on 2026/09/01 15:47:23 UTC
(iceberg) branch main updated: Spark 3.5: Add rewrite option to enable executor cache for delete files (#17909)
This is an automated email from the ASF dual-hosted git repository.
RussellSpitzer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 6a179f3128 Spark 3.5: Add rewrite option to enable executor cache for delete files (#17909)
6a179f3128 is described below
commit 6a179f3128b162e7b12de748d9f79628924a4249
Author: Anurag Mantripragada <am...@apple.com>
AuthorDate: Tue Sep 1 08:47:14 2026 -0700
Spark 3.5: Add rewrite option to enable executor cache for delete files (#17909)
---
.../spark/actions/RewriteDataFilesSparkAction.java | 29 ++++++++++++++---
.../iceberg/spark/TestSparkExecutorCache.java | 36 ++++++++++++++++++++++
.../spark/actions/TestRewriteDataFilesAction.java | 21 ++++++++++---
3 files changed, 76 insertions(+), 10 deletions(-)
diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
index 2f5cc25074..2bc51af06c 100644
--- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
+++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
@@ -70,6 +70,19 @@ public class RewriteDataFilesSparkAction
extends BaseSnapshotUpdateSparkAction<RewriteDataFilesSparkAction> implements RewriteDataFiles {
private static final Logger LOG = LoggerFactory.getLogger(RewriteDataFilesSparkAction.class);
+
+ /**
+ * When true, the executor cache is used for delete files while rewriting.
+ *
+ * <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 +95,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 +119,6 @@ public class RewriteDataFilesSparkAction
super(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;
}
@@ -204,7 +214,8 @@ public class RewriteDataFilesSparkAction
return result;
}
- private void init(long startingSnapshotId) {
+ @VisibleForTesting
+ void init(long startingSnapshotId) {
this.planner =
runner instanceof SparkShufflingFileRewriteRunner
? new SparkShufflingDataRewritePlanner(table, filter, startingSnapshotId, caseSensitive)
@@ -417,6 +428,14 @@ public class RewriteDataFilesSparkAction
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.",
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkExecutorCache.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkExecutorCache.java
index a411c3fc70..6c8295d638 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkExecutorCache.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkExecutorCache.java
@@ -49,6 +49,8 @@ import org.apache.iceberg.RowLevelOperationMode;
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.Maps;
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 class TestSparkExecutorCache extends TestBaseWithCatalog {
});
}
+ @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();
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
index 234c06550e..3c0b58d305 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
@@ -2646,15 +2646,26 @@ public class TestRewriteDataFilesAction extends TestBase {
}
@TestTemplate
- public void testExecutorCacheForDeleteFilesDisabled() {
+ void cacheDeleteFilesOnExecutorsDisabledByDefault() {
Table table = createTablePartitioned(1, 1);
RewriteDataFilesSparkAction action = SparkActions.get(spark).rewriteDataFiles(table);
+ action.init(0L);
- // The constructor should have set the configuration to false
SparkReadConf readConf = new SparkReadConf(action.spark(), table, Collections.emptyMap());
- assertThat(readConf.cacheDeleteFilesOnExecutors())
- .as("Executor cache for delete files should be disabled in RewriteDataFilesSparkAction")
- .isFalse();
+ assertThat(readConf.cacheDeleteFilesOnExecutors()).isFalse();
+ }
+
+ @TestTemplate
+ void cacheDeleteFilesOnExecutorsEnabledByOption() {
+ Table table = createTablePartitioned(1, 1);
+ RewriteDataFilesSparkAction action =
+ SparkActions.get(spark)
+ .rewriteDataFiles(table)
+ .option(RewriteDataFilesSparkAction.CACHE_DELETE_FILES, "true");
+ action.init(0L);
+
+ SparkReadConf readConf = new SparkReadConf(action.spark(), table, Collections.emptyMap());
+ assertThat(readConf.cacheDeleteFilesOnExecutors()).isTrue();
}
@TestTemplate