sdaberdaku commented on code in PR #17862:
URL: https://github.com/apache/iceberg/pull/17862#discussion_r3948034338
##########
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 fair enough — the provenance argument convinced me.
`Table.name()` is display output, and a delete action shouldn't pick
credentials based on what a string happens to look like. I've pushed the
explicit version: `catalogName(String)` on the action, and
`RemoveOrphanFilesProcedure` passes `tableCatalog().name()`. Without the call
the action keeps today's behaviour (session configuration only), so nothing
changes for existing `SparkActions` users unless they opt in.
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -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();
Review Comment:
Agreed, and this is gone now — the pushed version takes the catalog name
explicitly instead of parsing `table.name()`.
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -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)) {
Review Comment:
Good catch — I had assumed it was a passive lookup, but it goes through
`catalog(name)` and can initialize and cache the catalog. One more reason the
inference was the wrong idea. No catalog resolution happens in the action
anymore; it only reads the `spark.sql.catalog.<name>.hadoop.*` conf keys for
the name it's given.
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java:
##########
@@ -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 {
Review Comment:
Added the `CALL <catalog>.system.remove_orphan_files(...)` regression to
`TestRemoveOrphanFilesProcedure`; it registers the marker filesystem only in
the catalog's `hadoop.*` overrides and runs across all four test catalogs
(testhive, testhadoop, spark_catalog, testrest), so it covers the exact path
from #17860 including the session catalog. I didn't add the collision test —
with the inference removed there's no name-based selection left to demonstrate;
the action tests now pass the catalog explicitly.
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction3.java:
##########
@@ -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
Review Comment:
Done — the `@AfterEach` now unsets the `overridecat` keys as well, and the
new procedure test cleans up its override key too.
--
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]
