yuqi1129 commented on code in PR #14466:
URL: https://github.com/apache/gravitino/pull/14466#discussion_r4440715550
##########
iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java:
##########
@@ -244,7 +247,30 @@ public void dropTable(TableIdentifier tableIdentifier) {
public void purgeTable(TableIdentifier tableIdentifier) {
getMetadataCache().invalidate(tableIdentifier);
- CatalogHandlers.purgeTable(getCatalog(), tableIdentifier);
+ Table table = getCatalog().loadTable(tableIdentifier);
+ if (!(table instanceof HasTableOperations)) {
+ CatalogHandlers.purgeTable(getCatalog(), tableIdentifier);
+ return;
+ }
+ // Iceberg drops the catalog entry first and then deletes the files,
giving up on the first
+ // file it cannot read (for example a missing manifest list). The request
then fails although
+ // the table is gone, and no request is left that could delete the files.
Read the metadata
+ // first, drop the entry, then delete every file that can still be reached.
+ TableOperations ops = ((HasTableOperations) table).operations();
+ TableMetadata metadata = ops.current();
+ FileIO io = ops.io();
+ if (!getCatalog().dropTable(tableIdentifier, false)) {
Review Comment:
[Important] Preserve backend purge delegation, particularly for REST catalogs
`HasTableOperations` does not establish that it is safe to replace the
catalog's purge implementation. Tables loaded by Iceberg's REST catalog also
implement this interface. For that supported backend, this changes the upstream
request from `purgeRequested=true` to an ordinary drop, then tries to delete
files from this server. A proxy that can authenticate to the upstream REST
service but lacks direct storage/delete permissions will now leave files
behind; those deletion failures are logged and the request still succeeds.
Custom catalogs may also attach their own semantics to purge.
Please keep purge delegated for REST backends and avoid selecting this
replacement solely from the loaded table's type. The resilient cleanup needs an
explicit backend/capability boundary. This finding is based on the REST loading
and drop/purge implementations; I have not run a two-server test with separate
storage permissions.
##########
iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java:
##########
@@ -244,7 +247,30 @@ public void dropTable(TableIdentifier tableIdentifier) {
public void purgeTable(TableIdentifier tableIdentifier) {
getMetadataCache().invalidate(tableIdentifier);
- CatalogHandlers.purgeTable(getCatalog(), tableIdentifier);
+ Table table = getCatalog().loadTable(tableIdentifier);
Review Comment:
[Important] Do not make readable current metadata a prerequisite for
dropping a broken table
The new `loadTable` happens before the fallback and regresses JDBC purge
when the current metadata JSON is missing. Previously,
`JdbcCatalog.dropTable(..., true)` caught `NotFoundException` while reading the
metadata and still removed the catalog registration. Now this pre-read throws
first, leaving the registration in place and producing HTTP 404.
I reproduced this on SQLite: deleting the current metadata JSON makes the
new wrapper throw `NotFoundException`, whereas a subsequent direct
`catalog.dropTable(id, true)` returns true. Please preserve the backend's
existing handling of missing current metadata, and add a regression test for
this case.
##########
iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergPurgeUtil.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.iceberg.common.ops;
+
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.ManifestContent;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.StatisticsFile;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.io.FileIO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Deletes a dropped table's files when some of them can no longer be read. */
+final class IcebergPurgeUtil {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(IcebergPurgeUtil.class);
+
+ private IcebergPurgeUtil() {}
+
+ /**
+ * Deletes every data, delete, manifest, manifest list, statistics and
metadata file reachable
+ * from the metadata, skipping the parts that cannot be read. A missing
manifest list only loses
+ * the files listed in it.
+ *
+ * @param io the table's FileIO
+ * @param metadata the table metadata read before the table was dropped
+ */
+ static void deleteReachableFiles(FileIO io, TableMetadata metadata) {
+ Set<String> manifestLists = new HashSet<>();
+ Set<String> manifests = new HashSet<>();
+ Set<String> contentFiles = new HashSet<>();
+ for (Snapshot snapshot : metadata.snapshots()) {
+ if (snapshot.manifestListLocation() != null) {
+ manifestLists.add(snapshot.manifestListLocation());
+ }
+ try {
+ for (ManifestFile manifest : snapshot.allManifests(io)) {
+ if (manifests.add(manifest.path())) {
+ collectContentFiles(io, metadata, manifest, contentFiles);
+ }
+ }
+ } catch (RuntimeException e) {
+ LOG.warn(
+ "Cannot read the manifest list of snapshot {}; its files are not
deleted: {}",
+ snapshot.snapshotId(),
+ e.getMessage());
+ }
+ }
+ Set<String> metadataFiles = new HashSet<>();
+ metadata.previousFiles().forEach(entry -> metadataFiles.add(entry.file()));
+ if (metadata.metadataFileLocation() != null) {
+ metadataFiles.add(metadata.metadataFileLocation());
+ }
+ for (StatisticsFile stats : metadata.statisticsFiles()) {
+ metadataFiles.add(stats.path());
+ }
+ metadata.partitionStatisticsFiles().forEach(stats ->
metadataFiles.add(stats.path()));
+
+ deleteAll(io, contentFiles);
Review Comment:
[Blocker] Preserve `gc.enabled=false` in the fallback
If a manifest list is missing, this fallback deletes content files
unconditionally, including when the table has `gc.enabled=false`. Iceberg
1.11's `CatalogUtil.dropTableData` deliberately skips content-file deletion in
that configuration to avoid corrupting other tables that share those files. I
reproduced this with an append-only table, `gc.enabled=false`, and a missing
earlier snapshot manifest list: the fallback deletes the data file.
Please apply the same GC-property check before collecting/deleting data and
delete files, while retaining the appropriate metadata cleanup. A regression
test should assert that the shared content files survive this fallback.
##########
iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java:
##########
@@ -244,7 +247,30 @@ public void dropTable(TableIdentifier tableIdentifier) {
public void purgeTable(TableIdentifier tableIdentifier) {
getMetadataCache().invalidate(tableIdentifier);
- CatalogHandlers.purgeTable(getCatalog(), tableIdentifier);
+ Table table = getCatalog().loadTable(tableIdentifier);
+ if (!(table instanceof HasTableOperations)) {
+ CatalogHandlers.purgeTable(getCatalog(), tableIdentifier);
+ return;
+ }
+ // Iceberg drops the catalog entry first and then deletes the files,
giving up on the first
+ // file it cannot read (for example a missing manifest list). The request
then fails although
+ // the table is gone, and no request is left that could delete the files.
Read the metadata
+ // first, drop the entry, then delete every file that can still be reached.
+ TableOperations ops = ((HasTableOperations) table).operations();
+ TableMetadata metadata = ops.current();
+ FileIO io = ops.io();
+ if (!getCatalog().dropTable(tableIdentifier, false)) {
+ throw new NoSuchTableException("Table does not exist: %s",
tableIdentifier);
+ }
+ try {
+ CatalogUtil.dropTableData(io, metadata);
+ } catch (RuntimeException e) {
+ LOG.warn(
Review Comment:
[Nit] Retain the exception in cleanup warnings
This warning and the new utility's three catch blocks log only
`e.getMessage()`, discarding the exception type, cause chain, and stack trace.
Since the best-effort path may return success despite read/delete failures,
these warnings are the main diagnostic signal for storage permission failures,
SDK errors, and corrupt manifests. Please pass the Throwable to the logger as
well, and include the manifest-list location in the snapshot-read warning.
##########
iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergPurgeUtil.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.iceberg.common.ops;
+
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.ManifestContent;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.StatisticsFile;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.io.FileIO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Deletes a dropped table's files when some of them can no longer be read. */
+final class IcebergPurgeUtil {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(IcebergPurgeUtil.class);
+
+ private IcebergPurgeUtil() {}
+
+ /**
+ * Deletes every data, delete, manifest, manifest list, statistics and
metadata file reachable
+ * from the metadata, skipping the parts that cannot be read. A missing
manifest list only loses
+ * the files listed in it.
+ *
+ * @param io the table's FileIO
+ * @param metadata the table metadata read before the table was dropped
+ */
+ static void deleteReachableFiles(FileIO io, TableMetadata metadata) {
+ Set<String> manifestLists = new HashSet<>();
+ Set<String> manifests = new HashSet<>();
+ Set<String> contentFiles = new HashSet<>();
+ for (Snapshot snapshot : metadata.snapshots()) {
+ if (snapshot.manifestListLocation() != null) {
+ manifestLists.add(snapshot.manifestListLocation());
+ }
+ try {
+ for (ManifestFile manifest : snapshot.allManifests(io)) {
+ if (manifests.add(manifest.path())) {
+ collectContentFiles(io, metadata, manifest, contentFiles);
+ }
+ }
+ } catch (RuntimeException e) {
+ LOG.warn(
+ "Cannot read the manifest list of snapshot {}; its files are not
deleted: {}",
+ snapshot.snapshotId(),
+ e.getMessage());
+ }
+ }
+ Set<String> metadataFiles = new HashSet<>();
+ metadata.previousFiles().forEach(entry -> metadataFiles.add(entry.file()));
+ if (metadata.metadataFileLocation() != null) {
+ metadataFiles.add(metadata.metadataFileLocation());
+ }
+ for (StatisticsFile stats : metadata.statisticsFiles()) {
+ metadataFiles.add(stats.path());
+ }
+ metadata.partitionStatisticsFiles().forEach(stats ->
metadataFiles.add(stats.path()));
+
+ deleteAll(io, contentFiles);
+ deleteAll(io, manifests);
+ deleteAll(io, manifestLists);
+ deleteAll(io, metadataFiles);
+ }
+
+ private static void collectContentFiles(
+ FileIO io, TableMetadata metadata, ManifestFile manifest, Set<String>
paths) {
+ try {
+ if (manifest.content() == ManifestContent.DATA) {
+ try (ManifestReader<DataFile> reader =
+ ManifestFiles.read(manifest, io, metadata.specsById())) {
+ reader.forEach(file -> paths.add(file.location()));
Review Comment:
[Important] Include deleted manifest entries when collecting purge paths
`ManifestReader.forEach()` uses `iterator()`, which iterates only live
entries. The original Iceberg purge iterates all manifest entries, including
`DELETED`. For example: append file A, delete A in a later snapshot, then lose
the append snapshot's manifest list. A is still reachable through the later
readable manifest's deleted entry, but this fallback omits it and then deletes
the remaining manifests and metadata, leaving A orphaned. I reproduced that
sequence with Iceberg 1.11.0.
Please collect paths from all entries rather than only live entries, while
respecting `gc.enabled`. The delete-manifest branch at line 99 has the same
filtering issue and needs the corresponding treatment.
##########
iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableHookDispatcher.java:
##########
@@ -171,6 +184,25 @@ public void tearDown() throws IllegalAccessException {
FieldUtils.writeStaticField(holderClass, "INSTANCE", null, true);
}
+ @Test
+ public void testCreateTableHoldsTheNamespaceBranchLockForReading() {
+ // dropNamespace holds the same lock for writing, so a create and a drop
of one namespace are
+ // serialized on this server.
+ Namespace namespace = Namespace.of("A", "B");
+ CreateTableRequest request =
+
CreateTableRequest.builder().withName("t").withSchema(TABLE_SCHEMA).build();
+ when(mockDispatcher.createTable(mockContext, namespace, request))
+ .thenReturn(mock(LoadTableResponse.class));
+
+ hookDispatcher.createTable(mockContext, namespace, request);
+
+ ArgumentCaptor<NameIdentifier> lockId =
ArgumentCaptor.forClass(NameIdentifier.class);
Review Comment:
[Nit] Verify the lock mode and protected execution, not just the identifier
This test verifies which tree lock is created, but does not verify that it
is acquired for reading or remains held during the dispatcher call. Changing
READ to WRITE, or unlocking before executing the create, would not be caught by
these assertions.
Please add a test with a real LockManager and controlled latches that proves
namespace drop cannot run while create is executing, and checks both orderings.
The current test does fail when the locking change is removed entirely, but
does not establish the concurrency behavior claimed by its name.
--
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]
