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

### What changes were proposed in this pull request?

This PR moves Lance storage configuration into Gravitino-managed catalog
properties and teaches Lance REST to use those catalog defaults when
serving Spark clients. It removes the need to pass
spark.sql.catalog.lance.storage.* in the demo path, while keeping
table-level lance.storage.* overrides supported.

### Why are the changes needed?

Spark currently has to repeat MinIO/S3 settings that are already known
to Gravitino, which makes Lance integration verbose and error-prone.
This change keeps Gravitino as the source of truth for Lance storage
configuration and simplifies Spark usage for Gravitino-managed Lance
catalogs.

Fix: #11063

### Does this PR introduce _any_ user-facing change?

Yes.

- lance.storage.* is now supported as a catalog-level property family
for generic lakehouse catalogs.
- Lance REST responses now resolve storage options from catalog defaults
when table-level values are absent.
- The Spark demo no longer requires explicit Lance storage conf entries.
- Table-level lance.storage.* remains supported as an override.

### How was this patch tested?

- ./gradlew :lance:lance-common:test --tests
org.apache.gravitino.lance.common.utils.TestLancePropertiesUtils
-PskipITs -PskipDockerTests=true
- ./gradlew :catalogs:catalog-lakehouse-generic:test --tests
org.apache.gravitino.catalog.lakehouse.generic.TestPropertiesMetadata
-PskipITs -PskipDockerTests=true
- ./gradlew :lance:lance-rest-server:compileTestJava -PskipITs
-PskipDockerTests=true

---------

Co-authored-by: fanng <“fanng@apache.org”>

Author: FANNG1

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

@@ -44,6 +44,7 @@
 import org.apache.gravitino.SchemaChange;
 import org.apache.gravitino.catalog.ManagedSchemaOperations;
 import org.apache.gravitino.catalog.ManagedTableOperations;
+import org.apache.gravitino.catalog.lakehouse.lance.LanceTableOperations;
 import org.apache.gravitino.connector.CatalogInfo;
 import org.apache.gravitino.connector.CatalogOperations;
 import org.apache.gravitino.connector.HasPropertyMetadata;
@@ -77,6 +78,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 +128,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
@@ -246,7 +250,7 @@ public Table createTable(
     // Get the table operations for the specified table format.
     Supplier<ManagedTableOperations> tableOpsSupplier = tableOpsCache.get(format);
     Preconditions.checkArgument(tableOpsSupplier != null, "Unsupported table format: %s", format);
-    ManagedTableOperations tableOps = tableOpsSupplier.get();
+    ManagedTableOperations tableOps = configureTableOps(tableOpsSupplier.get());
 
     Table createdTable =
         tableOps.createTable(
@@ -339,7 +343,7 @@ private ManagedTableOperations tableOps(NameIdentifier tableIdent) {
                 return format.toLowerCase(Locale.ROOT);
               });
 
-      ManagedTableOperations ops = tableOpsCache.get(tableFormat).get();
+      ManagedTableOperations ops = configureTableOps(tableOpsCache.get(tableFormat).get());
       Preconditions.checkArgument(
           ops != null, "No table operations found for table format %s", tableFormat);
       return ops;
@@ -360,4 +364,12 @@ private ManagedTableOperations tableOps(NameIdentifier tableIdent) {
       }
     }
   }
+
+  private ManagedTableOperations configureTableOps(ManagedTableOperations ops) {
+    if (ops instanceof LanceTableOperations) {
+      ((LanceTableOperations) ops).setCatalogProperties(catalogProperties);
+    }
+
+    return ops;
+  }
 }

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/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java

@@ -53,6 +53,7 @@
 import org.apache.gravitino.rel.indexes.Index;
 import org.apache.gravitino.storage.IdGenerator;
 import org.lance.Dataset;
+import org.lance.ReadOptions;
 import org.lance.WriteParams;
 import org.lance.index.DistanceType;
 import org.lance.index.IndexOptions;
