Spark: Use the table's catalog Hadoop configuration when listing orphan files

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: <ul><li>ERROR - throw an exception. (default) </li><li>IGNORE - no action.</li><li>DELETE - delete files.</li></ul> |  
 | `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 |

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<DeleteOrphanFi
   DeleteOrphanFilesSparkAction(SparkSession spark, Table table) {
     super(spark);
 
-    this.hadoopConf = new SerializableConfiguration(spark.sessionState().newHadoopConf());
+    this.hadoopConf = new SerializableConfiguration(hadoopConfForTable(spark, table));
     this.listingParallelism = spark.sessionState().conf().parallelPartitionDiscoveryParallelism();
     this.table = table;
     this.location = table.location();
@@ -146,6 +148,28 @@ public class DeleteOrphanFilesSparkAction extends BaseSparkAction<DeleteOrphanFi
         "Cannot delete orphan files: GC is disabled (deleting files may corrupt other tables)");
   }
 
+  /**
+   * Resolves the Hadoop configuration for listing the table location: the session configuration
+   * plus the {@code spark.sql.catalog.<name>.hadoop.*} overrides of the catalog that owns the
+   * table.
+   *
+   * <p>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;

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;
+    }
   }
 }