You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@amoro.apache.org by na...@apache.org on 2026/08/10 08:57:17 UTC
(amoro) 04/06: feat: add HMS namespace catalog explorer
This is an automated email from the ASF dual-hosted git repository.
majin1102 pushed a commit to branch emr/hms-namespace-explorer
in repository https://gitbox.apache.org/repos/asf/amoro.git
commit 4090a2cab0980102c641a304bd1fe214bc642d8d
Author: majin.nathan <ma...@bytedance.com>
AuthorDate: Thu Aug 6 20:57:09 2026 +0800
feat: add HMS namespace catalog explorer
---
.../apache/amoro/server/AmoroManagementConf.java | 6 +
.../amoro/server/catalog/CatalogManager.java | 26 ++
.../server/catalog/DefaultCatalogManager.java | 62 +++++
.../amoro/server/dashboard/DashboardServer.java | 8 +
.../dashboard/controller/TableController.java | 33 ++-
.../amoro/server/las/LasCatalogSynchronizer.java | 266 +++++++++++++++++++++
.../amoro/server/las/LasIntegrationConfig.java | 12 +
.../amoro/server/las/LasIntegrationContext.java | 17 ++
.../apache/amoro/server/las/LasRestExtension.java | 11 +
.../persistence/SqlSessionFactoryProvider.java | 2 +
.../mapper/NamespaceAllowlistMapper.java | 43 ++++
.../resources/authorization/privilege_mapping.yaml | 15 ++
.../src/main/resources/derby/ams-derby-init.sql | 4 +
amoro-ams/src/main/resources/derby/upgrade.sql | 4 +
.../src/main/resources/mysql/ams-mysql-init.sql | 5 +
amoro-ams/src/main/resources/mysql/upgrade.sql | 4 +
amoro-ams/src/main/resources/openapi/openapi.yaml | 60 ++++-
.../main/resources/postgres/ams-postgres-init.sql | 5 +
amoro-ams/src/main/resources/postgres/upgrade.sql | 4 +
.../TestDefaultCatalogManagerNamespaces.java | 130 ++++++++++
.../controller/TestTableControllerNamespaces.java | 113 +++++++++
.../server/las/TestLasCatalogSynchronizer.java | 245 +++++++++++++++++++
.../amoro/properties/CatalogMetaProperties.java | 3 +
amoro-web/src/components/Sidebar.vue | 10 +-
amoro-web/src/services/table.service.ts | 10 +-
amoro-web/src/views/tables/components/Details.vue | 28 ++-
.../src/views/tables/components/TableExplorer.vue | 207 ++++++++++++++--
dist/src/main/amoro-bin/conf/config.yaml | 7 +
docs/configuration/ams-config.md | 2 +-
29 files changed, 1311 insertions(+), 31 deletions(-)
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java
index f38f6f5ac..82efcb281 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java
@@ -113,6 +113,12 @@ public class AmoroManagementConf {
.defaultValue(Duration.ofSeconds(60))
.withDescription("TTL for catalog metadata.");
+ public static final ConfigOption<Boolean> CATALOG_NAMESPACE_ENABLED =
+ ConfigOptions.key("catalog.namespace-enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Whether catalogs are grouped by an optional namespace.");
+
public static final ConfigOption<Integer> TABLE_MANIFEST_IO_THREAD_COUNT =
ConfigOptions.key("table-manifest-io.thread-count")
.intType()
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogManager.java b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogManager.java
index 7f331dcbf..5a1c05c19 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogManager.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogManager.java
@@ -22,6 +22,7 @@ import org.apache.amoro.AmoroTable;
import org.apache.amoro.api.CatalogMeta;
import org.apache.amoro.table.TableIdentifier;
+import java.util.Collections;
import java.util.List;
/** The CatalogManager interface defines the operations that can be performed on catalogs. */
@@ -33,6 +34,31 @@ public interface CatalogManager {
*/
List<CatalogMeta> listCatalogMetas();
+ /** Returns whether this catalog manager exposes a namespace level. */
+ default boolean supportNamespace() {
+ return false;
+ }
+
+ /** Lists namespaces, or the community-compatible default namespace when unsupported. */
+ default List<String> listNamespaces() {
+ return Collections.singletonList("default");
+ }
+
+ /** Lists catalogs in a namespace. Implementations without namespace support return all. */
+ default List<CatalogMeta> listCatalogMetas(String namespace) {
+ return listCatalogMetas();
+ }
+
+ /** Adds a namespace to the allowlist. */
+ default void addNamespace(String namespace) {
+ throw new UnsupportedOperationException("Catalog namespaces are not supported");
+ }
+
+ /** Removes a namespace from the allowlist. */
+ default void removeNamespace(String namespace) {
+ throw new UnsupportedOperationException("Catalog namespaces are not supported");
+ }
+
/**
* Gets the catalog metadata for the given catalog name.
*
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/DefaultCatalogManager.java b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/DefaultCatalogManager.java
index 6535283da..c815afc8a 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/DefaultCatalogManager.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/DefaultCatalogManager.java
@@ -29,12 +29,15 @@ import org.apache.amoro.config.Configurations;
import org.apache.amoro.exception.AlreadyExistsException;
import org.apache.amoro.exception.IllegalMetadataException;
import org.apache.amoro.exception.ObjectNotExistsException;
+import org.apache.amoro.exception.PersistenceException;
import org.apache.amoro.properties.CatalogMetaProperties;
import org.apache.amoro.server.AmoroManagementConf;
import org.apache.amoro.server.dashboard.utils.AmsUtil;
import org.apache.amoro.server.persistence.PersistentBase;
import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper;
+import org.apache.amoro.server.persistence.mapper.NamespaceAllowlistMapper;
import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting;
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
import org.apache.amoro.shade.guava32.com.google.common.cache.CacheBuilder;
import org.apache.amoro.shade.guava32.com.google.common.cache.CacheLoader;
import org.apache.amoro.shade.guava32.com.google.common.cache.LoadingCache;
@@ -46,6 +49,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -104,6 +108,64 @@ public class DefaultCatalogManager extends PersistentBase implements CatalogMana
.collect(Collectors.toList());
}
+ @Override
+ public boolean supportNamespace() {
+ return serverConfiguration.getBoolean(AmoroManagementConf.CATALOG_NAMESPACE_ENABLED);
+ }
+
+ @Override
+ public List<String> listNamespaces() {
+ if (!supportNamespace()) {
+ return Collections.singletonList("default");
+ }
+ return getAs(NamespaceAllowlistMapper.class, NamespaceAllowlistMapper::listNamespaces);
+ }
+
+ @Override
+ public List<CatalogMeta> listCatalogMetas(String namespace) {
+ if (namespace == null || namespace.trim().isEmpty()) {
+ return listCatalogMetas();
+ }
+ if (!supportNamespace()) {
+ return "default".equals(namespace) ? listCatalogMetas() : Collections.emptyList();
+ }
+ if (!listNamespaces().contains(namespace)) {
+ return Collections.emptyList();
+ }
+ return listCatalogMetas().stream()
+ .filter(
+ catalog ->
+ catalog.getCatalogProperties() != null
+ && namespace.equals(
+ catalog.getCatalogProperties().get(CatalogMetaProperties.NAMESPACE)))
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public void addNamespace(String namespace) {
+ Preconditions.checkState(supportNamespace(), "Catalog namespaces are not enabled");
+ try {
+ doAs(
+ NamespaceAllowlistMapper.class,
+ mapper -> {
+ if (mapper.getNamespace(namespace) == null) {
+ mapper.insertNamespace(namespace);
+ }
+ });
+ } catch (PersistenceException e) {
+ // Another AMS node may have inserted the same namespace after our existence check.
+ if (getAs(NamespaceAllowlistMapper.class, mapper -> mapper.getNamespace(namespace)) == null) {
+ throw e;
+ }
+ }
+ }
+
+ @Override
+ public void removeNamespace(String namespace) {
+ Preconditions.checkState(supportNamespace(), "Catalog namespaces are not enabled");
+ doAs(NamespaceAllowlistMapper.class, mapper -> mapper.deleteNamespace(namespace));
+ }
+
@Override
public CatalogMeta getCatalogMeta(String catalogName) {
return getCatalogMetaOptional(catalogName)
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/DashboardServer.java b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/DashboardServer.java
index f6e1ab29d..eeea24b5d 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/DashboardServer.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/DashboardServer.java
@@ -301,6 +301,14 @@ public class DashboardServer {
});
get("/upgrade/properties", tableController::getUpgradeHiveTableProperties);
+ path(
+ "/namespaces",
+ () -> {
+ get("", tableController::getNamespaces);
+ put("/{namespace}", tableController::addNamespace);
+ delete("/{namespace}", tableController::removeNamespace);
+ });
+
// catalog apis
path(
"/catalogs",
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/TableController.java b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/TableController.java
index 81d29a425..7d460bb8c 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/TableController.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/TableController.java
@@ -626,10 +626,41 @@ public class TableController {
* @param ctx - context for handling the request and response
*/
public void getCatalogs(Context ctx) {
- List<CatalogMeta> catalogs = catalogManager.listCatalogMetas();
+ String namespace = ctx.queryParam("namespace");
+ List<CatalogMeta> catalogs =
+ StringUtils.isBlank(namespace)
+ ? catalogManager.listCatalogMetas()
+ : catalogManager.listCatalogMetas(namespace.trim());
ctx.json(OkResponse.of(catalogs));
}
+ /** Lists the namespace level used by the catalog explorer. */
+ public void getNamespaces(Context ctx) {
+ ctx.json(OkResponse.of(catalogManager.listNamespaces()));
+ }
+
+ /** Adds a namespace to the synchronization allowlist. */
+ public void addNamespace(Context ctx) {
+ String namespace = validateNamespace(ctx.pathParam("namespace"));
+ catalogManager.addNamespace(namespace);
+ ctx.json(OkResponse.of(namespace));
+ }
+
+ /** Removes a namespace from the synchronization allowlist. */
+ public void removeNamespace(Context ctx) {
+ String namespace = validateNamespace(ctx.pathParam("namespace"));
+ catalogManager.removeNamespace(namespace);
+ ctx.json(OkResponse.of(namespace));
+ }
+
+ private static String validateNamespace(String namespace) {
+ String normalized = StringUtils.trimToEmpty(namespace);
+ Preconditions.checkArgument(!normalized.isEmpty(), "namespace cannot be empty");
+ Preconditions.checkArgument(normalized.length() <= 128, "namespace is too long");
+ Preconditions.checkArgument(!normalized.contains("@"), "namespace cannot contain @");
+ return normalized;
+ }
+
/**
* get single page query token.
*
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/las/LasCatalogSynchronizer.java b/amoro-ams/src/main/java/org/apache/amoro/server/las/LasCatalogSynchronizer.java
new file mode 100644
index 000000000..e52a1f5b4
--- /dev/null
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/las/LasCatalogSynchronizer.java
@@ -0,0 +1,266 @@
+/*
+ * 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.amoro.server.las;
+
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.api.CatalogMeta;
+import org.apache.amoro.properties.CatalogMetaProperties;
+import org.apache.amoro.server.catalog.CatalogManager;
+import org.apache.amoro.shade.guava32.com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.amoro.table.TableProperties;
+import org.apache.hadoop.conf.Configuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+
+/** Periodically projects allowlisted HMS3 catalogs into AMS CatalogMeta records. */
+public final class LasCatalogSynchronizer implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(LasCatalogSynchronizer.class);
+
+ static final String CATALOG_SOURCE = "catalog.source";
+ static final String CATALOG_SOURCE_LAS_HMS = "las-hms-sync";
+
+ private static final String HIVE_METASTORE_URIS = "hive.metastore.uris";
+ private static final String METASTORE_CATALOG_DEFAULT = "metastore.catalog.default";
+ private static final int MAX_CATALOG_NAME_LENGTH = 64;
+
+ private final LasIntegrationContext context;
+ private final LasHmsClient hmsClient;
+ private final CatalogManager catalogManager;
+ private final ScheduledExecutorService scheduler;
+ private final long intervalMillis;
+ private final AtomicBoolean started = new AtomicBoolean(false);
+
+ public LasCatalogSynchronizer(
+ LasIntegrationContext context, LasHmsClient hmsClient, CatalogManager catalogManager) {
+ this(
+ context,
+ hmsClient,
+ catalogManager,
+ Executors.newSingleThreadScheduledExecutor(
+ new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat("las-hms-catalog-sync-%d")
+ .build()));
+ }
+
+ LasCatalogSynchronizer(
+ LasIntegrationContext context,
+ LasHmsClient hmsClient,
+ CatalogManager catalogManager,
+ ScheduledExecutorService scheduler) {
+ this.context = Objects.requireNonNull(context, "context");
+ this.hmsClient = Objects.requireNonNull(hmsClient, "hmsClient");
+ this.catalogManager = Objects.requireNonNull(catalogManager, "catalogManager");
+ this.scheduler = Objects.requireNonNull(scheduler, "scheduler");
+ this.intervalMillis = context.catalogSyncInterval().toMillis();
+ }
+
+ /** Starts synchronization after one full interval; no HMS call is made during AMS startup. */
+ public void start() {
+ if (started.compareAndSet(false, true)) {
+ scheduler.scheduleWithFixedDelay(
+ this::syncSafely, intervalMillis, intervalMillis, TimeUnit.MILLISECONDS);
+ LOG.info("LAS HMS catalog synchronization scheduled every {} ms", intervalMillis);
+ }
+ }
+
+ void syncOnce() throws Exception {
+ List<String> discoveredCatalogs = hmsClient.listCatalogs();
+ if (discoveredCatalogs == null || discoveredCatalogs.stream().anyMatch(Objects::isNull)) {
+ throw new IllegalStateException("HMS getCatalogs returned an invalid snapshot");
+ }
+
+ Set<String> rawCatalogNames = new HashSet<>(discoveredCatalogs);
+ Set<String> allowlistedNamespaces = new HashSet<>(catalogManager.listNamespaces());
+ Map<String, CatalogMeta> existingCatalogs =
+ catalogManager.listCatalogMetas().stream()
+ .collect(Collectors.toMap(CatalogMeta::getCatalogName, catalog -> catalog));
+
+ int created = 0;
+ int updated = 0;
+ int removed = 0;
+ for (String physicalCatalogName : rawCatalogNames) {
+ Optional<String> namespace = parseNamespace(physicalCatalogName);
+ if (!namespace.isPresent() || !allowlistedNamespaces.contains(namespace.get())) {
+ continue;
+ }
+
+ CatalogMeta existing = existingCatalogs.get(physicalCatalogName);
+ if (existing == null) {
+ catalogManager.createCatalog(createCatalogMeta(physicalCatalogName, namespace.get()));
+ created++;
+ } else if (isManagedByThisSynchronizer(existing)) {
+ CatalogMeta refreshed = mergeCatalogMeta(existing, physicalCatalogName, namespace.get());
+ if (!existing.equals(refreshed)) {
+ catalogManager.updateCatalog(refreshed);
+ updated++;
+ }
+ } else {
+ LOG.warn(
+ "Skip HMS catalog {} because an unmanaged catalog has the same name",
+ physicalCatalogName);
+ }
+ }
+
+ for (CatalogMeta existing : existingCatalogs.values()) {
+ if (isManagedByThisSynchronizer(existing)
+ && !rawCatalogNames.contains(existing.getCatalogName())) {
+ catalogManager.dropCatalog(existing.getCatalogName());
+ removed++;
+ }
+ }
+
+ LOG.info(
+ "LAS HMS catalog synchronization completed: discovered={}, allowlistedNamespaces={}, created={}, updated={}, removed={}",
+ rawCatalogNames.size(),
+ allowlistedNamespaces.size(),
+ created,
+ updated,
+ removed);
+ }
+
+ static Optional<String> parseNamespace(String catalogName) {
+ if (catalogName == null || catalogName.length() > MAX_CATALOG_NAME_LENGTH) {
+ return Optional.empty();
+ }
+ int separator = catalogName.indexOf('@');
+ if (separator <= 0 || separator == catalogName.length() - 1) {
+ return Optional.empty();
+ }
+ return Optional.of(catalogName.substring(0, separator));
+ }
+
+ private CatalogMeta mergeCatalogMeta(
+ CatalogMeta existing, String physicalCatalogName, String namespace) {
+ CatalogMeta refreshed = createCatalogMeta(physicalCatalogName, namespace);
+ Map<String, String> properties = new HashMap<>(existing.getCatalogProperties());
+ properties.putAll(refreshed.getCatalogProperties());
+ refreshed.setCatalogProperties(properties);
+ return refreshed;
+ }
+
+ private CatalogMeta createCatalogMeta(String physicalCatalogName, String namespace) {
+ CatalogMeta catalogMeta = new CatalogMeta();
+ catalogMeta.setCatalogName(physicalCatalogName);
+ catalogMeta.setCatalogType(CatalogMetaProperties.CATALOG_TYPE_HIVE);
+
+ Map<String, String> catalogProperties = new HashMap<>();
+ catalogProperties.put(
+ CatalogMetaProperties.TABLE_FORMATS,
+ TableFormat.ICEBERG.name() + "," + TableFormat.PAIMON.name());
+ catalogProperties.put(CatalogMetaProperties.NAMESPACE, namespace);
+ catalogProperties.put(CATALOG_SOURCE, CATALOG_SOURCE_LAS_HMS);
+ catalogProperties.put(
+ CatalogMetaProperties.TABLE_PROPERTIES_PREFIX + TableProperties.ENABLE_SELF_OPTIMIZING,
+ "false");
+ catalogMeta.setCatalogProperties(catalogProperties);
+
+ Map<String, String> storageConfigs = new HashMap<>();
+ storageConfigs.put(
+ CatalogMetaProperties.STORAGE_CONFIGS_KEY_TYPE,
+ CatalogMetaProperties.STORAGE_CONFIGS_VALUE_TYPE_HADOOP);
+ storageConfigs.put(
+ CatalogMetaProperties.STORAGE_CONFIGS_KEY_HIVE_SITE,
+ encodeConfiguration(createHiveConfiguration(physicalCatalogName)));
+ storageConfigs.put(
+ CatalogMetaProperties.STORAGE_CONFIGS_KEY_CORE_SITE,
+ encodeConfiguration(createTosConfiguration(namespace)));
+ storageConfigs.put(
+ CatalogMetaProperties.STORAGE_CONFIGS_KEY_HDFS_SITE,
+ encodeConfiguration(new Configuration(false)));
+ catalogMeta.setStorageConfigs(storageConfigs);
+ catalogMeta.setAuthConfigs(Collections.emptyMap());
+ return catalogMeta;
+ }
+
+ private Configuration createHiveConfiguration(String physicalCatalogName) {
+ Configuration configuration = new Configuration(false);
+ configuration.set(HIVE_METASTORE_URIS, context.hmsUri().toString());
+ configuration.set(METASTORE_CATALOG_DEFAULT, physicalCatalogName);
+ return configuration;
+ }
+
+ private Configuration createTosConfiguration(String namespace) {
+ Configuration configuration = new Configuration(false);
+ configuration.set("fs.AbstractFileSystem.tos.impl", "io.proton.fs.ProtonFS");
+ configuration.set("fs.tos.impl", "io.proton.fs.ProtonFileSystem");
+ configuration.set("fs.tos.endpoint", context.tosEndpoint().toString());
+ configuration.set("proton.cache.enable", "false");
+ configuration.set(
+ "mapreduce.outputcommitter.factory.class", "io.proton.commit.CommitterFactory");
+ configuration.set(
+ "fs.tos.credentials.provider", "io.proton.tos.iam.AssumeIamRoleCredentialProvider");
+ configuration.set("fs.volc.openapi.host", context.iamEndpoint().getAuthority());
+ configuration.set("fs.volc.openapi.region", context.region());
+ configuration.set("fs.tos.http.maxConnections", "1024");
+ configuration.set(
+ "fs.tos.credential.sts.iam-role-trn",
+ String.format("trn:iam::%s:role/%s", namespace, context.iamDataRoleName()));
+ configuration.set(
+ "fs.tos.credential.sts.token.time-to-live",
+ String.valueOf(context.iamAssumeRoleTtl().getSeconds()));
+ return configuration;
+ }
+
+ private static String encodeConfiguration(Configuration configuration) {
+ try (ByteArrayOutputStream output = new ByteArrayOutputStream()) {
+ configuration.writeXml(output);
+ return Base64.getEncoder().encodeToString(output.toByteArray());
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private static boolean isManagedByThisSynchronizer(CatalogMeta catalogMeta) {
+ return catalogMeta.getCatalogProperties() != null
+ && CATALOG_SOURCE_LAS_HMS.equals(catalogMeta.getCatalogProperties().get(CATALOG_SOURCE));
+ }
+
+ private void syncSafely() {
+ try {
+ syncOnce();
+ } catch (Throwable t) {
+ LOG.error("LAS HMS catalog synchronization failed", t);
+ }
+ }
+
+ @Override
+ public void close() {
+ scheduler.shutdownNow();
+ }
+}
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/las/LasIntegrationConfig.java b/amoro-ams/src/main/java/org/apache/amoro/server/las/LasIntegrationConfig.java
index 92852712d..65bd5cc6a 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/las/LasIntegrationConfig.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/las/LasIntegrationConfig.java
@@ -40,6 +40,12 @@ public final class LasIntegrationConfig {
.noDefaultValue()
.withDescription("HMS3 Thrift URI used as the metadata source of truth.");
+ public static final ConfigOption<Duration> CATALOG_SYNC_INTERVAL =
+ ConfigOptions.key(PREFIX + "catalog-sync-interval")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(5))
+ .withDescription("Interval for synchronizing allowlisted HMS catalogs into AMS.");
+
public static final ConfigOption<String> TOS_ENDPOINT =
ConfigOptions.key(PREFIX + "tos-endpoint")
.stringType()
@@ -88,6 +94,12 @@ public final class LasIntegrationConfig {
.defaultValue(1000)
.withDescription("Maximum number of role credentials cached by AMS.");
+ public static final ConfigOption<String> IAM_DATA_ROLE_NAME =
+ ConfigOptions.key(PREFIX + "iam.data-role-name")
+ .stringType()
+ .defaultValue("ServiceRoleForLAS")
+ .withDescription("Tenant IAM role name used by Proton to access TOS data.");
+
public static final ConfigOption<String> EMR_SERVERLESS_ENDPOINT =
ConfigOptions.key(PREFIX + "emr-serverless-endpoint")
.stringType()
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/las/LasIntegrationContext.java b/amoro-ams/src/main/java/org/apache/amoro/server/las/LasIntegrationContext.java
index 94dcaf911..7c558747b 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/las/LasIntegrationContext.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/las/LasIntegrationContext.java
@@ -79,9 +79,11 @@ public final class LasIntegrationContext {
requiredString(configurations, LasIntegrationConfig.IAM_BOOTSTRAP_ACCESS_KEY);
requiredString(configurations, LasIntegrationConfig.IAM_BOOTSTRAP_SECRET_KEY);
requiredString(configurations, LasIntegrationConfig.IAM_ROLE_SESSION_NAME);
+ requiredString(configurations, LasIntegrationConfig.IAM_DATA_ROLE_NAME);
positiveDuration(configurations, LasIntegrationConfig.CONNECT_TIMEOUT);
positiveDuration(configurations, LasIntegrationConfig.READ_TIMEOUT);
positiveDuration(configurations, LasIntegrationConfig.IAM_ASSUME_ROLE_TTL);
+ positiveDuration(configurations, LasIntegrationConfig.CATALOG_SYNC_INTERVAL);
if (configurations.getInteger(LasIntegrationConfig.IAM_CREDENTIAL_CACHE_SIZE) <= 0) {
throw new IllegalArgumentException(
LasIntegrationConfig.IAM_CREDENTIAL_CACHE_SIZE.key() + " must be greater than zero");
@@ -155,6 +157,11 @@ public final class LasIntegrationContext {
return hmsUri;
}
+ public URI tosEndpoint() {
+ ensureEnabled();
+ return tosEndpoint;
+ }
+
public URI iamEndpoint() {
ensureEnabled();
return iamEndpoint;
@@ -191,6 +198,16 @@ public final class LasIntegrationContext {
return configurations.get(LasIntegrationConfig.IAM_ASSUME_ROLE_TTL);
}
+ public String iamDataRoleName() {
+ ensureEnabled();
+ return configurations.getString(LasIntegrationConfig.IAM_DATA_ROLE_NAME);
+ }
+
+ public Duration catalogSyncInterval() {
+ ensureEnabled();
+ return configurations.get(LasIntegrationConfig.CATALOG_SYNC_INTERVAL);
+ }
+
public int iamCredentialCacheSize() {
ensureEnabled();
return configurations.getInteger(LasIntegrationConfig.IAM_CREDENTIAL_CACHE_SIZE);
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/las/LasRestExtension.java b/amoro-ams/src/main/java/org/apache/amoro/server/las/LasRestExtension.java
index 224d29170..bb6bb2693 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/las/LasRestExtension.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/las/LasRestExtension.java
@@ -85,6 +85,7 @@ public class LasRestExtension implements RestExtension {
private CatalogManager catalogManager;
private TableManager tableManager;
private LasIamClient iamClient;
+ private LasCatalogSynchronizer catalogSynchronizer;
@Override
public RestExtensionFactory withServiceConfig(Configurations serviceConfig) {
@@ -116,6 +117,12 @@ public class LasRestExtension implements RestExtension {
iamClient = new LasIamClient(context);
hmsClient = new LasHmsClient(context);
sparkSqlManager = new ServerlessSparkSqlManager(context, iamClient);
+ if (catalogManager.supportNamespace()) {
+ catalogSynchronizer = new LasCatalogSynchronizer(context, hmsClient, catalogManager);
+ catalogSynchronizer.start();
+ } else {
+ LOG.info("LAS HMS catalog synchronization is disabled because namespaces are disabled");
+ }
}
return new LasRestExtension(
context, hmsClient, sparkSqlManager, catalogManager, tableManager);
@@ -128,6 +135,10 @@ public class LasRestExtension implements RestExtension {
@Override
public void close() {
+ if (catalogSynchronizer != null) {
+ catalogSynchronizer.close();
+ catalogSynchronizer = null;
+ }
if (iamClient != null) {
try {
iamClient.close();
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/SqlSessionFactoryProvider.java b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/SqlSessionFactoryProvider.java
index c6cb1bd53..03383dc2b 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/SqlSessionFactoryProvider.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/SqlSessionFactoryProvider.java
@@ -28,6 +28,7 @@ import org.apache.amoro.server.persistence.mapper.ApiTokensMapper;
import org.apache.amoro.server.persistence.mapper.BucketAssignMapper;
import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper;
import org.apache.amoro.server.persistence.mapper.HaLeaseMapper;
+import org.apache.amoro.server.persistence.mapper.NamespaceAllowlistMapper;
import org.apache.amoro.server.persistence.mapper.OptimizerMapper;
import org.apache.amoro.server.persistence.mapper.OptimizingProcessMapper;
import org.apache.amoro.server.persistence.mapper.PlatformFileMapper;
@@ -78,6 +79,7 @@ public class SqlSessionFactoryProvider {
configuration.addMapper(TableRuntimeMapper.class);
configuration.addMapper(HaLeaseMapper.class);
configuration.addMapper(BucketAssignMapper.class);
+ configuration.addMapper(NamespaceAllowlistMapper.class);
PageInterceptor interceptor = new PageInterceptor();
Properties interceptorProperties = new Properties();
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/NamespaceAllowlistMapper.java b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/NamespaceAllowlistMapper.java
new file mode 100644
index 000000000..4d2c867d6
--- /dev/null
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/NamespaceAllowlistMapper.java
@@ -0,0 +1,43 @@
+/*
+ * 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.amoro.server.persistence.mapper;
+
+import org.apache.ibatis.annotations.Delete;
+import org.apache.ibatis.annotations.Insert;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
+
+import java.util.List;
+
+/** Persistence mapper for namespaces that are allowed to synchronize external catalogs. */
+public interface NamespaceAllowlistMapper {
+ String TABLE_NAME = "namespace_allowlist";
+
+ @Select("SELECT namespace FROM " + TABLE_NAME + " ORDER BY namespace")
+ List<String> listNamespaces();
+
+ @Select("SELECT namespace FROM " + TABLE_NAME + " WHERE namespace = #{namespace}")
+ String getNamespace(@Param("namespace") String namespace);
+
+ @Insert("INSERT INTO " + TABLE_NAME + " (namespace) VALUES (#{namespace})")
+ void insertNamespace(@Param("namespace") String namespace);
+
+ @Delete("DELETE FROM " + TABLE_NAME + " WHERE namespace = #{namespace}")
+ int deleteNamespace(@Param("namespace") String namespace);
+}
diff --git a/amoro-ams/src/main/resources/authorization/privilege_mapping.yaml b/amoro-ams/src/main/resources/authorization/privilege_mapping.yaml
index b0a32fe9a..aa6fe981c 100644
--- a/amoro-ams/src/main/resources/authorization/privilege_mapping.yaml
+++ b/amoro-ams/src/main/resources/authorization/privilege_mapping.yaml
@@ -39,6 +39,21 @@ mappings:
resource-type: CATALOG
privilege: MANAGE_CATALOG
+ - prefixes:
+ - /api/ams/v1/namespaces
+ methods:
+ - GET
+ resource-type: CATALOG
+ privilege: VIEW_CATALOG
+
+ - prefixes:
+ - /api/ams/v1/namespaces
+ methods:
+ - PUT
+ - DELETE
+ resource-type: CATALOG
+ privilege: MANAGE_CATALOG
+
- prefixes:
- /api/ams/v1/tables
methods:
diff --git a/amoro-ams/src/main/resources/derby/ams-derby-init.sql b/amoro-ams/src/main/resources/derby/ams-derby-init.sql
index 72c43f4d8..1e9628ee8 100644
--- a/amoro-ams/src/main/resources/derby/ams-derby-init.sql
+++ b/amoro-ams/src/main/resources/derby/ams-derby-init.sql
@@ -276,3 +276,7 @@ CREATE TABLE bucket_assignments (
node_heartbeat_ts BIGINT NOT NULL DEFAULT 0,
PRIMARY KEY (cluster_name, node_key)
);
+
+CREATE TABLE namespace_allowlist (
+ namespace VARCHAR(128) NOT NULL PRIMARY KEY
+);
diff --git a/amoro-ams/src/main/resources/derby/upgrade.sql b/amoro-ams/src/main/resources/derby/upgrade.sql
index 0282c32ec..ccc8efee0 100644
--- a/amoro-ams/src/main/resources/derby/upgrade.sql
+++ b/amoro-ams/src/main/resources/derby/upgrade.sql
@@ -14,3 +14,7 @@
-- limitations under the License.
ALTER TABLE database_metadata ADD COLUMN properties CLOB(64m);
+
+CREATE TABLE namespace_allowlist (
+ namespace VARCHAR(128) NOT NULL PRIMARY KEY
+);
diff --git a/amoro-ams/src/main/resources/mysql/ams-mysql-init.sql b/amoro-ams/src/main/resources/mysql/ams-mysql-init.sql
index b7c423776..5d391b1c5 100644
--- a/amoro-ams/src/main/resources/mysql/ams-mysql-init.sql
+++ b/amoro-ams/src/main/resources/mysql/ams-mysql-init.sql
@@ -292,3 +292,8 @@ CREATE TABLE IF NOT EXISTS bucket_assignments (
node_heartbeat_ts BIGINT NOT NULL DEFAULT 0 COMMENT 'Per-node heartbeat timestamp updated only by the owning node (ms since epoch)',
PRIMARY KEY (cluster_name, node_key)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='Bucket ID assignments per AMS node for master-slave mode';
+
+CREATE TABLE IF NOT EXISTS `namespace_allowlist` (
+ `namespace` VARCHAR(128) NOT NULL COMMENT 'Catalog namespace allowed for synchronization',
+ PRIMARY KEY (`namespace`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='Catalog namespace synchronization allowlist';
diff --git a/amoro-ams/src/main/resources/mysql/upgrade.sql b/amoro-ams/src/main/resources/mysql/upgrade.sql
index e8752b1ca..349ee074e 100644
--- a/amoro-ams/src/main/resources/mysql/upgrade.sql
+++ b/amoro-ams/src/main/resources/mysql/upgrade.sql
@@ -178,3 +178,7 @@ ALTER TABLE `bucket_assignments` ADD COLUMN `node_heartbeat_ts` BIGINT NOT NULL
-- ADD properties to table database_metadata
ALTER TABLE `database_metadata` ADD COLUMN `properties` MEDIUMTEXT COMMENT 'Database properties';
+CREATE TABLE IF NOT EXISTS `namespace_allowlist` (
+ `namespace` VARCHAR(128) NOT NULL COMMENT 'Catalog namespace allowed for synchronization',
+ PRIMARY KEY (`namespace`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='Catalog namespace synchronization allowlist';
diff --git a/amoro-ams/src/main/resources/openapi/openapi.yaml b/amoro-ams/src/main/resources/openapi/openapi.yaml
index 2a37aceb9..1824e8867 100644
--- a/amoro-ams/src/main/resources/openapi/openapi.yaml
+++ b/amoro-ams/src/main/resources/openapi/openapi.yaml
@@ -39,11 +39,69 @@ tags:
- name: Overview
description: Overview operations
paths:
+ /api/ams/v1/namespaces:
+ get:
+ tags:
+ - Catalogs
+ summary: List catalog namespaces
+ description: Returns default when namespace support is disabled, otherwise the namespace allowlist.
+ responses:
+ '200':
+ description: Successful response
+ content:
+ application/json:
+ schema:
+ allOf:
+ - $ref: '#/components/schemas/Response'
+ - type: object
+ properties:
+ result:
+ type: array
+ items:
+ type: string
+ /api/ams/v1/namespaces/{namespace}:
+ parameters:
+ - name: namespace
+ in: path
+ required: true
+ schema:
+ type: string
+ description: The namespace allowlist entry
+ put:
+ tags:
+ - Catalogs
+ summary: Add a namespace to the allowlist
+ responses:
+ '200':
+ description: Namespace was added or already existed
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/Response'
+ delete:
+ tags:
+ - Catalogs
+ summary: Remove a namespace from the allowlist
+ description: Stops future synchronization and hides its catalogs without treating them as deleted in HMS.
+ responses:
+ '200':
+ description: Namespace was removed or did not exist
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/Response'
/api/ams/v1/catalogs:
get:
tags:
- Catalogs
summary: Get list of catalogs
+ parameters:
+ - name: namespace
+ in: query
+ required: false
+ schema:
+ type: string
+ description: Optional namespace filter. Omitting it preserves the legacy all-catalog response.
responses:
'200':
description: Successful response
@@ -979,4 +1037,4 @@ components:
storageConfigs:
type: object
additionalProperties:
- type: string
\ No newline at end of file
+ type: string
diff --git a/amoro-ams/src/main/resources/postgres/ams-postgres-init.sql b/amoro-ams/src/main/resources/postgres/ams-postgres-init.sql
index 5f1494bc0..87ca82fa9 100644
--- a/amoro-ams/src/main/resources/postgres/ams-postgres-init.sql
+++ b/amoro-ams/src/main/resources/postgres/ams-postgres-init.sql
@@ -477,3 +477,8 @@ COMMENT ON COLUMN server_info_json IS 'JSON encoded server info (AmsServerInfo)'
COMMENT ON COLUMN lease_expire_ts IS 'Lease expiration timestamp (ms since epoch)';
COMMENT ON COLUMN version IS 'Optimistic lock version of the lease row';
COMMENT ON COLUMN updated_at IS 'Last update timestamp (ms since epoch)';
+
+CREATE TABLE IF NOT EXISTS namespace_allowlist (
+ namespace VARCHAR(128) PRIMARY KEY
+);
+COMMENT ON TABLE namespace_allowlist IS 'Catalog namespace synchronization allowlist';
diff --git a/amoro-ams/src/main/resources/postgres/upgrade.sql b/amoro-ams/src/main/resources/postgres/upgrade.sql
index 8c290769e..cba35c8fb 100644
--- a/amoro-ams/src/main/resources/postgres/upgrade.sql
+++ b/amoro-ams/src/main/resources/postgres/upgrade.sql
@@ -240,3 +240,7 @@ ALTER TABLE bucket_assignments ADD COLUMN node_heartbeat_ts BIGINT NOT NULL DEFA
-- ADD properties to table database_metadata
ALTER TABLE database_metadata ADD COLUMN properties text;
COMMENT ON COLUMN database_metadata.properties IS 'Database properties';
+
+CREATE TABLE IF NOT EXISTS namespace_allowlist (
+ namespace VARCHAR(128) PRIMARY KEY
+);
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/catalog/TestDefaultCatalogManagerNamespaces.java b/amoro-ams/src/test/java/org/apache/amoro/server/catalog/TestDefaultCatalogManagerNamespaces.java
new file mode 100644
index 000000000..af8dbed85
--- /dev/null
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/catalog/TestDefaultCatalogManagerNamespaces.java
@@ -0,0 +1,130 @@
+/*
+ * 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.amoro.server.catalog;
+
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.api.CatalogMeta;
+import org.apache.amoro.catalog.CatalogTestHelpers;
+import org.apache.amoro.config.Configurations;
+import org.apache.amoro.properties.CatalogMetaProperties;
+import org.apache.amoro.server.AmoroManagementConf;
+import org.apache.amoro.server.persistence.PersistentBase;
+import org.apache.amoro.server.persistence.mapper.CatalogMetaMapper;
+import org.apache.amoro.server.persistence.mapper.NamespaceAllowlistMapper;
+import org.apache.amoro.server.table.DerbyPersistence;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class TestDefaultCatalogManagerNamespaces {
+
+ private static final TestPersistence PERSISTENCE = new TestPersistence();
+ private static final DirectDbAccess DB = new DirectDbAccess();
+
+ @BeforeAll
+ public static void initializePersistence() {
+ PERSISTENCE.initialize();
+ }
+
+ @AfterEach
+ public void cleanUp() {
+ DB.deleteCatalog("tenant-a@las");
+ DB.deleteCatalog("tenant-b@las");
+ DB.deleteNamespace("tenant-a");
+ DB.deleteNamespace("tenant-b");
+ }
+
+ @Test
+ public void testCommunityDefaultsRemainCompatible() {
+ DefaultCatalogManager manager = new DefaultCatalogManager(new Configurations());
+
+ Assertions.assertFalse(manager.supportNamespace());
+ Assertions.assertEquals(Collections.singletonList("default"), manager.listNamespaces());
+ Assertions.assertEquals(manager.listCatalogMetas(), manager.listCatalogMetas("default"));
+ Assertions.assertTrue(manager.listCatalogMetas("unknown").isEmpty());
+ }
+
+ @Test
+ public void testEnabledNamespaceAllowlistAndCatalogFiltering() {
+ Configurations configurations = new Configurations();
+ configurations.setBoolean(AmoroManagementConf.CATALOG_NAMESPACE_ENABLED, true);
+ DefaultCatalogManager manager = new DefaultCatalogManager(configurations);
+
+ Assertions.assertTrue(manager.listNamespaces().isEmpty());
+ manager.addNamespace("tenant-b");
+ manager.addNamespace("tenant-a");
+ manager.addNamespace("tenant-a");
+ Assertions.assertEquals(Arrays.asList("tenant-a", "tenant-b"), manager.listNamespaces());
+
+ DB.insertCatalog(catalog("tenant-a@las", "tenant-a"));
+ DB.insertCatalog(catalog("tenant-b@las", "tenant-b"));
+
+ Assertions.assertEquals(
+ Collections.singletonList("tenant-a@las"),
+ catalogNames(manager.listCatalogMetas("tenant-a")));
+ Assertions.assertEquals(
+ Collections.singletonList("tenant-b@las"),
+ catalogNames(manager.listCatalogMetas("tenant-b")));
+ Assertions.assertTrue(manager.listCatalogMetas("unknown").isEmpty());
+ Assertions.assertEquals(2, manager.listCatalogMetas().size());
+
+ manager.removeNamespace("tenant-a");
+ manager.removeNamespace("tenant-a");
+ Assertions.assertEquals(Collections.singletonList("tenant-b"), manager.listNamespaces());
+ Assertions.assertTrue(manager.listCatalogMetas("tenant-a").isEmpty());
+ }
+
+ private static CatalogMeta catalog(String catalogName, String namespace) {
+ HashMap<String, String> properties = new HashMap<>();
+ properties.put(CatalogMetaProperties.NAMESPACE, namespace);
+ return CatalogTestHelpers.buildCatalogMeta(
+ catalogName, CatalogMetaProperties.CATALOG_TYPE_HIVE, properties, TableFormat.ICEBERG);
+ }
+
+ private static List<String> catalogNames(List<CatalogMeta> catalogs) {
+ return catalogs.stream().map(CatalogMeta::getCatalogName).collect(Collectors.toList());
+ }
+
+ private static class TestPersistence extends DerbyPersistence {
+ void initialize() {
+ // Class initialization creates the shared Derby schema used by persistence tests.
+ }
+ }
+
+ private static class DirectDbAccess extends PersistentBase {
+ void insertCatalog(CatalogMeta catalogMeta) {
+ doAs(CatalogMetaMapper.class, mapper -> mapper.insertCatalog(catalogMeta));
+ }
+
+ void deleteCatalog(String catalogName) {
+ doAs(CatalogMetaMapper.class, mapper -> mapper.deleteCatalog(catalogName));
+ }
+
+ void deleteNamespace(String namespace) {
+ doAs(NamespaceAllowlistMapper.class, mapper -> mapper.deleteNamespace(namespace));
+ }
+ }
+}
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/dashboard/controller/TestTableControllerNamespaces.java b/amoro-ams/src/test/java/org/apache/amoro/server/dashboard/controller/TestTableControllerNamespaces.java
new file mode 100644
index 000000000..ffa3d9e54
--- /dev/null
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/dashboard/controller/TestTableControllerNamespaces.java
@@ -0,0 +1,113 @@
+/*
+ * 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.amoro.server.dashboard.controller;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import io.javalin.http.Context;
+import org.apache.amoro.api.CatalogMeta;
+import org.apache.amoro.config.Configurations;
+import org.apache.amoro.server.catalog.CatalogManager;
+import org.apache.amoro.server.dashboard.ServerTableDescriptor;
+import org.apache.amoro.server.dashboard.response.OkResponse;
+import org.apache.amoro.server.table.TableManager;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.Collections;
+
+public class TestTableControllerNamespaces {
+
+ @Test
+ public void testCatalogEndpointWithoutNamespaceKeepsLegacyCall() {
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ CatalogMeta catalog = new CatalogMeta();
+ when(catalogManager.listCatalogMetas()).thenReturn(Collections.singletonList(catalog));
+ Context context = context();
+ when(context.queryParam("namespace")).thenReturn(null);
+
+ controller(catalogManager).getCatalogs(context);
+
+ verify(catalogManager).listCatalogMetas();
+ verify(catalogManager, never()).listCatalogMetas(any());
+ Assertions.assertEquals(
+ Collections.singletonList(catalog), capturedResponse(context).getResult());
+ }
+
+ @Test
+ public void testCatalogEndpointFiltersByNamespace() {
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ CatalogMeta catalog = new CatalogMeta();
+ when(catalogManager.listCatalogMetas("tenant-a"))
+ .thenReturn(Collections.singletonList(catalog));
+ Context context = context();
+ when(context.queryParam("namespace")).thenReturn(" tenant-a ");
+
+ controller(catalogManager).getCatalogs(context);
+
+ verify(catalogManager).listCatalogMetas("tenant-a");
+ Assertions.assertEquals(
+ Collections.singletonList(catalog), capturedResponse(context).getResult());
+ }
+
+ @Test
+ public void testNamespaceAllowlistEndpointsDelegateToCatalogManager() {
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ when(catalogManager.listNamespaces()).thenReturn(Collections.singletonList("tenant-a"));
+ Context listContext = context();
+ controller(catalogManager).getNamespaces(listContext);
+ Assertions.assertEquals(
+ Collections.singletonList("tenant-a"), capturedResponse(listContext).getResult());
+
+ Context putContext = context();
+ when(putContext.pathParam("namespace")).thenReturn("tenant-a");
+ controller(catalogManager).addNamespace(putContext);
+ verify(catalogManager).addNamespace("tenant-a");
+
+ Context deleteContext = context();
+ when(deleteContext.pathParam("namespace")).thenReturn("tenant-a");
+ controller(catalogManager).removeNamespace(deleteContext);
+ verify(catalogManager).removeNamespace("tenant-a");
+ }
+
+ private static TableController controller(CatalogManager catalogManager) {
+ return new TableController(
+ catalogManager,
+ mock(TableManager.class),
+ mock(ServerTableDescriptor.class),
+ new Configurations());
+ }
+
+ private static Context context() {
+ Context context = mock(Context.class);
+ when(context.json(any())).thenReturn(context);
+ return context;
+ }
+
+ private static OkResponse<?> capturedResponse(Context context) {
+ ArgumentCaptor<Object> response = ArgumentCaptor.forClass(Object.class);
+ verify(context).json(response.capture());
+ return (OkResponse<?>) response.getValue();
+ }
+}
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/las/TestLasCatalogSynchronizer.java b/amoro-ams/src/test/java/org/apache/amoro/server/las/TestLasCatalogSynchronizer.java
new file mode 100644
index 000000000..24af35cc7
--- /dev/null
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/las/TestLasCatalogSynchronizer.java
@@ -0,0 +1,245 @@
+/*
+ * 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.amoro.server.las;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import org.apache.amoro.api.CatalogMeta;
+import org.apache.amoro.client.ClientPool;
+import org.apache.amoro.hive.HMSClient;
+import org.apache.amoro.hive.HMSClientPool;
+import org.apache.amoro.properties.CatalogMetaProperties;
+import org.apache.amoro.server.catalog.CatalogManager;
+import org.apache.amoro.table.TableProperties;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.thrift.TException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.io.ByteArrayInputStream;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public class TestLasCatalogSynchronizer {
+
+ @Test
+ public void testParseNamespaceUsesFirstSeparator() {
+ Assertions.assertEquals(
+ "tenant-a", LasCatalogSynchronizer.parseNamespace("tenant-a@las@archive").orElse(null));
+ Assertions.assertFalse(LasCatalogSynchronizer.parseNamespace("las").isPresent());
+ Assertions.assertFalse(LasCatalogSynchronizer.parseNamespace("@las").isPresent());
+ Assertions.assertFalse(LasCatalogSynchronizer.parseNamespace("tenant-a@").isPresent());
+ Assertions.assertFalse(
+ LasCatalogSynchronizer.parseNamespace(
+ "tenant-a@" + String.join("", Collections.nCopies(64, "x")))
+ .isPresent());
+ Assertions.assertFalse(LasCatalogSynchronizer.parseNamespace(null).isPresent());
+ }
+
+ @Test
+ public void testSyncUsesOneSnapshotAndProtectsManualAndFilteredCatalogs() throws Exception {
+ HMSClient hmsSdk = mock(HMSClient.class);
+ when(hmsSdk.getCatalogs())
+ .thenReturn(
+ Arrays.asList("tenant-a@las", "tenant-a@manual", "tenant-b@keep", "invalid-catalog"));
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ when(catalogManager.listNamespaces()).thenReturn(Collections.singletonList("tenant-a"));
+ CatalogMeta manual = catalog("tenant-a@manual", "tenant-a", false);
+ CatalogMeta removedFromAllowlist = catalog("tenant-b@keep", "tenant-b", true);
+ CatalogMeta stale = catalog("tenant-z@gone", "tenant-z", true);
+ when(catalogManager.listCatalogMetas())
+ .thenReturn(Arrays.asList(manual, removedFromAllowlist, stale));
+
+ LasCatalogSynchronizer synchronizer = synchronizer(hmsSdk, catalogManager);
+ synchronizer.syncOnce();
+
+ verify(hmsSdk, times(1)).getCatalogs();
+ ArgumentCaptor<CatalogMeta> created = ArgumentCaptor.forClass(CatalogMeta.class);
+ verify(catalogManager).createCatalog(created.capture());
+ Assertions.assertEquals("tenant-a@las", created.getValue().getCatalogName());
+ Assertions.assertEquals(
+ CatalogMetaProperties.CATALOG_TYPE_HIVE, created.getValue().getCatalogType());
+ Assertions.assertEquals(
+ "tenant-a", created.getValue().getCatalogProperties().get(CatalogMetaProperties.NAMESPACE));
+ Assertions.assertEquals(
+ "false",
+ created
+ .getValue()
+ .getCatalogProperties()
+ .get(
+ CatalogMetaProperties.TABLE_PROPERTIES_PREFIX
+ + TableProperties.ENABLE_SELF_OPTIMIZING));
+ Configuration hiveConfiguration =
+ decodeConfiguration(
+ created
+ .getValue()
+ .getStorageConfigs()
+ .get(CatalogMetaProperties.STORAGE_CONFIGS_KEY_HIVE_SITE));
+ Assertions.assertEquals(
+ "thrift://hms-service:9083", hiveConfiguration.get("hive.metastore.uris"));
+ Assertions.assertEquals("tenant-a@las", hiveConfiguration.get("metastore.catalog.default"));
+ Configuration tosConfiguration =
+ decodeConfiguration(
+ created
+ .getValue()
+ .getStorageConfigs()
+ .get(CatalogMetaProperties.STORAGE_CONFIGS_KEY_CORE_SITE));
+ Assertions.assertEquals(
+ "https://tos-cn-beijing.volces.com", tosConfiguration.get("fs.tos.endpoint"));
+ Assertions.assertEquals(
+ "io.proton.tos.iam.AssumeIamRoleCredentialProvider",
+ tosConfiguration.get("fs.tos.credentials.provider"));
+ Assertions.assertEquals(
+ "trn:iam::tenant-a:role/ServiceRoleForLAS",
+ tosConfiguration.get("fs.tos.credential.sts.iam-role-trn"));
+ verify(catalogManager).dropCatalog("tenant-z@gone");
+ verify(catalogManager, never()).dropCatalog("tenant-b@keep");
+ verify(catalogManager, never()).updateCatalog(manual);
+ }
+
+ @Test
+ public void testManagedCatalogIsUpdatedIdempotently() throws Exception {
+ HMSClient hmsSdk = mock(HMSClient.class);
+ when(hmsSdk.getCatalogs()).thenReturn(Collections.singletonList("tenant-a@las"));
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ when(catalogManager.listNamespaces()).thenReturn(Collections.singletonList("tenant-a"));
+ CatalogMeta existing = catalog("tenant-a@las", "tenant-a", true);
+ existing.getCatalogProperties().put("preserved", "value");
+ when(catalogManager.listCatalogMetas()).thenReturn(Collections.singletonList(existing));
+
+ LasCatalogSynchronizer synchronizer = synchronizer(hmsSdk, catalogManager);
+ synchronizer.syncOnce();
+
+ ArgumentCaptor<CatalogMeta> updated = ArgumentCaptor.forClass(CatalogMeta.class);
+ verify(catalogManager).updateCatalog(updated.capture());
+ Assertions.assertEquals("value", updated.getValue().getCatalogProperties().get("preserved"));
+ Assertions.assertEquals(
+ "ICEBERG,PAIMON",
+ updated.getValue().getCatalogProperties().get(CatalogMetaProperties.TABLE_FORMATS));
+ when(catalogManager.listCatalogMetas())
+ .thenReturn(Collections.singletonList(updated.getValue()));
+ synchronizer.syncOnce();
+ verify(catalogManager, times(1)).updateCatalog(any());
+ verify(catalogManager, never()).createCatalog(any());
+ verify(catalogManager, never()).dropCatalog(any());
+ }
+
+ @Test
+ public void testHmsFailureDoesNotMutateCatalogState() throws Exception {
+ HMSClient hmsSdk = mock(HMSClient.class);
+ when(hmsSdk.getCatalogs()).thenThrow(new TException("unavailable"));
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ LasCatalogSynchronizer synchronizer = synchronizer(hmsSdk, catalogManager);
+
+ Assertions.assertThrows(TException.class, synchronizer::syncOnce);
+ verifyNoInteractions(catalogManager);
+ }
+
+ @Test
+ public void testInvalidHmsSnapshotDoesNotMutateCatalogState() throws Exception {
+ HMSClient hmsSdk = mock(HMSClient.class);
+ when(hmsSdk.getCatalogs()).thenReturn(Arrays.asList("tenant-a@las", null));
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ LasCatalogSynchronizer synchronizer = synchronizer(hmsSdk, catalogManager);
+
+ Assertions.assertThrows(IllegalStateException.class, synchronizer::syncOnce);
+ verifyNoInteractions(catalogManager);
+ }
+
+ @Test
+ public void testFirstRunIsDelayedByOneFiveMinuteInterval() {
+ HMSClient hmsSdk = mock(HMSClient.class);
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ ScheduledExecutorService scheduler = mock(ScheduledExecutorService.class);
+ LasIntegrationContext context =
+ LasIntegrationContext.initialize(TestLasIntegrationContext.validConfigurations());
+ LasCatalogSynchronizer synchronizer =
+ new LasCatalogSynchronizer(
+ context,
+ new LasHmsClient(directPool(hmsSdk), ignored -> directPool(hmsSdk)),
+ catalogManager,
+ scheduler);
+
+ synchronizer.start();
+ synchronizer.start();
+
+ verify(scheduler, times(1))
+ .scheduleWithFixedDelay(
+ any(Runnable.class), eq(300_000L), eq(300_000L), eq(TimeUnit.MILLISECONDS));
+ verifyNoInteractions(hmsSdk);
+ }
+
+ private static LasCatalogSynchronizer synchronizer(
+ HMSClient hmsSdk, CatalogManager catalogManager) {
+ LasIntegrationContext context =
+ LasIntegrationContext.initialize(TestLasIntegrationContext.validConfigurations());
+ HMSClientPool pool = directPool(hmsSdk);
+ return new LasCatalogSynchronizer(
+ context, new LasHmsClient(pool, ignored -> pool), catalogManager);
+ }
+
+ private static CatalogMeta catalog(String name, String namespace, boolean managed) {
+ Map<String, String> properties = new HashMap<>();
+ properties.put(CatalogMetaProperties.NAMESPACE, namespace);
+ if (managed) {
+ properties.put(
+ LasCatalogSynchronizer.CATALOG_SOURCE, LasCatalogSynchronizer.CATALOG_SOURCE_LAS_HMS);
+ }
+ return new CatalogMeta(
+ name,
+ CatalogMetaProperties.CATALOG_TYPE_HIVE,
+ new HashMap<>(),
+ new HashMap<>(),
+ properties);
+ }
+
+ private static HMSClientPool directPool(HMSClient client) {
+ return new HMSClientPool() {
+ @Override
+ public <R> R run(ClientPool.Action<R, HMSClient, TException> action) throws TException {
+ return action.run(client);
+ }
+
+ @Override
+ public <R> R run(ClientPool.Action<R, HMSClient, TException> action, boolean retry)
+ throws TException {
+ return action.run(client);
+ }
+ };
+ }
+
+ private static Configuration decodeConfiguration(String encoded) {
+ Configuration configuration = new Configuration(false);
+ configuration.addResource(new ByteArrayInputStream(Base64.getDecoder().decode(encoded)));
+ return configuration;
+ }
+}
diff --git a/amoro-common/src/main/java/org/apache/amoro/properties/CatalogMetaProperties.java b/amoro-common/src/main/java/org/apache/amoro/properties/CatalogMetaProperties.java
index 10f055b80..df324b891 100644
--- a/amoro-common/src/main/java/org/apache/amoro/properties/CatalogMetaProperties.java
+++ b/amoro-common/src/main/java/org/apache/amoro/properties/CatalogMetaProperties.java
@@ -69,6 +69,9 @@ public class CatalogMetaProperties {
public static final String TABLE_FORMATS = "table-formats";
+ /** Optional namespace used to group catalogs in namespace-aware deployments. */
+ public static final String NAMESPACE = "namespace";
+
public static final String CLIENT_POOL_SIZE = "clients";
public static final int CLIENT_POOL_SIZE_DEFAULT = 20;
diff --git a/amoro-web/src/components/Sidebar.vue b/amoro-web/src/components/Sidebar.vue
index f5cf27526..058645e78 100644
--- a/amoro-web/src/components/Sidebar.vue
+++ b/amoro-web/src/components/Sidebar.vue
@@ -146,11 +146,18 @@ export default defineComponent({
let catalog: string | undefined
let db: string | undefined
let tableName: string | undefined
+ let namespace: string | undefined
try {
const stored = localStorage.getItem('easylake-menu-catalog-db-table')
if (stored) {
- const parsed = JSON.parse(stored) as { catalog?: string; database?: string; tableName?: string }
+ const parsed = JSON.parse(stored) as {
+ namespace?: string
+ catalog?: string
+ database?: string
+ tableName?: string
+ }
+ namespace = parsed.namespace
catalog = parsed.catalog
db = parsed.database
tableName = parsed.tableName
@@ -164,6 +171,7 @@ export default defineComponent({
router.replace({
path: '/tables',
query: {
+ ...(namespace && namespace !== 'default' ? { namespace } : {}),
catalog,
db,
table: tableName,
diff --git a/amoro-web/src/services/table.service.ts b/amoro-web/src/services/table.service.ts
index 0ff5ba928..2a5b51c7b 100644
--- a/amoro-web/src/services/table.service.ts
+++ b/amoro-web/src/services/table.service.ts
@@ -20,8 +20,14 @@
import type { ICatalogItem, IMap } from '@/types/common.type'
import request from '@/utils/request'
-export function getCatalogList(): Promise<ICatalogItem[]> {
- return request.get('api/ams/v1/catalogs')
+export function getNamespaceList(): Promise<string[]> {
+ return request.get('api/ams/v1/namespaces')
+}
+
+export function getCatalogList(namespace?: string): Promise<ICatalogItem[]> {
+ return request.get('api/ams/v1/catalogs', {
+ params: namespace ? { namespace } : undefined,
+ })
}
export function getDatabaseList(params: {
catalog: string
diff --git a/amoro-web/src/views/tables/components/Details.vue b/amoro-web/src/views/tables/components/Details.vue
index 93c26b737..19ce44d96 100644
--- a/amoro-web/src/views/tables/components/Details.vue
+++ b/amoro-web/src/views/tables/components/Details.vue
@@ -17,7 +17,7 @@ limitations under the License.
/ -->
<script setup lang="ts">
-import { computed, onMounted, reactive, shallowReactive, watch } from 'vue'
+import { computed, reactive, shallowReactive, watch } from 'vue'
import { useI18n } from 'vue-i18n'
import { useRoute, useRouter } from 'vue-router'
import type { ColumnProps } from 'ant-design-vue/es/table'
@@ -27,13 +27,14 @@ import { dateFormat } from '@/utils'
const emit = defineEmits<{
(e: 'setBaseDetailInfo', data: IBaseDetailInfo): void
- (e: 'tableNotFound', info: { catalog: string; db: string; table: string }): void
+ (e: 'tableNotFound', info: { catalog: string, db: string, table: string }): void
}>()
const { t } = useI18n()
const route = useRoute()
const router = useRouter()
const STORAGE_TABLE_KEY = 'easylake-menu-catalog-db-table'
+let detailRequestGeneration = 0
const params = computed(() => {
return {
@@ -44,6 +45,7 @@ const params = computed(() => {
watch(
() => route.query,
(val) => {
+ detailRequestGeneration += 1
val?.catalog && route.path === '/tables' && getTableDetails()
},
)
@@ -72,7 +74,7 @@ const state = reactive({
createTime: '',
tableFormat: '',
hasPartition: false, // Whether there is a partition, if there is no partition, the file list will be displayed
- comment: ''
+ comment: '',
} as IBaseDetailInfo,
pkList: [] as DetailColumnItem[],
partitionColumnList: [] as PartitionColumnItem[],
@@ -83,6 +85,7 @@ const state = reactive({
})
async function getTableDetails() {
+ const requestGeneration = ++detailRequestGeneration
const requestParams = { ...params.value }
const { catalog, db, table } = requestParams
if (!catalog || !db || !table) {
@@ -93,6 +96,9 @@ async function getTableDetails() {
const result = await getTableDetail({
...requestParams,
})
+ if (requestGeneration !== detailRequestGeneration) {
+ return
+ }
const { pkList = [], tableType, partitionColumnList = [], properties, changeMetrics, schema, createTime, tableIdentifier, baseMetrics, tableSummary, comment } = result
state.baseDetailInfo = {
...tableSummary,
@@ -100,7 +106,7 @@ async function getTableDetails() {
tableName: `${tableIdentifier?.catalog || ''}.${tableIdentifier?.database || ''}.${tableIdentifier?.tableName || ''}`,
createTime: createTime ? dateFormat(createTime) : '',
hasPartition: !!(partitionColumnList?.length),
- comment: comment || ''
+ comment: comment || '',
}
state.pkList = pkList || []
@@ -129,10 +135,15 @@ async function getTableDetails() {
setBaseDetailInfo()
}
catch (error) {
+ if (requestGeneration !== detailRequestGeneration) {
+ return
+ }
const errorMessage = (error as Error)?.message || ''
const isNotFoundError = /not exist|not found/i.test(errorMessage)
if (isNotFoundError) {
+ const namespace = (route.query.namespace as string) || 'default'
+ localStorage.removeItem(`${STORAGE_TABLE_KEY}:${namespace}`)
localStorage.removeItem(STORAGE_TABLE_KEY)
emit('tableNotFound', {
@@ -141,11 +152,16 @@ async function getTableDetails() {
table: table as string,
})
- router.replace({ path: '/tables', query: {} })
+ router.replace({
+ path: '/tables',
+ query: route.query.namespace ? { namespace: route.query.namespace } : {},
+ })
}
}
finally {
- state.detailLoading = false
+ if (requestGeneration === detailRequestGeneration) {
+ state.detailLoading = false
+ }
}
function setBaseDetailInfo() {
diff --git a/amoro-web/src/views/tables/components/TableExplorer.vue b/amoro-web/src/views/tables/components/TableExplorer.vue
index 96740e9de..f21946c9b 100755
--- a/amoro-web/src/views/tables/components/TableExplorer.vue
+++ b/amoro-web/src/views/tables/components/TableExplorer.vue
@@ -19,7 +19,7 @@ limitations under the License.
<script setup lang="ts">
import { computed, onBeforeMount, reactive, watch } from 'vue'
import { useRoute, useRouter } from 'vue-router'
-import { getCatalogList, getDatabaseList, getTableList } from '@/services/table.service'
+import { getCatalogList, getDatabaseList, getNamespaceList, getTableList } from '@/services/table.service'
import type { ICatalogItem } from '@/types/common.type'
// Node types: Catalog / Database / Table
@@ -47,7 +47,7 @@ const router = useRouter()
const route = useRoute()
const storageTableKey = 'easylake-menu-catalog-db-table'
-const expandedKeysSessionKey = 'tables_expanded_keys'
+const expandedKeysSessionKeyPrefix = 'tables_expanded_keys'
const state = reactive({
loading: false,
@@ -56,16 +56,34 @@ const state = reactive({
treeData: [] as TreeNode[],
expandedKeys: [] as string[],
selectedKeys: [] as string[],
+ namespaces: [] as string[],
+ selectedNamespace: '',
+ namespaceMode: false,
// Cache
catalogList: [] as string[],
dbListByCatalog: {} as Record<string, string[]>,
tablesByCatalogDb: {} as Record<string, TableItem[]>,
})
+let namespaceGeneration = 0
+let initialized = false
+
+function expandedKeysSessionKey() {
+ return `${expandedKeysSessionKeyPrefix}:${state.selectedNamespace || 'default'}`
+}
+
+function catalogDisplayName(catalog: string) {
+ if (!state.namespaceMode || !state.selectedNamespace) {
+ return catalog
+ }
+ const prefix = `${state.selectedNamespace}@`
+ return catalog.startsWith(prefix) ? catalog.slice(prefix.length) : catalog
+}
+
function buildCatalogNode(catalog: string): TreeNode {
return {
key: `catalog:${catalog}`,
- title: catalog,
+ title: catalogDisplayName(catalog),
isLeaf: false,
nodeType: 'catalog',
catalog,
@@ -114,18 +132,42 @@ function updateTreeNodeChildren(targetKey: string, children: TreeNode[]) {
}
async function initRootCatalogs() {
+ if (state.namespaceMode && !state.selectedNamespace) {
+ state.catalogList = []
+ state.treeData = []
+ return
+ }
+ const generation = namespaceGeneration
state.loading = true
try {
- const res = await getCatalogList()
+ const res = await getCatalogList(state.namespaceMode ? state.selectedNamespace : undefined)
+ if (generation !== namespaceGeneration) {
+ return
+ }
const catalogs = (res || []).map((item: ICatalogItem) => item.catalogName)
state.catalogList = catalogs
state.treeData = catalogs.map(catalog => buildCatalogNode(catalog))
}
finally {
- state.loading = false
+ if (generation === namespaceGeneration) {
+ state.loading = false
+ }
}
}
+function clearExplorerState() {
+ namespaceGeneration += 1
+ state.loading = false
+ state.searchKey = ''
+ state.filterKey = ''
+ state.treeData = []
+ state.expandedKeys = []
+ state.selectedKeys = []
+ state.catalogList = []
+ state.dbListByCatalog = {}
+ state.tablesByCatalogDb = {}
+}
+
async function loadChildren(node: any) {
const data = node?.dataRef || node
if (!data) {
@@ -133,6 +175,7 @@ async function loadChildren(node: any) {
}
const nodeType = data.nodeType as NodeType
+ const generation = namespaceGeneration
if (nodeType === 'catalog') {
const catalog = data.catalog as string
if (!catalog || state.dbListByCatalog[catalog]) {
@@ -142,6 +185,9 @@ async function loadChildren(node: any) {
state.loading = true
try {
const res = await getDatabaseList({ catalog, keywords: '' })
+ if (generation !== namespaceGeneration) {
+ return
+ }
const dbs = (res || []) as string[]
state.dbListByCatalog[catalog] = dbs
if (!dbs.length) {
@@ -171,6 +217,9 @@ async function loadChildren(node: any) {
state.loading = true
try {
const res = await getTableList({ catalog, db, keywords: '' })
+ if (generation !== namespaceGeneration) {
+ return
+ }
const tables = (res || []) as TableItem[]
state.tablesByCatalogDb[cacheKey] = tables
if (!tables.length) {
@@ -217,7 +266,7 @@ async function expandPathBySelected(catalog: string, db: string) {
state.expandedKeys = Array.from(nextExpandedKeys)
try {
- sessionStorage.setItem(expandedKeysSessionKey, JSON.stringify(state.expandedKeys))
+ sessionStorage.setItem(expandedKeysSessionKey(), JSON.stringify(state.expandedKeys))
}
catch (e) {
// ignore sessionStorage write errors
@@ -231,16 +280,21 @@ function handleSelectTable(catalog: string, db: string, tableName: string, table
const type = tableType || 'MIXED_ICEBERG'
- localStorage.setItem(storageTableKey, JSON.stringify({
+ const namespace = state.namespaceMode ? state.selectedNamespace : 'default'
+ const storedSelection = JSON.stringify({
+ namespace,
catalog,
database: db,
tableName,
- }))
+ })
+ localStorage.setItem(`${storageTableKey}:${namespace}`, storedSelection)
+ localStorage.setItem(storageTableKey, storedSelection)
const path = type === 'HIVE' ? '/hive-tables' : '/tables'
const pathQuery = {
path,
query: {
+ ...(state.namespaceMode ? { namespace: state.selectedNamespace } : {}),
catalog,
db,
table: tableName,
@@ -289,7 +343,7 @@ function handleTreeSelect(selectedKeys: (string | number)[], info: any) {
async function handleTreeExpand(expandedKeys: (string | number)[], info: any) {
state.expandedKeys = expandedKeys.map(key => String(key))
try {
- sessionStorage.setItem(expandedKeysSessionKey, JSON.stringify(state.expandedKeys))
+ sessionStorage.setItem(expandedKeysSessionKey(), JSON.stringify(state.expandedKeys))
}
catch (e) {
// ignore sessionStorage write errors
@@ -320,7 +374,7 @@ async function toggleNodeExpand(dataRef: TreeNode) {
state.expandedKeys = nextExpandedKeys
try {
- sessionStorage.setItem(expandedKeysSessionKey, JSON.stringify(state.expandedKeys))
+ sessionStorage.setItem(expandedKeysSessionKey(), JSON.stringify(state.expandedKeys))
}
catch (e) {
// ignore sessionStorage write errors
@@ -455,10 +509,34 @@ watch(
watch(
() => route.query,
async (value, oldValue) => {
- const { catalog, db, table } = value as any
- const { catalog: oldCatalog, db: oldDb, table: oldTable } = (oldValue || {}) as any
+ if (!initialized) {
+ return
+ }
- if (`${catalog || ''}${db || ''}${table || ''}` === `${oldCatalog || ''}${oldDb || ''}${oldTable || ''}`) {
+ const { namespace, catalog, db, table } = value as any
+ const {
+ namespace: oldNamespace,
+ catalog: oldCatalog,
+ db: oldDb,
+ table: oldTable,
+ } = (oldValue || {}) as any
+
+ if (state.namespaceMode && namespace !== oldNamespace) {
+ const nextNamespace = state.namespaces.includes(namespace as string) ? namespace as string : ''
+ if (nextNamespace !== state.selectedNamespace) {
+ clearExplorerState()
+ state.selectedNamespace = nextNamespace
+ if (nextNamespace) {
+ await initRootCatalogs()
+ await restoreExpandedState()
+ }
+ }
+ }
+
+ if (
+ namespace === oldNamespace
+ && `${catalog || ''}${db || ''}${table || ''}` === `${oldCatalog || ''}${oldDb || ''}${oldTable || ''}`
+ ) {
return
}
@@ -496,13 +574,11 @@ const searchResult = computed(() => {
const displayTreeData = computed(() => searchResult.value.tree)
const displayExpandedKeys = computed(() => searchResult.value.expandedKeys)
-onBeforeMount(async () => {
- await initRootCatalogs()
-
+async function restoreExpandedState() {
let restoredExpandedKeys: string[] = []
try {
- const stored = sessionStorage.getItem(expandedKeysSessionKey)
+ const stored = sessionStorage.getItem(expandedKeysSessionKey())
if (stored) {
const parsed = JSON.parse(stored)
if (Array.isArray(parsed)) {
@@ -538,7 +614,9 @@ onBeforeMount(async () => {
state.expandedKeys = restoredExpandedKeys
}
+}
+async function restoreRouteSelection() {
const query = route.query || {}
const queryCatalog = (query.catalog as string) || ''
const queryDb = (query.db as string) || ''
@@ -551,12 +629,72 @@ onBeforeMount(async () => {
state.selectedKeys = [tableKey]
}
}
+ else {
+ state.selectedKeys = []
+ }
+}
+
+async function initializeNamespaces() {
+ let namespaces: string[]
+ try {
+ namespaces = (await getNamespaceList()) || []
+ }
+ catch (e) {
+ // Fail closed: falling back to the legacy catalog endpoint could expose every namespace.
+ namespaces = []
+ }
+
+ state.namespaces = namespaces
+ state.namespaceMode = !(namespaces.length === 1 && namespaces[0] === 'default')
+ if (!state.namespaceMode) {
+ state.selectedNamespace = 'default'
+ return
+ }
+
+ const routeNamespace = (route.query.namespace as string) || ''
+ state.selectedNamespace = namespaces.includes(routeNamespace) ? routeNamespace : ''
+}
+
+async function handleNamespaceChange(namespace?: string) {
+ clearExplorerState()
+ state.selectedNamespace = namespace || ''
+
+ await router.replace({
+ path: route.path,
+ query: state.selectedNamespace ? { namespace: state.selectedNamespace } : {},
+ })
+
+ if (state.selectedNamespace) {
+ await initRootCatalogs()
+ await restoreExpandedState()
+ }
+}
+
+onBeforeMount(async () => {
+ await initializeNamespaces()
+ await initRootCatalogs()
+ await restoreExpandedState()
+ await restoreRouteSelection()
+ initialized = true
})
</script>
<template>
<div class="table-explorer">
<div class="table-explorer-header">
+ <div class="namespace-selector">
+ <span class="namespace-label">Namespace:</span>
+ <a-select
+ v-if="state.namespaceMode"
+ :value="state.selectedNamespace || undefined"
+ :options="state.namespaces.map(namespace => ({ label: namespace, value: namespace }))"
+ placeholder="Select namespace"
+ class="namespace-select"
+ allow-clear
+ @change="handleNamespaceChange"
+ />
+ <span v-else class="namespace-default">default</span>
+ </div>
<a-input
v-model:value="state.searchKey"
placeholder="Search catalog.database.table"
@@ -598,10 +736,10 @@ onBeforeMount(async () => {
</template>
</a-tree>
<div v-else class="empty-placeholder">
- <span>No results..</span>
+ <span v-if="state.namespaceMode && !state.selectedNamespace">Select a namespace to browse tables.</span>
+ <span v-else>No results.</span>
</div>
</div>
-
</div>
</template>
@@ -621,6 +759,37 @@ onBeforeMount(async () => {
padding: 0 7px 0 8px;
margin-bottom: 8px;
+ .namespace-selector {
+ display: flex;
+ align-items: center;
+ min-height: 24px;
+ margin-bottom: 8px;
+ font-size: 13px;
+
+ .namespace-label {
+ margin-right: 6px;
+ color: #666;
+ }
+
+ .namespace-default {
+ color: #262626;
+ }
+
+ .namespace-select {
+ flex: 1;
+ min-width: 0;
+
+ :deep(.ant-select-selector) {
+ height: 24px;
+ }
+
+ :deep(.ant-select-selection-item),
+ :deep(.ant-select-selection-placeholder) {
+ line-height: 22px;
+ }
+ }
+ }
+
.search-input {
width: 100%;
diff --git a/dist/src/main/amoro-bin/conf/config.yaml b/dist/src/main/amoro-bin/conf/config.yaml
index c14d19089..c23a8d4a2 100644
--- a/dist/src/main/amoro-bin/conf/config.yaml
+++ b/dist/src/main/amoro-bin/conf/config.yaml
@@ -82,6 +82,7 @@ ams:
integration:
enabled: false
# hms-uri: thrift://hms-service.<namespace>.svc.cluster.local:9083
+ catalog-sync-interval: 5min
# tos-endpoint: https://tos-cn-beijing.volces.com
# iam-endpoint: https://iam.volcengineapi.com
# emr-serverless-endpoint: https://open.volcengineapi.com
@@ -96,6 +97,7 @@ ams:
role-session-name: AmoroAssumeRoleSession
assume-role-ttl: 1h
credential-cache-size: 1000
+ data-role-name: ServiceRoleForLAS
cross-vpc:
enabled: false
# account-id: <account-id>
@@ -146,6 +148,11 @@ ams:
catalog-meta-cache:
expiration-interval: 60s
+ # Optional namespace level for catalog browsing. LAS deployments enable this and typically set
+ # refresh-external-catalogs.interval to 10min; the community defaults remain unchanged.
+ catalog:
+ namespace-enabled: false
+
# Support for encrypted sensitive configuration items
shade:
identifier: default # Built-in support for default/base64. Defaults to "default", indicating no encryption
diff --git a/docs/configuration/ams-config.md b/docs/configuration/ams-config.md
index 2a6c6b6e8..32003801a 100644
--- a/docs/configuration/ams-config.md
+++ b/docs/configuration/ams-config.md
@@ -46,6 +46,7 @@ table td:last-child, table th:last-child { width: 40%; word-break: break-all; }
| admin-username | admin | The administrator account name. |
| blocker.timeout | 1 min | Session timeout. Default unit is milliseconds if not specified. |
| catalog-meta-cache.expiration-interval | 1 min | TTL for catalog metadata. |
+| catalog.namespace-enabled | false | Whether catalogs are grouped by an optional namespace. |
| database.auto-create-tables | true | Auto init table schema when started |
| database.connection-pool-max-idle | 16 | Max idle connect count of database connect pool. |
| database.connection-pool-max-total | 20 | Max connect count of database connect pool. |
@@ -232,4 +233,3 @@ table td:last-child, table th:last-child { width: 40%; word-break: break-all; }
| shade.identifier | default | The identifier of the encryption method for decryption. Defaults to "default", indicating no encryption |
| shade.sensitive-keywords | admin-password;database.password | A semicolon-separated list of keywords for the configuration items to be decrypted. |
-