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

[sdaberdaku]

Closes #17860.

RemoveOrphanFiles lists the table location through the Hadoop FileSystem API with a configuration built from the Spark session:

// DeleteOrphanFilesSparkAction.java
this.hadoopConf = new SerializableConfiguration(spark.sessionState().newHadoopConf());

Everything else in the procedure reaches storage through table.io(), which the catalog built. When a catalog reaches its storage with different credentials than the cluster (for example AssumeRoleAwsClientFactory on a multi-account setup), expire_snapshots succeeds and remove_orphan_files gets a 403 on the same table, because the listing runs as the cluster identity. #17860 has the full report.

spark.sql.catalog.<name>.hadoop.* already exists to give a catalog its own Hadoop configuration, and SparkCatalog applies it when building the Iceberg catalog, but no action or procedure used it, so on this path the overrides were accepted and ignored.

This change makes DeleteOrphanFilesSparkAction build its Hadoop configuration with SparkUtil.hadoopConfCatalogOverrides for the catalog that owns the table. Catalog tables are named <catalog>.<namespace>.<table> by CatalogUtil.fullTableName, and the catalog part is the Spark catalog name (spark_catalog for SparkSessionCatalog), so the action resolves it from table.name() and applies the overrides only when that part is a registered Spark catalog. Path-based tables and any other name fall back to the session configuration, which is the current behaviour. No new public API, and RemoveOrphanFilesProcedure is unchanged; direct users of SparkActions.deleteOrphanFiles(table) get the same behaviour.

With this, the setup in #17860 is fixed per catalog, without the per-bucket session-wide workaround:

spark.sql.catalog.tenant.hadoop.fs.s3a.aws.credentials.provider           org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider
spark.sql.catalog.tenant.hadoop.fs.s3a.assumed.role.arn                   arn:aws:iam::111122223333:role/lakehouse-access
spark.sql.catalog.tenant.hadoop.fs.s3a.assumed.role.credentials.provider  com.amazonaws.auth.WebIdentityTokenCredentialsProvider

The remove_orphan_files docs now say which configuration the listing uses.

Tests

TestRemoveOrphanFilesAction3 gets two cases, one for SparkCatalog and one for SparkSessionCatalog, that register a FileSystem for a scheme under the catalog's hadoop.* only and list a table location on that scheme, so they pass only when the catalog's configuration reaches the walk. Both fail on main with No FileSystem for scheme "catalogscopedfs".

Verified against spark/v4.1. If the direction is agreed I will follow up with the v4.0 and v3.5 backports.

[bujjibabukatta]

Changes looks good.

[yangshangqing95]

I don't think the presence of client.assume-role.arn alone is enough to conclude that the S3/FileIO path is using assume-role credentials.

For S3FileIO, that property has AssumeRole semantics when it is consumed by an appropriate client factory, such as AssumeRoleAwsClientFactory. The actual client used by S3FileIO is selected via s3.client-factory-impl, falling back to client.factory, and the default AwsClientFactory does not switch to assume-role credentials merely because client.assume-role.arn is present.

As a result, a catalog using the default or a custom client factory could have this property present without using it to determine the FileIO credentials, while this code would still force the Hadoop/S3A path to assume that role.

Could we either base this derivation on the client factory actually selected for the FileIO path, or avoid inferring S3A authentication from the catalog AWS properties here?

[sdaberdaku]

Agreed, the property is the wrong signal. Removing the derivation entirely.

[yangshangqing95]

Even when the catalog is using AssumeRoleAwsClientFactory, translating only the role ARN doesn't reproduce the credentials used by table.io().

Iceberg's AssumeRole configuration can also include an external ID, session name, session duration, tags, and region. For example, a perfectly valid catalog that requires client.assume-role.external-id would still have table.io() succeed while this Hadoop path assumes the same ARN without the external ID and fails with AccessDenied.

That means the comment above that both paths reach storage as the same principal isn't generally true yet. I think we should either define and cover the complete mapping between the two AssumeRole configurations, or keep this fix limited to propagating the catalog's explicit hadoop.* overrides.

[sdaberdaku]

You're right, external ID is a clean counter-example. Dropping it. Worth noting for any follow-up that session tags have no S3A equivalent at all.

[yangshangqing95]

Could we avoid changing the semantics of hadoopConfCatalogOverrides itself here?

This helper is already used outside remove_orphan_files, including while constructing SparkCatalog / HadoopTables. Adding AWS AssumeRole derivation here means every existing caller starts getting a different Hadoop authentication configuration, so the behavioral scope is larger than the issue being fixed.

