[#8955] feat(lance-rest): Support register and deregister table in Lance REST server.

catalogs/catalog-generic-lakehouse/src/main/java/org/apache/gravitino/catalog/lakehouse/GenericLakehouseCatalogOperations.java

@@ -210,6 +210,8 @@ public Table loadTable(NameIdentifier ident) throws NoSuchTableException {
           .withName(tableEntity.name())
           .withComment(tableEntity.getComment())
           .build();
+    } catch (NoSuchEntityException e) {
+      throw new NoSuchTableException(e, "Table %s does not exist", ident);
     } catch (IOException e) {
       throw new RuntimeException("Failed to list tables under schema " + ident.namespace(), e);
     }
@@ -270,6 +272,34 @@ public Table createTable(
               .withAuditInfo(auditInfo)
               .build();
       store.put(entityToStore);
+
+      // Get the value of register in table properties
+      boolean register =
+          Boolean.parseBoolean(
+              properties.getOrDefault(
+                  GenericLakehouseTablePropertiesMetadata.LAKEHOUSE_REGISTER, "false"));
+      if (register) {
+        // Do not need to create the physical table if this is a registration operation.
+        // Whether we need to check the existence of the physical table?
+        GenericLakehouseTable.Builder builder = GenericLakehouseTable.builder();
+        return builder
+            .withName(ident.name())
+            .withColumns(columns)
+            .withComment(comment)
+            .withProperties(properties)
+            .withDistribution(distribution)
+            .withIndexes(indexes)
+            .withAuditInfo(
+                AuditInfo.builder()
+                    .withCreator(PrincipalUtils.getCurrentUserName())
+                    .withCreateTime(Instant.now())
+                    .build())
+            .withPartitioning(partitions)
+            .withSortOrders(sortOrders)
+            .withFormat(LakehouseTableFormat.LANCE.lowerName())
+            .build();
+      }
+
       LakehouseCatalogOperations lanceCatalogOperations =
           getLakehouseCatalogOperations(newProperties);
       return lanceCatalogOperations.createTable(
@@ -324,21 +354,19 @@ private String calculateTableLocation(
   @Override
   public Table alterTable(NameIdentifier ident, TableChange... changes)
       throws NoSuchTableException, IllegalArgumentException {
-    Namespace namespace = ident.namespace();
     try {
       TableEntity tableEntity = store.get(ident, Entity.EntityType.TABLE, TableEntity.class);
       Map<String, String> tableProperties = tableEntity.getProperties();
       LakehouseCatalogOperations lakehouseCatalogOperations =
           getLakehouseCatalogOperations(tableProperties);
       return lakehouseCatalogOperations.alterTable(ident, changes);
     } catch (IOException e) {
-      throw new RuntimeException("Failed to list tables under schema " + namespace, e);
+      throw new RuntimeException("Failed to alter table " + ident, e);
     }
   }
 
   @Override
   public boolean dropTable(NameIdentifier ident) {
-    Namespace namespace = ident.namespace();
     try {
       TableEntity tableEntity = store.get(ident, Entity.EntityType.TABLE, TableEntity.class);
       LakehouseCatalogOperations lakehouseCatalogOperations =
@@ -348,7 +376,20 @@ public boolean dropTable(NameIdentifier ident) {
       LOG.warn("Table {} does not exist, skip dropping it.", ident);
       return false;
     } catch (IOException e) {
-      throw new RuntimeException("Failed to list tables under schema " + namespace, e);
+      throw new RuntimeException("Failed to drop table: " + ident, e);
+    }
+  }
+
+  @Override
+  public boolean purgeTable(NameIdentifier ident) throws UnsupportedOperationException {
+    try {
+      // Only delete the metadata entry here. The physical data will not be deleted.
+      if (!tableExists(ident)) {
+        return false;
+      }
+      return store.delete(ident, Entity.EntityType.TABLE);
+    } catch (IOException e) {
+      throw new RuntimeException("Failed to purge table " + ident, e);
     }
   }
 

catalogs/catalog-generic-lakehouse/src/main/java/org/apache/gravitino/catalog/lakehouse/GenericLakehouseTablePropertiesMetadata.java

@@ -18,6 +18,7 @@
  */
 package org.apache.gravitino.catalog.lakehouse;
 
+import static org.apache.gravitino.connector.PropertyEntry.booleanPropertyEntry;
 import static org.apache.gravitino.connector.PropertyEntry.enumPropertyEntry;
 import static org.apache.gravitino.connector.PropertyEntry.stringOptionalPropertyEntry;
 
@@ -32,6 +33,7 @@ public class GenericLakehouseTablePropertiesMetadata extends BasePropertiesMetad
   public static final String LAKEHOUSE_LOCATION = "location";
   public static final String LAKEHOUSE_FORMAT = "format";
   public static final String LANCE_TABLE_STORAGE_OPTION_PREFIX = "lance.storage.";
+  public static final String LAKEHOUSE_REGISTER = "register";
 
   private static final Map<String, PropertyEntry<?>> PROPERTIES_METADATA;
 
@@ -59,7 +61,15 @@ public class GenericLakehouseTablePropertiesMetadata extends BasePropertiesMetad
                 false /* immutable */,
                 null /* default value*/,
                 false /* hidden */,
-                false /* reserved */));
+                false /* reserved */),
+            booleanPropertyEntry(
+                LAKEHOUSE_REGISTER,
+                "Whether this is a table registration operation.",
+                false,
+                true /* immutable */,
+                false /* defaultValue */,
+                false /* hidden */,
+                false));
 
     PROPERTIES_METADATA = Maps.uniqueIndex(propertyEntries, PropertyEntry::getName);
   }

