From d929afd7e30c49713cdbc7e38a03f5cd37c2cece Mon Sep 17 00:00:00 2001 From: WenLingzhang <1298877813@qq.com> Date: Thu, 3 Sep 2026 09:43:09 +0800 Subject: [PATCH 1/2] add lance-hive catalog --- .../amoro/server/catalog/CatalogBuilder.java | 14 +- .../server/catalog/DefaultCatalogManager.java | 5 +- .../controller/CatalogController.java | 7 + .../org/apache/amoro/utils/CatalogUtil.java | 28 ++- amoro-format-lance/pom.xml | 35 ++++ .../formats/lance/AbstractLanceCatalog.java | 163 ++++++++++++++++-- .../formats/lance/LanceCatalogFactory.java | 3 + .../lance/LanceDirectoryV1Catalog.java | 6 +- .../amoro/formats/lance/LanceHiveCatalog.java | 161 +++++++++++++++++ .../amoro/formats/lance/LanceRestCatalog.java | 141 +-------------- amoro-web/src/views/catalogs/Detail.vue | 53 ++++-- 11 files changed, 436 insertions(+), 180 deletions(-) create mode 100644 amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceHiveCatalog.java 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 408a8c34a8..b9c9101f7f 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)); @@ -86,7 +87,7 @@ public static ServerCatalog buildServerCatalog( "Table format %s is not supported for metastore type: %s", tableFormats, type); - checkRestLanceExclusivity(type, tableFormats); + checkLanceExclusivity(type, tableFormats); switch (type) { case CATALOG_TYPE_HADOOP: @@ -104,13 +105,14 @@ public static ServerCatalog buildServerCatalog( } /** Enforced on both catalog creation and metadata updates. */ - static void checkRestLanceExclusivity(String type, Set tableFormats) { + static void checkLanceExclusivity(String type, Set tableFormats) { + boolean lanceExclusiveType = CATALOG_TYPE_REST.equals(type) || CATALOG_TYPE_HIVE.equals(type); Preconditions.checkState( - !(CATALOG_TYPE_REST.equals(type) + !(lanceExclusiveType && tableFormats.contains(TableFormat.LANCE) && tableFormats.size() > 1), - "REST catalog serves a single protocol per uri," - + " Lance cannot be combined with other table formats: %s", + "Lance cannot be combined with other table formats in a %s catalog: %s", + type, tableFormats); } } 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 2ea31ff2a6..98f8114b32 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 @@ -276,11 +276,10 @@ private void validateCatalogUpdate(CatalogMeta oldMeta, CatalogMeta newMeta) { throw new IllegalMetadataException("Cannot update catalog type"); } - // A REST catalog serves a single protocol per uri, so Lance cannot be combined with other - // table formats. Reject the update before it is persisted. + // Lance cannot mix with other table formats in REST/Hive catalogs; reject before persisting. String type = newMeta.getCatalogType(); Set tableFormats = CatalogUtil.tableFormats(newMeta); - CatalogBuilder.checkRestLanceExclusivity(type, tableFormats); + CatalogBuilder.checkLanceExclusivity(type, tableFormats); } private void disposeCatalog(String name) { diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/CatalogController.java b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/CatalogController.java index b7dcd5651b..6d4c0b5eca 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/CatalogController.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/CatalogController.java @@ -217,6 +217,13 @@ public class CatalogController { CatalogDescriptor.of(CATALOG_TYPE_REST, STORAGE_CONFIGS_VALUE_TYPE_HADOOP, LANCE)); VALIDATE_CATALOGS.add( CatalogDescriptor.of(CATALOG_TYPE_REST, STORAGE_CONFIGS_VALUE_TYPE_OSS, LANCE)); + + VALIDATE_CATALOGS.add( + CatalogDescriptor.of(CATALOG_TYPE_HIVE, STORAGE_CONFIGS_VALUE_TYPE_S3, LANCE)); + VALIDATE_CATALOGS.add( + CatalogDescriptor.of(CATALOG_TYPE_HIVE, STORAGE_CONFIGS_VALUE_TYPE_OSS, LANCE)); + VALIDATE_CATALOGS.add( + CatalogDescriptor.of(CATALOG_TYPE_HIVE, STORAGE_CONFIGS_VALUE_TYPE_LOCAL, LANCE)); } private final PlatformFileManager platformFileInfoService; diff --git a/amoro-common/src/main/java/org/apache/amoro/utils/CatalogUtil.java b/amoro-common/src/main/java/org/apache/amoro/utils/CatalogUtil.java index 46285b9d0d..f499065918 100644 --- a/amoro-common/src/main/java/org/apache/amoro/utils/CatalogUtil.java +++ b/amoro-common/src/main/java/org/apache/amoro/utils/CatalogUtil.java @@ -121,11 +121,9 @@ public static Map mergeCatalogPropertiesToTable( /** Build {@link TableMetaStore} from catalog meta. */ public static TableMetaStore buildMetaStore(CatalogMeta catalogMeta) { - // REST + Lance manages storage/auth server-side, so a client-side TableMetaStore is - // unnecessary. Set a local configuration so build() returns a disabled-auth store. - if (CatalogMetaProperties.CATALOG_TYPE_REST.equalsIgnoreCase(catalogMeta.getCatalogType()) - && isLanceOnly(catalogMeta)) { - return TableMetaStore.builder().withConfiguration(new Configuration()).build(); + // Lance over REST/Hive catalog authenticates on its own; EMPTY is a disabled-auth store. + if (isLanceManagedExternally(catalogMeta)) { + return TableMetaStore.EMPTY; } // load storage configs @@ -222,9 +220,25 @@ && isLanceOnly(catalogMeta)) { return builder.build(); } - private static boolean isLanceOnly(CatalogMeta catalogMeta) { + private static boolean isLanceManagedExternally(CatalogMeta catalogMeta) { Set formats = tableFormats(catalogMeta); - return formats.size() == 1 && formats.contains(TableFormat.LANCE); + if (formats.size() != 1 || !formats.contains(TableFormat.LANCE)) { + return false; + } + + String type = catalogMeta.getCatalogType(); + Map authConfigs = catalogMeta.getAuthConfigs(); + if (CatalogMetaProperties.CATALOG_TYPE_HIVE.equalsIgnoreCase(type) + && authConfigs != null + && CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_KERBEROS.equalsIgnoreCase( + authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_TYPE))) { + throw new IllegalArgumentException( + "Kerberos auth is not supported for Lance catalogs backed by a Hive metastore: " + + catalogMeta.getCatalogName()); + } + + return CatalogMetaProperties.CATALOG_TYPE_REST.equalsIgnoreCase(type) + || CatalogMetaProperties.CATALOG_TYPE_HIVE.equalsIgnoreCase(type); } public static TableIdentifier tableId(TableMeta tableMeta) { diff --git a/amoro-format-lance/pom.xml b/amoro-format-lance/pom.xml index 4d546422ca..4ca948e620 100755 --- a/amoro-format-lance/pom.xml +++ b/amoro-format-lance/pom.xml @@ -71,6 +71,41 @@ lance-namespace-apache-client 0.8.0 + + + + org.lance + lance-namespace-hive3 + 0.4.1 + + + org.lance + lance-namespace-core + + + io.netty + netty + + + io.netty + netty-all + + + + org.apache.orc + orc-core + + + org.apache.hadoop + hadoop-common + + + org.apache.hadoop + hadoop-hdfs + + + org.apache.iceberg iceberg-aws diff --git a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/AbstractLanceCatalog.java b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/AbstractLanceCatalog.java index 420299c3ae..949a894434 100644 --- a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/AbstractLanceCatalog.java +++ b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/AbstractLanceCatalog.java @@ -18,36 +18,102 @@ package org.apache.amoro.formats.lance; +import org.apache.amoro.AlreadyExistsException; import org.apache.amoro.AmoroTable; import org.apache.amoro.FormatCatalog; import org.apache.amoro.NoSuchDatabaseException; import org.apache.amoro.NoSuchTableException; import org.apache.amoro.table.TableIdentifier; +import org.apache.arrow.util.Preconditions; +import org.apache.commons.lang3.StringUtils; import org.lance.Dataset; import org.lance.namespace.LanceNamespace; +import org.lance.namespace.errors.InvalidInputException; +import org.lance.namespace.errors.NamespaceAlreadyExistsException; +import org.lance.namespace.errors.NamespaceNotFoundException; import org.lance.namespace.errors.TableNotFoundException; +import org.lance.namespace.model.CreateNamespaceRequest; +import org.lance.namespace.model.DropNamespaceRequest; import org.lance.namespace.model.DropTableRequest; +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.ArrayList; import java.util.Collections; +import java.util.LinkedHashSet; import java.util.List; +import java.util.Map; +import java.util.Set; /** Base implementation of {@link FormatCatalog} for Lance. */ public abstract class AbstractLanceCatalog implements FormatCatalog { + protected static final String URI_PROPERTY = "uri"; + + private static final String CATALOG_PROPERTY = "catalog-name"; + protected final String catalogName; protected final LanceNamespace namespace; + private final List catalogPrefix; protected AbstractLanceCatalog(String catalogName, LanceNamespace namespace) { + this(catalogName, namespace, Collections.emptyList()); + } + + protected AbstractLanceCatalog( + String catalogName, LanceNamespace namespace, List catalogPrefix) { this.catalogName = catalogName; this.namespace = namespace; + this.catalogPrefix = catalogPrefix; + } + + protected static List catalogPrefix( + Map catalogProperties, String defaultCatalog) { + String catalog = catalogProperties == null ? null : catalogProperties.get(CATALOG_PROPERTY); + if (StringUtils.isEmpty(catalog)) { + catalog = defaultCatalog; + } + return StringUtils.isEmpty(catalog) + ? Collections.emptyList() + : Collections.singletonList(catalog); } /** Composes the namespace table id used to reference a single table. */ - protected abstract List tableId(String database, String table); + protected List tableId(String database, String table) { + return concat(catalogPrefix, database, table); + } /** Composes the namespace id used to list tables of a database. */ - protected abstract List tableIdForListTables(String database); + protected List namespaceId(String database) { + return concat(catalogPrefix, database); + } + + protected static List concat(List prefix, String... parts) { + List id = new ArrayList<>(prefix); + Collections.addAll(id, parts); + return id; + } + + protected static void validateProperties(Map properties) { + Preconditions.checkArgument( + properties != null && !properties.isEmpty(), "Catalog properties must be set."); + } + + protected AmoroTable openTable(String database, String table) { + namespace.tableExists(new TableExistsRequest().id(tableId(database, table))); + Dataset dataset = + Dataset.open().namespaceClient(namespace).tableId(tableId(database, table)).build(); + return new LanceTable( + TableIdentifier.of(catalogName, database, table), dataset, Collections.emptyMap()); + } + + protected void dropInNamespace(String database, String table, boolean purge) { + namespace.dropTable(new DropTableRequest().id(tableId(database, table))); + } @Override public boolean tableExists(String database, String table) { @@ -58,21 +124,20 @@ public boolean tableExists(String database, String table) { try { namespace.tableExists(new TableExistsRequest().id(tableId(database, table))); return true; - } catch (TableNotFoundException e) { + } catch (TableNotFoundException | InvalidInputException e) { return false; } } @Override - public AmoroTable loadTable(String database, String tableName) { - if (!tableExists(database, tableName)) { - throw new NoSuchTableException("Table: " + database + "." + tableName + " does not exist"); - } + public AmoroTable loadTable(String database, String table) { + validateDatabase(database); - TableIdentifier identifier = TableIdentifier.of(catalogName, database, tableName); - Dataset dataset = - Dataset.open().namespaceClient(namespace).tableId(tableId(database, tableName)).build(); - return new LanceTable(identifier, dataset, Collections.emptyMap()); + try { + return openTable(database, table); + } catch (TableNotFoundException | InvalidInputException e) { + throw new NoSuchTableException("Table: " + database + "." + table + " does not exist", e); + } } @Override @@ -80,22 +145,90 @@ public boolean dropTable(String database, String table, boolean purge) { validateDatabase(database); try { - namespace.dropTable(new DropTableRequest().id(tableId(database, table))); + dropInNamespace(database, table, purge); return true; - } catch (TableNotFoundException e) { + } catch (TableNotFoundException | InvalidInputException e) { return false; } } @Override - public abstract List listTables(String database); + public List listTables(String database) { + validateDatabase(database); + + Set tables = new LinkedHashSet<>(); + String pageToken = null; + do { + ListTablesRequest request = new ListTablesRequest().id(namespaceId(database)); + if (pageToken != null) { + request.pageToken(pageToken); + } + + ListTablesResponse response = namespace.listTables(request); + if (response == null) { + break; + } + if (response.getTables() != null) { + tables.addAll(response.getTables()); + } + pageToken = response.getPageToken(); + } while (StringUtils.isNotEmpty(pageToken)); + + return new ArrayList<>(tables); + } @Override - public abstract List listDatabases(); + public List listDatabases() { + Set namespaces = new LinkedHashSet<>(); + String pageToken = null; + do { + ListNamespacesRequest request = new ListNamespacesRequest().id(catalogPrefix); + if (pageToken != null) { + request.pageToken(pageToken); + } + + ListNamespacesResponse response = namespace.listNamespaces(request); + if (response == null) { + break; + } + if (response.getNamespaces() != null) { + namespaces.addAll(response.getNamespaces()); + } + + pageToken = response.getPageToken(); + } while (StringUtils.isNotEmpty(pageToken)); + + return new ArrayList<>(namespaces); + } protected void validateDatabase(String database) { if (!databaseExists(database)) { throw new NoSuchDatabaseException("Database: " + database + " does not exist"); } } + + @Override + public boolean databaseExists(String database) { + try { + namespace.namespaceExists(new NamespaceExistsRequest().id(namespaceId(database))); + return true; + } catch (NamespaceNotFoundException e) { + return false; + } + } + + @Override + public void createDatabase(String database) { + try { + namespace.createNamespace(new CreateNamespaceRequest().id(namespaceId(database))); + } catch (NamespaceAlreadyExistsException e) { + throw new AlreadyExistsException("Database: " + database + " already exists", e); + } + } + + @Override + public void dropDatabase(String database) { + validateDatabase(database); + namespace.dropNamespace(new DropNamespaceRequest().id(namespaceId(database))); + } } 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 a3faa8b2de..b0d3b7f3ea 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 @@ -42,6 +42,9 @@ public FormatCatalog create( if (CatalogMetaProperties.CATALOG_TYPE_REST.equals(metastoreType)) { return new LanceRestCatalog(catalogName, properties); } + if (CatalogMetaProperties.CATALOG_TYPE_HIVE.equals(metastoreType)) { + return new LanceHiveCatalog(catalogName, properties, metaStore); + } return new LanceDirectoryV1Catalog(catalogName, properties); } diff --git a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceDirectoryV1Catalog.java b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceDirectoryV1Catalog.java index fcabb8d71c..baabdf7651 100755 --- a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceDirectoryV1Catalog.java +++ b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceDirectoryV1Catalog.java @@ -52,9 +52,7 @@ public LanceDirectoryV1Catalog(String catalogName, Map catalogPr } private static LanceNamespace buildNamespace(Map catalogProperties) { - Preconditions.checkArgument( - catalogProperties != null && !catalogProperties.isEmpty(), - "Catalog properties must be set."); + validateProperties(catalogProperties); Map namespaceProperties = new HashMap<>(catalogProperties); String root = namespaceProperties.remove(CatalogMetaProperties.KEY_WAREHOUSE); String storageAccessKey = @@ -89,7 +87,7 @@ protected List tableId(String database, String table) { } @Override - protected List tableIdForListTables(String database) { + protected List namespaceId(String database) { return Collections.emptyList(); } diff --git a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceHiveCatalog.java b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceHiveCatalog.java new file mode 100644 index 0000000000..bbe771fa22 --- /dev/null +++ b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceHiveCatalog.java @@ -0,0 +1,161 @@ +/* + * 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.properties.CatalogMetaProperties; +import org.apache.amoro.table.TableIdentifier; +import org.apache.amoro.table.TableMetaStore; +import org.apache.amoro.utils.CatalogUtil; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.util.Preconditions; +import org.apache.commons.lang3.StringUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.aliyun.AliyunProperties; +import org.apache.iceberg.aws.AwsClientProperties; +import org.apache.iceberg.aws.s3.S3FileIOProperties; +import org.lance.Dataset; +import org.lance.ReadOptions; +import org.lance.namespace.LanceNamespace; +import org.lance.namespace.hive3.Hive3Namespace; +import org.lance.namespace.hive3.Hive3NamespaceConfig; +import org.lance.namespace.model.DeregisterTableRequest; +import org.lance.namespace.model.DescribeTableRequest; +import org.lance.namespace.model.DescribeTableResponse; +import org.lance.namespace.model.DropTableRequest; + +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; + +/** Lance catalog backed by a Hive metastore; metadata is delegated to {@link Hive3Namespace}. */ +public class LanceHiveCatalog extends AbstractLanceCatalog { + + private static final String HIVE_METASTORE_URIS = "hive.metastore.uris"; + + /** Escape hatch for storage options Amoro does not model, e.g. MinIO's aws_allow_http. */ + private static final String STORAGE_OPTIONS_PREFIX = "lance.storage-options."; + + private final Map catalogProperties; + + public LanceHiveCatalog( + String catalogName, Map catalogProperties, TableMetaStore metaStore) { + super( + catalogName, + buildNamespace(catalogProperties, metaStore), + catalogPrefix(catalogProperties, Hive3NamespaceConfig.CATALOG_NAME)); + this.catalogProperties = new HashMap<>(catalogProperties); + } + + @Override + protected AmoroTable openTable(String database, String table) { + DescribeTableResponse response = + namespace.describeTable(new DescribeTableRequest().id(tableId(database, table))); + return new LanceTable( + TableIdentifier.of(catalogName, database, table), + openDataset(response.getLocation()), + response.getProperties()); + } + + /** Purge deletes the table data; otherwise only the metastore entry is deregistered. */ + @Override + protected void dropInNamespace(String database, String table, boolean purge) { + if (purge) { + namespace.dropTable(new DropTableRequest().id(tableId(database, table))); + } else { + namespace.deregisterTable(new DeregisterTableRequest().id(tableId(database, table))); + } + } + + private Dataset openDataset(String location) { + ReadOptions readOptions = + new ReadOptions.Builder() + .setStorageOptions(storageOptions(catalogProperties, location)) + .build(); + return Dataset.open().uri(location).readOptions(readOptions).build(); + } + + private static LanceNamespace buildNamespace( + Map catalogProperties, TableMetaStore metaStore) { + validateProperties(catalogProperties); + + String root = catalogProperties.get(CatalogMetaProperties.KEY_WAREHOUSE); + Preconditions.checkArgument( + StringUtils.isNotBlank(root), "Warehouse must be set in catalogProperties."); + Preconditions.checkArgument( + !"hdfs".equals(scheme(root.trim())), "Unsupported warehouse scheme: " + root); + + String uri = catalogProperties.get(URI_PROPERTY); + Preconditions.checkArgument( + StringUtils.isNotBlank(uri), "Hive metastore uri must be set, e.g. thrift://host:9083."); + + Configuration conf = new Configuration(metaStore.getConfiguration()); + conf.set(HIVE_METASTORE_URIS, uri.trim()); + + Map namespaceProperties = new HashMap<>(); + namespaceProperties.put(Hive3NamespaceConfig.ROOT, root.trim()); + + Hive3Namespace namespace = new Hive3Namespace(); + namespace.setHadoopConf(conf); + namespace.initialize(namespaceProperties, new RootAllocator(Long.MAX_VALUE)); + return namespace; + } + + /** Maps the catalog credentials onto Lance storage_options of the location's scheme. */ + static Map storageOptions( + Map catalogProperties, String location) { + Map options = new HashMap<>(); + String scheme = scheme(location); + if ("s3".equals(scheme)) { + CatalogUtil.copyProperty( + catalogProperties, options, S3FileIOProperties.ACCESS_KEY_ID, "aws_access_key_id"); + CatalogUtil.copyProperty( + catalogProperties, + options, + S3FileIOProperties.SECRET_ACCESS_KEY, + "aws_secret_access_key"); + CatalogUtil.copyProperty( + catalogProperties, options, S3FileIOProperties.ENDPOINT, "aws_endpoint"); + CatalogUtil.copyProperty( + catalogProperties, options, AwsClientProperties.CLIENT_REGION, "aws_region"); + } else if ("oss".equals(scheme)) { + CatalogUtil.copyProperty( + catalogProperties, options, AliyunProperties.CLIENT_ACCESS_KEY_ID, "access_key_id"); + CatalogUtil.copyProperty( + catalogProperties, + options, + AliyunProperties.CLIENT_ACCESS_KEY_SECRET, + "access_key_secret"); + CatalogUtil.copyProperty( + catalogProperties, options, AliyunProperties.OSS_ENDPOINT, "endpoint"); + } + + // lance.storage-options.* overrides win over the translated defaults. + for (Map.Entry entry : catalogProperties.entrySet()) { + if (entry.getKey().startsWith(STORAGE_OPTIONS_PREFIX)) { + options.put(entry.getKey().substring(STORAGE_OPTIONS_PREFIX.length()), entry.getValue()); + } + } + return options; + } + + private static String scheme(String uri) { + return StringUtils.substringBefore(uri, "://").toLowerCase(Locale.ROOT); + } +} diff --git a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceRestCatalog.java b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceRestCatalog.java index c665f16f12..d33ec9021c 100644 --- a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceRestCatalog.java +++ b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceRestCatalog.java @@ -18,28 +18,12 @@ package org.apache.amoro.formats.lance; -import org.apache.amoro.AlreadyExistsException; import org.apache.arrow.memory.RootAllocator; import org.apache.arrow.util.Preconditions; -import org.apache.commons.lang3.StringUtils; import org.lance.namespace.LanceNamespace; -import org.lance.namespace.errors.NamespaceAlreadyExistsException; -import org.lance.namespace.errors.NamespaceNotFoundException; -import org.lance.namespace.model.CreateNamespaceRequest; -import org.lance.namespace.model.DropNamespaceRequest; -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 java.util.ArrayList; -import java.util.Collections; import java.util.HashMap; -import java.util.LinkedHashSet; -import java.util.List; import java.util.Map; -import java.util.Set; /** * REST catalog implementation for Lance. @@ -50,133 +34,20 @@ */ public class LanceRestCatalog extends AbstractLanceCatalog { - private static final String URI_PROPERTY = "uri"; - - /** - * Optional catalog name prepended to the REST namespace id, mapping Amoro's database/table onto - * the service's schema/table levels (catalog/schema/table hierarchy). Empty when the service has - * no catalog level. - */ - private static final String CATALOG_PROPERTY = "catalog-name"; - - private final List catalogPrefix; - public LanceRestCatalog(String catalogName, Map catalogProperties) { - super(catalogName, buildNamespace(catalogProperties)); - this.catalogPrefix = catalogPrefix(catalogProperties); - } - - private static List catalogPrefix(Map catalogProperties) { - String catalog = catalogProperties.get(CATALOG_PROPERTY); - return catalog == null || catalog.isEmpty() - ? Collections.emptyList() - : Collections.singletonList(catalog); + super( + catalogName, + buildNamespace(catalogProperties), + // The REST service may or may not expose a catalog level. + catalogPrefix(catalogProperties, null)); } private static LanceNamespace buildNamespace(Map namespaceProperties) { - Preconditions.checkArgument( - namespaceProperties != null && !namespaceProperties.isEmpty(), - "Catalog properties must be set."); + validateProperties(namespaceProperties); Preconditions.checkArgument( namespaceProperties.containsKey(URI_PROPERTY), "URI must be set in namespaceProperties for REST metastore type."); Map restProperties = new HashMap<>(namespaceProperties); return LanceNamespace.connect("rest", restProperties, new RootAllocator(Long.MAX_VALUE)); } - - @Override - protected List tableId(String database, String table) { - return concat(catalogPrefix, database, table); - } - - @Override - protected List tableIdForListTables(String database) { - return concat(catalogPrefix, database); - } - - private static List concat(List prefix, String... parts) { - List id = new ArrayList<>(prefix); - Collections.addAll(id, parts); - return id; - } - - /** - * Lists the tables of a database. The REST service pages results via pageToken, so follow the - * token until it is empty to avoid silently missing tables on large catalogs. - */ - @Override - public List listTables(String database) { - validateDatabase(database); - - Set tables = new LinkedHashSet<>(); - String pageToken = null; - do { - ListTablesRequest request = new ListTablesRequest().id(tableIdForListTables(database)); - if (pageToken != null) { - request.pageToken(pageToken); - } - ListTablesResponse response = namespace.listTables(request); - if (response == null) { - break; - } - if (response.getTables() != null) { - tables.addAll(response.getTables()); - } - - pageToken = response.getPageToken(); - } while (StringUtils.isNotEmpty(pageToken)); - - return new ArrayList<>(tables); - } - - /** Lists the schemas under the catalog, following the REST pageToken until it is empty. */ - @Override - public List listDatabases() { - Set namespaces = new LinkedHashSet<>(); - String pageToken = null; - do { - ListNamespacesRequest request = new ListNamespacesRequest().id(catalogPrefix); - if (pageToken != null) { - request.pageToken(pageToken); - } - ListNamespacesResponse response = namespace.listNamespaces(request); - if (response == null) { - break; - } - if (response.getNamespaces() != null) { - namespaces.addAll(response.getNamespaces()); - } - - pageToken = response.getPageToken(); - } while (StringUtils.isNotEmpty(pageToken)); - - return new ArrayList<>(namespaces); - } - - @Override - public boolean databaseExists(String database) { - try { - NamespaceExistsRequest request = - new NamespaceExistsRequest().id(concat(catalogPrefix, database)); - namespace.namespaceExists(request); - return true; - } catch (NamespaceNotFoundException e) { - return false; - } - } - - @Override - public void createDatabase(String database) { - try { - namespace.createNamespace(new CreateNamespaceRequest().id(concat(catalogPrefix, database))); - } catch (NamespaceAlreadyExistsException e) { - throw new AlreadyExistsException("Database: " + database + " already exists", e); - } - } - - @Override - public void dropDatabase(String database) { - validateDatabase(database); - namespace.dropNamespace(new DropNamespaceRequest().id(concat(catalogPrefix, database))); - } } diff --git a/amoro-web/src/views/catalogs/Detail.vue b/amoro-web/src/views/catalogs/Detail.vue index ce6bd55732..d14c20982a 100644 --- a/amoro-web/src/views/catalogs/Detail.vue +++ b/amoro-web/src/views/catalogs/Detail.vue @@ -191,6 +191,14 @@ const defaultPropertiesMap = { PAIMON: ['warehouse'], } +const LANCE_EXCLUSIVE_TYPES = ['rest', 'hive'] + +// Hive + Lance reads metastore uri and warehouse root from catalog properties. +const HIVE_LANCE_PROPERTIES = ['uri', 'warehouse'] +const isHiveLance = computed(() => { + return formState.catalog.type === 'hive' && formState.tableFormatList.includes(tableFormatMap.LANCE) +}) + watch(() => route.query, (value) => { value && initData() }, { @@ -250,9 +258,8 @@ async function loadMetastoreCapabilities(loadTableFormats: boolean) { const options = (formats || []) as string[] tableFormatOptions.value = options if (loadTableFormats) { - // REST catalog: Lance and Iceberg use different protocols (one protocol per uri), - // so they can't coexist. Default to Iceberg family; Lance stays unchecked (opt-in). - const defaultFormats = type === 'rest' + // Lance is opt-in: including it in the default would collapse the selection onto Lance. + const defaultFormats = LANCE_EXCLUSIVE_TYPES.includes(type) ? options.filter(format => format !== tableFormatMap.LANCE) : options formState.tableFormatList = defaultFormats.length > 0 ? [...defaultFormats] : [...options] @@ -361,7 +368,10 @@ async function getConfigInfo() { async function changeProperties() { const properties = await propertiesRef.value.getPropertiesWithoputValidation() - const catalogKeys = defaultPropertiesMap[formState.catalog.type as keyof typeof defaultPropertiesMap] || [] + const catalogKeys = [ + ...(defaultPropertiesMap[formState.catalog.type as keyof typeof defaultPropertiesMap] || []), + ...(isHiveLance.value ? HIVE_LANCE_PROPERTIES : []), + ] catalogKeys.forEach((key) => { if (key && !properties[key]) { properties[key] = '' @@ -406,12 +416,22 @@ const isRestLance = computed(() => { return formState.catalog.type === 'rest' && formState.tableFormatList.includes(tableFormatMap.LANCE) }) +// Lance whitelists S3/OSS/Local only; hide Hadoop while Lance is selected. +const storageTypeOptionsView = computed(() => { + return formState.tableFormatList.includes(tableFormatMap.LANCE) + ? storageTypeOptions.value.filter(option => option.value !== 'Hadoop') + : storageTypeOptions.value +}) + watch( () => formState.storageConfig['storage.type'], (storageType) => { if (storageType === 'Local') { if (isNewCatalog.value) { - formState.tableFormatList = [tableFormatMap.ICEBERG] + // Hive + Local only serves Lance; the other metastores still default to Iceberg. + formState.tableFormatList = formState.catalog.type === 'hive' + ? [tableFormatMap.LANCE] + : [tableFormatMap.ICEBERG] } if (!formState.authConfig['auth.type']) { formState.authConfig['auth.type'] = 'SIMPLE' @@ -426,17 +446,30 @@ watch( watch( () => formState.tableFormatList, - (formats, oldFormats) => { - const isRest = formState.catalog.type === 'rest' + async (formats, oldFormats) => { const hasLance = formats.includes(tableFormatMap.LANCE) - if (isRest && hasLance && formats.length > 1) { - // If Lance was just added, keep only Lance; otherwise the user switched away from Lance. + if (LANCE_EXCLUSIVE_TYPES.includes(formState.catalog.type as string) && hasLance && formats.length > 1) { + // If Lance was just added, keep only Lance; otherwise drop the deselected Lance. const hadLance = oldFormats?.includes(tableFormatMap.LANCE) formState.tableFormatList = hadLance ? formats.filter(format => format !== tableFormatMap.LANCE) : [tableFormatMap.LANCE] + return } + + // Skip while loading: changeProperties() would overwrite the just-loaded properties. + if (loading.value) { + return + } + + // Hadoop is not whitelisted for Lance; fall back to the first supported storage type. + if (hasLance && formState.storageConfig['storage.type'] === 'Hadoop') { + formState.storageConfig['storage.type'] = storageTypeOptionsView.value[0]?.value || '' + } + + // Use changeProperties(): a direct write drops the user's unsaved edits. + await changeProperties() }, ) @@ -686,7 +719,7 @@ onMounted(() => { {{ formState.storageConfig['storage.type'] }} From 1327f60af3dcd617c1975ed09e80b4368a4740ce Mon Sep 17 00:00:00 2001 From: WenLingzhang <1298877813@qq.com> Date: Wed, 21 Oct 2026 17:15:14 +0800 Subject: [PATCH 2/2] fixup --- amoro-format-lance/pom.xml | 22 ++ .../formats/lance/AbstractLanceCatalog.java | 6 +- .../amoro/formats/lance/LanceHiveCatalog.java | 21 +- .../formats/lance/TestLanceHiveCatalog.java | 210 ++++++++++++++++++ 4 files changed, 251 insertions(+), 8 deletions(-) create mode 100644 amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceHiveCatalog.java diff --git a/amoro-format-lance/pom.xml b/amoro-format-lance/pom.xml index 4ca948e620..9b86fc9f0d 100755 --- a/amoro-format-lance/pom.xml +++ b/amoro-format-lance/pom.xml @@ -114,6 +114,28 @@ org.apache.iceberg iceberg-aliyun + + + + org.apache.amoro + amoro-common + ${project.version} + test-jar + test + + + org.apache.derby + derby + test + + + + io.netty + netty-common + ${netty.version} + test + diff --git a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/AbstractLanceCatalog.java b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/AbstractLanceCatalog.java index 949a894434..4bff56f1e9 100644 --- a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/AbstractLanceCatalog.java +++ b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/AbstractLanceCatalog.java @@ -229,6 +229,10 @@ public void createDatabase(String database) { @Override public void dropDatabase(String database) { validateDatabase(database); - namespace.dropNamespace(new DropNamespaceRequest().id(namespaceId(database))); + try { + namespace.dropNamespace(new DropNamespaceRequest().id(namespaceId(database))); + } catch (InvalidInputException e) { + throw new IllegalStateException("Failed to drop database: " + database, e); + } } } diff --git a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceHiveCatalog.java b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceHiveCatalog.java index bbe771fa22..6d5f1db755 100644 --- a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceHiveCatalog.java +++ b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceHiveCatalog.java @@ -36,9 +36,9 @@ import org.lance.namespace.hive3.Hive3Namespace; import org.lance.namespace.hive3.Hive3NamespaceConfig; import org.lance.namespace.model.DeregisterTableRequest; +import org.lance.namespace.model.DeregisterTableResponse; import org.lance.namespace.model.DescribeTableRequest; import org.lance.namespace.model.DescribeTableResponse; -import org.lance.namespace.model.DropTableRequest; import java.util.HashMap; import java.util.Locale; @@ -46,7 +46,6 @@ /** Lance catalog backed by a Hive metastore; metadata is delegated to {@link Hive3Namespace}. */ public class LanceHiveCatalog extends AbstractLanceCatalog { - private static final String HIVE_METASTORE_URIS = "hive.metastore.uris"; /** Escape hatch for storage options Amoro does not model, e.g. MinIO's aws_allow_http. */ @@ -73,13 +72,21 @@ protected AmoroTable openTable(String database, String table) { response.getProperties()); } - /** Purge deletes the table data; otherwise only the metastore entry is deregistered. */ @Override protected void dropInNamespace(String database, String table, boolean purge) { - if (purge) { - namespace.dropTable(new DropTableRequest().id(tableId(database, table))); - } else { - namespace.deregisterTable(new DeregisterTableRequest().id(tableId(database, table))); + // delete the table metadata. + DeregisterTableResponse response = + namespace.deregisterTable(new DeregisterTableRequest().id(tableId(database, table))); + if (!purge) { + return; + } + + // delete the table data. + String location = response.getLocation(); + try { + Dataset.drop(location, storageOptions(catalogProperties, location)); + } catch (Exception e) { + throw new IllegalStateException("Failed to purge Lance data at " + location, e); } } diff --git a/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceHiveCatalog.java b/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceHiveCatalog.java new file mode 100644 index 0000000000..4815b02a30 --- /dev/null +++ b/amoro-format-lance/src/test/java/org/apache/amoro/formats/lance/TestLanceHiveCatalog.java @@ -0,0 +1,210 @@ +/* + * 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 static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import org.apache.amoro.hive.HMSMockServer; +import org.apache.amoro.table.TableMetaStore; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.FieldType; +import org.apache.arrow.vector.types.pojo.Schema; +import org.apache.commons.io.FileUtils; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.metastore.api.SerDeInfo; +import org.apache.hadoop.hive.metastore.api.StorageDescriptor; +import org.apache.hadoop.hive.metastore.api.Table; +import org.apache.iceberg.aws.s3.S3FileIOProperties; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.lance.Dataset; +import org.lance.namespace.LanceNamespace; +import org.lance.namespace.hive3.Hive3NamespaceConfig; +import org.lance.namespace.model.DeclareTableRequest; +import org.lance.namespace.model.DeclareTableResponse; +import org.lance.namespace.model.DeregisterTableRequest; +import org.lance.namespace.model.DeregisterTableResponse; + +import java.io.File; +import java.net.URI; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; + +public class TestLanceHiveCatalog { + + private static final String S3_ENDPOINT = "http://127.0.0.1:1"; + private static HMSMockServer hms; + private static RootAllocator allocator; + private static LanceHiveCatalog catalog; + private String database; + + private static final Schema SCHEMA = + new Schema( + Collections.singletonList( + new Field("id", FieldType.nullable(new ArrowType.Int(32, true)), null))); + + /** Registers the table in HMS and writes a Lance dataset at the returned location. */ + private String createTable(String table) { + DeclareTableResponse response = + catalog.namespace.declareTable( + new DeclareTableRequest() + .id(Arrays.asList(Hive3NamespaceConfig.CATALOG_NAME, database, table))); + String location = response.getLocation(); + Dataset.write().allocator(allocator).uri(location).schema(SCHEMA).execute().close(); + return location; + } + + private static boolean dataExists(String location) { + return new File(URI.create(location)).exists(); + } + + @BeforeAll + public static void startMetastore() { + hms = new HMSMockServer(new File("target/hive_warehouse")); + hms.start(); + allocator = new RootAllocator(Long.MAX_VALUE); + Map properties = new HashMap<>(); + properties.put("warehouse", hms.getWareHouseLocation()); + properties.put("uri", hms.hiveConf().get(HiveConf.ConfVars.METASTOREURIS.varname)); + properties.put(S3FileIOProperties.ENDPOINT, S3_ENDPOINT); + properties.put(S3FileIOProperties.ACCESS_KEY_ID, "test-key"); + properties.put(S3FileIOProperties.SECRET_ACCESS_KEY, "test-secret"); + catalog = new LanceHiveCatalog("lance_hive", properties, TableMetaStore.EMPTY); + } + + @BeforeEach + public void createDatabase() { + database = "db_" + UUID.randomUUID().toString().replace("-", ""); + catalog.createDatabase(database); + } + + @AfterEach + public void cleanUp() throws Exception { + hms.getClient().dropDatabase(database, true, true, true); + FileUtils.deleteQuietly(new File(hms.getDatabasePath(database))); + } + + @AfterAll + public static void stopMetastore() { + if (allocator != null) { + allocator.close(); + } + hms.stop(); + FileUtils.deleteQuietly(new File("target/hive_warehouse")); + } + + @Test + public void testDropTable() { + // 1、purge=true removes both the metastore entry and the Lance dataset directory. + String location = createTable("t1"); + Assertions.assertTrue(catalog.dropTable(database, "t1", true)); + Assertions.assertFalse(catalog.tableExists(database, "t1")); + Assertions.assertFalse(dataExists(location)); + // A second drop finds no metastore entry and reports false instead of failing. + Assertions.assertFalse(catalog.dropTable(database, "t1", true)); + + // 2、purge=false only deregisters the metastore entry, so the table data must still be there. + location = createTable("t1"); + Assertions.assertTrue(catalog.dropTable(database, "t1", false)); + Assertions.assertFalse(catalog.tableExists(database, "t1")); + Assertions.assertTrue(dataExists(location)); + } + + /** A failed data deletion must be reported with the location left behind. */ + @Test + public void testPurgeFailureIsReported() throws Exception { + String location = createTable("t2"); + // Replace the dataset directory with a file so Dataset.drop fails deterministically. + FileUtils.deleteDirectory(new File(URI.create(location))); + FileUtils.writeStringToFile(new File(URI.create(location)), "not a dataset", "UTF-8"); + + Exception error = + Assertions.assertThrows(Exception.class, () -> catalog.dropTable(database, "t2", true)); + // The entry is already dropped, so the message must carry the location to clean up. + Assertions.assertTrue(error.getMessage().contains(location), error.getMessage()); + Assertions.assertFalse(catalog.tableExists(database, "t2")); + } + + /** Purge must reach the catalog-only endpoint instead of an empty option map. */ + @Test + public void testPurgeUsesCatalogStorageOptions() throws Exception { + LanceNamespace stub = mock(LanceNamespace.class); + when(stub.deregisterTable(any(DeregisterTableRequest.class))) + .thenReturn(new DeregisterTableResponse().location("s3://bucket/t.lance")); + java.lang.reflect.Field field = AbstractLanceCatalog.class.getDeclaredField("namespace"); + field.setAccessible(true); + Object original = field.get(catalog); + field.set(catalog, stub); + try { + Exception error = + Assertions.assertThrows( + IllegalStateException.class, () -> catalog.dropTable(database, "t1", true)); + // Only the catalog-configured endpoint can put this host into the S3 error. + Assertions.assertTrue( + error.getCause().getMessage().contains(S3_ENDPOINT), error.getCause().getMessage()); + } finally { + field.set(catalog, original); + } + } + + @Test + public void testDropDatabase() { + // drop empty database + catalog.dropDatabase(database); + Assertions.assertFalse(catalog.databaseExists(database)); + + // drop no empty database. + catalog.createDatabase(database); + String location = createTable("t1"); + Assertions.assertThrows(IllegalStateException.class, () -> catalog.dropDatabase(database)); + // A rejected drop must leave the database, its table and the data untouched. + Assertions.assertTrue(catalog.databaseExists(database)); + Assertions.assertTrue(catalog.tableExists(database, "t1")); + Assertions.assertTrue(dataExists(location)); + } + + /** A non-Lance table in the same database must be left untouched by purge. */ + @Test + public void testDropNonLanceTableReturnsFalse() throws Exception { + Table table = new Table(); + table.setDbName(database); + table.setTableName("hive_tbl"); + table.setTableType("EXTERNAL_TABLE"); + StorageDescriptor sd = new StorageDescriptor(); + sd.setLocation(hms.getTablePath(database, "hive_tbl")); + sd.setCols(Collections.emptyList()); + sd.setSerdeInfo(new SerDeInfo()); + table.setSd(sd); + hms.getClient().createTable(table); + + Assertions.assertFalse(catalog.dropTable(database, "hive_tbl", true)); + Assertions.assertTrue(hms.getClient().tableExists(database, "hive_tbl")); + } +}