[#11063] fix(lance): manage storage config in Gravitino

Author: fanng

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

@@ -55,6 +55,8 @@
 import org.apache.gravitino.exceptions.NonEmptySchemaException;
 import org.apache.gravitino.exceptions.SchemaAlreadyExistsException;
 import org.apache.gravitino.exceptions.TableAlreadyExistsException;
+import org.apache.gravitino.lance.common.utils.LanceConstants;
+import org.apache.gravitino.lance.common.utils.LancePropertiesUtils;
 import org.apache.gravitino.meta.TableEntity;
 import org.apache.gravitino.rel.Column;
 import org.apache.gravitino.rel.Table;
@@ -77,6 +79,8 @@ public class GenericCatalogOperations implements CatalogOperations, SupportsSche
 
   private Optional<String> catalogLocation;
 
+  private Map<String, String> catalogProperties = Map.of();
+
   private HasPropertyMetadata propertiesMetadata;
 
   private final Cache<NameIdentifier, String> tableFormatCache;
@@ -125,6 +129,7 @@ protected EntityStore store() {
   public void initialize(
       Map<String, String> conf, CatalogInfo info, HasPropertyMetadata propertiesMetadata)
       throws RuntimeException {
+    this.catalogProperties = conf == null ? Map.of() : Maps.newHashMap(conf);
     String location =
         (String)
             propertiesMetadata
@@ -242,6 +247,13 @@ public Table createTable(
     Map<String, String> newProperties = Maps.newHashMap(properties);
     newProperties.put(Table.PROPERTY_LOCATION, tableLocation);
     newProperties.put(Table.PROPERTY_TABLE_FORMAT, format);
+    if ("lance".equals(format)) {
+      LancePropertiesUtils.getLanceStorageOptions(catalogProperties)
+          .forEach(
+              (key, value) ->
+                  newProperties.putIfAbsent(
+                      LanceConstants.LANCE_STORAGE_OPTIONS_PREFIX + key, value));
+    }
 
     // Get the table operations for the specified table format.
     Supplier<ManagedTableOperations> tableOpsSupplier = tableOpsCache.get(format);

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

@@ -20,6 +20,8 @@
 package org.apache.gravitino.catalog.lakehouse.generic;
 
 import static org.apache.gravitino.connector.PropertyEntry.stringOptionalPropertyEntry;
+import static org.apache.gravitino.connector.PropertyEntry.stringOptionalPropertyPrefixEntry;
+import static org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_STORAGE_OPTIONS_PREFIX;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Maps;
@@ -41,7 +43,14 @@ public class GenericCatalogPropertiesMetadata extends BaseCatalogPropertiesMetad
                 "The root directory of the generic catalog.",
                 false /* immutable */,
                 null, /* defaultValue */
-                false /* hidden */));
+                false /* hidden */),
+            stringOptionalPropertyPrefixEntry(
+                LANCE_STORAGE_OPTIONS_PREFIX,
+                "The Lance storage options managed by the catalog.",
+                false /* immutable */,
+                null, /* defaultValue */
+                false /* hidden */,
+                false /* reserved */));
 
     PROPERTIES_METADATA = Maps.uniqueIndex(propertyEntries, PropertyEntry::getName);
   }

catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/generic/TestPropertiesMetadata.java

@@ -41,13 +41,20 @@ void testCatalogPropertiesMetadata() {
     PropertiesMetadata catalogPropertiesMetadata = genericCatalog.catalogPropertiesMetadata();
     Assertions.assertNotNull(catalogPropertiesMetadata);
 
-    Map<String, String> catalogProperties = ImmutableMap.of("location", "/tmp/test1");
+    Map<String, String> catalogProperties =
+        ImmutableMap.of(
+            "location", "/tmp/test1",
+            "lance.storage.endpoint", "http://minio:9000");
 
     String catalogLocation =
         (String)
             catalogPropertiesMetadata.getOrDefault(
                 catalogProperties, GenericCatalog.PROPERTY_LOCATION);
     Assertions.assertEquals("/tmp/test1", catalogLocation);
+    Assertions.assertTrue(catalogPropertiesMetadata.containsProperty("lance.storage.endpoint"));
+    Assertions.assertEquals(
+        "http://minio:9000",
+        catalogPropertiesMetadata.getOrDefault(catalogProperties, "lance.storage.endpoint"));
   }
 
   @Test

docs/lakehouse-generic-lance-table.md