core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java

@@ -408,6 +408,10 @@ public boolean purgeTable(NameIdentifier ident) throws UnsupportedOperationExcep
                   RuntimeException.class,
                   UnsupportedOperationException.class);
 
+          if (isManagedTable(catalogIdent)) {
+            return droppedFromCatalog;
+          }
+
           // For unmanaged table, it could happen that the table:
           // 1. Is not found in the catalog (dropped directly from underlying sources)
           // 2. Is found in the catalog but not in the store (not managed by Gravitino)

lance/lance-common/build.gradle.kts

@@ -25,7 +25,8 @@ plugins {
 }
 
 dependencies {
-  implementation(project(":clients:client-java-runtime", configuration = "shadow"))
+  implementation(project(":clients:client-java"))
+  implementation(project(":api"))
   implementation(project(":common")) {
     exclude("*")
   }

lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/LanceTableOperations.java

@@ -19,7 +19,9 @@
 package org.apache.gravitino.lance.common.ops;
 
 import com.lancedb.lance.namespace.model.CreateTableResponse;
+import com.lancedb.lance.namespace.model.DeregisterTableResponse;
 import com.lancedb.lance.namespace.model.DescribeTableResponse;
+import com.lancedb.lance.namespace.model.RegisterTableResponse;
 import java.util.Map;
 
 public interface LanceTableOperations {
@@ -32,6 +34,10 @@ CreateTableResponse createTable(
       String delimiter,
       String tableLocation,
       Map<String, String> tableProperties,
-      String rootCatalog,
       byte[] arrowStreamBody);
+
+  RegisterTableResponse registerTable(
+      String tableId, String mode, String delimiter, Map<String, String> tableProperties);
+
+  DeregisterTableResponse deregisterTable(String tableId, String delimiter);
 }

lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceNamespaceWrapper.java

@@ -35,13 +35,16 @@
 import com.lancedb.lance.namespace.model.CreateNamespaceRequest.ModeEnum;
 import com.lancedb.lance.namespace.model.CreateNamespaceResponse;
 import com.lancedb.lance.namespace.model.CreateTableResponse;
+import com.lancedb.lance.namespace.model.DeregisterTableResponse;
 import com.lancedb.lance.namespace.model.DescribeNamespaceResponse;
 import com.lancedb.lance.namespace.model.DescribeTableResponse;
 import com.lancedb.lance.namespace.model.DropNamespaceRequest;
 import com.lancedb.lance.namespace.model.DropNamespaceResponse;
 import com.lancedb.lance.namespace.model.JsonArrowSchema;
 import com.lancedb.lance.namespace.model.ListNamespacesResponse;
 import com.lancedb.lance.namespace.model.ListTablesResponse;
+import com.lancedb.lance.namespace.model.RegisterTableRequest;
+import com.lancedb.lance.namespace.model.RegisterTableResponse;
 import com.lancedb.lance.namespace.util.CommonUtil;
 import com.lancedb.lance.namespace.util.JsonArrowSchemaConverter;
 import com.lancedb.lance.namespace.util.PageUtil;
@@ -492,7 +495,7 @@ public ListTablesResponse listTables(
       String namespaceId, String delimiter, String pageToken, Integer limit) {
     ObjectIdentifier nsId = ObjectIdentifier.of(namespaceId, Pattern.quote(delimiter));
     Preconditions.checkArgument(
-        nsId.levels() <= 2, "Expected at most 2-level namespace but got: %s", nsId.levels());
+        nsId.levels() == 2, "Expected 2-level namespace but got: %s", nsId.levels());
     String catalogName = nsId.levelAtListPos(0);
     Catalog catalog = loadAndValidateLakehouseCatalog(catalogName);
     String schemaName = nsId.levelAtListPos(1);
@@ -516,7 +519,7 @@ public ListTablesResponse listTables(
   public DescribeTableResponse describeTable(String tableId, String delimiter) {
     ObjectIdentifier nsId = ObjectIdentifier.of(tableId, Pattern.quote(delimiter));
     Preconditions.checkArgument(
-        nsId.levels() <= 3, "Expected at most 3-level namespace but got: %s", nsId.levels());
+        nsId.levels() == 3, "Expected at 3-level namespace but got: %s", nsId.levels());
 
     String catalogName = nsId.levelAtListPos(0);
     Catalog catalog = loadAndValidateLakehouseCatalog(catalogName);
@@ -538,17 +541,10 @@ public CreateTableResponse createTable(
       String delimiter,
       String tableLocation,
       Map<String, String> tableProperties,
-      String rootCatalog,
       byte[] arrowStreamBody) {
     ObjectIdentifier nsId = ObjectIdentifier.of(tableId, Pattern.quote(delimiter));
     Preconditions.checkArgument(
-        nsId.levels() <= 3, "Expected at most 3-level namespace but got: %s", nsId.levels());
-    if (rootCatalog != null) {
-      List<String> levels = nsId.listStyleId();
-      List<String> newLevels = Lists.newArrayList(rootCatalog);
-      newLevels.addAll(levels);
-      nsId = ObjectIdentifier.of(newLevels);
-    }
+        nsId.levels() == 3, "Expected at 3-level namespace but got: %s", nsId.levels());
 
     // Parser column information.
     List<Column> columns = Lists.newArrayList();
@@ -614,6 +610,68 @@ public CreateTableResponse createTable(
     return response;
   }
 
+  @Override
+  public RegisterTableResponse registerTable(
+      String tableId, String mode, String delimiter, Map<String, String> tableProperties) {
+    ObjectIdentifier nsId = ObjectIdentifier.of(tableId, Pattern.quote(delimiter));
+    Preconditions.checkArgument(
+        nsId.levels() == 3, "Expected at 3-level namespace but got: %s", nsId.levels());
+
+    String catalogName = nsId.levelAtListPos(0);
+    Catalog catalog = loadAndValidateLakehouseCatalog(catalogName);
+    NameIdentifier tableIdentifier =
+        NameIdentifier.of(nsId.levelAtListPos(1), nsId.levelAtListPos(2));
+
+    // TODO Support real register API
+    RegisterTableRequest.ModeEnum createMode =
+        RegisterTableRequest.ModeEnum.fromValue(mode.toUpperCase());
+    if (createMode == RegisterTableRequest.ModeEnum.CREATE
+        && catalog.asTableCatalog().tableExists(tableIdentifier)) {
+      throw LanceNamespaceException.conflict(
+          "Table already exists: " + tableId,
+          SchemaAlreadyExistsException.class.getSimpleName(),
+          tableId,
+          CommonUtil.formatCurrentStackTrace());
+    }
+
+    if (createMode == RegisterTableRequest.ModeEnum.OVERWRITE
+        && catalog.asTableCatalog().tableExists(tableIdentifier)) {
+      LOG.info("Overwriting existing table: {}", tableId);
+      catalog.asTableCatalog().dropTable(tableIdentifier);
+    }
+
+    Table t =
+        catalog.asTableCatalog().createTable(tableIdentifier, new Column[] {}, "", tableProperties);
+
+    RegisterTableResponse response = new RegisterTableResponse();
+    response.setProperties(t.properties());
+    response.setLocation(t.properties().get("location"));
+    return response;
+  }
+
+  @Override
+  public DeregisterTableResponse deregisterTable(String tableId, String delimiter) {
+
+    ObjectIdentifier nsId = ObjectIdentifier.of(tableId, Pattern.quote(delimiter));
+    Preconditions.checkArgument(
+        nsId.levels() == 3, "Expected at 3-level namespace but got: %s", nsId.levels());
+
+    String catalogName = nsId.levelAtListPos(0);
+    Catalog catalog = loadAndValidateLakehouseCatalog(catalogName);
+
+    NameIdentifier tableIdentifier =
+        NameIdentifier.of(nsId.levelAtListPos(1), nsId.levelAtListPos(2));
+    Table t = catalog.asTableCatalog().loadTable(tableIdentifier);
+    Map<String, String> properties = t.properties();
+    // TODO Support real deregister API.
+    catalog.asTableCatalog().purgeTable(tableIdentifier);
+
+    DeregisterTableResponse response = new DeregisterTableResponse();
+    response.setProperties(properties);
+    response.setLocation(properties.get("location"));
+    return response;
+  }
+
   private JsonArrowSchema toJsonArrowSchema(Column[] columns) {
     List<Field> fields =
         Arrays.stream(columns)
@@ -627,7 +685,6 @@ private JsonArrowSchema toJsonArrowSchema(Column[] columns) {
   @VisibleForTesting
   org.apache.arrow.vector.types.pojo.Schema parseArrowIpcStream(byte[] stream) {
     org.apache.arrow.vector.types.pojo.Schema schema;
-
     try (BufferAllocator allocator = new RootAllocator();
         ByteArrayInputStream bais = new ByteArrayInputStream(stream);
         ArrowStreamReader reader = new ArrowStreamReader(bais, allocator)) {

lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/LanceRESTService.java

@@ -127,6 +127,7 @@ private NamespaceWrapper loadNamespaceImpl(LanceConfig lanceConfig) {
     try {
       Constructor<? extends NamespaceWrapper> constructor =
           lanceNamespaceBackend.getWrapperClass().getConstructor(LanceConfig.class);
+
       return constructor.newInstance(lanceConfig);
     } catch (Exception e) {
       LOG.error("Error loading namespace implementation for backend type: {}", backendType, e);

lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/ServiceConstants.java

@@ -0,0 +1,29 @@
+/*
+ * 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.gravitino.lance.service;
+
+public class ServiceConstants {
+  public static final String LANCE_HTTP_HEADER_PREFIX = "x-lance-";
+
+  public static final String LANCE_TABLE_LOCATION_HEADER =
+      LANCE_HTTP_HEADER_PREFIX + "table-location";
+  public static final String LANCE_TABLE_PROPERTIES_PREFIX_HEADER =
+      LANCE_HTTP_HEADER_PREFIX + "table-properties";
+}