[AMORO-4280] Add support for Lance format on REST metastore

amoro-ams/src/main/java/org/apache/amoro/server/catalog/CatalogBuilder.java

@@ -59,7 +59,7 @@ public class CatalogBuilder {
           CATALOG_TYPE_GLUE,
           Sets.newHashSet(TableFormat.ICEBERG, TableFormat.MIXED_ICEBERG),
           CATALOG_TYPE_REST,
-          Sets.newHashSet(TableFormat.ICEBERG, TableFormat.MIXED_ICEBERG),
+          Sets.newHashSet(TableFormat.ICEBERG, TableFormat.MIXED_ICEBERG, TableFormat.LANCE),
           CATALOG_TYPE_CUSTOM,
           Sets.newHashSet(TableFormat.ICEBERG, TableFormat.MIXED_ICEBERG),
           CATALOG_TYPE_HIVE,
@@ -86,6 +86,13 @@ public static ServerCatalog buildServerCatalog(
         "Table format %s is not supported for metastore type: %s",
         tableFormats,
         type);
+    Preconditions.checkState(
+        !(CATALOG_TYPE_REST.equals(type)
+            && tableFormats.contains(TableFormat.LANCE)
+            && tableFormats.size() > 1),
+        "REST catalog serves a single protocol per uri,"
+            + " Lance cannot be combined with other table formats: %s",
+        tableFormats);
 
     switch (type) {
       case CATALOG_TYPE_HADOOP:

amoro-ams/src/main/java/org/apache/amoro/server/dashboard/controller/CatalogController.java

@@ -208,6 +208,15 @@ public class CatalogController {
         CatalogDescriptor.of(CATALOG_TYPE_REST, STORAGE_CONFIGS_VALUE_TYPE_OSS, ICEBERG));
     VALIDATE_CATALOGS.add(
         CatalogDescriptor.of(CATALOG_TYPE_REST, STORAGE_CONFIGS_VALUE_TYPE_OSS, MIXED_ICEBERG));
+
+    // Lance REST is a data proxy: the client never talks to object storage, so the storage config
+    // is not consumed, but it still reuses the standard REST storage types.
+    VALIDATE_CATALOGS.add(
+        CatalogDescriptor.of(CATALOG_TYPE_REST, STORAGE_CONFIGS_VALUE_TYPE_S3, LANCE));
+    VALIDATE_CATALOGS.add(
+        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));
   }
 
   private final PlatformFileManager platformFileInfoService;
@@ -473,8 +482,11 @@ private CatalogMeta constructCatalogMeta(CatalogRegisterInfo info, CatalogMeta o
       catalogMeta.putToCatalogProperties(
           CatalogProperties.CATALOG_IMPL, GlueCatalog.class.getName());
     } else if (CatalogMetaProperties.CATALOG_TYPE_REST.equals(metastoreType)) {
-      catalogMeta.putToCatalogProperties(
-          CatalogProperties.CATALOG_IMPL, RESTCatalog.class.getName());
+      // Only the Iceberg-family formats need the Iceberg RESTCatalog implementation.
+      if (!info.getTableFormatList().contains(LANCE.name())) {
+        catalogMeta.putToCatalogProperties(
+            CatalogProperties.CATALOG_IMPL, RESTCatalog.class.getName());
+      }
     }
 
     catalogMeta.putToCatalogProperties(

amoro-common/src/main/java/org/apache/amoro/utils/CatalogUtil.java

@@ -121,6 +121,13 @@ public static Map<String, String> 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();
+    }
+
     // load storage configs
     TableMetaStore.Builder builder = TableMetaStore.builder();
     boolean isLocalStorage = false;
@@ -215,6 +222,11 @@ public static TableMetaStore buildMetaStore(CatalogMeta catalogMeta) {
     return builder.build();
   }
 
