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 <[email protected]> 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. | -
