This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 30687f68ff [#12851] fix(iceberg-rest): Share the managed memory
catalog in auxiliary mode (#12852)
30687f68ff is described below
commit 30687f68ff8c2a276a4a0fbf13211435da36f7f9
Author: roryqi <[email protected]>
AuthorDate: Fri Sep 4 16:33:58 2026 +0800
[#12851] fix(iceberg-rest): Share the managed memory catalog in auxiliary
mode (#12852)
### What changes were proposed in this pull request?
- Expose the Iceberg backend owned by `IcebergCatalogOperations`.
- Reuse that backend when the auxiliary Iceberg REST service serves a
memory catalog.
- Resolve the REST `default_catalog` alias to the configured Gravitino
catalog.
- Prevent the REST wrapper from closing the borrowed catalog.
- Add regression tests covering namespace, table, and view creation for
both explicit catalog names
and the default catalog alias.
### Why are the changes needed?
The auxiliary Iceberg REST service currently creates a second
InMemoryCatalog. REST creates are
written to that instance, while metadata import uses the instance owned
by Gravitino CatalogManager.
Because in-memory state is local to each instance, the import cannot
find the newly created object
and leaves Iceberg and Gravitino metadata inconsistent.
Fix: #12851
### Does this PR introduce _any_ user-facing change?
Yes. Namespace, table, and view creation through the Iceberg REST API
now succeeds with a memory
backend, and the created objects are visible through the Gravitino
metadata API.
No public API or configuration property is added or removed.
### How was this patch tested?
- `./gradlew spotlessApply`
- `./gradlew :iceberg:iceberg-common:test --tests
org.apache.gravitino.iceberg.common.ops.TestIcebergCatalogWrapper`
- `./gradlew :catalogs:catalog-lakehouse-iceberg:compileJava`
- `./gradlew :iceberg:iceberg-rest-server:test --tests
org.apache.gravitino.iceberg.service.TestIcebergCatalogWrapperManagerForREST`
- `./gradlew :iceberg:iceberg-rest-server:test --tests
org.apache.gravitino.iceberg.service.TestIcebergCatalogWrapperManagerForREST
--tests
org.apache.gravitino.iceberg.service.dispatcher.TestIcebergNamespaceHookDispatcher
--tests
org.apache.gravitino.iceberg.service.dispatcher.TestIcebergTableHookDispatcher
--tests
org.apache.gravitino.iceberg.service.dispatcher.TestIcebergViewHookDispatcher`
- `./gradlew :catalogs:catalog-lakehouse-iceberg:test --tests
org.apache.gravitino.catalog.lakehouse.iceberg.integration.test.CatalogIcebergRestIT
-PskipTests -PtestMode=embedded -PjdbcBackend=h2 -PskipDockerTests=false
-PskipWeb=true`
---
.../lakehouse/iceberg/IcebergConstants.java | 3 +
.../iceberg/IcebergCatalogOperations.java | 16 +++-
.../iceberg/TestIcebergCatalogOperations.java | 23 ++++++
.../lakehouse/iceberg/TestIcebergSchema.java | 3 +-
.../apache/gravitino/catalog/CatalogManager.java | 12 +++
.../gravitino/connector/CatalogDropAware.java | 29 ++++++++
.../gravitino/catalog/TestCatalogManager.java | 13 ++++
.../iceberg/common/ops/IcebergCatalogWrapper.java | 6 +-
.../iceberg/common/utils/IcebergCatalogUtil.java | 35 +++++++++
.../common/ops/TestIcebergCatalogWrapper.java | 27 +++++++
.../common/utils/TestIcebergCatalogUtil.java | 86 ++++++++++++++++++++++
.../service/IcebergCatalogWrapperManager.java | 4 +
.../provider/DynamicIcebergConfigProvider.java | 13 ++--
.../TestIcebergCatalogWrapperManagerForREST.java | 46 +++++++++++-
.../provider/TestDynamicIcebergConfigProvider.java | 6 ++
15 files changed, 312 insertions(+), 10 deletions(-)
diff --git
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
index 354153b303..2d29e9e754 100644
---
a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
+++
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java
@@ -24,6 +24,9 @@ public class IcebergConstants {
public static final String CATALOG_BACKEND = "catalog-backend";
public static final String CATALOG_BACKEND_IMPL = "catalog-backend-impl";
+ /** Internal property containing the unique identifier of a Gravitino
catalog. */
+ public static final String CATALOG_UUID = "catalog_uuid";
+
public static final String GRAVITINO_JDBC_USER = "jdbc-user";
public static final String ICEBERG_JDBC_USER = "jdbc.user";
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java
index ea53a7fd0f..140db99ff0 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java
@@ -31,12 +31,14 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
+import javax.annotation.Nullable;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.SchemaChange;
import
org.apache.gravitino.catalog.lakehouse.iceberg.ops.IcebergCatalogWrapperHelper;
+import org.apache.gravitino.connector.CatalogDropAware;
import org.apache.gravitino.connector.CatalogInfo;
import org.apache.gravitino.connector.CatalogOperations;
import org.apache.gravitino.connector.HasPropertyMetadata;
@@ -56,6 +58,7 @@ import
org.apache.gravitino.iceberg.common.authentication.SupportsKerberos;
import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
import
org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper.IcebergTableChange;
import
org.apache.gravitino.iceberg.common.ops.KerberosAwareIcebergCatalogProxy;
+import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil;
import org.apache.gravitino.iceberg.common.utils.IcebergIdentifierUtils;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.rel.Column;
@@ -92,7 +95,7 @@ import org.slf4j.LoggerFactory;
/** Operations for interacting with an Apache Iceberg catalog in Apache
Gravitino. */
public class IcebergCatalogOperations
- implements CatalogOperations, SupportsSchemas, TableCatalog, ViewCatalog {
+ implements CatalogOperations, SupportsSchemas, TableCatalog, ViewCatalog,
CatalogDropAware {
private static final String ICEBERG_TABLE_DOES_NOT_EXIST_MSG = "Iceberg
table does not exist: %s";
@@ -100,6 +103,7 @@ public class IcebergCatalogOperations
@VisibleForTesting IcebergCatalogWrapper icebergCatalogWrapper;
+ @VisibleForTesting @Nullable String catalogUuid;
private IcebergCatalogWrapperHelper icebergCatalogWrapperHelper;
private IcebergViewCatalogOperations icebergViewCatalogOperations;
@@ -124,7 +128,8 @@ public class IcebergCatalogOperations
Map<String, String> resultConf = Maps.newHashMap(prefixMap);
resultConf.putAll(gravitinoConfig);
- resultConf.put("catalog_uuid", info.id().toString());
+ this.catalogUuid = info.id().toString();
+ resultConf.put(IcebergConstants.CATALOG_UUID, catalogUuid);
IcebergConfig icebergConfig = new IcebergConfig(resultConf);
IcebergCatalogWrapper rawWrapper = new
IcebergCatalogWrapper(icebergConfig);
@@ -154,6 +159,13 @@ public class IcebergCatalogOperations
}
}
+ @Override
+ public void onCatalogDropped() {
+ if (catalogUuid != null) {
+ IcebergCatalogUtil.removeMemoryCatalog(catalogUuid);
+ }
+ }
+
// Maps an Iceberg REST-client exception into Gravitino's taxonomy
@VisibleForTesting
static RuntimeException handleRestException(RESTException e) {
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogOperations.java
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogOperations.java
index 841d3ecfaa..9b42c1849f 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogOperations.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergCatalogOperations.java
@@ -31,6 +31,7 @@ import
org.apache.gravitino.exceptions.ConnectionFailedException;
import org.apache.gravitino.exceptions.GravitinoRuntimeException;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
+import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil;
import org.apache.iceberg.exceptions.BadRequestException;
import org.apache.iceberg.exceptions.RESTException;
import org.apache.iceberg.exceptions.ServiceUnavailableException;
@@ -148,4 +149,26 @@ public class TestIcebergCatalogOperations {
Assertions.assertTrue(Arrays.stream(result).anyMatch(id ->
"db1".equals(id.name())));
Assertions.assertTrue(Arrays.stream(result).anyMatch(id ->
"db2".equals(id.name())));
}
+
+ @Test
+ public void testCatalogDropRemovesInternedMemoryCatalog() {
+ String catalogUuid = "catalog-uuid";
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.CATALOG_UUID,
+ catalogUuid));
+ Object original =
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+
+ IcebergCatalogOperations catalogOperations = new
IcebergCatalogOperations();
+ catalogOperations.catalogUuid = catalogUuid;
+ catalogOperations.onCatalogDropped();
+
+ Object replacement =
+ IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY,
config);
+ Assertions.assertNotSame(original, replacement);
+ IcebergCatalogUtil.removeMemoryCatalog(catalogUuid);
+ }
}
diff --git
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergSchema.java
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergSchema.java
index 9c9f6f5de5..c028c24b2e 100644
---
a/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergSchema.java
+++
b/catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/TestIcebergSchema.java
@@ -187,7 +187,8 @@ public class TestIcebergSchema {
private IcebergCatalog initIcebergCatalog(String name) {
CatalogEntity entity =
CatalogEntity.builder()
- .withId(1L)
+ // Each test creates a different catalog, so it must also use a
distinct catalog ID.
+ .withId(Integer.toUnsignedLong(name.hashCode()))
.withName(name)
.withNamespace(Namespace.of(META_LAKE_NAME))
.withType(IcebergCatalog.Type.RELATIONAL)
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
index c78438838c..c3ca9701c9 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -79,6 +79,7 @@ import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
import org.apache.gravitino.StringIdentifier;
import org.apache.gravitino.connector.BaseCatalog;
+import org.apache.gravitino.connector.CatalogDropAware;
import org.apache.gravitino.connector.CatalogOperations;
import org.apache.gravitino.connector.HasPropertyMetadata;
import org.apache.gravitino.connector.SupportsSchemas;
@@ -1142,6 +1143,17 @@ public class CatalogManager implements
CatalogDispatcher, Closeable {
boolean deleted = store.delete(ident, EntityType.CATALOG, true);
if (deleted) {
markLocalMutation(ident);
+ try {
+ catalogWrapper.doWithCatalogOps(
+ operations -> {
+ if (operations instanceof CatalogDropAware) {
+ ((CatalogDropAware) operations).onCatalogDropped();
+ }
+ return null;
+ });
+ } catch (Exception e) {
+ LOG.warn("Failed to clean up resources for dropped catalog
{}", ident, e);
+ }
// Unmanaged: schemas removed only via store cascade — clean
secrets captured above.
if (!managedStorage) {
for (Map<String, String> schemaProperties :
unmanagedSchemaSecrets) {
diff --git
a/core/src/main/java/org/apache/gravitino/connector/CatalogDropAware.java
b/core/src/main/java/org/apache/gravitino/connector/CatalogDropAware.java
new file mode 100644
index 0000000000..2972808c1c
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/connector/CatalogDropAware.java
@@ -0,0 +1,29 @@
+/*
+ * 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.connector;
+
+import org.apache.gravitino.annotation.Evolving;
+
+/** Supports cleanup that must run only when a catalog is permanently dropped.
*/
+@Evolving
+public interface CatalogDropAware {
+
+ /** Performs cleanup after the catalog metadata has been permanently
dropped. */
+ void onCatalogDropped();
+}
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
index 7697431211..26380a5833 100644
--- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
+++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
@@ -62,6 +62,7 @@ import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
import org.apache.gravitino.connector.BaseCatalog;
+import org.apache.gravitino.connector.CatalogDropAware;
import org.apache.gravitino.connector.CatalogOperations;
import org.apache.gravitino.connector.HiddenPropertyMaskUtils;
import org.apache.gravitino.connector.TestCatalogOperations;
@@ -1369,10 +1370,21 @@ public class TestCatalogManager {
Mockito.mock(CatalogManager.CatalogWrapper.class,
Mockito.RETURNS_DEEP_STUBS);
Capability capability = Mockito.mock(Capability.class);
CapabilityResult unsupportedResult = CapabilityResult.unsupported("Not
managed");
+ CatalogOperations operations =
+ Mockito.mock(
+ CatalogOperations.class,
+ Mockito.withSettings().extraInterfaces(CatalogDropAware.class));
Mockito.doReturn(catalogWrapper).when(catalogManager).loadCatalogAndWrap(ident);
Mockito.doReturn(catalog).when(catalogWrapper).catalog();
Mockito.doReturn(capability).when(catalogWrapper).capabilities();
Mockito.doReturn(unsupportedResult).when(capability).managedStorage(any());
+ Mockito.doAnswer(
+ invocation -> {
+ ThrowableFunction<CatalogOperations, ?> function =
invocation.getArgument(0);
+ return function.apply(operations);
+ })
+ .when(catalogWrapper)
+ .doWithCatalogOps(any());
catalogManager.getCatalogCache().put(ident, catalogWrapper);
boolean dropped = catalogManager.dropCatalog(ident);
@@ -1380,6 +1392,7 @@ public class TestCatalogManager {
Assertions.assertTrue(dropped);
Assertions.assertFalse(entityStore.exists(ident, EntityType.CATALOG));
Assertions.assertNull(catalogManager.getCatalogCache().getIfPresent(ident));
+ Mockito.verify((CatalogDropAware) operations).onCatalogDropped();
}
@Test
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
index 8d2196ecca..ed58b2885b 100644
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
@@ -27,6 +27,7 @@ import lombok.Setter;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.catalog.hadoop.fs.FileSystemUtils;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.cache.SupportsMetadataLocation;
import org.apache.gravitino.iceberg.common.cache.TableMetadataCache;
@@ -389,7 +390,10 @@ public class IcebergCatalogWrapper implements
AutoCloseable {
} else {
LOG.info("Closing IcebergCatalogWrapper before catalog is initialized");
}
- if (loadedCatalog instanceof AutoCloseable) {
+ boolean internedMemoryCatalog =
+ catalogBackend == IcebergCatalogBackend.MEMORY
+ &&
icebergConfig.getAllConfig().containsKey(IcebergConstants.CATALOG_UUID);
+ if (!internedMemoryCatalog && loadedCatalog instanceof AutoCloseable) {
// JdbcCatalog and ClosableHiveCatalog implement AutoCloseable and will
handle their own
// cleanup
((AutoCloseable) loadedCatalog).close();
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
index 5674b6c1a7..745b6b1d4f 100644
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java
@@ -23,11 +23,13 @@ import static
org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
+import java.io.IOException;
import java.sql.SQLException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.exceptions.ConnectionFailedException;
@@ -63,7 +65,19 @@ public class IcebergCatalogUtil {
*/
private static final String ICEBERG_TYPE_COLUMN = "iceberg_type";
+ private static final ConcurrentHashMap<String, InMemoryCatalog>
MEMORY_CATALOGS =
+ new ConcurrentHashMap<>();
+
private static InMemoryCatalog loadMemoryCatalog(IcebergConfig
icebergConfig) {
+ String catalogUuid =
icebergConfig.getAllConfig().get(IcebergConstants.CATALOG_UUID);
+ if (catalogUuid == null) {
+ return createMemoryCatalog(icebergConfig);
+ }
+ return MEMORY_CATALOGS.computeIfAbsent(
+ catalogUuid, ignored -> createMemoryCatalog(icebergConfig));
+ }
+
+ private static InMemoryCatalog createMemoryCatalog(IcebergConfig
icebergConfig) {
String icebergCatalogName = icebergConfig.getCatalogBackendName();
InMemoryCatalog memoryCatalog = new
MemoryCatalogWithMetadataLocationSupport();
Map<String, String> resultProperties =
icebergConfig.getIcebergCatalogProperties();
@@ -75,6 +89,27 @@ public class IcebergCatalogUtil {
return memoryCatalog;
}
+ /**
+ * Removes the in-memory Iceberg catalog associated with a permanently
dropped Gravitino catalog.
+ *
+ * @param catalogUuid the unique Gravitino catalog identifier
+ */
+ public static void removeMemoryCatalog(String catalogUuid) {
+ InMemoryCatalog memoryCatalog = MEMORY_CATALOGS.remove(catalogUuid);
+ if (memoryCatalog != null) {
+ try {
+ memoryCatalog.close();
+ } catch (IOException e) {
+ LOG.warn("Failed to close dropped in-memory Iceberg catalog {}",
catalogUuid, e);
+ }
+ }
+ }
+
+ @VisibleForTesting
+ static void clearMemoryCatalogs() {
+ MEMORY_CATALOGS.clear();
+ }
+
private static HiveCatalog loadHiveCatalog(IcebergConfig icebergConfig) {
ClosableHiveCatalog hiveCatalog = new
HiveCatalogWithMetadataLocationSupport();
HdfsConfiguration hdfsConfiguration = new HdfsConfiguration();
diff --git
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/ops/TestIcebergCatalogWrapper.java
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/ops/TestIcebergCatalogWrapper.java
index e34f1becb3..e6928e56a8 100644
---
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/ops/TestIcebergCatalogWrapper.java
+++
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/ops/TestIcebergCatalogWrapper.java
@@ -30,8 +30,12 @@ import
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.cache.SupportsMetadataLocation;
import org.apache.gravitino.iceberg.common.cache.TableMetadataCache;
+import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil;
import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -57,6 +61,29 @@ public class TestIcebergCatalogWrapper {
});
}
+ @Test
+ public void testMemoryCatalogSurvivesWrapperClose() throws Exception {
+ String catalogUuid = "catalog-1";
+ IcebergConfig config =
+ new IcebergConfig(
+ Map.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.CATALOG_UUID,
+ catalogUuid));
+ IcebergCatalogWrapper firstWrapper = new IcebergCatalogWrapper(config);
+ Catalog firstCatalog = firstWrapper.getCatalog();
+ Namespace namespace = Namespace.of("preserved");
+
firstWrapper.createNamespace(CreateNamespaceRequest.builder().withNamespace(namespace).build());
+
+ firstWrapper.close();
+ IcebergCatalogWrapper secondWrapper = new IcebergCatalogWrapper(config);
+
+ Assertions.assertSame(firstCatalog, secondWrapper.getCatalog());
+ Assertions.assertTrue(secondWrapper.namespaceExists(namespace));
+ IcebergCatalogUtil.removeMemoryCatalog(catalogUuid);
+ }
+
@Test
public void testMetadataCacheShouldInitializeOnFirstAccessAndClose(@TempDir
Path warehouseDir)
throws Exception {
diff --git
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
index a575352d3e..f5ee1b8cf0 100644
---
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
+++
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java
@@ -40,6 +40,7 @@ import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.jdbc.JdbcCatalogWithMetadataLocationSupport;
import org.apache.iceberg.jdbc.UncheckedSQLException;
import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -48,6 +49,91 @@ public class TestIcebergCatalogUtil {
@TempDir private Path warehouse;
+ @AfterEach
+ void tearDown() {
+ IcebergCatalogUtil.clearMemoryCatalogs();
+ }
+
+ @Test
+ void testMemoryCatalogIsInternedByCatalogUuid() {
+ Map<String, String> properties =
+ Map.of(
+ IcebergConstants.CATALOG_BACKEND, "memory",
IcebergConstants.CATALOG_UUID, "catalog-1");
+ IcebergConfig config = new IcebergConfig(properties);
+
+ InMemoryCatalog first =
+ (InMemoryCatalog)
+
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+ Namespace namespace = Namespace.of("shared");
+ first.createNamespace(namespace);
+ InMemoryCatalog second =
+ (InMemoryCatalog)
+
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+
+ Assertions.assertSame(first, second);
+ Assertions.assertTrue(second.namespaceExists(namespace));
+ }
+
+ @Test
+ void testMemoryCatalogsWithDifferentUuidsAreIsolated() {
+ IcebergConfig firstConfig =
+ new IcebergConfig(
+ Map.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.CATALOG_UUID,
+ "catalog-1"));
+ IcebergConfig secondConfig =
+ new IcebergConfig(
+ Map.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.CATALOG_UUID,
+ "catalog-2"));
+
+ Catalog first =
+ IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY,
firstConfig);
+ Catalog second =
+ IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY,
secondConfig);
+
+ Assertions.assertNotSame(first, second);
+ }
+
+ @Test
+ void testMemoryCatalogWithoutUuidIsNotInterned() {
+ IcebergConfig config = new
IcebergConfig(Map.of(IcebergConstants.CATALOG_BACKEND, "memory"));
+
+ Catalog first =
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+ Catalog second =
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+
+ Assertions.assertNotSame(first, second);
+ }
+
+ @Test
+ void testDroppedMemoryCatalogIsRemoved() {
+ String catalogUuid = "catalog-1";
+ IcebergConfig config =
+ new IcebergConfig(
+ Map.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.CATALOG_UUID,
+ catalogUuid));
+ InMemoryCatalog original =
+ (InMemoryCatalog)
+
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+ Namespace namespace = Namespace.of("removed");
+ original.createNamespace(namespace);
+
+ IcebergCatalogUtil.removeMemoryCatalog(catalogUuid);
+ InMemoryCatalog replacement =
+ (InMemoryCatalog)
+
IcebergCatalogUtil.loadCatalogBackend(IcebergCatalogBackend.MEMORY, config);
+
+ Assertions.assertNotSame(original, replacement);
+ Assertions.assertFalse(replacement.namespaceExists(namespace));
+ }
+
@Test
void testLoadCatalog() {
Catalog catalog;
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
index d5a2d27206..5d8fbf52f1 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
@@ -30,6 +30,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig;
@@ -87,6 +88,9 @@ public class IcebergCatalogWrapperManager implements
AutoCloseable {
ident -> {
if (ident.namespace().level(0).equals(metalakeName)) {
catalogWrapperCache.invalidate(ident.name());
+ if
(ident.name().equals(configProvider.getDefaultCatalogName())) {
+
catalogWrapperCache.invalidate(IcebergConstants.ICEBERG_REST_DEFAULT_CATALOG);
+ }
}
});
}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/provider/DynamicIcebergConfigProvider.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/provider/DynamicIcebergConfigProvider.java
index b4f724626f..fc9efbf8ab 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/provider/DynamicIcebergConfigProvider.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/provider/DynamicIcebergConfigProvider.java
@@ -112,11 +112,14 @@ public class DynamicIcebergConfigProvider implements
IcebergConfigProvider {
private static Map<String, String> resolveProps(Catalog catalog) {
if (catalog instanceof BaseCatalog) {
- return new HashMap<>(
- GravitinoEnv.getInstance()
- .secretManager()
- .toPlaintextProperties(
- ((BaseCatalog<?>)
catalog).propertiesWithCredentialProviders()));
+ BaseCatalog<?> baseCatalog = (BaseCatalog<?>) catalog;
+ Map<String, String> props =
+ new HashMap<>(
+ GravitinoEnv.getInstance()
+ .secretManager()
+
.toPlaintextProperties(baseCatalog.propertiesWithCredentialProviders()));
+ props.put(IcebergConstants.CATALOG_UUID,
baseCatalog.entity().id().toString());
+ return props;
}
Map<String, String> props =
new HashMap<>(catalog.properties() == null ? Map.of() :
catalog.properties());
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
index 2d4386b730..45eb62f69e 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
@@ -21,9 +21,13 @@ package org.apache.gravitino.iceberg.service;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.catalog.CatalogManager;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.iceberg.common.IcebergConfig;
@@ -42,11 +46,12 @@ import org.mockito.Mockito;
public class TestIcebergCatalogWrapperManagerForREST {
private static final String DEFAULT_CATALOG = "memory";
+ private static CatalogManager mockCatalogManager;
@BeforeAll
public static void setup() throws IllegalAccessException {
// Mock CatalogManager for GravitinoEnv to avoid initialization errors
- CatalogManager mockCatalogManager = Mockito.mock(CatalogManager.class);
+ mockCatalogManager = Mockito.mock(CatalogManager.class);
FieldUtils.writeField(GravitinoEnv.getInstance(), "catalogManager",
mockCatalogManager, true);
}
@@ -145,6 +150,45 @@ public class TestIcebergCatalogWrapperManagerForREST {
Assertions.assertEquals(CatalogWrapperForREST.class, wrapper.getClass());
}
+ @Test
+ public void testDefaultCatalogAliasInvalidatedWhenCatalogRemoved() throws
Exception {
+ Mockito.clearInvocations(mockCatalogManager);
+ AtomicReference<Consumer<NameIdentifier>> removeListener = new
AtomicReference<>();
+ Mockito.doAnswer(
+ invocation -> {
+ removeListener.set(invocation.getArgument(0));
+ return null;
+ })
+ .when(mockCatalogManager)
+ .addCatalogCacheRemoveListener(Mockito.any());
+
+ IcebergConfigProvider configProvider =
Mockito.mock(IcebergConfigProvider.class);
+ Mockito.when(configProvider.getDefaultCatalogName()).thenReturn("test");
+ IcebergConfig icebergConfig =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ Mockito.when(
+
configProvider.getIcebergCatalogConfig(IcebergConstants.ICEBERG_REST_DEFAULT_CATALOG))
+ .thenReturn(Optional.of(icebergConfig));
+
+ try (IcebergCatalogWrapperManager manager =
+ new IcebergCatalogWrapperManager(Maps.newHashMap(), configProvider,
true, "metalake")) {
+ IcebergRESTServerContext.create(configProvider, false, false, true,
manager);
+ CatalogWrapperForREST original =
+
manager.getCatalogWrapper(IcebergConstants.ICEBERG_REST_DEFAULT_CATALOG);
+
+ removeListener.get().accept(NameIdentifier.of("metalake", "test"));
+
+ CatalogWrapperForREST replacement =
+
manager.getCatalogWrapper(IcebergConstants.ICEBERG_REST_DEFAULT_CATALOG);
+ Assertions.assertNotSame(original, replacement);
+ }
+ }
+
private static IcebergCatalogWrapperManager newManager() {
Map<String, String> config = Maps.newHashMap();
IcebergConfigProvider configProvider =
IcebergConfigProviderFactory.create(config);
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/provider/TestDynamicIcebergConfigProvider.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/provider/TestDynamicIcebergConfigProvider.java
index 42807c1816..3389768d54 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/provider/TestDynamicIcebergConfigProvider.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/provider/TestDynamicIcebergConfigProvider.java
@@ -45,6 +45,7 @@ import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
+import org.apache.gravitino.meta.CatalogEntity;
import org.apache.gravitino.secret.SecretBinding;
import org.apache.gravitino.secret.SecretManager;
import org.apache.gravitino.secret.SecretMaterial;
@@ -760,8 +761,11 @@ public class TestDynamicIcebergConfigProvider {
@SuppressWarnings("unchecked")
BaseCatalog<?> baseCatalog = Mockito.mock(BaseCatalog.class);
+ CatalogEntity catalogEntity = Mockito.mock(CatalogEntity.class);
Mockito.when(baseCatalog.provider()).thenReturn("lakehouse-iceberg");
Mockito.when(baseCatalog.propertiesWithCredentialProviders()).thenReturn(entityProps);
+ Mockito.when(baseCatalog.entity()).thenReturn(catalogEntity);
+ Mockito.when(catalogEntity.id()).thenReturn(9L);
FieldUtils.writeField(GravitinoEnv.getInstance(), "secretManager", sm,
true);
@@ -781,6 +785,8 @@ public class TestDynamicIcebergConfigProvider {
Assertions.assertEquals(
"root",
config.get().getIcebergCatalogProperties().get(IcebergConstants.GRAVITINO_JDBC_USER));
+ Assertions.assertEquals(
+ "9",
config.get().getIcebergCatalogProperties().get(IcebergConstants.CATALOG_UUID));
}
}