+  private static boolean isLanceOnly(CatalogMeta catalogMeta) {
+    Set<TableFormat> formats = tableFormats(catalogMeta);
+    return formats.size() == 1 && formats.contains(TableFormat.LANCE);
+  }
+
   public static TableIdentifier tableId(TableMeta tableMeta) {
     return TableIdentifier.of(
         tableMeta.getTableIdentifier().getCatalog(),

amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/AbstractLanceCatalog.java

@@ -0,0 +1,111 @@
+/*
+ * 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.NoSuchDatabaseException;
+import org.apache.amoro.NoSuchTableException;
+import org.apache.amoro.table.TableIdentifier;
+import org.lance.Dataset;
+import org.lance.namespace.LanceNamespace;
+import org.lance.namespace.errors.TableNotFoundException;
+import org.lance.namespace.model.DropTableRequest;
+import org.lance.namespace.model.ListTablesRequest;
+import org.lance.namespace.model.ListTablesResponse;
+import org.lance.namespace.model.TableExistsRequest;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** Base implementation of {@link FormatCatalog} for Lance. */
+public abstract class AbstractLanceCatalog implements FormatCatalog {
+
+  protected final String catalogName;
+  protected final LanceNamespace namespace;
+
+  protected AbstractLanceCatalog(String catalogName, LanceNamespace namespace) {
+    this.catalogName = catalogName;
+    this.namespace = namespace;
+  }
+
+  /** Composes the namespace table id used to reference a single table. */
+  protected abstract List<String> tableId(String database, String table);
+
+  /** Composes the namespace id used to list tables of a database. */
+  protected abstract List<String> tableIdForListTables(String database);
+
+  @Override
+  public boolean tableExists(String database, String table) {
+    if (!databaseExists(database)) {
+      return false;
+    }
+
+    try {
+      namespace.tableExists(new TableExistsRequest().id(tableId(database, table)));
+      return true;
+    } catch (TableNotFoundException e) {
+      return false;
+    }
+  }
+
+  @Override
+  public AmoroTable<?> loadTable(String database, String tableName) {
+    if (!tableExists(database, tableName)) {
+      throw new NoSuchTableException("Table: " + database + "." + tableName + " does not exist");
+    }
+
+    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());
+  }
+
+  @Override
+  public boolean dropTable(String database, String table, boolean purge) {
+    validateDatabase(database);
+
+    try {
+      namespace.dropTable(new DropTableRequest().id(tableId(database, table)));
+      return true;
+    } catch (TableNotFoundException e) {
+      return false;
+    }
+  }
+
+  @Override
+  public List<String> listTables(String database) {
+    if (!databaseExists(database)) {
+      return Collections.emptyList();
+    }
+    ListTablesRequest request = new ListTablesRequest().id(tableIdForListTables(database));
+    ListTablesResponse response = namespace.listTables(request);
+    if (response == null) {
+      return Collections.emptyList();
+    }
+    return new ArrayList<>(response.getTables());
+  }
+
+  protected void validateDatabase(String database) {
+    if (!databaseExists(database)) {
+      throw new NoSuchDatabaseException("Database: " + database + " does not exist");
+    }
+  }
+}

amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceCatalogFactory.java

@@ -21,6 +21,7 @@
 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 FormatCatalog create(
       String metastoreType,
       Map<String, String> properties,
       TableMetaStore metaStore) {
+    if (CatalogMetaProperties.CATALOG_TYPE_REST.equals(metastoreType)) {
+      return new LanceRestCatalog(catalogName, properties);
+    }
     return new LanceDirectoryV1Catalog(catalogName, properties);
   }
 

amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceDirectoryV1Catalog.java

@@ -18,26 +18,13 @@
 
 package org.apache.amoro.formats.lance;
 
-import org.apache.amoro.AmoroTable;
-import org.apache.amoro.FormatCatalog;
-import org.apache.amoro.NoSuchDatabaseException;
-import org.apache.amoro.NoSuchTableException;
 import org.apache.amoro.properties.CatalogMetaProperties;
-import org.apache.amoro.table.TableIdentifier;
-import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.RootAllocator;
 import org.apache.arrow.util.Preconditions;
 import org.apache.iceberg.aliyun.AliyunProperties;
 import org.apache.iceberg.aws.s3.S3FileIOProperties;
-import org.lance.Dataset;
 import org.lance.namespace.LanceNamespace;
-import org.lance.namespace.errors.TableNotFoundException;
-import org.lance.namespace.model.DropTableRequest;
-import org.lance.namespace.model.ListTablesRequest;
-import org.lance.namespace.model.ListTablesResponse;
-import org.lance.namespace.model.TableExistsRequest;
 
-import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -50,22 +37,22 @@
  * root directory, each immediate subdirectory whose name ends with ".lance" is treated as a Lance
  * dataset. All tables live in a single logical database named "default".
  */
-public class LanceDirectoryV1Catalog implements FormatCatalog {
+public class LanceDirectoryV1Catalog extends AbstractLanceCatalog {
 
   private static final String DEFAULT_DATABASE = "default";
   private static final String STORAGE_ACCESS_KEY_ID = "storage.access_key_id";
   private static final String STORAGE_SECRET_ACCESS_KEY = "storage.secret_access_key";
   private static final String STORAGE_ENDPOINT = "storage.endpoint";
-  private final String catalogName;
-  private final Map<String, String> namespaceProperties;
-  private final LanceNamespace namespace;
 
   public LanceDirectoryV1Catalog(String catalogName, Map<String, String> catalogProperties) {
+    super(catalogName, buildNamespace(catalogProperties));
+  }
+
+  private static LanceNamespace buildNamespace(Map<String, String> catalogProperties) {
     Preconditions.checkArgument(
         catalogProperties != null && !catalogProperties.isEmpty(),
         "Catalog properties must be set.");
-    this.catalogName = catalogName;
-    this.namespaceProperties = new HashMap<>(catalogProperties);
+    Map<String, String> namespaceProperties = new HashMap<>(catalogProperties);
     String root = namespaceProperties.remove(CatalogMetaProperties.KEY_WAREHOUSE);
     String storageAccessKey =
         removeFirstNonNull(
@@ -80,42 +67,37 @@ public LanceDirectoryV1Catalog(String catalogName, Map<String, String> catalogPr
 
     Preconditions.checkArgument(
         root != null && !root.isEmpty(), "Warehouse must be set in catalogProperties.");
-    this.namespaceProperties.put("manifest_enabled", "false");
-    this.namespaceProperties.put("vend_input_storage_options", "true");
-    this.namespaceProperties.put("root", root);
+    namespaceProperties.put("manifest_enabled", "false");
+    namespaceProperties.put("vend_input_storage_options", "true");
+    namespaceProperties.put("root", root);
     if (storageAccessKey != null) {
-      this.namespaceProperties.put(STORAGE_ACCESS_KEY_ID, storageAccessKey);
+      namespaceProperties.put(STORAGE_ACCESS_KEY_ID, storageAccessKey);
     }
     if (storageSecretKey != null) {
-      this.namespaceProperties.put(STORAGE_SECRET_ACCESS_KEY, storageSecretKey);
+      namespaceProperties.put(STORAGE_SECRET_ACCESS_KEY, storageSecretKey);
     }
     putStorageEndpointIfPresent(namespaceProperties);
-    this.namespace = initializeNamespace(new RootAllocator(Long.MAX_VALUE));
+    return LanceNamespace.connect("dir", namespaceProperties, new RootAllocator(Long.MAX_VALUE));
   }
 
   @Override
-  public List<String> listDatabases() {
-    return Collections.singletonList(DEFAULT_DATABASE);
+  protected List<String> tableId(String database, String table) {
+    return Collections.singletonList(table);
   }
 
   @Override
-  public boolean databaseExists(String database) {
-    return DEFAULT_DATABASE.equals(database);
+  protected List<String> tableIdForListTables(String database) {
+    return Collections.emptyList();
   }
 
   @Override
-  public boolean tableExists(String database, String table) {
-    if (!databaseExists(database)) {
-      return false;
-    }
+  public List<String> listDatabases() {
+    return Collections.singletonList(DEFAULT_DATABASE);
+  }
 
-    try {
-      TableExistsRequest request = new TableExistsRequest().id(Collections.singletonList(table));
-      namespace.tableExists(request);
-      return true;
-    } catch (TableNotFoundException e) {
-      return false;
-    }
+  @Override
+  public boolean databaseExists(String database) {
+    return DEFAULT_DATABASE.equals(database);
   }
 
   @Override
@@ -128,59 +110,6 @@ public void dropDatabase(String database) {
     throw new UnsupportedOperationException("Dropping Lance databases is not supported.");
   }
 
-  @Override
-  public AmoroTable<?> loadTable(String database, String tableName) {
-    if (!databaseExists(database) || !tableExists(database, tableName)) {
-      throw new NoSuchTableException("Table: " + database + "." + tableName + " does not exist");
-    }
-
-    TableIdentifier identifier = TableIdentifier.of(catalogName, database, tableName);
-    Dataset dataset =
-        Dataset.open()
-            .namespaceClient(namespace)
-            .tableId(Collections.singletonList(tableName))
-            .build();
-    return new LanceTable(identifier, dataset, Collections.emptyMap());
-  }
-
-  @Override
-  public boolean dropTable(String database, String table, boolean purge) {
-    validateDatabase(database);
-
-    try {
-      namespace.dropTable(new DropTableRequest().id(Collections.singletonList(table)));
-      return true;
-    } catch (TableNotFoundException e) {
-      return false;
-    }
-  }
-
-  @Override
-  public List<String> listTables(String database) {
-    if (!databaseExists(database)) {
-      return Collections.emptyList();
-    }
-    return listTablesFromNamespace();
-  }
-
-  private List<String> listTablesFromNamespace() {
-    if (namespace == null) {
-      return Collections.emptyList();
-    }
-
-    ListTablesRequest request = new ListTablesRequest().id(Collections.emptyList());
-    ListTablesResponse response = namespace.listTables(request);
-    if (response == null) {
-      return Collections.emptyList();
-    }
-
-    return new ArrayList<>(response.getTables());
-  }
-
-  private LanceNamespace initializeNamespace(BufferAllocator allocator) {
-    return LanceNamespace.connect("dir", namespaceProperties, allocator);
-  }
-
   private static String removeFirstNonNull(Map<String, String> properties, String... keys) {
     String value = null;
     for (String key : keys) {
@@ -205,10 +134,4 @@ private static void putStorageEndpointIfPresent(Map<String, String> properties)
       properties.put(STORAGE_ENDPOINT, endpoint);
     }
   }
-
-  private void validateDatabase(String database) {
-    if (!databaseExists(database)) {
-      throw new NoSuchDatabaseException("Database: " + database + " does not exist");
-    }
-  }
 }