From 1b12b367de9ab6ea2d25363718a5e6fa4b69c77d Mon Sep 17 00:00:00 2001 From: Sebastian Daberdaku Date: Fri, 28 Aug 2026 10:57:09 +0200 Subject: [PATCH] Spark: Use the table's catalog Hadoop configuration when listing orphan files RemoveOrphanFiles lists the table location through the Hadoop FileSystem API, which is configured from the Spark session, while the rest of the procedure reaches storage through the catalog's FileIO. A catalog that reaches its storage with credentials other than the cluster's therefore has its listing performed as the cluster identity, which fails when that identity has no access to the catalog's storage. spark.sql.catalog..hadoop.* already exists to give a catalog its own Hadoop configuration, but no action used it, so on this path the overrides were accepted and ignored. Build the action's Hadoop configuration with SparkUtil.hadoopConfCatalogOverrides for the catalog that owns the table. Catalog tables are named catalog.namespace.table with the Spark catalog name first, so the action resolves it from the table name and applies the overrides only when that part is a registered Spark catalog. Path-based tables and other names keep the session configuration. --- docs/docs/spark-procedures.md | 2 + .../actions/DeleteOrphanFilesSparkAction.java | 26 ++++- .../actions/TestRemoveOrphanFilesAction3.java | 110 ++++++++++++++++++ 3 files changed, 137 insertions(+), 1 deletion(-) diff --git a/docs/docs/spark-procedures.md b/docs/docs/spark-procedures.md index ced96eb9cd6a..0a96b8c19183 100644 --- a/docs/docs/spark-procedures.md +++ b/docs/docs/spark-procedures.md @@ -320,6 +320,8 @@ Used to remove files which are not referenced in any metadata files of an Iceber | `prefix_mismatch_mode` | | string | Action behavior when location prefixes (schemes/authorities) mismatch: | | `prefix_listing` | | boolean | When true, use prefix-based file listing via the `SupportsPrefixOperations` interface. The Table FileIO implementation must support `SupportsPrefixOperations` when this flag is enabled (defaults to false) | +Unless `prefix_listing` is enabled, files are listed through the Hadoop `FileSystem` API, configured from the Spark session and the [catalog specific Hadoop configuration](spark-configuration.md#using-catalog-specific-hadoop-configuration-values) of the catalog that owns the table. A catalog that reaches its storage with credentials other than the cluster's can be given the same credentials for the listing with `spark.sql.catalog.(catalog-name).hadoop.fs.s3a.*`. + #### Output | Output Name | Type | Description | diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java index b47922820d21..dc7240d8fb2c 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java @@ -32,6 +32,7 @@ import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import java.util.function.Predicate; +import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; @@ -51,6 +52,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.spark.JobGroupInfo; +import org.apache.iceberg.spark.SparkUtil; import org.apache.iceberg.util.FileSystemWalker; import org.apache.iceberg.util.Pair; import org.apache.iceberg.util.PropertyUtil; @@ -136,7 +138,7 @@ public class DeleteOrphanFilesSparkAction extends BaseSparkAction.hadoop.*} overrides of the catalog that owns the + * table. + * + *

