yangshangqing95 commented on code in PR #17862:
URL: https://github.com/apache/iceberg/pull/17862#discussion_r3895748674
##########
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:
I'm still concerned about using Table.name() to recover the Spark catalog
here.
Table.name() gives us a table name, but it doesn't provide provenance that
the first segment identifies the Spark catalog that produced this particular
Table object. Standard Iceberg catalogs may currently produce names like
<catalog>.<namespace>.<table>, but that looks like an implementation convention
rather than a contract that SparkActions.deleteOrphanFiles(Table) can rely on
for arbitrary Table implementations.
For example, a table passed directly to SparkActions could happen to be
named tenant.db.orders while having been loaded from a different/custom
catalog. If a Spark catalog named tenant is registered in the session, this
code would silently apply spark.sql.catalog.tenant.hadoop.* to that unrelated
table, potentially making the Hadoop listing credentials diverge from
table.io() again.
Given that this is a delete action, I think explicit catalog context is
safer than inferring storage configuration from the table name. With the
earlier discussion, I'd prefer the typed catalogName(String) approach and have
the procedure pass tableCatalog().name() explicitly.
##########
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:
One additional concern with using isCatalogRegistered as validation here:
this isn't necessarily a passive membership check. Spark's
CatalogManager.isCatalogRegistered resolves the catalog by name, which may load
and cache the catalog if it hasn't been initialized yet.
That means a false positive first segment from table.name() can have side
effects beyond selecting the wrong Hadoop configuration: constructing this
action may initialize an unrelated Spark catalog, and could even fail because
that unrelated catalog is misconfigured.
This seems like another reason not to infer catalog provenance from the
table name. If the caller already knows the catalog, passing that context
explicitly avoids both the ambiguity and the catalog loading side effect.
##########
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:
Nit: could we also unset the spark.sql.catalog.overridecat.* settings added
by the other new test?
##########
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:
These tests cover the positive cases where the table name happens to encode
the expected Spark catalog, but I think the risky case for this implementation
is the negative one.
Could we add coverage for a Table whose name starts with an existing Spark
catalog name but which was not loaded from that catalog? The current
implementation would still select that catalog's Hadoop configuration, which
would demonstrate the provenance ambiguity discussed above.
I'd also still like to see one regression through CALL
<catalog>.system.remove_orphan_files(...), since that's the user-facing path
reported in #17860.
--
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]
