Skip to content

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

Open
sdaberdaku wants to merge 1 commit into
apache:mainfrom
sdaberdaku:spark-orphan-files-catalog-hadoop-conf
Open

Spark: Use the table's catalog Hadoop configuration when listing orphan files#17862
sdaberdaku wants to merge 1 commit into
apache:mainfrom
sdaberdaku:spark-orphan-files-catalog-hadoop-conf

Conversation

@sdaberdaku

@sdaberdaku sdaberdaku commented Aug 28, 2026

Copy link
Copy Markdown

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.

@github-actions github-actions Bot added the spark label Aug 28, 2026
@sdaberdaku
sdaberdaku force-pushed the spark-orphan-files-catalog-hadoop-conf branch from a29e31a to fd564d1 Compare August 28, 2026 09:20

@bujjibabukatta bujjibabukatta left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changes looks good.

*/
private static void applyCatalogAssumeRole(
SparkSession spark, String catalogName, Configuration conf) {
String roleArn =

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

}

conf.set(S3A_CREDENTIALS_PROVIDER, S3A_ASSUMED_ROLE_PROVIDER);
conf.set(S3A_ASSUMED_ROLE_ARN, roleArn);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

final String hadoopConfCatalogPrefix = hadoopConfPrefixForCatalog(catalogName);
final Configuration conf = spark.sessionState().newHadoopConf();

applyCatalogAssumeRole(spark, catalogName, conf);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed. Reverting the helper to its existing contract.

}

String sessionProvider = conf.get(S3A_CREDENTIALS_PROVIDER);
if (sessionProvider != null && !S3A_ASSUMED_ROLE_PROVIDER.equals(sessionProvider)) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

DeleteOrphanFiles.Result results =
SparkActions.get()
.deleteOrphanFiles(table.table())
.catalogName("overridecat")

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

* @param newCatalogName the name of the catalog that holds the table
* @return this for method chaining
*/
public DeleteOrphanFilesSparkAction catalogName(String newCatalogName) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@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.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Author

@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.

…an 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.<name>.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.
@sdaberdaku
sdaberdaku force-pushed the spark-orphan-files-catalog-hadoop-conf branch from fd564d1 to 1b12b36 Compare August 28, 2026 20:41
@github-actions github-actions Bot added the docs label Aug 28, 2026
@sdaberdaku sdaberdaku changed the title Spark: Reach catalog storage as the catalog's role when listing orphan files Spark: Use the table's catalog Hadoop configuration when listing orphan files Aug 28, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Spark: remove_orphan_files lists the table location with session credentials, not the catalog's

3 participants