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 269587f9017e51b2dd71ff4c1c0abfb3613fed87 Author: majin.nathan <[email protected]> AuthorDate: Mon Aug 10 16:57:01 2026 +0800 feat: support HMS namespaces and Lance tables --- .../apache/amoro/server/AmoroManagementConf.java | 8 + .../apache/amoro/server/AmoroServiceContainer.java | 17 +- .../amoro/server/catalog/CatalogBuilder.java | 3 +- .../amoro/server/catalog/CatalogManager.java | 5 + .../server/catalog/DefaultCatalogManager.java | 23 +- .../amoro/server/catalog/ExternalCatalog.java | 11 + .../amoro/server/las/LasCatalogSynchronizer.java | 19 +- .../amoro/server/las/LasIntegrationConfig.java | 13 +- .../amoro/server/las/LasIntegrationContext.java | 50 +++- .../TestAmoroServiceContainerEnvironment.java | 40 +++ .../TestDefaultCatalogManagerNamespaces.java | 20 ++ .../server/las/TestLasCatalogSynchronizer.java | 28 +- .../server/las/TestLasIntegrationContext.java | 30 ++ .../org/apache/amoro/CommonUnifiedCatalog.java | 60 ++-- amoro-format-lance/pom.xml | 35 ++- .../amoro/formats/lance/LanceCatalogFactory.java | 4 + .../amoro/formats/lance/LanceHms3Catalog.java | 302 +++++++++++++++++++++ .../formats/lance/LanceStorageOptionsProvider.java | 205 ++++++++++++++ .../formats/lance/TestLanceCatalogFactory.java | 67 +++++ .../amoro/formats/lance/TestLanceHms3Catalog.java | 144 ++++++++++ .../lance/TestLanceHms3CatalogIntegration.java | 53 ++++ .../lance/TestLanceStorageOptionsProvider.java | 91 +++++++ .../amoro/formats/paimon/PaimonCatalogFactory.java | 28 ++ .../formats/paimon/TestPaimonCatalogFactory.java | 74 +++++ amoro-web/mock/modules/catalogs.js | 37 ++- amoro-web/mock/modules/common.js | 6 +- amoro-web/mock/modules/table.js | 35 ++- amoro-web/src/components/Sidebar.vue | 4 + amoro-web/src/views/catalogs/index.vue | 17 +- amoro-web/src/views/hive-details/index.vue | 11 +- .../src/views/tables/components/TableExplorer.vue | 174 +++++++++--- amoro-web/src/views/tables/index.vue | 84 +++--- dist/src/main/amoro-bin/bin/load-config.sh | 10 + dist/src/main/amoro-bin/conf/config.yaml | 11 +- pom.xml | 12 + 35 files changed, 1572 insertions(+), 159 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 c122cee2b..90338dd5d 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 @@ -119,6 +119,14 @@ public class AmoroManagementConf { .defaultValue(true) .withDescription("Whether catalogs are grouped by an optional namespace."); + public static final ConfigOption<Boolean> CATALOG_NAMESPACE_ALLOWLIST_ENABLED = + ConfigOptions.key("catalog.namespace-allowlist-enabled") + .booleanType() + .defaultValue(true) + .withDescription( + "Whether namespace discovery and HMS catalog synchronization are restricted to the" + + " namespace allowlist."); + 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/AmoroServiceContainer.java b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java index b9eb95f2d..c84a1d77d 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java @@ -575,6 +575,21 @@ public class AmoroServiceContainer { .build(); } + @VisibleForTesting + static String redactEnvironmentValue(String key, String value) { + String normalizedKey = key.toUpperCase(Locale.ROOT); + if (normalizedKey.endsWith("_AK") + || normalizedKey.endsWith("_SK") + || normalizedKey.contains("ACCESS_KEY") + || normalizedKey.contains("SECRET") + || normalizedKey.contains("TOKEN") + || normalizedKey.contains("PASSWORD") + || normalizedKey.contains("CREDENTIAL")) { + return "<redacted>"; + } + return value; + } + private class ConfigurationHelper { private JsonNode yamlConfig; @@ -614,7 +629,7 @@ public class AmoroServiceContainer { private Map<String, Object> initEnvConfig() { LOG.info("initializing system env configuration..."); Map<String, String> envs = System.getenv(); - envs.forEach((k, v) -> LOG.info("export {}={}", k, v)); + envs.forEach((k, v) -> LOG.info("export {}={}", k, redactEnvironmentValue(k, v))); String prefix = AmoroManagementConf.SYSTEM_CONFIG.toUpperCase(); return ConfigHelpers.convertConfigurationKeys(prefix, System.getenv()); } diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogBuilder.java b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogBuilder.java index af39ba533..6eedd1e8f 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogBuilder.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogBuilder.java @@ -68,7 +68,8 @@ public class CatalogBuilder { TableFormat.MIXED_ICEBERG, TableFormat.MIXED_HIVE, TableFormat.PAIMON, - TableFormat.HUDI), + TableFormat.HUDI, + TableFormat.LANCE), CATALOG_TYPE_AMS, Sets.newHashSet(TableFormat.ICEBERG, TableFormat.MIXED_ICEBERG)); 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 5a1c05c19..5163d28f9 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 @@ -39,6 +39,11 @@ public interface CatalogManager { return false; } + /** Returns whether namespace discovery is restricted to an explicit allowlist. */ + default boolean namespaceAllowlistEnabled() { + return true; + } + /** Lists namespaces, or the community-compatible default namespace when unsupported. */ default List<String> listNamespaces() { return Collections.singletonList("default"); 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 c815afc8a..6bb1b3f4e 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 @@ -52,6 +52,7 @@ import java.time.Duration; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; @@ -113,11 +114,27 @@ public class DefaultCatalogManager extends PersistentBase implements CatalogMana return serverConfiguration.getBoolean(AmoroManagementConf.CATALOG_NAMESPACE_ENABLED); } + @Override + public boolean namespaceAllowlistEnabled() { + return serverConfiguration.getBoolean(AmoroManagementConf.CATALOG_NAMESPACE_ALLOWLIST_ENABLED); + } + @Override public List<String> listNamespaces() { if (!supportNamespace()) { return Collections.singletonList("default"); } + if (!namespaceAllowlistEnabled()) { + return listCatalogMetas().stream() + .map(CatalogMeta::getCatalogProperties) + .filter(Objects::nonNull) + .map(properties -> properties.get(CatalogMetaProperties.NAMESPACE)) + .filter(Objects::nonNull) + .filter(namespace -> !namespace.trim().isEmpty()) + .distinct() + .sorted() + .collect(Collectors.toList()); + } return getAs(NamespaceAllowlistMapper.class, NamespaceAllowlistMapper::listNamespaces); } @@ -129,7 +146,7 @@ public class DefaultCatalogManager extends PersistentBase implements CatalogMana if (!supportNamespace()) { return "default".equals(namespace) ? listCatalogMetas() : Collections.emptyList(); } - if (!listNamespaces().contains(namespace)) { + if (namespaceAllowlistEnabled() && !listNamespaces().contains(namespace)) { return Collections.emptyList(); } return listCatalogMetas().stream() @@ -144,6 +161,8 @@ public class DefaultCatalogManager extends PersistentBase implements CatalogMana @Override public void addNamespace(String namespace) { Preconditions.checkState(supportNamespace(), "Catalog namespaces are not enabled"); + Preconditions.checkState( + namespaceAllowlistEnabled(), "Catalog namespace allowlist is not enabled"); try { doAs( NamespaceAllowlistMapper.class, @@ -163,6 +182,8 @@ public class DefaultCatalogManager extends PersistentBase implements CatalogMana @Override public void removeNamespace(String namespace) { Preconditions.checkState(supportNamespace(), "Catalog namespaces are not enabled"); + Preconditions.checkState( + namespaceAllowlistEnabled(), "Catalog namespace allowlist is not enabled"); doAs(NamespaceAllowlistMapper.class, mapper -> mapper.deleteNamespace(namespace)); } diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/ExternalCatalog.java b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/ExternalCatalog.java index 8148f64a3..5b17c42df 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/catalog/ExternalCatalog.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/catalog/ExternalCatalog.java @@ -129,6 +129,17 @@ public class ExternalCatalog extends ServerCatalog { return doAs(() -> unifiedCatalog.loadTable(database, tableName)); } + @Override + public void dispose() { + if (unifiedCatalog instanceof AutoCloseable) { + try { + ((AutoCloseable) unifiedCatalog).close(); + } catch (Exception e) { + throw new IllegalStateException("Failed to dispose external catalog " + name(), e); + } + } + } + private void updateDatabaseFilter(CatalogMeta metadata) { String databaseFilter = metadata.getCatalogProperties().get(CatalogMetaProperties.KEY_DATABASE_FILTER); 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 index e52a1f5b4..060742cf7 100644 --- 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 @@ -46,7 +46,7 @@ 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. */ +/** Periodically projects HMS3 catalogs into AMS CatalogMeta records. */ public final class LasCatalogSynchronizer implements AutoCloseable { private static final Logger LOG = LoggerFactory.getLogger(LasCatalogSynchronizer.class); @@ -106,7 +106,9 @@ public final class LasCatalogSynchronizer implements AutoCloseable { } Set<String> rawCatalogNames = new HashSet<>(discoveredCatalogs); - Set<String> allowlistedNamespaces = new HashSet<>(catalogManager.listNamespaces()); + boolean allowlistEnabled = catalogManager.namespaceAllowlistEnabled(); + Set<String> allowlistedNamespaces = + allowlistEnabled ? new HashSet<>(catalogManager.listNamespaces()) : Collections.emptySet(); Map<String, CatalogMeta> existingCatalogs = catalogManager.listCatalogMetas().stream() .collect(Collectors.toMap(CatalogMeta::getCatalogName, catalog -> catalog)); @@ -116,7 +118,8 @@ public final class LasCatalogSynchronizer implements AutoCloseable { int removed = 0; for (String physicalCatalogName : rawCatalogNames) { Optional<String> namespace = parseNamespace(physicalCatalogName); - if (!namespace.isPresent() || !allowlistedNamespaces.contains(namespace.get())) { + if (!namespace.isPresent() + || (allowlistEnabled && !allowlistedNamespaces.contains(namespace.get()))) { continue; } @@ -146,8 +149,10 @@ public final class LasCatalogSynchronizer implements AutoCloseable { } LOG.info( - "LAS HMS catalog synchronization completed: discovered={}, allowlistedNamespaces={}, created={}, updated={}, removed={}", + "LAS HMS catalog synchronization completed: discovered={}, allowlistEnabled={}," + + " allowlistedNamespaces={}, created={}, updated={}, removed={}", rawCatalogNames.size(), + allowlistEnabled, allowlistedNamespaces.size(), created, updated, @@ -182,7 +187,11 @@ public final class LasCatalogSynchronizer implements AutoCloseable { Map<String, String> catalogProperties = new HashMap<>(); catalogProperties.put( CatalogMetaProperties.TABLE_FORMATS, - TableFormat.ICEBERG.name() + "," + TableFormat.PAIMON.name()); + TableFormat.ICEBERG.name() + + "," + + TableFormat.PAIMON.name() + + "," + + TableFormat.LANCE.name()); catalogProperties.put(CatalogMetaProperties.NAMESPACE, namespace); catalogProperties.put(CATALOG_SOURCE, CATALOG_SOURCE_LAS_HMS); catalogProperties.put( 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 65bd5cc6a..4e373fb24 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 @@ -28,6 +28,9 @@ public final class LasIntegrationConfig { private static final String PREFIX = "las.integration."; + public static final String LAS_SERVICE_ACCESS_KEY_ENV = "LAS_SERVICE_AK"; + public static final String LAS_SERVICE_SECRET_KEY_ENV = "LAS_SERVICE_SK"; + public static final ConfigOption<Boolean> ENABLED = ConfigOptions.key(PREFIX + "enabled") .booleanType() @@ -62,13 +65,19 @@ public final class LasIntegrationConfig { ConfigOptions.key(PREFIX + "iam.bootstrap-access-key") .stringType() .noDefaultValue() - .withDescription("Access key of the AMS workload identity used to call AssumeRole."); + .withDescription( + "Optional access key override for the AMS workload identity. LAS deployments use " + + LAS_SERVICE_ACCESS_KEY_ENV + + "."); public static final ConfigOption<String> IAM_BOOTSTRAP_SECRET_KEY = ConfigOptions.key(PREFIX + "iam.bootstrap-secret-key") .stringType() .noDefaultValue() - .withDescription("Secret key of the AMS workload identity used to call AssumeRole."); + .withDescription( + "Optional secret key override for the AMS workload identity. LAS deployments use " + + LAS_SERVICE_SECRET_KEY_ENV + + "."); public static final ConfigOption<String> IAM_BOOTSTRAP_SESSION_TOKEN = ConfigOptions.key(PREFIX + "iam.bootstrap-session-token") 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 7c558747b..ecbd99233 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 @@ -34,6 +34,8 @@ import java.time.Duration; import java.util.Arrays; import java.util.Collections; import java.util.HashSet; +import java.util.Map; +import java.util.Objects; import java.util.Set; /** Validated network configuration and client factories shared by future LAS OpenAPI handlers. */ @@ -50,15 +52,30 @@ public final class LasIntegrationContext { private final String emrServerlessService; private final Duration connectTimeout; private final Duration readTimeout; + private final String bootstrapAccessKey; + private final String bootstrapSecretKey; + private final String bootstrapSessionToken; - private LasIntegrationContext(Configurations configurations) { + private LasIntegrationContext( + Configurations configurations, Map<String, String> environmentVariables) { this.configurations = configurations; + Objects.requireNonNull(environmentVariables, "environmentVariables"); this.enabled = configurations.getBoolean(LasIntegrationConfig.ENABLED); this.region = configurations.getString(LasIntegrationConfig.REGION); this.emrServerlessService = configurations.getString(LasIntegrationConfig.EMR_SERVERLESS_SERVICE); this.connectTimeout = configurations.get(LasIntegrationConfig.CONNECT_TIMEOUT); this.readTimeout = configurations.get(LasIntegrationConfig.READ_TIMEOUT); + this.bootstrapAccessKey = + firstNonBlank( + configurations.getString(LasIntegrationConfig.IAM_BOOTSTRAP_ACCESS_KEY), + environmentVariables.get(LasIntegrationConfig.LAS_SERVICE_ACCESS_KEY_ENV)); + this.bootstrapSecretKey = + firstNonBlank( + configurations.getString(LasIntegrationConfig.IAM_BOOTSTRAP_SECRET_KEY), + environmentVariables.get(LasIntegrationConfig.LAS_SERVICE_SECRET_KEY_ENV)); + this.bootstrapSessionToken = + configurations.getString(LasIntegrationConfig.IAM_BOOTSTRAP_SESSION_TOKEN); if (!enabled) { this.hmsUri = null; @@ -76,8 +93,8 @@ public final class LasIntegrationContext { requiredUri(configurations, LasIntegrationConfig.EMR_SERVERLESS_ENDPOINT, HTTP_SCHEMES); requiredString(configurations, LasIntegrationConfig.REGION); requiredString(configurations, LasIntegrationConfig.EMR_SERVERLESS_SERVICE); - requiredString(configurations, LasIntegrationConfig.IAM_BOOTSTRAP_ACCESS_KEY); - requiredString(configurations, LasIntegrationConfig.IAM_BOOTSTRAP_SECRET_KEY); + requiredCredential(LasIntegrationConfig.LAS_SERVICE_ACCESS_KEY_ENV, bootstrapAccessKey); + requiredCredential(LasIntegrationConfig.LAS_SERVICE_SECRET_KEY_ENV, bootstrapSecretKey); requiredString(configurations, LasIntegrationConfig.IAM_ROLE_SESSION_NAME); requiredString(configurations, LasIntegrationConfig.IAM_DATA_ROLE_NAME); positiveDuration(configurations, LasIntegrationConfig.CONNECT_TIMEOUT); @@ -98,7 +115,12 @@ public final class LasIntegrationContext { } public static LasIntegrationContext initialize(Configurations configurations) { - return new LasIntegrationContext(configurations); + return new LasIntegrationContext(configurations, System.getenv()); + } + + static LasIntegrationContext initialize( + Configurations configurations, Map<String, String> environmentVariables) { + return new LasIntegrationContext(configurations, environmentVariables); } public boolean enabled() { @@ -143,13 +165,9 @@ public final class LasIntegrationContext { public Credential bootstrapCredential() { ensureEnabled(); - String accessKey = configurations.getString(LasIntegrationConfig.IAM_BOOTSTRAP_ACCESS_KEY); - String secretKey = configurations.getString(LasIntegrationConfig.IAM_BOOTSTRAP_SECRET_KEY); - String sessionToken = - configurations.getString(LasIntegrationConfig.IAM_BOOTSTRAP_SESSION_TOKEN); - return StringUtils.isBlank(sessionToken) - ? new Credential(accessKey, secretKey) - : new Credential(accessKey, secretKey, sessionToken); + return StringUtils.isBlank(bootstrapSessionToken) + ? new Credential(bootstrapAccessKey, bootstrapSecretKey) + : new Credential(bootstrapAccessKey, bootstrapSecretKey, bootstrapSessionToken); } public URI hmsUri() { @@ -263,6 +281,16 @@ public final class LasIntegrationContext { return value; } + private static String firstNonBlank(String preferred, String fallback) { + return StringUtils.isNotBlank(preferred) ? preferred : fallback; + } + + private static void requiredCredential(String environmentVariable, String value) { + if (StringUtils.isBlank(value)) { + throw new IllegalArgumentException(environmentVariable + " must be configured"); + } + } + private static Duration positiveDuration( Configurations configurations, ConfigOption<Duration> option) { Duration value = configurations.get(option); diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroServiceContainerEnvironment.java b/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroServiceContainerEnvironment.java new file mode 100644 index 000000000..6442b87d1 --- /dev/null +++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestAmoroServiceContainerEnvironment.java @@ -0,0 +1,40 @@ +/* + * 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; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class TestAmoroServiceContainerEnvironment { + + @Test + public void testSensitiveEnvironmentValuesAreRedacted() { + Assertions.assertEquals( + "<redacted>", AmoroServiceContainer.redactEnvironmentValue("LAS_SERVICE_AK", "ak")); + Assertions.assertEquals( + "<redacted>", AmoroServiceContainer.redactEnvironmentValue("LAS_SERVICE_SK", "sk")); + Assertions.assertEquals( + "<redacted>", AmoroServiceContainer.redactEnvironmentValue("ASSUME_ROLE_ACCESS_KEY", "ak")); + Assertions.assertEquals( + "<redacted>", AmoroServiceContainer.redactEnvironmentValue("SESSION_TOKEN", "token")); + Assertions.assertEquals( + "thrift://hms:9083", + AmoroServiceContainer.redactEnvironmentValue("HMS_URI", "thrift://hms:9083")); + } +} 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 index a8b91fc05..1e57f7368 100644 --- 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 @@ -107,6 +107,26 @@ public class TestDefaultCatalogManagerNamespaces { Assertions.assertTrue(manager.listCatalogMetas("tenant-a").isEmpty()); } + @Test + public void testDisabledAllowlistDiscoversNamespacesFromCatalogs() { + Configurations configurations = new Configurations(); + configurations.setBoolean(AmoroManagementConf.CATALOG_NAMESPACE_ENABLED, true); + configurations.setBoolean(AmoroManagementConf.CATALOG_NAMESPACE_ALLOWLIST_ENABLED, false); + DefaultCatalogManager manager = new DefaultCatalogManager(configurations); + + DB.insertCatalog(catalog("tenant-b@las", "tenant-b")); + DB.insertCatalog(catalog("tenant-a@las", "tenant-a")); + + Assertions.assertTrue(manager.supportNamespace()); + Assertions.assertFalse(manager.namespaceAllowlistEnabled()); + Assertions.assertEquals(Arrays.asList("tenant-a", "tenant-b"), manager.listNamespaces()); + Assertions.assertEquals( + Collections.singletonList("tenant-a@las"), + catalogNames(manager.listCatalogMetas("tenant-a"))); + Assertions.assertTrue(manager.listCatalogMetas("unknown").isEmpty()); + Assertions.assertThrows(IllegalStateException.class, () -> manager.addNamespace("tenant-c")); + } + private static CatalogMeta catalog(String catalogName, String namespace) { HashMap<String, String> properties = new HashMap<>(); properties.put(CatalogMetaProperties.NAMESPACE, namespace); 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 index 24af35cc7..dfcfddd32 100644 --- 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 @@ -45,9 +45,11 @@ import java.util.Arrays; import java.util.Base64; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; public class TestLasCatalogSynchronizer { @@ -72,6 +74,7 @@ public class TestLasCatalogSynchronizer { .thenReturn( Arrays.asList("tenant-a@las", "tenant-a@manual", "tenant-b@keep", "invalid-catalog")); CatalogManager catalogManager = mock(CatalogManager.class); + when(catalogManager.namespaceAllowlistEnabled()).thenReturn(true); 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); @@ -131,6 +134,7 @@ public class TestLasCatalogSynchronizer { HMSClient hmsSdk = mock(HMSClient.class); when(hmsSdk.getCatalogs()).thenReturn(Collections.singletonList("tenant-a@las")); CatalogManager catalogManager = mock(CatalogManager.class); + when(catalogManager.namespaceAllowlistEnabled()).thenReturn(true); when(catalogManager.listNamespaces()).thenReturn(Collections.singletonList("tenant-a")); CatalogMeta existing = catalog("tenant-a@las", "tenant-a", true); existing.getCatalogProperties().put("preserved", "value"); @@ -143,7 +147,7 @@ public class TestLasCatalogSynchronizer { verify(catalogManager).updateCatalog(updated.capture()); Assertions.assertEquals("value", updated.getValue().getCatalogProperties().get("preserved")); Assertions.assertEquals( - "ICEBERG,PAIMON", + "ICEBERG,PAIMON,LANCE", updated.getValue().getCatalogProperties().get(CatalogMetaProperties.TABLE_FORMATS)); when(catalogManager.listCatalogMetas()) .thenReturn(Collections.singletonList(updated.getValue())); @@ -164,6 +168,28 @@ public class TestLasCatalogSynchronizer { verifyNoInteractions(catalogManager); } + @Test + public void testDisabledAllowlistSynchronizesAllValidCatalogs() throws Exception { + HMSClient hmsSdk = mock(HMSClient.class); + when(hmsSdk.getCatalogs()) + .thenReturn(Arrays.asList("tenant-a@las", "tenant-b@hive", "invalid-catalog")); + CatalogManager catalogManager = mock(CatalogManager.class); + when(catalogManager.namespaceAllowlistEnabled()).thenReturn(false); + when(catalogManager.listCatalogMetas()).thenReturn(Collections.emptyList()); + + LasCatalogSynchronizer synchronizer = synchronizer(hmsSdk, catalogManager); + synchronizer.syncOnce(); + + ArgumentCaptor<CatalogMeta> created = ArgumentCaptor.forClass(CatalogMeta.class); + verify(catalogManager, times(2)).createCatalog(created.capture()); + Assertions.assertEquals( + new HashSet<>(Arrays.asList("tenant-a@las", "tenant-b@hive")), + created.getAllValues().stream() + .map(CatalogMeta::getCatalogName) + .collect(Collectors.toSet())); + verify(catalogManager, never()).listNamespaces(); + } + @Test public void testInvalidHmsSnapshotDoesNotMutateCatalogState() throws Exception { HMSClient hmsSdk = mock(HMSClient.class); diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/las/TestLasIntegrationContext.java b/amoro-ams/src/test/java/org/apache/amoro/server/las/TestLasIntegrationContext.java index fdeebaf5e..7e8530b87 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/las/TestLasIntegrationContext.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/las/TestLasIntegrationContext.java @@ -66,6 +66,36 @@ public class TestLasIntegrationContext { Assertions.assertTrue(exception.getMessage().contains(LasIntegrationConfig.HMS_URI.key())); } + @Test + public void testUsesExistingLasServiceCredentialEnvironmentNames() { + Configurations configurations = validConfigurations(); + configurations.removeConfig(LasIntegrationConfig.IAM_BOOTSTRAP_ACCESS_KEY); + configurations.removeConfig(LasIntegrationConfig.IAM_BOOTSTRAP_SECRET_KEY); + Map<String, String> environment = new HashMap<>(); + environment.put(LasIntegrationConfig.LAS_SERVICE_ACCESS_KEY_ENV, "las-service-ak"); + environment.put(LasIntegrationConfig.LAS_SERVICE_SECRET_KEY_ENV, "las-service-sk"); + + LasIntegrationContext context = LasIntegrationContext.initialize(configurations, environment); + bytedance.olap.iam.Credential credential = context.bootstrapCredential(); + + Assertions.assertEquals("las-service-ak", credential.getAccessKeyId()); + Assertions.assertEquals("las-service-sk", credential.getSecretAccessKey()); + } + + @Test + public void testExplicitCredentialConfigurationOverridesEnvironment() { + Map<String, String> environment = new HashMap<>(); + environment.put(LasIntegrationConfig.LAS_SERVICE_ACCESS_KEY_ENV, "las-service-ak"); + environment.put(LasIntegrationConfig.LAS_SERVICE_SECRET_KEY_ENV, "las-service-sk"); + + LasIntegrationContext context = + LasIntegrationContext.initialize(validConfigurations(), environment); + bytedance.olap.iam.Credential credential = context.bootstrapCredential(); + + Assertions.assertEquals("bootstrap-ak", credential.getAccessKeyId()); + Assertions.assertEquals("bootstrap-sk", credential.getSecretAccessKey()); + } + @Test public void testCrossVpcConfigurationIsAtomic() { Configurations configurations = validConfigurations(); diff --git a/amoro-common/src/main/java/org/apache/amoro/CommonUnifiedCatalog.java b/amoro-common/src/main/java/org/apache/amoro/CommonUnifiedCatalog.java index 281911c9a..f15a214d1 100644 --- a/amoro-common/src/main/java/org/apache/amoro/CommonUnifiedCatalog.java +++ b/amoro-common/src/main/java/org/apache/amoro/CommonUnifiedCatalog.java @@ -33,7 +33,7 @@ import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.Stream; -public class CommonUnifiedCatalog implements UnifiedCatalog { +public class CommonUnifiedCatalog implements UnifiedCatalog, AutoCloseable { private final String catalogName; private final String metaStoreType; @@ -116,12 +116,12 @@ public class CommonUnifiedCatalog implements UnifiedCatalog { @Override public AmoroTable<?> loadTable(String database, String table) { return formatCatalogAsOrder( + TableFormat.LANCE, TableFormat.MIXED_HIVE, TableFormat.MIXED_ICEBERG, TableFormat.ICEBERG, TableFormat.PAIMON, - TableFormat.HUDI, - TableFormat.LANCE) + TableFormat.HUDI) .map( formatCatalog -> { try { @@ -144,12 +144,12 @@ public class CommonUnifiedCatalog implements UnifiedCatalog { public List<TableIDWithFormat> listTables(String database) { TableFormat[] formats = new TableFormat[] { + TableFormat.LANCE, TableFormat.MIXED_HIVE, TableFormat.MIXED_ICEBERG, TableFormat.ICEBERG, TableFormat.PAIMON, - TableFormat.HUDI, - TableFormat.LANCE + TableFormat.HUDI }; Map<String, TableFormat> tableNameToFormat = Maps.newHashMap(); @@ -206,19 +206,47 @@ public class CommonUnifiedCatalog implements UnifiedCatalog { ServiceLoader<FormatCatalogFactory> loader = ServiceLoader.load(FormatCatalogFactory.class); String normalizedMetastoreType = CatalogUtil.normalizeMetastoreType(metaStoreType); Set<TableFormat> formats = CatalogUtil.tableFormats(normalizedMetastoreType, catalogProperties); - Map<TableFormat, FormatCatalog> formatCatalogs = Maps.newConcurrentMap(); - for (FormatCatalogFactory factory : loader) { - if (formats.contains(factory.format())) { - Map<String, String> formatCatalogProperties = - factory.convertCatalogProperties( - name(), normalizedMetastoreType, this.catalogProperties); - FormatCatalog catalog = - factory.create( - name(), normalizedMetastoreType, formatCatalogProperties, tableMetaStore); - formatCatalogs.put(factory.format(), catalog); + Map<TableFormat, FormatCatalog> newFormatCatalogs = Maps.newConcurrentMap(); + try { + for (FormatCatalogFactory factory : loader) { + if (formats.contains(factory.format())) { + Map<String, String> formatCatalogProperties = + factory.convertCatalogProperties( + name(), normalizedMetastoreType, this.catalogProperties); + FormatCatalog catalog = + factory.create( + name(), normalizedMetastoreType, formatCatalogProperties, tableMetaStore); + newFormatCatalogs.put(factory.format(), catalog); + } } + } catch (RuntimeException e) { + closeFormatCatalogs(newFormatCatalogs); + throw e; } - this.formatCatalogs = formatCatalogs; + Map<TableFormat, FormatCatalog> oldFormatCatalogs = this.formatCatalogs; + this.formatCatalogs = newFormatCatalogs; + closeFormatCatalogs(oldFormatCatalogs); + } + + @Override + public void close() { + Map<TableFormat, FormatCatalog> catalogs = this.formatCatalogs; + this.formatCatalogs = Maps.newHashMap(); + closeFormatCatalogs(catalogs); + } + + private static void closeFormatCatalogs(Map<TableFormat, FormatCatalog> catalogs) { + catalogs.values().stream() + .filter(AutoCloseable.class::isInstance) + .map(AutoCloseable.class::cast) + .forEach( + catalog -> { + try { + catalog.close(); + } catch (Exception e) { + throw new IllegalStateException("Failed to close format catalog", e); + } + }); } /** get format catalogs as given format order */ diff --git a/amoro-format-lance/pom.xml b/amoro-format-lance/pom.xml index 4d546422c..5955279d4 100755 --- a/amoro-format-lance/pom.xml +++ b/amoro-format-lance/pom.xml @@ -42,12 +42,19 @@ <groupId>org.apache.arrow</groupId> <artifactId>arrow-vector</artifactId> </dependency> + <!-- Keep Arrow's Netty buffer and common classes on one version. Without this direct + dependency Hadoop's older transitive netty-common wins and RootAllocator fails. --> + <dependency> + <groupId>io.netty</groupId> + <artifactId>netty-common</artifactId> + <version>${netty.version}</version> + </dependency> <!-- Lance format Java API --> <dependency> <groupId>org.lance</groupId> <artifactId>lance-core</artifactId> - <version>7.0.0</version> + <version>7.0.0-ve-6</version> <exclusions> <exclusion> <groupId>org.lance</groupId> @@ -71,6 +78,32 @@ <artifactId>lance-namespace-apache-client</artifactId> <version>0.8.0</version> </dependency> + <dependency> + <groupId>org.lance</groupId> + <artifactId>lance-namespace-hive3</artifactId> + <version>0.4.1</version> + <scope>runtime</scope> + <exclusions> + <!-- Use the versions already managed by Amoro. Hive3-only calls are isolated + behind reflection so the hadoop2 profile remains compilable. --> + <exclusion> + <groupId>org.lance</groupId> + <artifactId>lance-namespace-core</artifactId> + </exclusion> + <exclusion> + <groupId>org.lance</groupId> + <artifactId>lance-namespace-apache-client</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.hive</groupId> + <artifactId>hive-standalone-metastore</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.hive</groupId> + <artifactId>hive-exec</artifactId> + </exclusion> + </exclusions> + </dependency> <dependency> <groupId>org.apache.iceberg</groupId> <artifactId>iceberg-aws</artifactId> diff --git a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceCatalogFactory.java b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceCatalogFactory.java index a78fa1ae3..aeca11224 100755 --- a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceCatalogFactory.java +++ b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceCatalogFactory.java @@ -21,6 +21,7 @@ package org.apache.amoro.formats.lance; import org.apache.amoro.FormatCatalog; import org.apache.amoro.FormatCatalogFactory; import org.apache.amoro.TableFormat; +import org.apache.amoro.properties.CatalogMetaProperties; import org.apache.amoro.table.TableMetaStore; import java.util.HashMap; @@ -38,6 +39,9 @@ public class LanceCatalogFactory implements FormatCatalogFactory { String metastoreType, Map<String, String> properties, TableMetaStore metaStore) { + if (CatalogMetaProperties.CATALOG_TYPE_HIVE.equalsIgnoreCase(metastoreType)) { + return new LanceHms3Catalog(catalogName, properties, metaStore); + } return new LanceDirectoryV1Catalog(catalogName, properties); } diff --git a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceHms3Catalog.java b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceHms3Catalog.java new file mode 100644 index 000000000..e7a7fdedc --- /dev/null +++ b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceHms3Catalog.java @@ -0,0 +1,302 @@ +/* + * 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.formats.lance; + +import org.apache.amoro.AmoroTable; +import org.apache.amoro.FormatCatalog; +import org.apache.amoro.NoSuchTableException; +import org.apache.amoro.table.TableIdentifier; +import org.apache.amoro.table.TableMetaStore; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.commons.lang3.StringUtils; +import org.apache.hadoop.conf.Configuration; +import org.lance.Dataset; +import org.lance.namespace.LanceNamespace; +import org.lance.namespace.errors.NamespaceNotFoundException; +import org.lance.namespace.errors.TableNotFoundException; +import org.lance.namespace.model.DescribeTableResponse; +import org.lance.namespace.model.ListNamespacesRequest; +import org.lance.namespace.model.ListNamespacesResponse; +import org.lance.namespace.model.ListTablesRequest; +import org.lance.namespace.model.ListTablesResponse; +import org.lance.namespace.model.NamespaceExistsRequest; +import org.lance.namespace.model.TableExistsRequest; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Proxy; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.TreeSet; + +/** Read-only Lance catalog backed by a Hive 3 metastore. */ +public class LanceHms3Catalog implements FormatCatalog, AutoCloseable { + + private static final String HIVE3_NAMESPACE_CLASS = "org.lance.namespace.hive3.Hive3Namespace"; + private static final int PAGE_SIZE = 1_000; + + private final String catalogName; + private final TableMetaStore metaStore; + private final BufferAllocator allocator; + private final LanceNamespace delegateNamespace; + private final LanceNamespace datasetNamespace; + + public LanceHms3Catalog( + String catalogName, Map<String, String> properties, TableMetaStore metaStore) { + BufferAllocator newAllocator = new RootAllocator(Long.MAX_VALUE); + LanceNamespace newNamespace; + try { + newNamespace = + createHive3Namespace( + properties == null ? Collections.emptyMap() : properties, metaStore, newAllocator); + } catch (RuntimeException | Error e) { + newAllocator.close(); + throw e; + } + + this.catalogName = catalogName; + this.metaStore = metaStore; + this.allocator = newAllocator; + this.delegateNamespace = newNamespace; + this.datasetNamespace = + withStorageOptions( + newNamespace, new LanceStorageOptionsProvider(metaStore.getConfiguration())); + } + + LanceHms3Catalog( + String catalogName, + TableMetaStore metaStore, + BufferAllocator allocator, + LanceNamespace delegateNamespace, + LanceStorageOptionsProvider storageOptionsProvider) { + this.catalogName = catalogName; + this.metaStore = metaStore; + this.allocator = allocator; + this.delegateNamespace = delegateNamespace; + this.datasetNamespace = withStorageOptions(delegateNamespace, storageOptionsProvider); + } + + @Override + public List<String> listDatabases() { + TreeSet<String> databases = new TreeSet<>(); + String pageToken = null; + do { + ListNamespacesRequest request = + new ListNamespacesRequest() + .id(Collections.singletonList(catalogName)) + .limit(PAGE_SIZE) + .pageToken(pageToken); + ListNamespacesResponse response = + metaStore.doAs(() -> delegateNamespace.listNamespaces(request)); + if (response == null) { + break; + } + if (response.getNamespaces() != null) { + databases.addAll(response.getNamespaces()); + } + pageToken = nextPageToken(pageToken, response.getPageToken()); + } while (pageToken != null); + return new ArrayList<>(databases); + } + + @Override + public boolean databaseExists(String database) { + try { + NamespaceExistsRequest request = new NamespaceExistsRequest().id(identifier(database)); + metaStore.doAs( + () -> { + delegateNamespace.namespaceExists(request); + return null; + }); + return true; + } catch (NamespaceNotFoundException e) { + return false; + } + } + + @Override + public boolean tableExists(String database, String table) { + try { + TableExistsRequest request = new TableExistsRequest().id(identifier(database, table)); + metaStore.doAs( + () -> { + delegateNamespace.tableExists(request); + return null; + }); + return true; + } catch (NamespaceNotFoundException | TableNotFoundException e) { + return false; + } + } + + @Override + public void createDatabase(String database) { + throw readOnly(); + } + + @Override + public void dropDatabase(String database) { + throw readOnly(); + } + + @Override + public AmoroTable<?> loadTable(String database, String table) { + if (!tableExists(database, table)) { + throw new NoSuchTableException( + "Lance table " + catalogName + "." + database + "." + table + " does not exist"); + } + + List<String> tableId = identifier(database, table); + try { + Dataset dataset = + metaStore.doAs( + () -> + Dataset.open() + .allocator(allocator) + .namespaceClient(datasetNamespace) + .tableId(tableId) + .build()); + return new LanceTable( + TableIdentifier.of(catalogName, database, table), dataset, Collections.emptyMap()); + } catch (RuntimeException e) { + throw new IllegalStateException("Failed to open Lance table " + String.join(".", tableId), e); + } + } + + @Override + public boolean dropTable(String database, String table, boolean purge) { + throw readOnly(); + } + + @Override + public List<String> listTables(String database) { + if (!databaseExists(database)) { + return Collections.emptyList(); + } + + TreeSet<String> tables = new TreeSet<>(); + String pageToken = null; + do { + ListTablesRequest request = + new ListTablesRequest() + .id(identifier(database)) + // HMS is the metadata source of truth. Do not make table discovery depend on an + // object-store probe, which runs before temporary TOS credentials are vended. + .includeDeclared(true) + .limit(PAGE_SIZE) + .pageToken(pageToken); + ListTablesResponse response = metaStore.doAs(() -> delegateNamespace.listTables(request)); + if (response == null) { + break; + } + if (response.getTables() != null) { + tables.addAll(response.getTables()); + } + pageToken = nextPageToken(pageToken, response.getPageToken()); + } while (pageToken != null); + return new ArrayList<>(tables); + } + + @Override + public void close() { + try { + if (delegateNamespace instanceof AutoCloseable) { + ((AutoCloseable) delegateNamespace).close(); + } + } catch (Exception e) { + throw new IllegalStateException("Failed to close Lance HMS3 namespace", e); + } finally { + allocator.close(); + } + } + + private static LanceNamespace createHive3Namespace( + Map<String, String> properties, TableMetaStore metaStore, BufferAllocator allocator) { + try { + Class<?> namespaceClass = Class.forName(HIVE3_NAMESPACE_CLASS); + Object namespaceObject = namespaceClass.getDeclaredConstructor().newInstance(); + if (!(namespaceObject instanceof LanceNamespace)) { + throw new IllegalStateException( + HIVE3_NAMESPACE_CLASS + " does not implement the configured LanceNamespace API"); + } + namespaceClass + .getMethod("setHadoopConf", Configuration.class) + .invoke(namespaceObject, metaStore.getConfiguration()); + LanceNamespace namespace = (LanceNamespace) namespaceObject; + namespace.initialize(new HashMap<>(properties), allocator); + return namespace; + } catch (ClassNotFoundException e) { + throw new IllegalStateException( + "Lance HMS3 support requires the lance-namespace-hive3 runtime", e); + } catch (InvocationTargetException e) { + throw new IllegalStateException("Failed to initialize Lance HMS3 namespace", e.getCause()); + } catch (ReflectiveOperationException | LinkageError e) { + throw new IllegalStateException( + "Lance HMS3 namespace is unavailable with the current Hive runtime", e); + } + } + + private static LanceNamespace withStorageOptions( + LanceNamespace delegate, LanceStorageOptionsProvider storageOptionsProvider) { + return (LanceNamespace) + Proxy.newProxyInstance( + LanceNamespace.class.getClassLoader(), + new Class<?>[] {LanceNamespace.class}, + (proxy, method, args) -> { + try { + Object result = method.invoke(delegate, args); + if (result instanceof DescribeTableResponse) { + DescribeTableResponse response = (DescribeTableResponse) result; + String location = response.getLocation(); + Map<String, String> storageOptions = new HashMap<>(); + if (response.getStorageOptions() != null) { + storageOptions.putAll(response.getStorageOptions()); + } + storageOptions.putAll(storageOptionsProvider.storageOptions(location)); + response.setLocation(storageOptionsProvider.datasetLocation(location)); + response.setStorageOptions(storageOptions); + } + return result; + } catch (InvocationTargetException e) { + throw e.getCause(); + } + }); + } + + private List<String> identifier(String... parts) { + List<String> identifier = new ArrayList<>(parts.length + 1); + identifier.add(catalogName); + Collections.addAll(identifier, parts); + return identifier; + } + + private static String nextPageToken(String previous, String next) { + if (StringUtils.isBlank(next) || next.equals(previous)) { + return null; + } + return next; + } + + private static UnsupportedOperationException readOnly() { + return new UnsupportedOperationException("Lance HMS3 catalog is read-only"); + } +} diff --git a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceStorageOptionsProvider.java b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceStorageOptionsProvider.java new file mode 100644 index 000000000..ceebaf3ba --- /dev/null +++ b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceStorageOptionsProvider.java @@ -0,0 +1,205 @@ +/* + * 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.formats.lance; + +import org.apache.hadoop.conf.Configuration; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.net.URI; +import java.time.Duration; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +/** Converts Hadoop TOS configuration and credentials into Lance storage options. */ +class LanceStorageOptionsProvider { + + static final String TOS_ENDPOINT = "fs.tos.endpoint"; + static final String TOS_REGION = "fs.tos.region"; + static final String VOLC_REGION = "fs.volc.openapi.region"; + static final String TOS_ACCESS_KEY = "fs.tos.access-key-id"; + static final String TOS_SECRET_KEY = "fs.tos.secret-access-key"; + static final String TOS_SESSION_TOKEN = "fs.tos.session-token"; + static final String TOS_CREDENTIAL_PROVIDER = "fs.tos.credentials.provider"; + static final String TOS_CREDENTIAL_TTL = "fs.tos.credential.sts.token.time-to-live"; + + private static final long DEFAULT_CREDENTIAL_TTL_SECONDS = Duration.ofHours(1).getSeconds(); + private static final long MAX_ADVERTISED_LIFETIME_MILLIS = Duration.ofMinutes(15).toMillis(); + + private final Configuration configuration; + private final CredentialLoader credentialLoader; + + LanceStorageOptionsProvider(Configuration configuration) { + this(configuration, new ProtonCredentialLoader(configuration)); + } + + LanceStorageOptionsProvider(Configuration configuration, CredentialLoader credentialLoader) { + this.configuration = configuration; + this.credentialLoader = credentialLoader; + } + + Map<String, String> storageOptions(String location) { + if (location == null || !"tos".equalsIgnoreCase(URI.create(location).getScheme())) { + return Collections.emptyMap(); + } + + Map<String, String> options = new HashMap<>(); + putIfNotBlank(options, "endpoint", configuration.getTrimmed(TOS_ENDPOINT)); + putIfNotBlank( + options, + "region", + firstNonBlank(configuration.getTrimmed(TOS_REGION), configuration.getTrimmed(VOLC_REGION))); + + TemporaryCredential credential = credential(location); + if (credential != null) { + putIfNotBlank(options, "access_key_id", credential.accessKeyId); + putIfNotBlank(options, "secret_access_key", credential.secretAccessKey); + putIfNotBlank(options, "security_token", credential.sessionToken); + putRefreshOptions(options); + } + return options; + } + + String datasetLocation(String location) { + // The ByteDance Lance runtime has a native tos:// object-store implementation backed by + // Proton. Keeping the HMS location intact selects that implementation; rewriting it to s3:// + // would bypass Proton and route requests through the generic S3 client. + return location; + } + + private TemporaryCredential credential(String location) { + if (configuration.getTrimmed(TOS_CREDENTIAL_PROVIDER) != null) { + return credentialLoader.load(bucket(location)); + } + + String accessKey = configuration.getTrimmed(TOS_ACCESS_KEY); + String secretKey = configuration.getTrimmed(TOS_SECRET_KEY); + if (accessKey == null || secretKey == null) { + return null; + } + return new TemporaryCredential( + accessKey, secretKey, configuration.getTrimmed(TOS_SESSION_TOKEN)); + } + + private void putRefreshOptions(Map<String, String> options) { + long configuredTtlSeconds = + configuration.getLong(TOS_CREDENTIAL_TTL, DEFAULT_CREDENTIAL_TTL_SECONDS); + long advertisedLifetime = + Math.max( + 1_000L, + Math.min( + Duration.ofSeconds(configuredTtlSeconds).toMillis() / 2, + MAX_ADVERTISED_LIFETIME_MILLIS)); + long refreshOffset = Math.max(1_000L, Math.min(60_000L, advertisedLifetime / 5)); + options.put( + "expires_at_millis", String.valueOf(System.currentTimeMillis() + advertisedLifetime)); + options.put("refresh_offset_millis", String.valueOf(refreshOffset)); + } + + private static String bucket(String location) { + URI uri = URI.create(location); + return uri.getHost() == null ? uri.getAuthority() : uri.getHost(); + } + + private static String firstNonBlank(String first, String second) { + return first == null || first.isEmpty() ? second : first; + } + + private static void putIfNotBlank(Map<String, String> options, String key, String value) { + if (value != null && !value.isEmpty()) { + options.put(key, value); + } + } + + interface CredentialLoader { + TemporaryCredential load(String bucket); + } + + static class TemporaryCredential { + private final String accessKeyId; + private final String secretAccessKey; + private final String sessionToken; + + TemporaryCredential(String accessKeyId, String secretAccessKey, String sessionToken) { + this.accessKeyId = accessKeyId; + this.secretAccessKey = secretAccessKey; + this.sessionToken = sessionToken; + } + } + + /** Uses Proton through reflection so the community Lance module has no Proton dependency. */ + private static class ProtonCredentialLoader implements CredentialLoader { + private static final String CONF_CLASS = "io.proton.common.conf.Conf"; + private static final String PROVIDER_FACTORY_CLASS = + "io.proton.common.object.auth.ProviderFactory"; + + private final Configuration configuration; + private final ConcurrentMap<String, Object> providers = new ConcurrentHashMap<>(); + + private ProtonCredentialLoader(Configuration configuration) { + this.configuration = configuration; + } + + @Override + public TemporaryCredential load(String bucket) { + try { + Object provider = providers.computeIfAbsent(bucket, this::createProvider); + Object credential = provider.getClass().getMethod("expirableCredential").invoke(provider); + return new TemporaryCredential( + invokeString(credential, "accessKeyId"), + invokeString(credential, "accessKeySecret"), + invokeString(credential, "sessionToken")); + } catch (InvocationTargetException e) { + throw credentialFailure(bucket, e.getCause()); + } catch (ReflectiveOperationException | RuntimeException e) { + throw credentialFailure(bucket, e); + } + } + + private Object createProvider(String bucket) { + try { + Class<?> confClass = Class.forName(CONF_CLASS); + Object protonConf = + confClass.getMethod("copyOf", Iterable.class).invoke(null, configuration); + Class<?> factoryClass = Class.forName(PROVIDER_FACTORY_CLASS); + Method createProvider = + factoryClass.getMethod("createProvider", confClass, String.class, String.class); + return createProvider.invoke(null, protonConf, bucket, "tos"); + } catch (InvocationTargetException e) { + throw credentialFailure(bucket, e.getCause()); + } catch (ReflectiveOperationException e) { + throw credentialFailure(bucket, e); + } + } + + private static String invokeString(Object target, String method) + throws ReflectiveOperationException { + Object value = target.getClass().getMethod(method).invoke(target); + return value == null ? null : value.toString(); + } + + private static IllegalStateException credentialFailure(String bucket, Throwable cause) { + return new IllegalStateException( + "Failed to obtain temporary TOS credentials for bucket " + bucket, cause); + } + } +} diff --git a/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceCatalogFactory.java b/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceCatalogFactory.java new file mode 100644 index 000000000..32ccb87c3 --- /dev/null +++ b/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceCatalogFactory.java @@ -0,0 +1,67 @@ +/* + * 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.formats.lance; + +import org.apache.amoro.FormatCatalog; +import org.apache.amoro.properties.CatalogMetaProperties; +import org.apache.amoro.table.TableMetaStore; +import org.apache.hadoop.conf.Configuration; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +public class TestLanceCatalogFactory { + + @Test + public void testHiveMetastoreUsesHms3Catalog() throws Exception { + Configuration configuration = new Configuration(false); + configuration.set("hive.metastore.uris", "thrift://127.0.0.1:1"); + TableMetaStore metaStore = TableMetaStore.builder().withConfiguration(configuration).build(); + + FormatCatalog catalog = + new LanceCatalogFactory() + .create( + "tenant@catalog", + CatalogMetaProperties.CATALOG_TYPE_HIVE, + Collections.emptyMap(), + metaStore); + + Assertions.assertInstanceOf(LanceHms3Catalog.class, catalog); + ((AutoCloseable) catalog).close(); + } + + @Test + public void testFilesystemMetastoreKeepsDirectoryCatalog() { + Map<String, String> properties = new HashMap<>(); + properties.put(CatalogMetaProperties.KEY_WAREHOUSE, "file:/tmp/lance-catalog"); + + FormatCatalog catalog = + new LanceCatalogFactory() + .create( + "filesystem-catalog", + CatalogMetaProperties.CATALOG_TYPE_FILESYSTEM, + properties, + TableMetaStore.EMPTY); + + Assertions.assertInstanceOf(LanceDirectoryV1Catalog.class, catalog); + } +} diff --git a/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceHms3Catalog.java b/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceHms3Catalog.java new file mode 100644 index 000000000..6af85b496 --- /dev/null +++ b/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceHms3Catalog.java @@ -0,0 +1,144 @@ +/* + * 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.formats.lance; + +import org.apache.amoro.table.TableMetaStore; +import org.apache.arrow.memory.RootAllocator; +import org.apache.hadoop.conf.Configuration; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.lance.namespace.LanceNamespace; +import org.lance.namespace.errors.NamespaceNotFoundException; +import org.lance.namespace.errors.TableNotFoundException; +import org.lance.namespace.model.ListNamespacesRequest; +import org.lance.namespace.model.ListNamespacesResponse; +import org.lance.namespace.model.ListTablesRequest; +import org.lance.namespace.model.ListTablesResponse; +import org.lance.namespace.model.NamespaceExistsRequest; +import org.lance.namespace.model.TableExistsRequest; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; + +public class TestLanceHms3Catalog { + + @Test + public void testPhysicalCatalogScopesDatabaseAndTableRequests() { + TestingNamespace namespace = new TestingNamespace(); + TableMetaStore metaStore = metaStore(); + try (LanceHms3Catalog catalog = + new LanceHms3Catalog( + "tenant@catalog", + metaStore, + new RootAllocator(Long.MAX_VALUE), + namespace, + new LanceStorageOptionsProvider(metaStore.getConfiguration()))) { + Assertions.assertEquals(Arrays.asList("db_a", "db_b"), catalog.listDatabases()); + Assertions.assertEquals(Arrays.asList("table_a", "table_b"), catalog.listTables("db_a")); + Assertions.assertTrue(catalog.databaseExists("db_a")); + Assertions.assertFalse(catalog.databaseExists("missing")); + Assertions.assertTrue(catalog.tableExists("db_a", "table_a")); + Assertions.assertFalse(catalog.tableExists("db_a", "missing")); + + Assertions.assertEquals(Collections.singletonList("tenant@catalog"), namespace.databaseId); + Assertions.assertEquals(Arrays.asList("tenant@catalog", "db_a"), namespace.tableListId); + Assertions.assertTrue(namespace.includeDeclared); + Assertions.assertEquals( + Arrays.asList("tenant@catalog", "db_a", "missing"), namespace.tableExistsId); + } + Assertions.assertTrue(namespace.closed); + } + + @Test + public void testMutationsAreRejected() { + TestingNamespace namespace = new TestingNamespace(); + TableMetaStore metaStore = metaStore(); + try (LanceHms3Catalog catalog = + new LanceHms3Catalog( + "tenant@catalog", + metaStore, + new RootAllocator(Long.MAX_VALUE), + namespace, + new LanceStorageOptionsProvider(metaStore.getConfiguration()))) { + Assertions.assertThrows( + UnsupportedOperationException.class, () -> catalog.createDatabase("db")); + Assertions.assertThrows( + UnsupportedOperationException.class, () -> catalog.dropDatabase("db")); + Assertions.assertThrows( + UnsupportedOperationException.class, () -> catalog.dropTable("db", "table", false)); + } + } + + private static TableMetaStore metaStore() { + return TableMetaStore.builder().withConfiguration(new Configuration(false)).build(); + } + + private static class TestingNamespace implements LanceNamespace, AutoCloseable { + private List<String> databaseId; + private List<String> tableListId; + private List<String> tableExistsId; + private boolean includeDeclared; + private boolean closed; + + @Override + public void initialize( + Map<String, String> properties, org.apache.arrow.memory.BufferAllocator allocator) {} + + @Override + public String namespaceId() { + return "test"; + } + + @Override + public ListNamespacesResponse listNamespaces(ListNamespacesRequest request) { + databaseId = request.getId(); + return new ListNamespacesResponse().namespaces(new HashSet<>(Arrays.asList("db_b", "db_a"))); + } + + @Override + public void namespaceExists(NamespaceExistsRequest request) { + if (request.getId().contains("missing")) { + throw new NamespaceNotFoundException("missing"); + } + } + + @Override + public ListTablesResponse listTables(ListTablesRequest request) { + tableListId = request.getId(); + includeDeclared = Boolean.TRUE.equals(request.getIncludeDeclared()); + return new ListTablesResponse().tables(new HashSet<>(Arrays.asList("table_b", "table_a"))); + } + + @Override + public void tableExists(TableExistsRequest request) { + tableExistsId = request.getId(); + if (request.getId().contains("missing")) { + throw new TableNotFoundException("missing"); + } + } + + @Override + public void close() { + closed = true; + } + } +} diff --git a/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceHms3CatalogIntegration.java b/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceHms3CatalogIntegration.java new file mode 100644 index 000000000..e7af47f84 --- /dev/null +++ b/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceHms3CatalogIntegration.java @@ -0,0 +1,53 @@ +/* + * 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.formats.lance; + +import org.apache.amoro.table.TableMetaStore; +import org.apache.hadoop.conf.Configuration; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Assumptions; +import org.junit.jupiter.api.Test; + +import java.util.Collections; + +/** Opt-in smoke test for a real Hive 3 metastore. */ +public class TestLanceHms3CatalogIntegration { + + @Test + public void testRealHmsMetadata() { + String hmsUri = System.getProperty("lance.hms.uri"); + String catalogName = System.getProperty("lance.hms.catalog"); + String database = System.getProperty("lance.hms.database"); + String table = System.getProperty("lance.hms.table"); + Assumptions.assumeTrue( + hmsUri != null && catalogName != null && database != null && table != null, + "Real HMS coordinates were not provided"); + + Configuration configuration = new Configuration(false); + configuration.set("hive.metastore.uris", hmsUri); + TableMetaStore metaStore = TableMetaStore.builder().withConfiguration(configuration).build(); + + try (LanceHms3Catalog catalog = + new LanceHms3Catalog(catalogName, Collections.emptyMap(), metaStore)) { + Assertions.assertTrue(catalog.listDatabases().contains(database)); + Assertions.assertTrue(catalog.tableExists(database, table)); + Assertions.assertTrue(catalog.listTables(database).contains(table)); + } + } +} diff --git a/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceStorageOptionsProvider.java b/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceStorageOptionsProvider.java new file mode 100644 index 000000000..a06557d81 --- /dev/null +++ b/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceStorageOptionsProvider.java @@ -0,0 +1,91 @@ +/* + * 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.formats.lance; + +import org.apache.hadoop.conf.Configuration; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; + +public class TestLanceStorageOptionsProvider { + + @Test + public void testTemporaryCredentialsAreMappedForTos() { + Configuration configuration = new Configuration(false); + configuration.set(LanceStorageOptionsProvider.TOS_ENDPOINT, "https://tos.example.com";); + configuration.set(LanceStorageOptionsProvider.VOLC_REGION, "cn-test"); + configuration.set(LanceStorageOptionsProvider.TOS_CREDENTIAL_PROVIDER, "test-provider"); + configuration.setLong(LanceStorageOptionsProvider.TOS_CREDENTIAL_TTL, 3_600L); + AtomicReference<String> requestedBucket = new AtomicReference<>(); + + LanceStorageOptionsProvider provider = + new LanceStorageOptionsProvider( + configuration, + bucket -> { + requestedBucket.set(bucket); + return new LanceStorageOptionsProvider.TemporaryCredential("ak", "sk", "token"); + }); + + long before = System.currentTimeMillis(); + Map<String, String> options = provider.storageOptions("tos://test-bucket/path/table.lance"); + + Assertions.assertEquals("test-bucket", requestedBucket.get()); + Assertions.assertEquals("https://tos.example.com";, options.get("endpoint")); + Assertions.assertEquals("cn-test", options.get("region")); + Assertions.assertEquals("ak", options.get("access_key_id")); + Assertions.assertEquals("sk", options.get("secret_access_key")); + Assertions.assertEquals("token", options.get("security_token")); + Assertions.assertTrue(Long.parseLong(options.get("expires_at_millis")) > before); + Assertions.assertTrue(Long.parseLong(options.get("refresh_offset_millis")) > 0); + Assertions.assertEquals( + "tos://test-bucket/path/table.lance", + provider.datasetLocation("tos://test-bucket/path/table.lance")); + } + + @Test + public void testStaticCredentialsRemainSupported() { + Configuration configuration = new Configuration(false); + configuration.set(LanceStorageOptionsProvider.TOS_ACCESS_KEY, "static-ak"); + configuration.set(LanceStorageOptionsProvider.TOS_SECRET_KEY, "static-sk"); + + Map<String, String> options = + new LanceStorageOptionsProvider(configuration) + .storageOptions("tos://test-bucket/table.lance"); + + Assertions.assertEquals("static-ak", options.get("access_key_id")); + Assertions.assertEquals("static-sk", options.get("secret_access_key")); + } + + @Test + public void testNonTosLocationHasNoOptions() { + Configuration configuration = new Configuration(false); + configuration.set(LanceStorageOptionsProvider.TOS_ACCESS_KEY, "ak"); + configuration.set(LanceStorageOptionsProvider.TOS_SECRET_KEY, "sk"); + + Assertions.assertTrue( + new LanceStorageOptionsProvider(configuration) + .storageOptions("s3://bucket/table.lance") + .isEmpty()); + Assertions.assertEquals( + "s3://bucket/table.lance", + new LanceStorageOptionsProvider(configuration).datasetLocation("s3://bucket/table.lance")); + } +} diff --git a/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/PaimonCatalogFactory.java b/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/PaimonCatalogFactory.java index 8c9b4bbba..6cfe9dfbb 100644 --- a/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/PaimonCatalogFactory.java +++ b/amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/PaimonCatalogFactory.java @@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.net.URL; +import java.util.Arrays; import java.util.Map; import java.util.Optional; @@ -48,6 +49,8 @@ public class PaimonCatalogFactory implements FormatCatalogFactory { public static final String PAIMON_OSS_ACCESS_KEY = "fs.oss.accessKeyId"; public static final String PAIMON_OSS_SECRET_KEY = "fs.oss.accessKeySecret"; public static final String PAIMON_OSS_ENDPOINT = "fs.oss.endpoint"; + static final String HMS_DEFAULT_CATALOG = "metastore.catalog.default"; + static final String HMS_DEFAULT_CATALOG_CACHE_KEY = "conf:" + HMS_DEFAULT_CATALOG; @Override public PaimonCatalog create( @@ -57,6 +60,7 @@ public class PaimonCatalogFactory implements FormatCatalogFactory { // if format table enabled, paimon will load hive orc/parquet/csv table to paimon table catalogProperties.put(CatalogOptions.FORMAT_TABLE_ENABLED.key(), "false"); catalogProperties.putAll(properties); + configureHiveClientPoolCache(metastoreType, catalogProperties, metaStore.getConfiguration()); hiveSiteLocation.ifPresent( url -> catalogProperties.put( @@ -88,6 +92,30 @@ public class PaimonCatalogFactory implements FormatCatalogFactory { return CatalogFactory.createCatalog(catalogContext); } + static void configureHiveClientPoolCache( + String metastoreType, Map<String, String> catalogProperties, Configuration configuration) { + if (!CatalogMetaProperties.CATALOG_TYPE_HIVE.equalsIgnoreCase(metastoreType) + || configuration.get(HMS_DEFAULT_CATALOG) == null) { + return; + } + + String cacheKeys = catalogProperties.get(HiveCatalogOptions.CLIENT_POOL_CACHE_KEYS.key()); + boolean alreadyConfigured = + cacheKeys != null + && Arrays.stream(cacheKeys.split(",")) + .map(String::trim) + .anyMatch(HMS_DEFAULT_CATALOG_CACHE_KEY::equalsIgnoreCase); + if (alreadyConfigured) { + return; + } + + catalogProperties.put( + HiveCatalogOptions.CLIENT_POOL_CACHE_KEYS.key(), + cacheKeys == null || cacheKeys.trim().isEmpty() + ? HMS_DEFAULT_CATALOG_CACHE_KEY + : cacheKeys + "," + HMS_DEFAULT_CATALOG_CACHE_KEY); + } + @Override public TableFormat format() { return TableFormat.PAIMON; diff --git a/amoro-format-paimon/src/test/java/org/apache/amoro/formats/paimon/TestPaimonCatalogFactory.java b/amoro-format-paimon/src/test/java/org/apache/amoro/formats/paimon/TestPaimonCatalogFactory.java new file mode 100644 index 000000000..886dd2568 --- /dev/null +++ b/amoro-format-paimon/src/test/java/org/apache/amoro/formats/paimon/TestPaimonCatalogFactory.java @@ -0,0 +1,74 @@ +/* + * 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.formats.paimon; + +import org.apache.amoro.properties.CatalogMetaProperties; +import org.apache.hadoop.conf.Configuration; +import org.apache.paimon.hive.HiveCatalogOptions; +import org.junit.Assert; +import org.junit.Test; + +import java.util.HashMap; +import java.util.Map; + +public class TestPaimonCatalogFactory { + + @Test + public void testHmsDefaultCatalogIsIncludedInClientPoolCacheKey() { + Configuration configuration = new Configuration(false); + configuration.set(PaimonCatalogFactory.HMS_DEFAULT_CATALOG, "tenant@catalog"); + Map<String, String> properties = new HashMap<>(); + + PaimonCatalogFactory.configureHiveClientPoolCache( + CatalogMetaProperties.CATALOG_TYPE_HIVE, properties, configuration); + + Assert.assertEquals( + PaimonCatalogFactory.HMS_DEFAULT_CATALOG_CACHE_KEY, + properties.get(HiveCatalogOptions.CLIENT_POOL_CACHE_KEYS.key())); + } + + @Test + public void testExistingClientPoolCacheKeysArePreserved() { + Configuration configuration = new Configuration(false); + configuration.set(PaimonCatalogFactory.HMS_DEFAULT_CATALOG, "tenant@catalog"); + Map<String, String> properties = new HashMap<>(); + properties.put(HiveCatalogOptions.CLIENT_POOL_CACHE_KEYS.key(), "ugi"); + + PaimonCatalogFactory.configureHiveClientPoolCache( + CatalogMetaProperties.CATALOG_TYPE_HIVE, properties, configuration); + PaimonCatalogFactory.configureHiveClientPoolCache( + CatalogMetaProperties.CATALOG_TYPE_HIVE, properties, configuration); + + Assert.assertEquals( + "ugi," + PaimonCatalogFactory.HMS_DEFAULT_CATALOG_CACHE_KEY, + properties.get(HiveCatalogOptions.CLIENT_POOL_CACHE_KEYS.key())); + } + + @Test + public void testNonHiveCatalogIsUnchanged() { + Configuration configuration = new Configuration(false); + configuration.set(PaimonCatalogFactory.HMS_DEFAULT_CATALOG, "tenant@catalog"); + Map<String, String> properties = new HashMap<>(); + + PaimonCatalogFactory.configureHiveClientPoolCache( + CatalogMetaProperties.CATALOG_TYPE_HADOOP, properties, configuration); + + Assert.assertFalse(properties.containsKey(HiveCatalogOptions.CLIENT_POOL_CACHE_KEYS.key())); + } +} diff --git a/amoro-web/mock/modules/catalogs.js b/amoro-web/mock/modules/catalogs.js index 79195ae47..178d3d808 100644 --- a/amoro-web/mock/modules/catalogs.js +++ b/amoro-web/mock/modules/catalogs.js @@ -17,6 +17,15 @@ */ export default [ + { + url: '/mock/api/ams/v1/namespaces', + method: 'get', + response: () => ({ + "message": "success", + "code": 200, + "result": ["123456789", "987654321"] + }), + }, { url: '/mock/api/ams/v1/catalogs', method: 'get', @@ -25,7 +34,7 @@ export default [ "code": 200, "result": [ { - "catalogName": "test_catalog", + "catalogName": "123456789@las", "catalogType": "hadoop", "storageConfigs": { "storage.type": "Hadoop", @@ -62,17 +71,17 @@ export default [ }, { - url: '/mock/api/ams/v1/catalogs/test_catalog/databases', + url: '/mock/api/ams/v1/catalogs/123456789@las/databases', method: 'get', response: () => { - return { "message": "success", "code": 200, "result": ["db", "test", "acc"] } + return { "message": "success", "code": 200, "result": ["warehouse", "analytics", "index_demo"] } }, }, { - url: '/mock/api/ams/v1/catalogs/test_catalog/databases/db/tables', + url: '/mock/api/ams/v1/catalogs/123456789@las/databases/warehouse/tables', method: 'get', response: () => { - return { "message": "success", "code": 200, "result": [{ "name": "user", "type": "ICEBERG" },{ "name": "wf", "type": "ICEBERG" }, { "name": "xcvz", "type": "ICEBERG" }] }; + return { "message": "success", "code": 200, "result": [{ "name": "jk_large_table", "type": "ICEBERG" }, { "name": "orders", "type": "ICEBERG" }, { "name": "events", "type": "PAIMON" }] }; }, }, { @@ -160,6 +169,24 @@ export default [ ] }), }, + { + url: '/mock/api/ams/v1/catalogs/metastore/:type/table-formats', + method: 'get', + response: () => ({ + "message": "success", + "code": 200, + "result": ["ICEBERG", "PAIMON"] + }), + }, + { + url: '/mock/api/ams/v1/catalogs/metastore/:type/storage-types', + method: 'get', + response: () => ({ + "message": "success", + "code": 200, + "result": ["Hadoop"] + }), + }, { url: '/mock/api/ams/v1/tables/catalogs/:catalog/dbs/:dbId/tables/:tableName/optimizing-processes/:processesId/tasks', method: 'get', diff --git a/amoro-web/mock/modules/common.js b/amoro-web/mock/modules/common.js index 8a2267fd2..1c04794a5 100644 --- a/amoro-web/mock/modules/common.js +++ b/amoro-web/mock/modules/common.js @@ -27,7 +27,8 @@ export default [ "userName": "admin", "loginTime": "1703839452053", "role": "SERVICE_ADMIN", - "roles": ["SERVICE_ADMIN"] + "roles": ["SERVICE_ADMIN"], + "privileges": ["VIEW_SYSTEM", "VIEW_CATALOG", "VIEW_TABLE", "VIEW_OPTIMIZER", "MANAGE_CATALOG", "MANAGE_TABLE", "MANAGE_OPTIMIZER", "EXECUTE_SQL", "MANAGE_PLATFORM"] } }), }, @@ -41,7 +42,8 @@ export default [ userName: 'admin', loginTime: '1703839452053', role: 'SERVICE_ADMIN', - roles: ['SERVICE_ADMIN'] + roles: ['SERVICE_ADMIN'], + privileges: ['VIEW_SYSTEM', 'VIEW_CATALOG', 'VIEW_TABLE', 'VIEW_OPTIMIZER', 'MANAGE_CATALOG', 'MANAGE_TABLE', 'MANAGE_OPTIMIZER', 'EXECUTE_SQL', 'MANAGE_PLATFORM'] } }), }, diff --git a/amoro-web/mock/modules/table.js b/amoro-web/mock/modules/table.js index 119cf7b3f..fbaf68bdd 100644 --- a/amoro-web/mock/modules/table.js +++ b/amoro-web/mock/modules/table.js @@ -18,17 +18,17 @@ export default [ { - url: '/mock/api/ams/v1/tables/catalogs/test_catalog/dbs/db/tables/user/details', + url: '/mock/api/ams/v1/tables/catalogs/:catalog/dbs/:dbId/tables/:tableName/details', method: 'get', - response: () => ({ + response: ({ params }) => ({ "message": "success", "code": 200, "result": { "tableType": "ICEBERG", "tableIdentifier": { - "catalog": "test_catalog", - "database": "db", - "tableName": "user" + "catalog": params.catalog, + "database": params.dbId, + "tableName": params.tableName }, "schema": [ { @@ -91,7 +91,7 @@ export default [ }), }, { - url: '/mock/api/ams/v1/tables/catalogs/test_catalog/dbs/db/tables/user/partitions', + url: '/mock/api/ams/v1/tables/catalogs/:catalog/dbs/:dbId/tables/:tableName/partitions', method: 'get', response: () => ({ "message": "success", @@ -120,7 +120,7 @@ export default [ }), }, { - url: '/mock/api/ams/v1/tables/catalogs/test_catalog/dbs/db/tables/user/branches', + url: '/mock/api/ams/v1/tables/catalogs/:catalog/dbs/:dbId/tables/:tableName/branches', method: 'get', response: () => ({ "message": "success", @@ -141,12 +141,17 @@ export default [ }), }, { - url: '/mock/api/ams/v1/tables/catalogs/test_catalog/dbs/db/tables/user/tags', + url: '/mock/api/ams/v1/tables/catalogs/:catalog/dbs/:dbId/tables/:tableName/tags', + method: 'get', + response: () => ({ "message": "success", "code": 200, "result": { "list": [], "total": 0 } }), + }, + { + url: '/mock/api/ams/v1/tables/catalogs/:catalog/dbs/:dbId/tables/:tableName/consumers', method: 'get', response: () => ({ "message": "success", "code": 200, "result": { "list": [], "total": 0 } }), }, { - url: '/mock/api/ams/v1/tables/catalogs/test_catalog/dbs/db/tables/user/snapshots', + url: '/mock/api/ams/v1/tables/catalogs/:catalog/dbs/:dbId/tables/:tableName/snapshots', method: 'get', response: () => ({ "message": "success", @@ -193,12 +198,12 @@ export default [ }), }, { - url: '/mock/api/ams/v1/tables/catalogs/test_catalog/dbs/db/tables/user/operations', + url: '/mock/api/ams/v1/tables/catalogs/:catalog/dbs/:dbId/tables/:tableName/operations', method: 'get', response: () => ({ "message": "success", "code": 200, "result": { "list": [], "total": 0 } }), }, { - url: '/mock/api/ams/v1/tables/catalogs/test_catalog/dbs/db/tables/user/partitions/:filter/files', + url: '/mock/api/ams/v1/tables/catalogs/:catalog/dbs/:dbId/tables/:tableName/partitions/:filter/files', method: 'get', response: () => ({ "message": "success", @@ -223,7 +228,7 @@ export default [ }), }, { - url: '/mock/api/ams/v1/tables/catalogs/test_catalog/dbs/db/tables/user/snapshots/:snapshotId/detail', + url: '/mock/api/ams/v1/tables/catalogs/:catalog/dbs/:dbId/tables/:tableName/snapshots/:snapshotId/detail', method: 'get', response: () => ({ "message": "success", @@ -260,7 +265,7 @@ export default [ }), }, { - url: '/mock/api/ams/v1/tables/catalogs/test_catalog/dbs/db/tables/user/optimizing-processes', + url: '/mock/api/ams/v1/tables/catalogs/:catalog/dbs/:dbId/tables/:tableName/optimizing-processes', method: 'get', response: () => ({ "message": "success", @@ -307,7 +312,7 @@ export default [ }), }, { - url: '/mock/api/ams/v1/tables/catalogs/test_catalog/dbs/db/tables/user/process-types', + url: '/mock/api/ams/v1/tables/catalogs/:catalog/dbs/:dbId/tables/:tableName/process-types', method: 'get', response: ({ query }) => { const processCategory = query?.processCategory || 'OPTIMIZING' @@ -335,7 +340,7 @@ export default [ }, }, { - url: '/mock/api/ams/v1/tables/catalogs/test_catalog/dbs/db/tables/user/operations', + url: '/mock/api/ams/v1/tables/catalogs/:catalog/dbs/:dbId/tables/:tableName/operations', method: 'get', response: () => ({ "message": "success", diff --git a/amoro-web/src/components/Sidebar.vue b/amoro-web/src/components/Sidebar.vue index 058645e78..64335f7d3 100644 --- a/amoro-web/src/components/Sidebar.vue +++ b/amoro-web/src/components/Sidebar.vue @@ -147,6 +147,7 @@ export default defineComponent({ let db: string | undefined let tableName: string | undefined let namespace: string | undefined + let type: string | undefined try { const stored = localStorage.getItem('easylake-menu-catalog-db-table') @@ -156,11 +157,13 @@ export default defineComponent({ catalog?: string database?: string tableName?: string + type?: string } namespace = parsed.namespace catalog = parsed.catalog db = parsed.database tableName = parsed.tableName + type = parsed.type } } catch (e) { @@ -175,6 +178,7 @@ export default defineComponent({ catalog, db, table: tableName, + ...(type ? { type } : {}), }, }) } diff --git a/amoro-web/src/views/catalogs/index.vue b/amoro-web/src/views/catalogs/index.vue index 743d12313..bf843c187 100644 --- a/amoro-web/src/views/catalogs/index.vue +++ b/amoro-web/src/views/catalogs/index.vue @@ -65,7 +65,7 @@ function initSelectCatalog() { catalogName: '', catalogType: '', } - if (decodeURIComponent(catalogname as string) === NEW_CATALOG) { + if (catalogname === NEW_CATALOG) { addCatalog() return } @@ -98,7 +98,7 @@ async function selectCatalog(item: ICatalogItem) { await router.replace({ path: '/catalogs', query: { - catalogname: encodeURIComponent(curCatalog.catalogName), + catalogname: curCatalog.catalogName, type: curCatalog.catalogType, }, }) @@ -138,7 +138,7 @@ function addCatalog() { addNewCatalog() } } - async function addNewCatalog() { +async function addNewCatalog() { const item: ICatalogItem = { catalogName: NEW_CATALOG, catalogType: '', @@ -174,7 +174,7 @@ onBeforeRouteLeave((_to, _form, next) => { </script> <template> - <div class="page-scroll" ref="pageScrollRef"> + <div ref="pageScrollRef" class="page-scroll"> <div class="catalogs-wrap g-flex"> <div class="catalog-list-left"> <div class="catalog-header"> @@ -202,12 +202,9 @@ onBeforeRouteLeave((_to, _form, next) => { height: 100%; overflow-y: auto; } - - - - .catalogs-wrap { - height: 100%; - padding: 16px 24px; +.catalogs-wrap { + height: 100%; + padding: 16px 24px; .catalog-list-left { width: 200px; height: 100%; diff --git a/amoro-web/src/views/hive-details/index.vue b/amoro-web/src/views/hive-details/index.vue index a23781ab1..542554fc8 100644 --- a/amoro-web/src/views/hive-details/index.vue +++ b/amoro-web/src/views/hive-details/index.vue @@ -153,7 +153,7 @@ export default defineComponent({ () => route.query, (val, old) => { const { catalog, db, table } = val - if (route.path === '/hive-tables' && (catalog !== old.catalog || db !== old.db || table !== old.table)) { + if (!isSecondaryNav.value && (catalog !== old.catalog || db !== old.db || table !== old.table)) { init() } }, @@ -164,6 +164,15 @@ export default defineComponent({ }) onMounted(() => { + if (route.path === '/hive-tables') { + router.replace({ + path: '/tables', + query: { + ...route.query, + }, + }) + return + } init() }) diff --git a/amoro-web/src/views/tables/components/TableExplorer.vue b/amoro-web/src/views/tables/components/TableExplorer.vue index f21946c9b..77546ef0b 100755 --- a/amoro-web/src/views/tables/components/TableExplorer.vue +++ b/amoro-web/src/views/tables/components/TableExplorer.vue @@ -41,6 +41,7 @@ interface TreeNode { db?: string table?: string tableType?: string + catalogNamespace?: string } const router = useRouter() @@ -48,6 +49,7 @@ const route = useRoute() const storageTableKey = 'easylake-menu-catalog-db-table' const expandedKeysSessionKeyPrefix = 'tables_expanded_keys' +const allNamespaceValue = '__all__' const state = reactive({ loading: false, @@ -58,6 +60,7 @@ const state = reactive({ selectedKeys: [] as string[], namespaces: [] as string[], selectedNamespace: '', + selectedFormat: '', namespaceMode: false, // Cache catalogList: [] as string[], @@ -65,6 +68,13 @@ const state = reactive({ tablesByCatalogDb: {} as Record<string, TableItem[]>, }) +const formatOptions = [ + { label: 'Iceberg', value: 'ICEBERG' }, + { label: 'Paimon', value: 'PAIMON' }, + { label: 'Hive', value: 'HIVE' }, + { label: 'Hudi', value: 'HUDI' }, +] + let namespaceGeneration = 0 let initialized = false @@ -72,21 +82,35 @@ function expandedKeysSessionKey() { return `${expandedKeysSessionKeyPrefix}:${state.selectedNamespace || 'default'}` } -function catalogDisplayName(catalog: string) { +function catalogPresentation(catalog: string) { + if (state.namespaceMode && state.selectedNamespace === allNamespaceValue) { + const separatorIndex = catalog.indexOf('@') + if (separatorIndex > 0 && separatorIndex < catalog.length - 1) { + return { + title: catalog.slice(separatorIndex + 1), + namespace: catalog.slice(0, separatorIndex), + } + } + } + if (!state.namespaceMode || !state.selectedNamespace) { - return catalog + return { title: catalog } } const prefix = `${state.selectedNamespace}@` - return catalog.startsWith(prefix) ? catalog.slice(prefix.length) : catalog + return { + title: catalog.startsWith(prefix) ? catalog.slice(prefix.length) : catalog, + } } function buildCatalogNode(catalog: string): TreeNode { + const presentation = catalogPresentation(catalog) return { key: `catalog:${catalog}`, - title: catalogDisplayName(catalog), + title: presentation.title, isLeaf: false, nodeType: 'catalog', catalog, + catalogNamespace: presentation.namespace, } } @@ -140,7 +164,10 @@ async function initRootCatalogs() { const generation = namespaceGeneration state.loading = true try { - const res = await getCatalogList(state.namespaceMode ? state.selectedNamespace : undefined) + const namespace = state.namespaceMode && state.selectedNamespace !== allNamespaceValue + ? state.selectedNamespace + : undefined + const res = await getCatalogList(namespace) if (generation !== namespaceGeneration) { return } @@ -160,6 +187,7 @@ function clearExplorerState() { state.loading = false state.searchKey = '' state.filterKey = '' + state.selectedFormat = '' state.treeData = [] state.expandedKeys = [] state.selectedKeys = [] @@ -168,6 +196,10 @@ function clearExplorerState() { state.tablesByCatalogDb = {} } +function handleFormatChange(format?: string) { + state.selectedFormat = format || '' +} + async function loadChildren(node: any) { const data = node?.dataRef || node if (!data) { @@ -286,13 +318,13 @@ function handleSelectTable(catalog: string, db: string, tableName: string, table catalog, database: db, tableName, + type, }) localStorage.setItem(`${storageTableKey}:${namespace}`, storedSelection) localStorage.setItem(storageTableKey, storedSelection) - const path = type === 'HIVE' ? '/hive-tables' : '/tables' const pathQuery = { - path, + path: '/tables', query: { ...(state.namespaceMode ? { namespace: state.selectedNamespace } : {}), catalog, @@ -399,7 +431,10 @@ function filterBySingleKeyword(nodes: TreeNode[], keyword: string, expandedSet: const result: TreeNode[] = [] nodes.forEach((node) => { - const titleMatch = node.title.toLowerCase().includes(keyword) + const searchText = node.catalogNamespace + ? `${node.title} @${node.catalogNamespace}` + : node.title + const titleMatch = searchText.toLowerCase().includes(keyword) let childrenMatches: TreeNode[] = [] if (node.children && node.children.length) { @@ -427,7 +462,10 @@ function filterByHierarchical(nodes: TreeNode[], parts: string[], expandedSet: S if (catalogNode.nodeType !== 'catalog') { return } - if (!catalogNode.title.toLowerCase().includes(catalogPart)) { + const catalogSearchText = catalogNode.catalogNamespace + ? `${catalogNode.title} @${catalogNode.catalogNamespace}` + : catalogNode.title + if (!catalogSearchText.toLowerCase().includes(catalogPart)) { return } @@ -492,6 +530,24 @@ function filterTree(source: TreeNode[], rawKeyword: string): { tree: TreeNode[], } } +function filterTreeByFormat(source: TreeNode[], format: string): TreeNode[] { + return source.reduce<TreeNode[]>((result, node) => { + if (node.nodeType === 'table') { + const tableType = (node.tableType || '').toUpperCase() + if (tableType === format || tableType.endsWith(`_${format}`)) { + result.push(node) + } + return result + } + + result.push({ + ...node, + children: node.children ? filterTreeByFormat(node.children, format) : node.children, + }) + return result + }, []) +} + let searchTimer: any = null watch( @@ -522,7 +578,9 @@ watch( } = (oldValue || {}) as any if (state.namespaceMode && namespace !== oldNamespace) { - const nextNamespace = state.namespaces.includes(namespace as string) ? namespace as string : '' + const nextNamespace = namespace === allNamespaceValue || state.namespaces.includes(namespace as string) + ? namespace as string + : '' if (nextNamespace !== state.selectedNamespace) { clearExplorerState() state.selectedNamespace = nextNamespace @@ -561,14 +619,17 @@ watch( ) const searchResult = computed(() => { + const source = state.selectedFormat + ? filterTreeByFormat(state.treeData, state.selectedFormat) + : state.treeData const keyword = normalizeKeyword(state.filterKey) if (!keyword) { return { - tree: state.treeData, + tree: source, expandedKeys: state.expandedKeys, } } - return filterTree(state.treeData, keyword) + return filterTree(source, keyword) }) const displayTreeData = computed(() => searchResult.value.tree) @@ -652,7 +713,9 @@ async function initializeNamespaces() { } const routeNamespace = (route.query.namespace as string) || '' - state.selectedNamespace = namespaces.includes(routeNamespace) ? routeNamespace : '' + state.selectedNamespace = routeNamespace === allNamespaceValue || namespaces.includes(routeNamespace) + ? routeNamespace + : '' } async function handleNamespaceChange(namespace?: string) { @@ -682,18 +745,29 @@ onBeforeMount(async () => { <template> <div class="table-explorer"> <div class="table-explorer-header"> - <div class="namespace-selector"> - <span class="namespace-label">Namespace:</span> + <div class="explorer-filter-row"> <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 + :value="state.namespaceMode ? (state.selectedNamespace || undefined) : 'default'" + :options="state.namespaceMode + ? [ + { label: 'All', value: allNamespaceValue }, + ...state.namespaces.map(namespace => ({ label: namespace, value: namespace })), + ] + : [{ label: 'default', value: 'default' }]" + :disabled="!state.namespaceMode" + :allow-clear="state.namespaceMode" + placeholder="Namespace" + class="explorer-filter-select namespace-filter-select" @change="handleNamespaceChange" /> - <span v-else class="namespace-default">default</span> + <a-select + :value="state.selectedFormat || undefined" + :options="formatOptions" + placeholder="Format" + class="explorer-filter-select format-filter-select" + allow-clear + @change="handleFormatChange" + /> </div> <a-input v-model:value="state.searchKey" @@ -732,6 +806,9 @@ onBeforeMount(async () => { <span class="tree-node-text"> {{ dataRef.title }} </span> + <span v-if="dataRef.catalogNamespace" class="catalog-namespace-label"> + @{{ dataRef.catalogNamespace }} + </span> </span> </template> </a-tree> @@ -759,43 +836,43 @@ onBeforeMount(async () => { padding: 0 7px 0 8px; margin-bottom: 8px; - .namespace-selector { + .explorer-filter-row { display: flex; - align-items: center; - min-height: 24px; + gap: 8px; margin-bottom: 8px; - font-size: 13px; - - .namespace-label { - margin-right: 6px; - color: #666; - } - .namespace-default { - color: #262626; - } - - .namespace-select { - flex: 1; + .explorer-filter-select { min-width: 0; :deep(.ant-select-selector) { - height: 24px; + height: 32px; + display: flex; + align-items: center; } :deep(.ant-select-selection-item), :deep(.ant-select-selection-placeholder) { - line-height: 22px; + line-height: 30px; + font-size: 14px; + } + + :deep(.ant-select-selection-placeholder) { + color: #ccc; } } + + .namespace-filter-select { + flex: 3; + } + + .format-filter-select { + flex: 2; + } } .search-input { width: 100%; - - :deep(.ant-input-affix-wrapper) { - height: 24px; - } + height: 32px; :deep(.ant-input-prefix) { display: flex; @@ -806,11 +883,11 @@ onBeforeMount(async () => { :deep(.search-input-prefix-icon) { font-size: 16px; - line-height: 24px; + line-height: 30px; } :deep(.ant-input) { - line-height: 24px; + line-height: 30px; font-size: 14px; &::placeholder { @@ -847,6 +924,13 @@ onBeforeMount(async () => { .tree-node-icon { margin-right: 6px; } + + .catalog-namespace-label { + margin-left: 4px; + color: #999; + font-size: 11px; + font-weight: normal; + } } .empty-placeholder { diff --git a/amoro-web/src/views/tables/index.vue b/amoro-web/src/views/tables/index.vue index b3413d982..0eb630be4 100644 --- a/amoro-web/src/views/tables/index.vue +++ b/amoro-web/src/views/tables/index.vue @@ -28,6 +28,7 @@ import UCleanup from './components/Cleanup.vue' import UProfiling from './components/Profiling.vue' import UHealthScore from './components/HealthScoreDetails.vue' import TableExplorer from './components/TableExplorer.vue' +import HiveTableDetails from '@/views/hive-details/index.vue' import useStore from '@/store/index' import type { IBaseDetailInfo } from '@/types/common.type' import { usePageScroll } from '@/hooks/usePageScroll' @@ -44,6 +45,7 @@ export default defineComponent({ UProfiling, UHealthScore, TableExplorer, + HiveTableDetails, }, setup() { const router = useRouter() @@ -139,6 +141,10 @@ export default defineComponent({ return state.baseInfo.tableType === 'ICEBERG' }) + const isHiveTable = computed(() => { + return String(route.query?.type || '').toUpperCase() === 'HIVE' + }) + const hasSelectedTable = computed(() => !!(route.query?.catalog && route.query?.db && route.query?.table)) const setBaseDetailInfo = (baseInfo: IBaseDetailInfo & { comment?: string }) => { @@ -232,6 +238,7 @@ export default defineComponent({ tabConfigs, store, isIceberg, + isHiveTable, hasSelectedTable, setBaseDetailInfo, handleTableNotFound, @@ -246,7 +253,7 @@ export default defineComponent({ </script> <template> - <div class="page-scroll" ref="pageScrollRef"> + <div ref="pageScrollRef" class="page-scroll"> <div class="tables-wrap"> <div v-if="!isSecondaryNav" class="tables-content"> <div @@ -258,45 +265,48 @@ export default defineComponent({ <div class="tables-divider" aria-hidden="true" @mousedown="startSidebarResize" /> <div class="tables-main"> <template v-if="hasSelectedTable"> - <div class="tables-main-header g-flex-jsb"> - <div class="g-flex-col"> - <div class="g-flex"> - <span :title="baseInfo.tableName" class="table-name g-text-nowrap">{{ baseInfo.tableName }}</span> - </div> - <div v-if="baseInfo.comment" class="table-info g-flex-ac"> - <p>{{ $t('Comment') }}: <span class="text-color">{{ baseInfo.comment }}</span></p> - </div> - <div class="table-info g-flex-ac"> - <p>{{ $t('optimizingStatus') }}: <span class="text-color">{{ baseInfo.optimizingStatus }}</span></p> - <a-divider type="vertical" /> - <p>{{ $t('records') }}: <span class="text-color">{{ baseInfo.records }}</span></p> - <a-divider type="vertical" /> - <template v-if="!isIceberg"> - <p>{{ $t('createTime') }}: <span class="text-color">{{ baseInfo.createTime }}</span></p> + <HiveTableDetails v-if="isHiveTable" /> + <template v-else> + <div class="tables-main-header g-flex-jsb"> + <div class="g-flex-col"> + <div class="g-flex"> + <span :title="baseInfo.tableName" class="table-name g-text-nowrap">{{ baseInfo.tableName }}</span> + </div> + <div v-if="baseInfo.comment" class="table-info g-flex-ac"> + <p>{{ $t('Comment') }}: <span class="text-color">{{ baseInfo.comment }}</span></p> + </div> + <div class="table-info g-flex-ac"> + <p>{{ $t('optimizingStatus') }}: <span class="text-color">{{ baseInfo.optimizingStatus }}</span></p> <a-divider type="vertical" /> - </template> - <p>{{ $t('tableFormat') }}: <span class="text-color">{{ baseInfo.tableFormat }}</span></p> - <a-divider type="vertical" /> - <p> - {{ $t('healthScore') }}: - <UHealthScore :base-info="baseInfo" /> - </p> + <p>{{ $t('records') }}: <span class="text-color">{{ baseInfo.records }}</span></p> + <a-divider type="vertical" /> + <template v-if="!isIceberg"> + <p>{{ $t('createTime') }}: <span class="text-color">{{ baseInfo.createTime }}</span></p> + <a-divider type="vertical" /> + </template> + <p>{{ $t('tableFormat') }}: <span class="text-color">{{ baseInfo.tableFormat }}</span></p> + <a-divider type="vertical" /> + <p> + {{ $t('healthScore') }}: + <UHealthScore :base-info="baseInfo" /> + </p> + </div> </div> </div> - </div> - <div class="tables-main-body"> - <a-tabs v-model:activeKey="activeKey" destroy-inactive-tab-pane @change="onChangeTab"> - <a-tab-pane key="Details" :tab="$t('details')" force-render> - <UDetails ref="detailRef" @set-base-detail-info="setBaseDetailInfo" @table-not-found="handleTableNotFound" /> - </a-tab-pane> - <a-tab-pane v-if="detailLoaded" key="Files" :tab="$t('files')"> - <UFiles :has-partition="baseInfo.hasPartition" /> - </a-tab-pane> - <a-tab-pane v-for="tab in tabConfigs" :key="tab.key" :tab="$t(tab.label)"> - <component :is="`U${tab.key}`" /> - </a-tab-pane> - </a-tabs> - </div> + <div class="tables-main-body"> + <a-tabs v-model:activeKey="activeKey" destroy-inactive-tab-pane @change="onChangeTab"> + <a-tab-pane key="Details" :tab="$t('details')" force-render> + <UDetails ref="detailRef" @set-base-detail-info="setBaseDetailInfo" @table-not-found="handleTableNotFound" /> + </a-tab-pane> + <a-tab-pane v-if="detailLoaded" key="Files" :tab="$t('files')"> + <UFiles :has-partition="baseInfo.hasPartition" /> + </a-tab-pane> + <a-tab-pane v-for="tab in tabConfigs" :key="tab.key" :tab="$t(tab.label)"> + <component :is="`U${tab.key}`" /> + </a-tab-pane> + </a-tabs> + </div> + </template> </template> <div v-else class="empty-page" /> </div> diff --git a/dist/src/main/amoro-bin/bin/load-config.sh b/dist/src/main/amoro-bin/bin/load-config.sh index 69dc95462..aad3adadd 100644 --- a/dist/src/main/amoro-bin/bin/load-config.sh +++ b/dist/src/main/amoro-bin/bin/load-config.sh @@ -70,6 +70,16 @@ export JVM_EXTRA_CONFIG test -f ${AMORO_ENV_FILE} && source ${AMORO_ENV_FILE} +# LAS deployments expose the existing service credential names. Proton's built-in +# AssumeIamRoleCredentialProvider uses fixed environment names, so keep those as internal aliases +# without requiring a second pair of K8S Secret keys. +if [ -n "${LAS_SERVICE_AK}" ] && [ -z "${ASSUME_ROLE_ACCESS_KEY}" ]; then + export ASSUME_ROLE_ACCESS_KEY="${LAS_SERVICE_AK}" +fi +if [ -n "${LAS_SERVICE_SK}" ] && [ -z "${ASSUME_ROLE_SECRET_KEY}" ]; then + export ASSUME_ROLE_SECRET_KEY="${LAS_SERVICE_SK}" +fi + # set env variable amoro-addition-classpath if not exists if [ -z "${AMORO_ADDITION_CLASSPATH}" ]; then export AMORO_ADDITION_CLASSPATH= diff --git a/dist/src/main/amoro-bin/conf/config.yaml b/dist/src/main/amoro-bin/conf/config.yaml index 2c2ad6543..4aff943b0 100644 --- a/dist/src/main/amoro-bin/conf/config.yaml +++ b/dist/src/main/amoro-bin/conf/config.yaml @@ -75,9 +75,8 @@ ams: # Values may also be supplied through environment variables, for example: # AMS_LAS_INTEGRATION_ENABLED=true # AMS_LAS_INTEGRATION_HMS__URI=thrift://hms-service.<namespace>.svc.cluster.local:9083 - # Keep IAM bootstrap credentials in secret-backed environment variables: - # AMS_LAS_INTEGRATION_IAM_BOOTSTRAP__ACCESS__KEY and - # AMS_LAS_INTEGRATION_IAM_BOOTSTRAP__SECRET__KEY. + # Reuse the existing LAS service credential environment variables. Inject both from a K8S + # Secret; do not put their values in this file: LAS_SERVICE_AK and LAS_SERVICE_SK. las: integration: enabled: false @@ -91,8 +90,9 @@ ams: connect-timeout: 30s read-timeout: 30s iam: - # bootstrap-access-key: supplied by a secret-backed environment variable - # bootstrap-secret-key: supplied by a secret-backed environment variable + # Optional explicit overrides. LAS deployments should use LAS_SERVICE_AK/LAS_SERVICE_SK. + # bootstrap-access-key: supplied by a secret-backed configuration + # bootstrap-secret-key: supplied by a secret-backed configuration # bootstrap-session-token: optional, supplied with temporary bootstrap credentials role-session-name: AmoroAssumeRoleSession assume-role-ttl: 1h @@ -152,6 +152,7 @@ ams: # catalog explorer. LAS deployments typically set refresh-external-catalogs.interval to 10min. catalog: namespace-enabled: true + namespace-allowlist-enabled: true # Support for encrypted sensitive configuration items shade: diff --git a/pom.xml b/pom.xml index edba971c8..e8b666070 100644 --- a/pom.xml +++ b/pom.xml @@ -1456,6 +1456,18 @@ <url>https://repository.apache.org/content/repositories/releases/</url> </repository> + <repository> + <releases> + <enabled>true</enabled> + </releases> + <snapshots> + <enabled>false</enabled> + </snapshots> + <id>bytedance-releases</id> + <name>ByteDance Maven Releases</name> + <url>https://maven.byted.org/repository/releases/</url> + </repository> + <!-- Uncomment the following "repository" tag if your maven fails to download amoro-shade snapshot files. We comment these tags by default because downloading from apache repositories is very slow for both github and developers.
