yangshangqing95 commented on code in PR #17862:
URL: https://github.com/apache/iceberg/pull/17862#discussion_r3883225671
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java:
##########
@@ -170,6 +182,35 @@ public static Configuration
hadoopConfCatalogOverrides(SparkSession spark, Strin
return conf;
}
+ /**
+ * Points the S3A file system at the role the catalog's AWS clients assume,
so that both reach the
+ * catalog's storage as the same principal.
+ *
+ * <p>The identity resolved for the session stays in place as the one that
authenticates the
+ * AssumeRole call itself. Catalogs that declare no role, or that set S3A
properties of their own
+ * under {@code spark.sql.catalog.$catalogName.hadoop.*}, are left alone.
+ */
+ private static void applyCatalogAssumeRole(
+ SparkSession spark, String catalogName, Configuration conf) {
+ String roleArn =
Review Comment:
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?
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java:
##########
@@ -155,6 +165,8 @@ public static Configuration
hadoopConfCatalogOverrides(SparkSession spark, Strin
final String hadoopConfCatalogPrefix =
hadoopConfPrefixForCatalog(catalogName);
final Configuration conf = spark.sessionState().newHadoopConf();
+ applyCatalogAssumeRole(spark, catalogName, conf);
Review Comment:
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.<name>.hadoop.* overrides, and keep any additional action
specific behavior separate. That would also make the 3.5/4.0 backports much
narrower.
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java:
##########
@@ -170,6 +182,35 @@ public static Configuration
hadoopConfCatalogOverrides(SparkSession spark, Strin
return conf;
}
+ /**
+ * Points the S3A file system at the role the catalog's AWS clients assume,
so that both reach the
+ * catalog's storage as the same principal.
+ *
+ * <p>The identity resolved for the session stays in place as the one that
authenticates the
+ * AssumeRole call itself. Catalogs that declare no role, or that set S3A
properties of their own
+ * under {@code spark.sql.catalog.$catalogName.hadoop.*}, are left alone.
+ */
+ private static void applyCatalogAssumeRole(
+ SparkSession spark, String catalogName, Configuration conf) {
+ String roleArn =
+ spark
+ .sessionState()
+ .conf()
+ .settings()
+ .get(DOT.join(SPARK_CATALOG_CONF_PREFIX, catalogName,
CLIENT_ASSUME_ROLE_ARN));
+ if (roleArn == null) {
+ return;
+ }
+
+ String sessionProvider = conf.get(S3A_CREDENTIALS_PROVIDER);
+ if (sessionProvider != null &&
!S3A_ASSUMED_ROLE_PROVIDER.equals(sessionProvider)) {
Review Comment:
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?
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java:
##########
@@ -170,6 +182,35 @@ public static Configuration
hadoopConfCatalogOverrides(SparkSession spark, Strin
return conf;
}
+ /**
+ * Points the S3A file system at the role the catalog's AWS clients assume,
so that both reach the
+ * catalog's storage as the same principal.
+ *
+ * <p>The identity resolved for the session stays in place as the one that
authenticates the
+ * AssumeRole call itself. Catalogs that declare no role, or that set S3A
properties of their own
+ * under {@code spark.sql.catalog.$catalogName.hadoop.*}, are left alone.
+ */
+ private static void applyCatalogAssumeRole(
+ SparkSession spark, String catalogName, Configuration conf) {
+ String roleArn =
+ spark
+ .sessionState()
+ .conf()
+ .settings()
+ .get(DOT.join(SPARK_CATALOG_CONF_PREFIX, catalogName,
CLIENT_ASSUME_ROLE_ARN));
+ if (roleArn == null) {
+ return;
+ }
+
+ String sessionProvider = conf.get(S3A_CREDENTIALS_PROVIDER);
+ if (sessionProvider != null &&
!S3A_ASSUMED_ROLE_PROVIDER.equals(sessionProvider)) {
+ conf.set(S3A_ASSUMED_ROLE_CREDENTIALS_PROVIDER, sessionProvider);
+ }
+
+ conf.set(S3A_CREDENTIALS_PROVIDER, S3A_ASSUMED_ROLE_PROVIDER);
+ conf.set(S3A_ASSUMED_ROLE_ARN, roleArn);
Review Comment:
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.
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java:
##########
@@ -190,10 +193,68 @@ 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())
+ .catalogName("overridecat")
Review Comment:
This test manually supplies .catalogName("overridecat"), so reverting the
production change in RemoveOrphanFilesProcedure would still leave this test
green.
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -222,6 +223,24 @@ public DeleteOrphanFilesSparkAction
usePrefixListing(boolean newUsePrefixListing
return this;
}
+ /**
+ * Configures the listing to use the Hadoop configuration of the given
catalog.
+ *
+ * <p>Listing the table location goes through the Hadoop {@link
org.apache.hadoop.fs.FileSystem}
+ * API, which is configured from the session and so reaches storage as a
different principal than
+ * the catalog does. Setting the catalog makes the listing follow the
catalog's own configuration
+ * instead.
+ *
+ * @param newCatalogName the name of the catalog that holds the table
+ * @return this for method chaining
+ */
+ public DeleteOrphanFilesSparkAction catalogName(String newCatalogName) {
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