Catalog tables are named {@code catalog.namespace.table}, where the catalog part is the + * Spark catalog name. The overrides are applied only when that part names a registered Spark + * catalog; path-based tables and unknown names fall back to the session configuration. + */ + private static Configuration hadoopConfForTable(SparkSession spark, Table table) { + String name = table.name(); + int dot = name.indexOf('.'); + if (dot > 0 && !name.contains("/") && !name.contains(":")) { + String catalogName = name.substring(0, dot); + if (spark.sessionState().catalogManager().isCatalogRegistered(catalogName)) { + return SparkUtil.hadoopConfCatalogOverrides(spark, catalogName); + } + } + + return spark.sessionState().newHadoopConf(); + } + @Override protected DeleteOrphanFilesSparkAction self() { return this; diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java index 88ac800b158f..3483c4057f4b 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java @@ -21,8 +21,11 @@ import static org.assertj.core.api.Assertions.assertThat; import java.io.File; +import java.net.URI; import java.util.stream.StreamSupport; +import org.apache.hadoop.fs.RawLocalFileSystem; import org.apache.iceberg.actions.DeleteOrphanFiles; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.spark.SparkCatalog; import org.apache.iceberg.spark.SparkSchemaUtil; import org.apache.iceberg.spark.SparkSessionCatalog; @@ -190,10 +193,117 @@ public void testSparkSessionCatalogHiveTable() throws Exception { assertThat(results.orphanFilesCount()).as("trash file should be removed").isEqualTo(1L); } + @TestTemplate + public void catalogHadoopConfOverridesApplyToListing() throws Exception { + spark.conf().set("spark.sql.catalog.overridecat", "org.apache.iceberg.spark.SparkCatalog"); + spark.conf().set("spark.sql.catalog.overridecat.type", "hadoop"); + spark.conf().set("spark.sql.catalog.overridecat.warehouse", tableLocation); + // registered for this catalog alone, so the location below resolves only when the catalog's + // Hadoop overrides reach the listing + spark + .conf() + .set( + String.format( + "spark.sql.catalog.overridecat.hadoop.fs.%s.impl", CatalogScopedFileSystem.SCHEME), + CatalogScopedFileSystem.class.getName()); + SparkCatalog cat = (SparkCatalog) spark.sessionState().catalogManager().catalog("overridecat"); + + String[] database = {"default"}; + Identifier id = Identifier.of(database, randomName("table")); + Transform[] transforms = {}; + cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms, properties); + SparkTable table = (SparkTable) cat.loadTable(id); + + sql("INSERT INTO overridecat.default.%s VALUES (1,1,1)", id.name()); + + String location = table.table().location().replaceFirst("file:", ""); + String trashFile = randomName("/data/trashfile"); + new File(location + trashFile).createNewFile(); + + DeleteOrphanFiles.Result results = + SparkActions.get() + .deleteOrphanFiles(table.table()) + .location(CatalogScopedFileSystem.SCHEME + "://" + location) + .equalSchemes(ImmutableMap.of(CatalogScopedFileSystem.SCHEME, "file")) + .deleteWith(file -> {}) + .olderThan(System.currentTimeMillis() + 1000) + .execute(); + + assertThat(StreamSupport.stream(results.orphanFileLocations().spliterator(), false)) + .as("trash file should be found") + .anyMatch(file -> file.endsWith(trashFile)); + assertThat(results.orphanFilesCount()).as("only the trash file is an orphan").isEqualTo(1L); + } + + @TestTemplate + public void sessionCatalogHadoopConfOverridesApplyToListing() throws Exception { + spark + .conf() + .set("spark.sql.catalog.spark_catalog", "org.apache.iceberg.spark.SparkSessionCatalog"); + spark.conf().set("spark.sql.catalog.spark_catalog.type", "hadoop"); + spark.conf().set("spark.sql.catalog.spark_catalog.warehouse", tableLocation); + spark + .conf() + .set( + String.format( + "spark.sql.catalog.spark_catalog.hadoop.fs.%s.impl", + CatalogScopedFileSystem.SCHEME), + CatalogScopedFileSystem.class.getName()); + SparkSessionCatalog cat = + (SparkSessionCatalog) spark.sessionState().catalogManager().v2SessionCatalog(); + + String[] database = {"default"}; + Identifier id = Identifier.of(database, randomName("table")); + Transform[] transforms = {}; + cat.createTable(id, SparkSchemaUtil.convert(SCHEMA), transforms, properties); + SparkTable table = (SparkTable) cat.loadTable(id); + + sql("INSERT INTO default.%s VALUES (1,1,1)", id.name()); + + String location = table.table().location().replaceFirst("file:", ""); + String trashFile = randomName("/data/trashfile"); + new File(location + trashFile).createNewFile(); + + DeleteOrphanFiles.Result results = + SparkActions.get() + .deleteOrphanFiles(table.table()) + .location(CatalogScopedFileSystem.SCHEME + "://" + location) + .equalSchemes(ImmutableMap.of(CatalogScopedFileSystem.SCHEME, "file")) + .deleteWith(file -> {}) + .olderThan(System.currentTimeMillis() + 1000) + .execute(); + + assertThat(StreamSupport.stream(results.orphanFileLocations().spliterator(), false)) + .as("trash file should be found") + .anyMatch(file -> file.endsWith(trashFile)); + assertThat(results.orphanFilesCount()).as("only the trash file is an orphan").isEqualTo(1L); + } + @AfterEach public void resetSparkSessionCatalog() { spark.conf().unset("spark.sql.catalog.spark_catalog"); spark.conf().unset("spark.sql.catalog.spark_catalog.type"); spark.conf().unset("spark.sql.catalog.spark_catalog.warehouse"); + spark + .conf() + .unset( + String.format( + "spark.sql.catalog.spark_catalog.hadoop.fs.%s.impl", + CatalogScopedFileSystem.SCHEME)); + } + + /** Local file system reachable only under its own scheme, to tell the two configs apart. */ + public static class CatalogScopedFileSystem extends RawLocalFileSystem { + static final String SCHEME = "catalogscopedfs"; + + @Override + public URI getUri() { + return URI.create(SCHEME + ":///"); + } + + @Override + public String getScheme() { + return SCHEME; + } } }