It seems safer for hadoopConfCatalogOverrides to keep its existing contract: session Hadoop configuration plus explicit spark.sql.catalog..hadoop.* overrides, and keep any additional action specific behavior separate. That would also make the 3.5/4.0 backports much narrower.

[sdaberdaku]

Agreed. Reverting the helper to its existing contract.

[yangshangqing95]

I don't think an exact string comparison is sufficient here. fs.s3a.aws.credentials.provider may contain a provider chain, so a value such as:

AssumedRoleCredentialProvider,SomeFallbackProvider

would pass this condition and then be copied into fs.s3a.assumed.role.credentials.provider. S3A explicitly rejects an inner credentials-provider chain that contains AssumedRoleCredentialProvider.

Also, conf.get(...) preserves the configured provider expression rather than necessarily the identity that has already been "resolved" for the session.

Could we handle provider lists explicitly, or avoid synthesizing the inner provider chain here?

[sdaberdaku]

Good catch. Copying that setting was never a faithful mirror anyway, since Iceberg's own STS client uses the SDK default chain. Moot now that the derivation is gone.

[yangshangqing95]

This test manually supplies .catalogName("overridecat"), so reverting the production change in RemoveOrphanFilesProcedure would still leave this test green.

[sdaberdaku]

Fair. I'll add a procedure-level test in TestRemoveOrphanFilesProcedure that sets the catalog hadoop.* override in the session and goes through CALL remove_orphan_files.

[yangshangqing95]

This looks like Spark catalog context needed internally by the procedure rather than an action option. Exposing it also makes it possible to construct the action with a table from catalog A while using catalog B's Hadoop configuration for the filesystem walk, which is a particularly surprising combination for a delete action.

If possible, I'd prefer to keep this plumbing internal rather than expose a new public configuration knob.

[sdaberdaku]

@yangshangqing95 the procedure and the action are in different packages, so anything the procedure calls has to be public. My preference is to keep catalogName(String) with Javadoc saying it selects the Hadoop configuration for the listing; the action already accepts an arbitrary location, so listing outside the table's catalog isn't new. The alternative is passing it through the existing generic option(...) map, which avoids a new method but is just a stringly-typed version of the same thing. Happy to go either way, let me know which you'd prefer.

[sdaberdaku]

Thinking about this more, there's a way to avoid the knob entirely. Every catalog-loaded table is named <catalog>.<namespace>.<table> by CatalogUtil.fullTableName, and the catalog part is the Spark catalog name that SparkCatalog.buildIcebergCatalog passed in (spark_catalog for SparkSessionCatalog). So the action can resolve the catalog itself: take the first segment of table.name(), check it's a registered catalog via CatalogManager.isCatalogRegistered, and if so build the Hadoop configuration with SparkUtil.hadoopConfCatalogOverrides. Anything else, including path-based tables, falls back to the session configuration, which is today's behaviour.

That drops catalogName(...) and the procedure change altogether; RemoveOrphanFilesProcedure stays as it is on main, and direct users of SparkActions.deleteOrphanFiles(table) get the same fix. It's the same first-segment convention Spark3Util.catalogAndIdentifier relies on, but without the current-catalog fallback, since applying another catalog's overrides to a foreign table would be wrong.

The only gap I see is a custom SparkCatalog subclass that gives its Iceberg catalog a different name; that just falls back to session behaviour. I'll push this shape unless you'd rather not rely on the table name.

[sdaberdaku]

@yangshangqing95 thanks for the review. I agree the derivation was guessing at what table.io() does rather than mirroring it, so I'll narrow this to the second option you suggested:

• Drop applyCatalogAssumeRole. hadoopConfCatalogOverrides keeps its current contract.
• The action builds its Hadoop configuration from that helper instead of the session, and the procedure passes the catalog it was invoked on.
• Add a procedure-level test in TestRemoveOrphanFilesProcedure that fails if the wiring is reverted.
• Document that the listing honours spark.sql.catalog.<name>.hadoop.fs.s3a.*, so a catalog that assumes a role can get the same role on the S3A side, per catalog.

That limits the change to "the orphan file walk sees the catalog's own Hadoop overrides", which is what the issue is really about, and keeps the backports small. If there's interest in deriving the role later, I think it should read table.io().properties(), gate on the client factory, and map the full property set; I can open a follow-up issue for that.

One open question on the catalogName(...) thread before I push.