@@ -78,6 +79,8 @@ public enum CreationMode {
 
   private final IdGenerator idGenerator;
 
+  private volatile Map<String, String> catalogProperties = Map.of();
+
   public LanceTableOperations(
       EntityStore store, ManagedSchemaOperations schemaOps, IdGenerator idGenerator) {
     this.store = store;
@@ -100,6 +103,16 @@ protected IdGenerator idGenerator() {
     return idGenerator;
   }
 
+  /**
+   * Sets the catalog properties used to resolve Lance storage defaults at runtime.
+   *
+   * @param catalogProperties the catalog properties
+   */
+  public void setCatalogProperties(Map<String, String> catalogProperties) {
+    this.catalogProperties =
+        catalogProperties == null ? Map.of() : ImmutableMap.copyOf(catalogProperties);
+  }
+
   @Override
   public Table createTable(
       NameIdentifier ident,
@@ -197,7 +210,9 @@ public boolean purgeTable(NameIdentifier ident) {
       // Otherwise, we should not delete the dataset.
       if (purged) {
         // Delete the Lance dataset at the location
-        Dataset.drop(location, LancePropertiesUtils.getLanceStorageOptions(table.properties()));
+        Dataset.drop(
+            location,
+            LancePropertiesUtils.resolveLanceStorageOptions(catalogProperties, table.properties()));
         LOG.info("Deleted Lance dataset at location {}", location);
       }
 
@@ -231,7 +246,9 @@ public boolean dropTable(NameIdentifier ident) {
         String location = table.properties().get(Table.PROPERTY_LOCATION);
 
         // Delete the Lance dataset at the location
-        Dataset.drop(location, LancePropertiesUtils.getLanceStorageOptions(table.properties()));
+        Dataset.drop(
+            location,
+            LancePropertiesUtils.resolveLanceStorageOptions(catalogProperties, table.properties()));
         LOG.info("Deleted Lance dataset at location {}", location);
       }
 
@@ -277,7 +294,8 @@ Table createTableInternal(
           ident, columns, comment, properties, partitions, distribution, sortOrders, indexes);
     }
 
-    Map<String, String> storageProps = LancePropertiesUtils.getLanceStorageOptions(properties);
+    Map<String, String> storageProps =
+        LancePropertiesUtils.resolveLanceStorageOptions(catalogProperties, properties);
     try (Dataset ignored =
         Dataset.write()
             .allocator(new RootAllocator())
@@ -308,7 +326,7 @@ Table createTableInternal(
     } catch (TableAlreadyExistsException e) {
       // If the table metadata already exists, but the underlying lance table was just created
       // successfully, we need to clean up the created lance table to avoid orphaned datasets.
-      Dataset.drop(location, LancePropertiesUtils.getLanceStorageOptions(properties));
+      Dataset.drop(location, storageProps);
       throw e;
     } catch (IllegalArgumentException e) {
       if (e.getMessage().contains("Dataset already exists")) {
@@ -344,7 +362,9 @@ private org.apache.arrow.vector.types.pojo.Schema convertColumnsToArrowSchema(Co
    */
   long handleLanceTableChange(Table table, TableChange[] changes) {
     String location = table.properties().get(Table.PROPERTY_LOCATION);
-    try (Dataset dataset = openDataset(location)) {
+    Map<String, String> storageOptions =
+        LancePropertiesUtils.resolveLanceStorageOptions(catalogProperties, table.properties());
+    try (Dataset dataset = openDataset(location, storageOptions)) {
       for (TableChange change : changes) {
         if (change instanceof TableChange.DeleteColumn deleteColumn) {
           dataset.dropColumns(List.of(String.join(".", deleteColumn.fieldName())));
@@ -385,7 +405,15 @@ long handleLanceTableChange(Table table, TableChange[] changes) {
   }
 
   Dataset openDataset(String location) {
-    return Dataset.open().allocator(new RootAllocator()).uri(location).build();
+    return openDataset(location, Map.of());
+  }
+
+  Dataset openDataset(String location, Map<String, String> storageOptions) {
+    return Dataset.open()
+        .allocator(new RootAllocator())
+        .uri(location)
+        .readOptions(new ReadOptions.Builder().setStorageOptions(storageOptions).build())
+        .build();
   }
 
   private IndexParams getIndexParamsByIndexType(IndexType indexType) {

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

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

@@ -19,6 +19,7 @@
 package org.apache.gravitino.catalog.lakehouse.lance;
 
 import static org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_CREATION_MODE;
+import static org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_STORAGE_OPTIONS_PREFIX;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.Mockito.mock;
@@ -108,7 +109,7 @@ public void testHandleLanceTableChangeRespectsOrder() {
     Version version = mock(Version.class);
     when(dataset.getVersion()).thenReturn(version);
     when(version.getId()).thenReturn(7L);
-    Mockito.doReturn(dataset).when(lanceTableOps).openDataset("location");
+    Mockito.doReturn(dataset).when(lanceTableOps).openDataset("location", Map.of());
 
     TableChange[] changes =
         new TableChange[] {
@@ -126,4 +127,45 @@ public void testHandleLanceTableChangeRespectsOrder() {
     inOrder.verify(dataset).dropColumns(anyList());
     inOrder.verify(dataset).getVersion();
   }
+
+  @Test
+  public void testHandleLanceTableChangeUsesCatalogStorageOptions() {
+    lanceTableOps.setCatalogProperties(
+        Map.of(
+            LANCE_STORAGE_OPTIONS_PREFIX + "endpoint", "http://catalog-endpoint",
+            LANCE_STORAGE_OPTIONS_PREFIX + "secret_access_key", "catalog-secret"));
+
+    Table table = mock(Table.class);
+    when(table.properties())
+        .thenReturn(
+            Map.of(
+                Table.PROPERTY_LOCATION,
+                "location",
+                LANCE_STORAGE_OPTIONS_PREFIX + "access_key_id",
+                "table-key"));
+
+    Dataset dataset = mock(Dataset.class);
+    Version version = mock(Version.class);
+    when(dataset.getVersion()).thenReturn(version);
+    when(version.getId()).thenReturn(9L);
+    Mockito.doReturn(dataset)
+        .when(lanceTableOps)
+        .openDataset(
+            "location",
+            Map.of(
+                "endpoint",
+                "http://catalog-endpoint",
+                "secret_access_key",
+                "catalog-secret",
+                "access_key_id",
+                "table-key"));
+
+    TableChange[] changes =
+        new TableChange[] {TableChange.deleteColumn(new String[] {"col1"}, false)};
+    long returnedVersion = lanceTableOps.handleLanceTableChange(table, changes);
+
+    Assertions.assertEquals(9L, returnedVersion);
+    Mockito.verify(dataset).dropColumns(anyList());
+    Mockito.verify(dataset).getVersion();
+  }
 }

docs/lakehouse-generic-lance-table.md

@@ -303,9 +303,26 @@ 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" \
+  -H "Content-Type: application/json" \
+  -d '{
+  "name": "lance_catalog",
+  "type": "RELATIONAL",
+  "provider": "lakehouse-generic",
+  "comment": "catalog for Lance tables on MinIO",
+  "properties": {
+    "location": "s3://bucket1/lance",
+    "lance.storage.endpoint": "http://minio:9000",
+    "lance.storage.access_key_id": "ak",
+    "lance.storage.secret_access_key": "sk",
+    "lance.storage.allow_http": "true",
+    "lance.storage.region": "us-east-1"
+  }
+}' http://localhost:8090/api/metalakes/test/catalogs
+
 curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
   -H "Content-Type: application/json" -d '{
   "name": "lance_orders",
@@ -320,13 +337,10 @@ curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
   ],
   "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,28 @@ 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.
+
+For example, create the Gravitino catalog with catalog-level Lance storage properties:
+
+```shell
+curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
+  -H "Content-Type: application/json" \
+  -d '{
+  "name": "lance_catalog",
+  "type": "RELATIONAL",
+  "provider": "lakehouse-generic",
+  "comment": "catalog for Lance tables on MinIO",
+  "properties": {
+    "location": "s3://bucket/tmp",
+    "lance.storage.endpoint": "http://minio:9000",
+    "lance.storage.access_key_id": "ak",
+    "lance.storage.secret_access_key": "sk",
+    "lance.storage.allow_http": "true",
+    "lance.storage.region": "us-east-1"
+  }
+}' http://localhost:8090/api/metalakes/test/catalogs
+```
 
 ```python
 spark.sql("""
@@ -136,16 +157,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 +247,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;
   }
 
@@ -179,11 +180,11 @@ public CreateTableResponse createTable(
             .createTable(
                 tableIdentifier, columns.toArray(new Column[0]), null, createTableProperties);
     Map<String, String> properties = t.properties();
+    Map<String, String> effectiveStorageOptions =
+        LancePropertiesUtils.resolveLanceStorageOptions(catalog.properties(), 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));