@@ -303,7 +303,7 @@ done
 Other table operations (load, alter, drop, truncate) follow standard relational catalog patterns. See [Table Operations](./manage-relational-metadata-using-gravitino.md#table-operations) for details.
 
 ### Using Lance table with MinIO
-To use Lance tables stored in MinIO with Gravitino, ensure that the MinIO storage backend is properly configured. Below is an example of how to set up and use Lance tables with MinIO.
+To use Lance tables stored in MinIO with Gravitino, configure the MinIO storage backend once on the Lance catalog. Gravitino will then return those storage options to Lance clients and Spark does not need to repeat them.
 
 ```shell
 curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
@@ -318,15 +318,12 @@ curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
       "nullable": false
     }
   ],
-  "properties": {
+    "properties": {
     "format": "lance",
-    "location": "s3://bucket1/lance_orders",
-    "lance.storage.access_key_id": "ak",
-    "lance.storage.endpoint": "http://minio:9000",
-    "lance.storage.secret_access_key": "sk",
-    "lance.storage.allow_http": "true"
-  }
+    "location": "s3://bucket1/lance_orders"
+    }
 }' http://localhost:8090/api/metalakes/test/catalogs/lance_catalog/schemas/sales/tables
 
 ```
 
+If you need to override storage on a single table, `lance.storage.*` table properties are still supported.

docs/lance-rest-integration.md

@@ -126,7 +126,7 @@ spark.sql("SELECT * FROM sales.orders").show()
 The `LOCATION` clause in the `CREATE TABLE` statement is optional. When omitted, lance-spark automatically determines an appropriate storage location based on catalog properties.
 For detailed information on location resolution logic, refer to the [Lakehouse Generic Catalog documentation](./lakehouse-generic-catalog.md#key-property-location).
 
-For cloud storage backends such as Amazon S3 or MinIO, specify credentials and endpoint configuration in the table properties:
+For Gravitino-managed Lance catalogs, put the storage configuration in the Gravitino catalog properties so Spark does not need to repeat it.
 
 ```python
 spark.sql("""
@@ -136,16 +136,12 @@ spark.sql("""
     )
     USING lance
     LOCATION 's3://bucket/tmp/sales/orders.lance/'
-    TBLPROPERTIES (
-        'format' = 'lance',
-        'lance.storage.access_key_id' = 'your_access_key',
-        'lance.storage.secret_access_key' = 'your_secret_key',
-        'lance.storage.endpoint' = 'http://minio:9000',
-        'lance.storage.allow_http' = 'true'
-    )
+    TBLPROPERTIES ('format' = 'lance')
 """)
 ```
 
+If you need a per-table override, `lance.storage.*` table properties are still supported and take precedence over catalog defaults.
+
 ## Ray Integration
 
 ### Installation
@@ -230,4 +226,4 @@ Refer to each engine's specific documentation for detailed configuration paramet
 - [Lance REST Service Documentation](./lance-rest-service)
 - [Lance Format Specification](https://lance.org/)
 - [Apache Gravitino Documentation](https://gravitino.apache.org/)
-- [Lakehouse Generic Catalog Guide](./lakehouse-generic-catalog.md)
+- [Lakehouse Generic Catalog Guide](./lakehouse-generic-catalog.md)

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

@@ -131,7 +131,8 @@ public DescribeTableResponse describeTable(
         Optional.ofNullable(table.properties().get(LANCE_TABLE_VERSION))
             .map(Long::valueOf)
             .orElse(null));
-    response.setStorageOptions(LancePropertiesUtils.getLanceStorageOptions(table.properties()));
+    response.setStorageOptions(
+        LancePropertiesUtils.resolveLanceStorageOptions(catalog.properties(), table.properties()));
     return response;
   }
 
@@ -171,6 +172,9 @@ public CreateTableResponse createTable(
 
     // Pass creation mode as property to delegate handling to LanceTableOperations
     createTableProperties.put(LANCE_CREATION_MODE, normalizeCreateMode(mode, tableId));
+    Map<String, String> effectiveStorageOptions =
+        LancePropertiesUtils.resolveLanceStorageOptions(
+            catalog.properties(), createTableProperties);
 
     // Single call - mode is handled server-side
     Table t =
@@ -181,9 +185,7 @@ public CreateTableResponse createTable(
     Map<String, String> properties = t.properties();
 
     CreateTableResponse response = new CreateTableResponse();
-    // Extract storage options from table properties. All storage options stores in table
-    // properties.
-    response.setStorageOptions(LancePropertiesUtils.getLanceStorageOptions(properties));
+    response.setStorageOptions(effectiveStorageOptions);
     response.setVersion(
         Optional.ofNullable(properties.get(LANCE_TABLE_VERSION)).map(Long::valueOf).orElse(null));
     response.setLocation(properties.get(LANCE_LOCATION));

lance/lance-common/src/main/java/org/apache/gravitino/lance/common/utils/LancePropertiesUtils.java

@@ -21,25 +21,51 @@
 
 import static org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_STORAGE_OPTIONS_PREFIX;
 
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.stream.Collectors;
 
-public class LancePropertiesUtils {
+/** Utility methods for Lance storage properties. */
+public final class LancePropertiesUtils {
 
   private LancePropertiesUtils() {
-    // Private constructor to prevent instantiation
+    // Utility class.
   }
 
+  /**
+   * Extracts Lance storage options from a property map.
+   *
+   * @param tableProperties the source properties
+   * @return the Lance storage options without the `lance.storage.` prefix
+   */
   public static Map<String, String> getLanceStorageOptions(Map<String, String> tableProperties) {
-    if (tableProperties == null) {
+    if (tableProperties == null || tableProperties.isEmpty()) {
       return Map.of();
     }
 
     return tableProperties.entrySet().stream()
-        .filter(e -> e.getKey().startsWith(LANCE_STORAGE_OPTIONS_PREFIX))
+        .filter(entry -> entry.getKey().startsWith(LANCE_STORAGE_OPTIONS_PREFIX))
         .collect(
             Collectors.toMap(
-                e -> e.getKey().substring(LANCE_STORAGE_OPTIONS_PREFIX.length()),
-                Map.Entry::getValue));
+                entry -> entry.getKey().substring(LANCE_STORAGE_OPTIONS_PREFIX.length()),
+                Map.Entry::getValue,
+                (left, right) -> right,
+                LinkedHashMap::new));
+  }
+
+  /**
+   * Resolves the effective Lance storage options using table properties first and catalog
+   * properties as defaults.
+   *
+   * @param catalogProperties the catalog properties
+   * @param tableProperties the table properties
+   * @return the effective storage options
+   */
+  public static Map<String, String> resolveLanceStorageOptions(
+      Map<String, String> catalogProperties, Map<String, String> tableProperties) {
+    Map<String, String> effectiveStorageOptions =
+        new LinkedHashMap<>(getLanceStorageOptions(catalogProperties));
+    effectiveStorageOptions.putAll(getLanceStorageOptions(tableProperties));
+    return effectiveStorageOptions;
   }
 }

lance/lance-common/src/test/java/org/apache/gravitino/lance/common/utils/TestLancePropertiesUtils.java

@@ -0,0 +1,63 @@
+/*
+ * 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.common.utils;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestLancePropertiesUtils {
+
+  @Test
+  public void testGetLanceStorageOptions() {
+    Map<String, String> properties =
+        ImmutableMap.of(
+            "lance.storage.endpoint", "http://minio:9000",
+            "lance.storage.access_key_id", "ak",
+            "not.storage.key", "ignored");
+
+    Map<String, String> storageOptions = LancePropertiesUtils.getLanceStorageOptions(properties);
+
+    Assertions.assertEquals(2, storageOptions.size());
+    Assertions.assertEquals("http://minio:9000", storageOptions.get("endpoint"));
+    Assertions.assertEquals("ak", storageOptions.get("access_key_id"));
+    Assertions.assertFalse(storageOptions.containsKey("not.storage.key"));
+  }
+
+  @Test
+  public void testResolveLanceStorageOptionsPrefersTableProperties() {
+    Map<String, String> catalogProperties =
+        ImmutableMap.of(
+            "lance.storage.endpoint", "http://catalog:9000",
+            "lance.storage.region", "us-east-1");
+    Map<String, String> tableProperties =
+        ImmutableMap.of(
+            "lance.storage.endpoint", "http://table:9000",
+            "lance.storage.access_key_id", "table-ak");
+
+    Map<String, String> storageOptions =
+        LancePropertiesUtils.resolveLanceStorageOptions(catalogProperties, tableProperties);
+
+    Assertions.assertEquals(3, storageOptions.size());
+    Assertions.assertEquals("http://table:9000", storageOptions.get("endpoint"));
+    Assertions.assertEquals("us-east-1", storageOptions.get("region"));
+    Assertions.assertEquals("table-ak", storageOptions.get("access_key_id"));
+  }
+}

lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceRESTServiceIT.java

@@ -95,6 +95,10 @@ public class LanceRESTServiceIT extends BaseIT {
   private static final String CATALOG_NAME = GravitinoITUtils.genRandomName("lance_rest_catalog");
   private static final String SCHEMA_NAME = GravitinoITUtils.genRandomName("lance_rest_schema");
   private static final String DELIMITER = ".";
+  private static final String MINIO_ENDPOINT = "http://127.0.0.1:9000";
+  private static final String MINIO_REGION = "us-east-1";
+  private static final String MINIO_ACCESS_KEY = "minioadmin";
+  private static final String MINIO_SECRET_KEY = "minioadmin";
 
   private GravitinoMetalake metalake;
   private Catalog catalog;
@@ -661,6 +665,43 @@ void testCreateTable() throws IOException {
             .contains("Column non_existing_column does not exist in the dataset"));
   }
 
+  @Test
+  void testCreateTableUsesCatalogStorageOptions() throws IOException {
+    catalog = createCatalog(GravitinoITUtils.genRandomName("lance_rest_catalog"));
+    createSchema();
+
+    String location = tempDir + "/" + "catalog_storage_table/";
+    List<String> ids = List.of(catalog.name(), SCHEMA_NAME, "catalog_storage_table");
+    org.apache.arrow.vector.types.pojo.Schema schema =
+        new org.apache.arrow.vector.types.pojo.Schema(
+            Arrays.asList(
+                Field.nullable("id", new ArrowType.Int(32, true)),
+                Field.nullable("value", new ArrowType.Utf8())));
+    byte[] body = ArrowUtils.generateIpcStream(schema);
+
+    CreateTableResponse response =
+        createTable(ids, location, ImmutableMap.of("key1", "v1"), body, /* mode= */ null);
+    Assertions.assertNotNull(response);
+    Assertions.assertEquals(location, response.getLocation());
+    Assertions.assertEquals(MINIO_ENDPOINT, response.getStorageOptions().get("endpoint"));
+    Assertions.assertEquals("true", response.getStorageOptions().get("allow_http"));
+    Assertions.assertEquals(MINIO_ACCESS_KEY, response.getStorageOptions().get("access_key_id"));
+    Assertions.assertEquals(
+        MINIO_SECRET_KEY, response.getStorageOptions().get("secret_access_key"));
+    Assertions.assertEquals(MINIO_REGION, response.getStorageOptions().get("region"));
+
+    DescribeTableRequest describeTableRequest = new DescribeTableRequest();
+    describeTableRequest.setId(ids);
+    DescribeTableResponse loadTable = ns.describeTable(describeTableRequest);
+    Assertions.assertNotNull(loadTable);
+    Assertions.assertEquals(MINIO_ENDPOINT, loadTable.getStorageOptions().get("endpoint"));
+    Assertions.assertEquals("true", loadTable.getStorageOptions().get("allow_http"));
+    Assertions.assertEquals(MINIO_ACCESS_KEY, loadTable.getStorageOptions().get("access_key_id"));
+    Assertions.assertEquals(
+        MINIO_SECRET_KEY, loadTable.getStorageOptions().get("secret_access_key"));
+    Assertions.assertEquals(MINIO_REGION, loadTable.getStorageOptions().get("region"));
+  }
+
   @Test
   void testAlterColumns() throws Exception {
     catalog = createCatalog(CATALOG_NAME);
@@ -1025,12 +1066,23 @@ private GravitinoMetalake createMetalake(String metalakeName) {
   }
 
   private Catalog createCatalog(String catalogName) {
+    Map<String, String> catalogProperties =
+        ImmutableMap.<String, String>builder()
+            .putAll(properties)
+            .put(Catalog.PROPERTY_LOCATION, tempDir.toString())
+            .put("lance.storage.endpoint", MINIO_ENDPOINT)
+            .put("lance.storage.allow_http", "true")
+            .put("lance.storage.access_key_id", MINIO_ACCESS_KEY)
+            .put("lance.storage.secret_access_key", MINIO_SECRET_KEY)
+            .put("lance.storage.region", MINIO_REGION)
+            .build();
+
     return metalake.createCatalog(
         catalogName,
         Catalog.Type.RELATIONAL,
         "lakehouse-generic",
         "catalog for lance rest service tests",
-        properties);
+        catalogProperties);
   }
 
   private void createSchema() {