sdaberdaku commented on code in PR #17862:
URL: https://github.com/apache/iceberg/pull/17862#discussion_r3883785805
##########
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:
Agreed, the property is the wrong signal. Removing the derivation entirely.
##########
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:
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.
##########
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:
Agreed. Reverting the helper to its existing contract.
##########
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:
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.
##########
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:
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`.
##########
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:
@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.
--
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]
