This is an automated email from the ASF dual-hosted git repository.
FANNG1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new b497691bc0 [#12525] feat(spark-connector): Add lakehouse REST catalog
discovery for lance (#12541)
b497691bc0 is described below
commit b497691bc026ddb3f2b76693089439f48a88b479
Author: FANNG <[email protected]>
AuthorDate: Fri Sep 18 15:15:06 2026 +0800
[#12525] feat(spark-connector): Add lakehouse REST catalog discovery for
lance (#12541)
### What changes were proposed in this pull request?
Add `GravitinoLakehouseRESTDiscoveryPlugin`, a Spark plugin that
discovers
lakehouse REST catalogs and registers them during driver startup.
This change:
- Adds a built-in Lance provider mapping and a catalog registration
policy API.
- Enforces user configuration precedence and atomic Spark configuration
updates.
- Discovers Lance catalogs using the Lance Namespace API with pagination
support.
- Injects the required Lance Spark extension.
- Treats catalogs already configured in Spark as normal skips in
`GravitinoDriverPlugin`.
- Adds unit tests, Lance integration coverage, and user documentation.
### Why are the changes needed?
Spark users currently need to configure every Lance REST catalog
manually.
Automatic discovery allows the Lance REST server to provide the
authoritative
catalog list while preserving user-defined Spark catalog configuration.
Fix: #12525
### Does this PR introduce _any_ user-facing change?
Yes.
Users can enable Lance REST catalog discovery with:
```text
spark.plugins=org.apache.gravitino.spark.connector.plugin.GravitinoLakehouseRESTDiscoveryPlugin
spark.sql.gravitino.lanceREST.uri=http://127.0.0.1:9101/lance
```
The change also introduces:
- `CatalogRegistrationPolicy`
- `spark.sql.gravitino.REST.registrationPolicy`
- `spark.sql.gravitino.lanceREST.catalogProperties.<key>`
Authenticated Lance REST listing is not supported in this change.
### How was this patch tested?
- `./gradlew :spark-connector:spark-common:check`
- Lance Spark integration matrix with bundle versions 0.2.0, 0.4.0, and
0.5.1.
- Spark 3.3, 3.4, and 3.5 runtime shadow JAR builds.
- Packaged runtime smoke test using only the shadow runtime and external
Lance bundle.
- Verified the packaged provider has no shaded Apache HttpClient
linkage.
- `git diff --check`
---
design-docs/spark-rest-catalog-registration.md | 49 +-
docs/spark-connector/spark-connector.md | 1 +
.../spark-rest-catalog-discovery.md | 113 ++++
gradle/libs.versions.toml | 1 +
lance/lance-rest-server/build.gradle.kts | 2 +
.../integration/test/LanceSparkRESTServiceIT.java | 49 +-
.../connector/plugin/GravitinoDriverPlugin.java | 7 +
.../restcatalog/BuiltinRESTCatalogProviders.java | 38 ++
.../restcatalog/CatalogRegistrationPolicy.java | 52 ++
...ravitinoLakehouseRESTDiscoveryDriverPlugin.java | 504 +++++++++++++++
.../GravitinoLakehouseRESTDiscoveryPlugin.java | 38 ++
.../restcatalog/LakehouseRESTCatalogProvider.java | 66 ++
.../lance/LanceRESTCatalogProvider.java | 147 +++++
.../spark/connector/utils/ConnectorUtil.java | 18 +-
.../plugin/TestGravitinoDriverPlugin.java | 16 +
...ravitinoLakehouseRESTDiscoveryDriverPlugin.java | 687 +++++++++++++++++++++
.../lance/TestLanceRESTCatalogProvider.java | 196 ++++++
.../spark/connector/utils/TestConnectorUtil.java | 5 +
spark-connector/v3.5/spark/build.gradle.kts | 2 +
spark-connector/v4.0/spark/build.gradle.kts | 2 +
20 files changed, 1944 insertions(+), 49 deletions(-)
diff --git a/design-docs/spark-rest-catalog-registration.md
b/design-docs/spark-rest-catalog-registration.md
index 1495c7e84f..bc7fac8924 100644
--- a/design-docs/spark-rest-catalog-registration.md
+++ b/design-docs/spark-rest-catalog-registration.md
@@ -66,10 +66,11 @@ spark.sql.catalog.iceberg_prod.warehouse=iceberg_prod
## Proposal
-Two new pieces — a **catalog-listing endpoint** on the Iceberg REST server and
a single Spark
-plugin, `GravitinoLakehouseRESTDiscoveryPlugin`, that consumes it — plus one
**ordering rule** that
-makes the interaction with `GravitinoSparkPlugin` deterministic. Each
lakehouse format plugs into
-the plugin as a provider; V1 ships Iceberg.
+The Spark plugin `GravitinoLakehouseRESTDiscoveryPlugin` and a catalog-listing
endpoint for
+formats that need one, plus one **ordering rule**, make the interaction with
+`GravitinoSparkPlugin` deterministic. Each lakehouse format plugs into the
plugin as a provider.
+Lance is the first supported provider; an Iceberg provider and its
catalog-listing endpoint remain
+future work.
### Catalog-listing endpoint
@@ -109,16 +110,16 @@ hand-write them. Which catalogs are registered, and under
what Spark name, is de
`CatalogRegistrationPolicy`.
```text
-spark.plugins=org.apache.gravitino.spark.connector.plugin.GravitinoLakehouseRESTDiscoveryPlugin
+spark.plugins=org.apache.gravitino.spark.connector.plugin.restcatalog.GravitinoLakehouseRESTDiscoveryPlugin
```
-The plugin is format-agnostic. Each lakehouse format is a **provider** —
Iceberg in V1, Lance later
-(see [Lance support](#lance-support)) — carrying its own engine runtime and
config prefix
-(`icebergREST.*`, `lanceREST.*`). A provider is active only when its `uri` is
set, so the URI
-doubles as the per-format switch and no `enable*` flag is needed; if a `uri`
is set but its provider
-is not on the classpath, the plugin fails fast. Dependency isolation is
preserved — a user who needs
-only Iceberg puts only the Iceberg provider on the classpath — while
`spark.plugins` lists one
-plugin and there is a single ordering rule.
+The plugin is format-agnostic. Each lakehouse format is a **provider** — Lance
is supported now and
+Iceberg is planned separately (see [Lance support](#lance-support)) — carrying
its own engine
+runtime and config prefix (`icebergREST.*`, `lanceREST.*`). A provider is
active only when its `uri`
+is set, so the URI doubles as the per-format switch and no `enable*` flag is
needed; if a `uri` is
+set but its provider is not on the classpath, the plugin fails fast.
Dependency isolation is
+preserved — a user who needs only Lance puts only the Lance provider on the
classpath — while
+`spark.plugins` lists one plugin and there is a single ordering rule.
The registration policy and user-configuration precedence below are shared by
every provider; the
listing client, generated entries, and credential handling are
provider-specific.
@@ -236,7 +237,7 @@ reach Spark is intentionally left to the two existing
Iceberg REST paths:
When both plugins are configured, `GravitinoLakehouseRESTDiscoveryPlugin`
**must be listed first**:
```text
-spark.plugins=org.apache.gravitino.spark.connector.plugin.GravitinoLakehouseRESTDiscoveryPlugin,\
+spark.plugins=org.apache.gravitino.spark.connector.plugin.restcatalog.GravitinoLakehouseRESTDiscoveryPlugin,\
org.apache.gravitino.spark.connector.plugin.GravitinoSparkPlugin
```
@@ -257,7 +258,7 @@ which would now fire once per REST-registered catalog and
read as an error.
2. Add to Spark configuration:
```text
-
spark.plugins=org.apache.gravitino.spark.connector.plugin.GravitinoLakehouseRESTDiscoveryPlugin
+
spark.plugins=org.apache.gravitino.spark.connector.plugin.restcatalog.GravitinoLakehouseRESTDiscoveryPlugin
spark.sql.gravitino.icebergREST.uri=http://127.0.0.1:9001/iceberg/
```
@@ -293,9 +294,9 @@ enforcement point for authorization and audit.
## Lance support
-A design sketch, not part of V1. Lance plugs into the same plugin as a second
provider, reusing the
-registration policy, the ordering rule, and the user-configuration precedence,
and differing only in
-discovery and the generated entries.
+Lance is the first provider in the plugin. It reuses the registration policy,
the ordering rule, and
+the user-configuration precedence, and differs from future providers only in
discovery and the
+generated entries.
**Discovery needs no new API.** The Lance Namespace protocol's root list,
`GET {lance-rest-base}/v1/namespace/list`, already returns the
`lakehouse-generic` catalogs from
@@ -311,9 +312,8 @@ model is identical to Iceberg: point the plugin at a REST
server, get its catalo
Gravitino splits that into two systems (Gravitino lists them, the REST server
serves them) plus the
unstated assumption that the names line up, and makes Lance discovery behave
differently from
Iceberg. That mismatch stays invisible until a name does not line up, which
fails at query time
-rather than at configuration time. Discovery therefore stays
server-authoritative; Gravitino's
-per-principal authorization, the one thing this path would have reused, is
instead added on the
-Lance server (see below).
+rather than at configuration time. Discovery therefore stays
server-authoritative and does not
+depend on Gravitino's per-principal authorization filtering for the root
namespace list.
**The Lance provider.** It uses the `spark.sql.gravitino.lanceREST.*` prefix,
mirroring the Iceberg
keys (`uri`, `catalogProperties.<key>`), and lists catalogs with
@@ -343,10 +343,11 @@ delivers it per table in
`DescribeTableResponse.storageOptions`, resolved from t
table's `lance.storage.*` properties. The plugin generates nothing either way;
unlike Iceberg
vending, secrets in catalog properties reach the client, with no vending
equivalent today.
-One gap must close before Lance ships: the root list is authenticated but not
-authorization-filtered, so every authenticated caller sees all catalog names.
Filtering it with
-`MetadataAuthzHelper`, as the Iceberg server already does for namespace
listing, is a prerequisite —
-not a follow-up.
+The root list requires authentication but is not authorization-filtered: every
authenticated caller
+that can access the Lance REST service can discover all catalog names served
by that endpoint. This
+is an intentional discovery contract, not a prerequisite for Lance support.
Table and namespace
+operations continue to enforce their existing authorization rules.
Per-principal filtering of the
+root list may be added later if catalog-name visibility needs to be restricted.
---
diff --git a/docs/spark-connector/spark-connector.md
b/docs/spark-connector/spark-connector.md
index b51ca6c262..7e5fbe8cd6 100644
--- a/docs/spark-connector/spark-connector.md
+++ b/docs/spark-connector/spark-connector.md
@@ -14,6 +14,7 @@ The Apache Gravitino Spark connector leverages the Spark
DataSourceV2 interface
1. Supports [Hive catalog](spark-catalog-hive.md), [Iceberg
catalog](spark-catalog-iceberg.md), [Paimon catalog](spark-catalog-paimon.md),
[Jdbc catalog](spark-catalog-jdbc.md), and [AWS Glue
catalog](spark-catalog-glue.md).
2. Supports federation query.
3. Supports most DDL and DML SQLs.
+4. Supports automatic [lakehouse REST catalog
discovery](spark-rest-catalog-discovery.md).
## Requirement
diff --git a/docs/spark-connector/spark-rest-catalog-discovery.md
b/docs/spark-connector/spark-rest-catalog-discovery.md
new file mode 100644
index 0000000000..0bf57eed98
--- /dev/null
+++ b/docs/spark-connector/spark-rest-catalog-discovery.md
@@ -0,0 +1,113 @@
+---
+title: "Spark Lakehouse REST Catalog Discovery"
+slug: "/spark-connector/spark-rest-catalog-discovery"
+keyword: "spark lakehouse REST catalog discovery Lance"
+license: "This software is licensed under the Apache License version 2."
+---
+
+## Overview
+
+The Gravitino Spark connector can discover the catalogs served by a lakehouse
REST server and
+register them in Spark automatically. A Spark session only needs the discovery
plugin and one REST
+server URI per lakehouse format. Catalogs added to the server become available
after the Spark
+session is restarted.
+
+Lance REST discovery is supported. The corresponding Iceberg REST provider is
planned separately.
+
+## Requirements
+
+Add both of the following artifacts to the Spark application classpath:
+
+- The Gravitino Spark connector runtime matching the Spark version.
+- A compatible `lance-spark-bundle` containing
`org.lance.spark.LanceNamespaceSparkCatalog` and
+ `org.lance.spark.extensions.LanceSparkSessionExtensions`.
+
+The discovery provider uses the Lance Namespace client shipped in the Lance
Spark bundle. Gravitino
+does not package another copy of that client in its Spark connector runtime.
+
+## Configuration
+
+Configure the discovery plugin and the Lance REST server URI:
+
+```text
+spark.plugins=org.apache.gravitino.spark.connector.plugin.restcatalog.GravitinoLakehouseRESTDiscoveryPlugin
+spark.sql.gravitino.lanceREST.uri=http://127.0.0.1:9101/lance
+```
+
+When the regular Gravitino Spark plugin is also required, the discovery plugin
must be listed first:
+
+```text
+spark.plugins=org.apache.gravitino.spark.connector.plugin.restcatalog.GravitinoLakehouseRESTDiscoveryPlugin,\
+ org.apache.gravitino.spark.connector.plugin.GravitinoSparkPlugin
+```
+
+Spark startup fails if the order is reversed. Initializing discovery first
allows the plugin to
+distinguish user-written catalog configuration from entries generated by the
regular plugin.
+
+For every catalog discovered from the Lance REST server, the plugin generates:
+
+```text
+spark.sql.catalog.<catalogName>=org.lance.spark.LanceNamespaceSparkCatalog
+spark.sql.catalog.<catalogName>.impl=rest
+spark.sql.catalog.<catalogName>.uri=<spark.sql.gravitino.lanceREST.uri>
+spark.sql.catalog.<catalogName>.parent=<catalogName>
+```
+
+It also adds `org.lance.spark.extensions.LanceSparkSessionExtensions` to
+`spark.sql.extensions`, without adding a duplicate when the extension is
already configured.
+
+### Catalog properties
+
+Properties under the following prefix are copied to every discovered Lance
catalog:
+
+```text
+spark.sql.gravitino.lanceREST.catalogProperties.<key>=<value>
+```
+
+Configuration precedence, from highest to lowest, is:
+
+1. A user-defined `spark.sql.catalog.<name>` owns the complete catalog name
and disables automatic
+ registration for that discovered name.
+2. A user-defined `spark.sql.catalog.<name>.<key>` overrides the value for
that catalog.
+3. Provider-generated routing properties such as `impl`, `uri`, and `parent`.
+4. Global `lanceREST.catalogProperties.<key>` defaults.
+
+The discovery provider does not generate `lance.storage.*` configuration.
Storage options continue
+to come from the Lance REST service or from explicit Spark catalog properties.
+
+Set `spark.sql.gravitino.lanceREST.catalogProperties.namespaceDelimiter` when
the Lance REST
+server uses a namespace delimiter other than the default `$`. The property is
used for the
+discovery request and is also copied to each discovered catalog.
+
+## Registration policy
+
+By default, every discovered catalog is registered under its discovered name.
To filter or rename
+catalogs, implement
+`org.apache.gravitino.spark.connector.plugin.restcatalog.CatalogRegistrationPolicy`
with a public
+no-argument constructor and configure its class name:
+
+```text
+spark.sql.gravitino.REST.registrationPolicy=com.example.MyCatalogRegistrationPolicy
+```
+
+The policy receives the format token (`lance`) and the discovered catalog
name. A renamed Spark
+catalog still uses the discovered name as its `parent`, so REST routing
remains server-authoritative.
+Spark startup fails if a policy returns a blank name, an invalid identifier, a
duplicate name, or a
+name already owned by user configuration.
+
+Gravitino accepts catalog names that Spark cannot reference unquoted, such as
names containing a
+hyphen. Without a configured policy, such a catalog is logged at WARN and
skipped rather than
+failing Spark startup, so one unusable name does not prevent the session from
starting. Configure a
+policy to rename it to an identifier Spark accepts.
+
+## Limitations
+
+- Each format supports one REST server URI per Spark session.
+- Discovery is a startup dependency with no fallback. If the REST server is
unreachable while the
+ driver starts, the plugin fails and the whole Spark session fails to start.
+- Automatic Lance discovery does not support authenticated REST listing.
Catalog properties are
+ applied to generated Spark catalogs but are not used to authenticate the
discovery request.
+- Discovery runs only during driver startup; restart the Spark session to
observe catalog additions
+ or removals.
+- A catalog registered by Spark is loaded lazily, so `SHOW CATALOGS` may not
display it until it has
+ been referenced.
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 02d82c09a2..6951b93f4e 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -210,6 +210,7 @@ log4j-layout-template-json = { group =
"org.apache.logging.log4j", name = "log4j
arrow-vector = { group = "org.apache.arrow", name = "arrow-vector",
version.ref = "arrow" }
lance = { group = "org.lance", name = "lance-core", version.ref = "lance" }
lance-namespace-core = { group = "org.lance", name = "lance-namespace-core",
version.ref = "lance-namespace" }
+lance-namespace-apache-client = { group = "org.lance", name =
"lance-namespace-apache-client", version.ref = "lance-namespace" }
delta-kernel = { group = "io.delta", name = "delta-kernel-api", version.ref =
"delta-kernel" }
delta-kernel-defaults = { group = "io.delta", name = "delta-kernel-defaults",
version.ref = "delta-kernel" }
jakarta-validation-api = { group = "jakarta.validation", name =
"jakarta.validation-api", version.ref = "jakarta-validation" }
diff --git a/lance/lance-rest-server/build.gradle.kts
b/lance/lance-rest-server/build.gradle.kts
index d55180fa15..8f35a43966 100644
--- a/lance/lance-rest-server/build.gradle.kts
+++ b/lance/lance-rest-server/build.gradle.kts
@@ -102,9 +102,11 @@ dependencies {
testImplementation(project(":clients:client-java"))
testImplementation(project(":server"))
+ testImplementation(project(":spark-connector:spark-3.5"))
testImplementation(project(":integration-test-common", "testArtifacts"))
testImplementation(project(":server-common", "testArtifacts"))
testImplementation(libs.lance)
+ testImplementation(libs.lance.namespace.apache.client)
lanceSparkBundleVersions.forEach { version ->
add(
diff --git
a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceSparkRESTServiceIT.java
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceSparkRESTServiceIT.java
index 62977fcb19..9a2d2d42bd 100644
---
a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceSparkRESTServiceIT.java
+++
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceSparkRESTServiceIT.java
@@ -46,6 +46,7 @@ import org.apache.gravitino.integration.test.util.BaseIT;
import org.apache.gravitino.integration.test.util.GravitinoITUtils;
import org.apache.gravitino.lance.common.utils.LanceConstants;
import org.apache.gravitino.rel.Table;
+import
org.apache.gravitino.spark.connector.plugin.restcatalog.GravitinoLakehouseRESTDiscoveryPlugin;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.AnalysisException;
import org.apache.spark.sql.Row;
@@ -58,10 +59,10 @@ import org.junit.jupiter.api.Test;
public class LanceSparkRESTServiceIT extends BaseIT {
- private static final String LANCE_SPARK_CATALOG = "lance";
- private static final String HISTORY_LANCE_SPARK_CATALOG = "lance_history";
private static final String LANCE_SPARK_CATALOG_CLASS =
"org.lance.spark.LanceNamespaceSparkCatalog";
+ private static final String LANCE_SPARK_EXTENSIONS =
+ "org.lance.spark.extensions.LanceSparkSessionExtensions";
private static final String LANCE_SPARK_BUNDLE_JAR_PATH_PROPERTY =
"gravitino.lance.spark.bundle.jar";
private static final String JAVA_MODULE_OPEN_OPTIONS =
@@ -148,11 +149,27 @@ public class LanceSparkRESTServiceIT extends BaseIT {
createdSchemas.clear();
for (String schemaName : createdHistorySchemas) {
- dropSchema(HISTORY_LANCE_SPARK_CATALOG, schemaName, historyCatalog);
+ dropSchema(HISTORY_CATALOG_NAME, schemaName, historyCatalog);
}
createdHistorySchemas.clear();
}
+ @Test
+ public void testCatalogRegisteredByDiscoveryPlugin() {
+ SparkConf sparkConf = sparkSession.sparkContext().getConf();
+
+ Assertions.assertEquals(
+ LANCE_SPARK_CATALOG_CLASS, sparkConf.get("spark.sql.catalog." +
CATALOG_NAME));
+ Assertions.assertEquals("rest", sparkConf.get("spark.sql.catalog." +
CATALOG_NAME + ".impl"));
+ Assertions.assertEquals(
+ getLanceRestServiceUrl(), sparkConf.get("spark.sql.catalog." +
CATALOG_NAME + ".uri"));
+ Assertions.assertEquals(
+ CATALOG_NAME, sparkConf.get("spark.sql.catalog." + CATALOG_NAME +
".parent"));
+ Assertions.assertTrue(
+ Arrays.asList(sparkConf.get("spark.sql.extensions").split(","))
+ .contains(LANCE_SPARK_EXTENSIONS));
+ }
+
@Test
public void testCreateAndListNamespacesViaSpark() {
String schemaName = createSchema("spark_ns_list");
@@ -224,8 +241,7 @@ public class LanceSparkRESTServiceIT extends BaseIT {
public void testReadHistoricalVersionViaSpark() {
String schemaName = createHistorySchema("spark_history_read");
String tableName = newTableName("orders");
- String tableIdentifier =
- String.format("%s.%s.%s", HISTORY_LANCE_SPARK_CATALOG, schemaName,
tableName);
+ String tableIdentifier = String.format("%s.%s.%s", HISTORY_CATALOG_NAME,
schemaName, tableName);
sql(
"CREATE TABLE %s (id INT, score FLOAT) USING lance TBLPROPERTIES
('format'='lance')",
@@ -351,6 +367,7 @@ public class LanceSparkRESTServiceIT extends BaseIT {
Assertions.assertThrows(
RuntimeException.class, () ->
createLanceTable(nonExistentSchemaName, tableName));
assertFailureContainsAll(exception, nonExistentSchemaName, "does not
exist");
+ assertFailureContainsAny(exception, "NoSuchSchemaException", "Namespace
not found");
Assertions.assertFalse(catalog.asSchemas().schemaExists(nonExistentSchemaName));
}
@@ -444,19 +461,9 @@ public class LanceSparkRESTServiceIT extends BaseIT {
.set("spark.jars", bundleJarPath)
.set("spark.driver.extraClassPath", bundleJarPath)
.set("spark.executor.extraClassPath", bundleJarPath)
- .set("spark.sql.catalog." + LANCE_SPARK_CATALOG,
LANCE_SPARK_CATALOG_CLASS)
- .set("spark.sql.catalog." + LANCE_SPARK_CATALOG + ".impl", "rest")
- .set("spark.sql.catalog." + LANCE_SPARK_CATALOG + ".uri",
getLanceRestServiceUrl())
- .set("spark.sql.catalog." + LANCE_SPARK_CATALOG + ".parent",
CATALOG_NAME)
- .set("spark.sql.catalog." + HISTORY_LANCE_SPARK_CATALOG,
LANCE_SPARK_CATALOG_CLASS)
- .set("spark.sql.catalog." + HISTORY_LANCE_SPARK_CATALOG + ".impl",
"rest")
- .set(
- "spark.sql.catalog." + HISTORY_LANCE_SPARK_CATALOG + ".uri",
- getLanceRestServiceUrl())
- .set(
- "spark.sql.catalog." + HISTORY_LANCE_SPARK_CATALOG + ".parent",
- HISTORY_CATALOG_NAME)
- .set("spark.sql.defaultCatalog", LANCE_SPARK_CATALOG)
+ .set("spark.plugins",
GravitinoLakehouseRESTDiscoveryPlugin.class.getName())
+ .set("spark.sql.gravitino.lanceREST.uri", getLanceRestServiceUrl())
+ .set("spark.sql.defaultCatalog", CATALOG_NAME)
.set("spark.driver.extraJavaOptions", JAVA_MODULE_OPEN_OPTIONS)
.set("spark.executor.extraJavaOptions", JAVA_MODULE_OPEN_OPTIONS)
.set("spark.ui.enabled", "false");
@@ -547,7 +554,7 @@ public class LanceSparkRESTServiceIT extends BaseIT {
private String createHistorySchema(String schemaNamePrefix) {
String schemaName = newSchemaName(schemaNamePrefix);
- sql("CREATE DATABASE %s.%s", HISTORY_LANCE_SPARK_CATALOG, schemaName);
+ sql("CREATE DATABASE %s.%s", HISTORY_CATALOG_NAME, schemaName);
createdHistorySchemas.add(schemaName);
return schemaName;
}
@@ -559,7 +566,7 @@ public class LanceSparkRESTServiceIT extends BaseIT {
}
private void dropSchema(String schemaName) {
- dropSchema(LANCE_SPARK_CATALOG, schemaName, catalog);
+ dropSchema(CATALOG_NAME, schemaName, catalog);
}
private void dropSchema(String sparkCatalogName, String schemaName, Catalog
sourceCatalog) {
@@ -580,7 +587,7 @@ public class LanceSparkRESTServiceIT extends BaseIT {
}
private List<String> listNamespaces() {
- return sql("SHOW NAMESPACES IN %s", LANCE_SPARK_CATALOG).stream()
+ return sql("SHOW NAMESPACES IN %s", CATALOG_NAME).stream()
.map(row -> row.getString(0))
.collect(Collectors.toList());
}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/GravitinoDriverPlugin.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/GravitinoDriverPlugin.java
index 85e6602b32..6950599eb2 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/GravitinoDriverPlugin.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/GravitinoDriverPlugin.java
@@ -173,6 +173,13 @@ public class GravitinoDriverPlugin implements DriverPlugin
{
if (SparkCatalogKind.LAKEHOUSE_PAIMON.equals(kind) &&
!enablePaimonSupport) {
return;
}
+ String sparkCatalogConfigName = "spark.sql.catalog." +
catalogName;
+ if (sparkConf.contains(sparkCatalogConfigName)) {
+ LOG.info(
+ "Skip registering catalog {} because it is already
configured in Spark.",
+ catalogName);
+ return;
+ }
try {
registerCatalog(sparkConf, catalogName, kind);
} catch (Exception e) {
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/BuiltinRESTCatalogProviders.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/BuiltinRESTCatalogProviders.java
new file mode 100644
index 0000000000..ebbc2b9501
--- /dev/null
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/BuiltinRESTCatalogProviders.java
@@ -0,0 +1,38 @@
+/*
+ * 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.spark.connector.plugin.restcatalog;
+
+import java.util.Collections;
+import java.util.Map;
+
+final class BuiltinRESTCatalogProviders {
+
+ private static final String LANCE_FORMAT = "lance";
+ private static final String LANCE_PROVIDER_CLASS_NAME =
+
"org.apache.gravitino.spark.connector.plugin.restcatalog.lance.LanceRESTCatalogProvider";
+ private static final Map<String, String> PROVIDER_CLASS_NAMES =
+ Collections.singletonMap(LANCE_FORMAT, LANCE_PROVIDER_CLASS_NAME);
+
+ private BuiltinRESTCatalogProviders() {}
+
+ static Map<String, String> providerClassNames() {
+ return PROVIDER_CLASS_NAMES;
+ }
+}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/CatalogRegistrationPolicy.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/CatalogRegistrationPolicy.java
new file mode 100644
index 0000000000..3e7033b3bb
--- /dev/null
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/CatalogRegistrationPolicy.java
@@ -0,0 +1,52 @@
+/*
+ * 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.spark.connector.plugin.restcatalog;
+
+import org.apache.gravitino.annotation.DeveloperApi;
+
+/**
+ * Decides whether a discovered REST catalog is registered, and under what
Spark name.
+ *
+ * <p>If no policy is configured, every discovered catalog is registered under
its discovered name.
+ * A configured implementation must provide a public no-argument constructor.
+ */
+@DeveloperApi
+public interface CatalogRegistrationPolicy {
+
+ /**
+ * Returns whether to register a discovered catalog automatically.
+ *
+ * @param format the lakehouse format that discovered the catalog
+ * @param catalogName the catalog name returned by the format's REST server
+ * @return true to register the catalog, false to skip it
+ */
+ boolean shouldRegister(String format, String catalogName);
+
+ /**
+ * Returns the Spark catalog name for an accepted catalog.
+ *
+ * @param format the lakehouse format that discovered the catalog
+ * @param catalogName the accepted catalog name
+ * @return the Spark catalog name
+ */
+ default String registeredCatalogName(String format, String catalogName) {
+ return catalogName;
+ }
+}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/GravitinoLakehouseRESTDiscoveryDriverPlugin.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/GravitinoLakehouseRESTDiscoveryDriverPlugin.java
new file mode 100644
index 0000000000..596b27e438
--- /dev/null
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/GravitinoLakehouseRESTDiscoveryDriverPlugin.java
@@ -0,0 +1,504 @@
+/*
+ * 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.spark.connector.plugin.restcatalog;
+
+import static org.apache.gravitino.spark.connector.ConnectorConstants.COMMA;
+import static
org.apache.gravitino.spark.connector.utils.ConnectorUtil.removeDuplicateSparkExtensions;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.spark.connector.plugin.GravitinoSparkPlugin;
+import org.apache.spark.SparkConf;
+import org.apache.spark.SparkContext;
+import org.apache.spark.api.plugin.DriverPlugin;
+import org.apache.spark.api.plugin.PluginContext;
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser$;
+import org.apache.spark.sql.catalyst.parser.ParseException;
+import org.apache.spark.sql.catalyst.parser.ParserInterface;
+import org.apache.spark.sql.internal.StaticSQLConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Tuple2;
+import scala.collection.Seq;
+
+class GravitinoLakehouseRESTDiscoveryDriverPlugin implements DriverPlugin {
+
+ @VisibleForTesting
+ static final String REGISTRATION_POLICY_CONFIG =
"spark.sql.gravitino.REST.registrationPolicy";
+
+ private static final Logger LOG =
+
LoggerFactory.getLogger(GravitinoLakehouseRESTDiscoveryDriverPlugin.class);
+ private static final String GRAVITINO_PREFIX = "spark.sql.gravitino.";
+ private static final String SPARK_CATALOG_PREFIX = "spark.sql.catalog.";
+ private static final String URI_SUFFIX = "REST.uri";
+ private static final String CATALOG_PROPERTIES_INFIX =
"REST.catalogProperties.";
+ private static final Pattern PROVIDER_URI_PATTERN =
+
Pattern.compile("^spark\\.sql\\.gravitino\\.([A-Za-z][A-Za-z0-9]*)REST\\.uri$");
+ // Matches keys that look like a discovery URI but are spelled differently,
so a typo such as
+ // `spark.sql.gravitino.lanceRest.uri` is reported instead of silently
disabling discovery.
+ private static final Pattern SIMILAR_URI_PATTERN =
+ Pattern.compile("^spark\\.sql\\.gravitino\\..*rest\\.uri$",
Pattern.CASE_INSENSITIVE);
+ private static final CatalogRegistrationPolicy DEFAULT_POLICY = (format,
catalogName) -> true;
+
+ GravitinoLakehouseRESTDiscoveryDriverPlugin() {}
+
+ @Override
+ public Map<String, String> init(SparkContext sc, PluginContext
pluginContext) {
+ initialize(sc.conf());
+ return Collections.emptyMap();
+ }
+
+ @VisibleForTesting
+ void initialize(SparkConf sparkConf) {
+ initialize(sparkConf, BuiltinRESTCatalogProviders.providerClassNames());
+ }
+
+ @VisibleForTesting
+ void initialize(SparkConf sparkConf, Map<String, String> providerClassNames)
{
+ validatePluginOrder(sparkConf);
+ SparkConf userConf = sparkConf.clone();
+ Map<String, String> activeFormats = findActiveFormats(userConf);
+ if (activeFormats.isEmpty()) {
+ warnNoActiveFormat(userConf);
+ return;
+ }
+
+ ClassLoader classLoader = contextClassLoader();
+ CatalogRegistrationPolicy policy = loadRegistrationPolicy(userConf,
classLoader);
+ List<CatalogRegistration> registrations = new ArrayList<>();
+ Set<String> registeredNames = new LinkedHashSet<>();
+ Set<String> extensions = new LinkedHashSet<>();
+
+ activeFormats.forEach(
+ (format, uri) -> {
+ LakehouseRESTCatalogProvider provider =
+ loadProvider(format, providerClassNames, classLoader);
+ validateProviderRuntime(provider, classLoader);
+
+ Map<String, String> globalProperties =
extractCatalogProperties(userConf, format);
+ List<String> discoveredCatalogs = listCatalogs(provider, format,
uri, globalProperties);
+ if (discoveredCatalogs.isEmpty()) {
+ LOG.warn("Discovered no {} REST catalog from {}.", format, uri);
+ } else {
+ LOG.info("Discovered {} REST catalogs {} from {}.", format,
discoveredCatalogs, uri);
+ }
+
+ for (String catalogName : discoveredCatalogs) {
+ addRegistration(
+ userConf,
+ provider,
+ policy,
+ format,
+ uri,
+ catalogName,
+ globalProperties,
+ registeredNames,
+ registrations);
+ }
+ extensions.addAll(Arrays.asList(provider.sparkExtensions()));
+ });
+
+ applyRegistrations(sparkConf, userConf, registrations);
+ registerSqlExtensions(sparkConf, extensions);
+ }
+
+ private static void validatePluginOrder(SparkConf sparkConf) {
+ String configuredPlugins = sparkConf.get("spark.plugins", "");
+ List<String> plugins = new ArrayList<>();
+ for (String plugin : configuredPlugins.split(COMMA)) {
+ if (StringUtils.isNotBlank(plugin)) {
+ plugins.add(plugin.trim());
+ }
+ }
+
+ int discoveryPluginIndex =
+ plugins.indexOf(GravitinoLakehouseRESTDiscoveryPlugin.class.getName());
+ int gravitinoPluginIndex =
plugins.indexOf(GravitinoSparkPlugin.class.getName());
+ Preconditions.checkArgument(
+ discoveryPluginIndex < 0
+ || gravitinoPluginIndex < 0
+ || discoveryPluginIndex < gravitinoPluginIndex,
+ "%s must be listed before %s in spark.plugins",
+ GravitinoLakehouseRESTDiscoveryPlugin.class.getName(),
+ GravitinoSparkPlugin.class.getName());
+ }
+
+ private static Map<String, String> findActiveFormats(SparkConf userConf) {
+ Map<String, String> activeFormats = new TreeMap<>();
+ for (Tuple2<String, String> entry : userConf.getAll()) {
+ Matcher matcher = PROVIDER_URI_PATTERN.matcher(entry._1);
+ if (matcher.matches()) {
+ String format = matcher.group(1);
+ String uri = entry._2;
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(uri),
+ "%s%s%s must not be blank",
+ GRAVITINO_PREFIX,
+ format,
+ URI_SUFFIX);
+ activeFormats.put(format, uri);
+ }
+ }
+ return activeFormats;
+ }
+
+ @VisibleForTesting
+ static List<String> similarUriKeys(SparkConf userConf) {
+ List<String> similarKeys = new ArrayList<>();
+ for (Tuple2<String, String> entry : userConf.getAll()) {
+ if (SIMILAR_URI_PATTERN.matcher(entry._1).matches()
+ && !PROVIDER_URI_PATTERN.matcher(entry._1).matches()) {
+ similarKeys.add(entry._1);
+ }
+ }
+ return similarKeys;
+ }
+
+ private static void warnNoActiveFormat(SparkConf userConf) {
+ List<String> similarKeys = similarUriKeys(userConf);
+ if (similarKeys.isEmpty()) {
+ LOG.warn(
+ "{} is configured but no {}<format>{} is set, no lakehouse REST
catalog is discovered.",
+ GravitinoLakehouseRESTDiscoveryPlugin.class.getName(),
+ GRAVITINO_PREFIX,
+ URI_SUFFIX);
+ } else {
+ LOG.warn(
+ "{} is configured but no {}<format>{} is set, no lakehouse REST
catalog is discovered. "
+ + "Keys that look misspelled: {}",
+ GravitinoLakehouseRESTDiscoveryPlugin.class.getName(),
+ GRAVITINO_PREFIX,
+ URI_SUFFIX,
+ similarKeys);
+ }
+ }
+
+ private static List<String> listCatalogs(
+ LakehouseRESTCatalogProvider provider,
+ String format,
+ String uri,
+ Map<String, String> globalProperties) {
+ List<String> discoveredCatalogs;
+ try {
+ discoveredCatalogs =
+ provider.listCatalogs(uri,
Collections.unmodifiableMap(globalProperties));
+ } catch (RuntimeException | LinkageError e) {
+ throw new IllegalStateException(
+ String.format("Failed to list %s REST catalogs from %s", format,
uri), e);
+ }
+ Preconditions.checkState(
+ discoveredCatalogs != null,
+ "Lakehouse REST catalog provider %s returned a null catalog list",
+ format);
+ return discoveredCatalogs;
+ }
+
+ private static CatalogRegistrationPolicy loadRegistrationPolicy(
+ SparkConf userConf, ClassLoader classLoader) {
+ if (!userConf.contains(REGISTRATION_POLICY_CONFIG)) {
+ return DEFAULT_POLICY;
+ }
+
+ String policyClassName = userConf.get(REGISTRATION_POLICY_CONFIG);
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(policyClassName),
+ "%s must not be blank",
+ REGISTRATION_POLICY_CONFIG);
+ try {
+ Class<?> policyClass = Class.forName(policyClassName, true, classLoader);
+ Preconditions.checkArgument(
+ CatalogRegistrationPolicy.class.isAssignableFrom(policyClass),
+ "%s does not implement %s",
+ policyClassName,
+ CatalogRegistrationPolicy.class.getName());
+ return
CatalogRegistrationPolicy.class.cast(policyClass.getConstructor().newInstance());
+ } catch (ReflectiveOperationException | LinkageError e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Failed to instantiate catalog registration policy %s configured
by %s",
+ policyClassName, REGISTRATION_POLICY_CONFIG),
+ e);
+ }
+ }
+
+ private static LakehouseRESTCatalogProvider loadProvider(
+ String format, Map<String, String> providerClassNames, ClassLoader
classLoader) {
+ String providerClassName = providerClassNames.get(format);
+ Preconditions.checkArgument(
+ providerClassName != null,
+ "No lakehouse REST catalog provider found for configured format: %s",
+ format);
+ try {
+ Class<?> providerClass = Class.forName(providerClassName, true,
classLoader);
+ Preconditions.checkArgument(
+ LakehouseRESTCatalogProvider.class.isAssignableFrom(providerClass),
+ "%s does not implement %s",
+ providerClassName,
+ LakehouseRESTCatalogProvider.class.getName());
+ return providerClass
+ .asSubclass(LakehouseRESTCatalogProvider.class)
+ .getConstructor()
+ .newInstance();
+ } catch (ReflectiveOperationException | LinkageError e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Failed to instantiate lakehouse REST catalog provider %s for
format %s",
+ providerClassName, format),
+ e);
+ }
+ }
+
+ private static void validateProviderRuntime(
+ LakehouseRESTCatalogProvider provider, ClassLoader classLoader) {
+ validateRuntimeClass(provider.format(), provider.catalogClassName(),
classLoader);
+ String[] providerExtensions = provider.sparkExtensions();
+ Preconditions.checkState(
+ providerExtensions != null,
+ "Lakehouse REST catalog provider %s returned null Spark extensions",
+ provider.format());
+ for (String extension : providerExtensions) {
+ validateRuntimeClass(provider.format(), extension, classLoader);
+ }
+ }
+
+ private static void validateRuntimeClass(
+ String format, String className, ClassLoader classLoader) {
+ Preconditions.checkState(
+ StringUtils.isNotBlank(className),
+ "Lakehouse REST catalog provider %s returned a blank runtime class",
+ format);
+ try {
+ Class.forName(className, false, classLoader);
+ } catch (ClassNotFoundException | LinkageError e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Required runtime class %s for lakehouse REST format %s is not
available",
+ className, format),
+ e);
+ }
+ }
+
+ private static Map<String, String> extractCatalogProperties(SparkConf
userConf, String format) {
+ String prefix = GRAVITINO_PREFIX + format + CATALOG_PROPERTIES_INFIX;
+ Map<String, String> properties = new LinkedHashMap<>();
+ for (Tuple2<String, String> entry : userConf.getAllWithPrefix(prefix)) {
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(entry._1), "%s must include a property name",
prefix);
+ properties.put(entry._1, entry._2);
+ }
+ return properties;
+ }
+
+ private static void addRegistration(
+ SparkConf userConf,
+ LakehouseRESTCatalogProvider provider,
+ CatalogRegistrationPolicy policy,
+ String format,
+ String uri,
+ String discoveredCatalogName,
+ Map<String, String> globalProperties,
+ Set<String> registeredNames,
+ List<CatalogRegistration> registrations) {
+ Preconditions.checkState(
+ StringUtils.isNotBlank(discoveredCatalogName),
+ "Lakehouse REST catalog provider %s discovered a blank catalog name",
+ format);
+ if (userConf.contains(SPARK_CATALOG_PREFIX + discoveredCatalogName)) {
+ LOG.info(
+ "Skip auto-registering {} catalog {} because it is configured by the
user.",
+ format,
+ discoveredCatalogName);
+ return;
+ }
+ if (!policy.shouldRegister(format, discoveredCatalogName)) {
+ LOG.info(
+ "Skip auto-registering {} catalog {} because catalog registration
policy {} rejected it.",
+ format,
+ discoveredCatalogName,
+ policy.getClass().getName());
+ return;
+ }
+
+ String registeredCatalogName = policy.registeredCatalogName(format,
discoveredCatalogName);
+ // A blank name can only come from a policy: a blank discovered name
already failed above.
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(registeredCatalogName),
+ "Catalog registration policy returned a blank catalog name");
+ if (!isValidCatalogName(registeredCatalogName)) {
+ // A user-configured policy chose this name, so failing fast points at
code the user owns. The
+ // default policy passes through what the REST server advertised, where
one name Spark cannot
+ // reference must not abort the whole Spark session.
+ Preconditions.checkArgument(
+ isDefaultPolicy(policy),
+ "Catalog registration policy returned invalid Spark identifier: %s",
+ registeredCatalogName);
+ LOG.warn(
+ "Skip auto-registering {} catalog {} because it is not a valid Spark
identifier.",
+ format,
+ discoveredCatalogName);
+ return;
+ }
+ Preconditions.checkArgument(
+ !userConf.contains(SPARK_CATALOG_PREFIX + registeredCatalogName),
+ "Catalog registration policy returned name %s, which is configured by
the user",
+ registeredCatalogName);
+ Preconditions.checkArgument(
+ registeredNames.add(registeredCatalogName),
+ "Catalog registration policy returned duplicate name: %s",
+ registeredCatalogName);
+
+ Map<String, String> generatedProperties =
+ provider.generatedCatalogProperties(uri, discoveredCatalogName);
+ Preconditions.checkState(
+ generatedProperties != null,
+ "Lakehouse REST catalog provider %s returned null generated
properties",
+ format);
+ Map<String, String> mergedProperties = new
LinkedHashMap<>(globalProperties);
+ generatedProperties.forEach(
+ (key, value) -> {
+ Preconditions.checkState(
+ StringUtils.isNotBlank(key),
+ "Lakehouse REST catalog provider %s returned a blank property
name",
+ format);
+ Preconditions.checkState(
+ value != null,
+ "Lakehouse REST catalog provider %s returned null for property
%s",
+ format,
+ key);
+ mergedProperties.put(key, value);
+ });
+
+ registrations.add(
+ new CatalogRegistration(
+ format,
+ discoveredCatalogName,
+ registeredCatalogName,
+ provider.catalogClassName(),
+ mergedProperties));
+ }
+
+ /**
+ * Returns whether Spark can reference {@code catalogName} unquoted.
Gravitino accepts catalog
+ * names that Spark's parser rejects, such as names containing a hyphen, so
the answer is a value
+ * rather than an exception: the caller decides whether that is a failure or
a skip.
+ */
+ private static boolean isValidCatalogName(String catalogName) {
+ if (StringUtils.isBlank(catalogName)) {
+ return false;
+ }
+ // Called through ParserInterface, the type that declares `throws
ParseException`, so the catch
+ // can name that exception instead of swallowing every RuntimeException
the parser may raise.
+ ParserInterface parser = CatalystSqlParser$.MODULE$;
+ try {
+ Seq<String> parts = parser.parseMultipartIdentifier(catalogName);
+ return parts.size() == 1 && catalogName.equals(parts.apply(0));
+ } catch (ParseException e) {
+ return false;
+ }
+ }
+
+ private static boolean isDefaultPolicy(CatalogRegistrationPolicy policy) {
+ return policy == DEFAULT_POLICY;
+ }
+
+ private static void applyRegistrations(
+ SparkConf sparkConf, SparkConf userConf, List<CatalogRegistration>
registrations) {
+ for (CatalogRegistration registration : registrations) {
+ String catalogPrefix = SPARK_CATALOG_PREFIX +
registration.registeredCatalogName;
+ sparkConf.set(catalogPrefix, registration.catalogClassName);
+ registration.properties.forEach(
+ (key, value) -> {
+ String sparkConfigKey = catalogPrefix + "." + key;
+ if (!userConf.contains(sparkConfigKey)) {
+ sparkConf.set(sparkConfigKey, value);
+ }
+ });
+ if
(!registration.discoveredCatalogName.equals(registration.registeredCatalogName))
{
+ LOG.info(
+ "Register {} REST catalog {} as Spark catalog {}.",
+ registration.format,
+ registration.discoveredCatalogName,
+ registration.registeredCatalogName);
+ } else {
+ LOG.info(
+ "Register {} REST catalog {} in Spark.",
+ registration.format,
+ registration.discoveredCatalogName);
+ }
+ }
+ }
+
+ private static void registerSqlExtensions(SparkConf sparkConf, Set<String>
extensions) {
+ if (extensions.isEmpty()) {
+ return;
+ }
+
+ String extensionsKey = StaticSQLConf.SPARK_SESSION_EXTENSIONS().key();
+ String[] providerExtensions = extensions.toArray(new String[0]);
+ if (sparkConf.contains(extensionsKey) &&
StringUtils.isNotBlank(sparkConf.get(extensionsKey))) {
+ sparkConf.set(
+ extensionsKey,
+ removeDuplicateSparkExtensions(
+ providerExtensions, sparkConf.get(extensionsKey).split(COMMA)));
+ } else {
+ sparkConf.set(extensionsKey, String.join(COMMA, providerExtensions));
+ }
+ }
+
+ private static ClassLoader contextClassLoader() {
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+ return classLoader == null
+ ? GravitinoLakehouseRESTDiscoveryDriverPlugin.class.getClassLoader()
+ : classLoader;
+ }
+
+ private static class CatalogRegistration {
+ private final String format;
+ private final String discoveredCatalogName;
+ private final String registeredCatalogName;
+ private final String catalogClassName;
+ private final Map<String, String> properties;
+
+ private CatalogRegistration(
+ String format,
+ String discoveredCatalogName,
+ String registeredCatalogName,
+ String catalogClassName,
+ Map<String, String> properties) {
+ this.format = format;
+ this.discoveredCatalogName = discoveredCatalogName;
+ this.registeredCatalogName = registeredCatalogName;
+ this.catalogClassName = catalogClassName;
+ this.properties = properties;
+ }
+ }
+}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/GravitinoLakehouseRESTDiscoveryPlugin.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/GravitinoLakehouseRESTDiscoveryPlugin.java
new file mode 100644
index 0000000000..4df1a5bea2
--- /dev/null
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/GravitinoLakehouseRESTDiscoveryPlugin.java
@@ -0,0 +1,38 @@
+/*
+ * 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.spark.connector.plugin.restcatalog;
+
+import org.apache.spark.api.plugin.DriverPlugin;
+import org.apache.spark.api.plugin.ExecutorPlugin;
+import org.apache.spark.api.plugin.SparkPlugin;
+
+/** Discovers lakehouse REST catalogs and registers them in Spark at driver
startup. */
+public class GravitinoLakehouseRESTDiscoveryPlugin implements SparkPlugin {
+
+ @Override
+ public DriverPlugin driverPlugin() {
+ return new GravitinoLakehouseRESTDiscoveryDriverPlugin();
+ }
+
+ @Override
+ public ExecutorPlugin executorPlugin() {
+ return null;
+ }
+}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/LakehouseRESTCatalogProvider.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/LakehouseRESTCatalogProvider.java
new file mode 100644
index 0000000000..3cc08a25be
--- /dev/null
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/LakehouseRESTCatalogProvider.java
@@ -0,0 +1,66 @@
+/*
+ * 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.spark.connector.plugin.restcatalog;
+
+import java.util.List;
+import java.util.Map;
+
+/** Provides format-specific discovery and Spark configuration for lakehouse
REST catalogs. */
+public interface LakehouseRESTCatalogProvider {
+
+ /**
+ * Returns the format token used in {@code
spark.sql.gravitino.<format>REST.*} configuration.
+ *
+ * @return the format token
+ */
+ String format();
+
+ /**
+ * Lists the catalog names discovered from the format's REST server.
+ *
+ * @param uri the configured REST server URI
+ * @param catalogProperties global catalog properties configured for the
format
+ * @return discovered catalog names
+ */
+ List<String> listCatalogs(String uri, Map<String, String> catalogProperties);
+
+ /**
+ * Returns the Spark catalog implementation class name.
+ *
+ * @return the Spark catalog implementation class name
+ */
+ String catalogClassName();
+
+ /**
+ * Returns provider-generated Spark catalog property suffixes and values.
+ *
+ * @param uri the configured REST server URI
+ * @param discoveredCatalogName the catalog name discovered from the REST
server
+ * @return generated property suffixes and values
+ */
+ Map<String, String> generatedCatalogProperties(String uri, String
discoveredCatalogName);
+
+ /**
+ * Returns Spark session extension class names required by the provider.
+ *
+ * @return Spark session extension class names
+ */
+ String[] sparkExtensions();
+}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/lance/LanceRESTCatalogProvider.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/lance/LanceRESTCatalogProvider.java
new file mode 100644
index 0000000000..83f1993270
--- /dev/null
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/restcatalog/lance/LanceRESTCatalogProvider.java
@@ -0,0 +1,147 @@
+/*
+ * 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.spark.connector.plugin.restcatalog.lance;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.commons.lang3.StringUtils;
+import
org.apache.gravitino.spark.connector.plugin.restcatalog.LakehouseRESTCatalogProvider;
+import org.lance.namespace.client.apache.ApiClient;
+import org.lance.namespace.client.apache.ApiException;
+import org.lance.namespace.client.apache.api.NamespaceApi;
+import org.lance.namespace.model.ListNamespacesResponse;
+
+/** Discovers and configures Lance REST catalogs. */
+public class LanceRESTCatalogProvider implements LakehouseRESTCatalogProvider {
+
+ static final String FORMAT = "lance";
+ static final String CATALOG_CLASS =
"org.lance.spark.LanceNamespaceSparkCatalog";
+ static final String SPARK_EXTENSIONS =
"org.lance.spark.extensions.LanceSparkSessionExtensions";
+
+ private static final String ROOT_NAMESPACE_ID = "$";
+ private static final String NAMESPACE_DELIMITER = "$";
+ static final String NAMESPACE_DELIMITER_PROPERTY = "namespaceDelimiter";
+
+ @Override
+ public String format() {
+ return FORMAT;
+ }
+
+ @Override
+ public List<String> listCatalogs(String uri, Map<String, String>
catalogProperties) {
+ List<String> catalogs = new ArrayList<>();
+ Set<String> seenPageTokens = new HashSet<>();
+ String namespaceDelimiter =
+ catalogProperties.getOrDefault(NAMESPACE_DELIMITER_PROPERTY,
NAMESPACE_DELIMITER);
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(namespaceDelimiter), "Lance namespace delimiter
must not be blank");
+ String pageToken = null;
+
+ ApiClient apiClient = new ApiClient().setBasePath(normalizeUri(uri));
+ try (Closeable httpClient = getHttpClient(apiClient)) {
+ NamespaceApi namespaceApi = new NamespaceApi(apiClient);
+ do {
+ ListNamespacesResponse response =
+ namespaceApi.listNamespaces(ROOT_NAMESPACE_ID, namespaceDelimiter,
pageToken, null);
+ Preconditions.checkState(response != null, "Lance REST server returned
an empty response");
+ Preconditions.checkState(
+ response.getNamespaces() != null,
+ "Lance REST server returned a response without namespaces");
+ catalogs.addAll(response.getNamespaces());
+
+ pageToken = StringUtils.trimToNull(response.getPageToken());
+ Preconditions.checkState(
+ pageToken == null || seenPageTokens.add(pageToken),
+ "Lance REST server returned repeated page token: %s",
+ pageToken);
+ } while (pageToken != null);
+ } catch (ApiException e) {
+ throw new IllegalStateException(
+ String.format(
+ "Failed to list catalogs from Lance REST server %s: HTTP %d, %s",
+ uri, e.getCode(), describeResponse(e)),
+ e);
+ } catch (IOException e) {
+ throw new IllegalStateException("Failed to list catalogs from Lance REST
server " + uri, e);
+ }
+
+ return catalogs;
+ }
+
+ @Override
+ public String catalogClassName() {
+ return CATALOG_CLASS;
+ }
+
+ @Override
+ public Map<String, String> generatedCatalogProperties(String uri, String
discoveredCatalogName) {
+ return ImmutableMap.of("impl", "rest", "uri", uri, "parent",
discoveredCatalogName);
+ }
+
+ @Override
+ public String[] sparkExtensions() {
+ return new String[] {SPARK_EXTENSIONS};
+ }
+
+ /**
+ * The Lance client uses the raw response body as the exception message,
which is empty for most
+ * error responses, so fall back to the reason phrase and finally to an
explicit placeholder
+ * instead of reporting a bare {@code ApiException}.
+ */
+ private static String describeResponse(ApiException e) {
+ if (StringUtils.isNotBlank(e.getResponseBody())) {
+ return e.getResponseBody();
+ }
+ if (StringUtils.isNotBlank(e.getMessage())) {
+ return e.getMessage();
+ }
+ return "no response body";
+ }
+
+ private static String normalizeUri(String uri) {
+ String normalized = uri;
+ while (normalized.endsWith("/") && !normalized.endsWith("://")) {
+ normalized = normalized.substring(0, normalized.length() - 1);
+ }
+ return normalized;
+ }
+
+ private static Closeable getHttpClient(ApiClient apiClient) {
+ try {
+ // Use reflection to keep Apache HttpClient types out of this class's
bytecode. Those types
+ // are relocated in Spark runtime JARs but remain unshaded in external
Lance bundles, so a
+ // direct call can encode an incompatible method descriptor and cause a
NoSuchMethodError.
+ Object httpClient =
ApiClient.class.getMethod("getHttpClient").invoke(apiClient);
+ Preconditions.checkState(
+ httpClient instanceof Closeable,
+ "Lance Namespace ApiClient returned an HTTP client that is not
closeable");
+ return (Closeable) httpClient;
+ } catch (ReflectiveOperationException e) {
+ throw new IllegalStateException("Failed to access the Lance Namespace
HTTP client", e);
+ }
+ }
+}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/utils/ConnectorUtil.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/utils/ConnectorUtil.java
index 672369854a..f5df6cb1bb 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/utils/ConnectorUtil.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/utils/ConnectorUtil.java
@@ -32,10 +32,9 @@ public class ConnectorUtil {
public static String removeDuplicateSparkExtensions(
String[] extensions, String[] addedExtensions) {
- Set<String> uniqueElements = new
LinkedHashSet<>(Arrays.asList(extensions));
- if (addedExtensions != null && StringUtils.isNoneBlank(addedExtensions)) {
- uniqueElements.addAll(Arrays.asList(addedExtensions));
- }
+ Set<String> uniqueElements = new LinkedHashSet<>();
+ addSparkExtensions(uniqueElements, extensions);
+ addSparkExtensions(uniqueElements, addedExtensions);
return uniqueElements.stream()
.reduce((element1, element2) -> element1 + ConnectorConstants.COMMA +
element2)
.orElse("");
@@ -49,4 +48,15 @@ public class ConnectorUtil {
seq.foreach(javaList::add);
return javaList;
}
+
+ private static void addSparkExtensions(Set<String> uniqueElements, String[]
extensions) {
+ if (extensions == null) {
+ return;
+ }
+
+ Arrays.stream(extensions)
+ .filter(StringUtils::isNotBlank)
+ .map(String::trim)
+ .forEach(uniqueElements::add);
+ }
}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/plugin/TestGravitinoDriverPlugin.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/plugin/TestGravitinoDriverPlugin.java
index bc17c21a48..ca581ce1ee 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/plugin/TestGravitinoDriverPlugin.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/plugin/TestGravitinoDriverPlugin.java
@@ -22,6 +22,8 @@ package org.apache.gravitino.spark.connector.plugin;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import com.google.common.collect.ImmutableMap;
import java.io.IOException;
@@ -29,6 +31,7 @@ import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Arrays;
+import java.util.Collections;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.auth.AuthProperties;
import org.apache.gravitino.client.GravitinoClient;
@@ -382,6 +385,19 @@ public class TestGravitinoDriverPlugin {
client.close();
}
+ @Test
+ void testSkipsCatalogAlreadyConfiguredInSpark() {
+ SparkConf sparkConf =
+ new SparkConf(false).set("spark.sql.catalog.existing",
"example.UserCatalog");
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.provider()).thenReturn("hive");
+
+ new GravitinoDriverPlugin(withoutPaimon())
+ .registerGravitinoCatalogs(sparkConf,
Collections.singletonMap("existing", catalog));
+
+ assertEquals("example.UserCatalog",
sparkConf.get("spark.sql.catalog.existing"));
+ }
+
private static SparkConf tokenAuthConf() {
SparkConf sparkConf = new SparkConf(false);
sparkConf.set(GravitinoSparkConfig.GRAVITINO_AUTH_TYPE,
AuthProperties.TOKEN_AUTH_TYPE);
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/plugin/restcatalog/TestGravitinoLakehouseRESTDiscoveryDriverPlugin.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/plugin/restcatalog/TestGravitinoLakehouseRESTDiscoveryDriverPlugin.java
new file mode 100644
index 0000000000..55daf0cef1
--- /dev/null
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/plugin/restcatalog/TestGravitinoLakehouseRESTDiscoveryDriverPlugin.java
@@ -0,0 +1,687 @@
+/*
+ * 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.spark.connector.plugin.restcatalog;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.gravitino.spark.connector.plugin.GravitinoSparkPlugin;
+import org.apache.spark.SparkConf;
+import org.apache.spark.sql.internal.StaticSQLConf;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class TestGravitinoLakehouseRESTDiscoveryDriverPlugin {
+
+ private static final String DISCOVERY_PLUGIN =
+ GravitinoLakehouseRESTDiscoveryPlugin.class.getName();
+ private static final String GRAVITINO_PLUGIN =
GravitinoSparkPlugin.class.getName();
+ private static final String URI_CONFIG = "spark.sql.gravitino.fakeREST.uri";
+ private static final String CATALOG_PREFIX = "spark.sql.catalog.";
+
+ @BeforeEach
+ void resetPolicies() {
+ TrackingPolicy.invocationCount = 0;
+ }
+
+ @Test
+ void testRequiresDiscoveryPluginBeforeGravitinoPlugin() {
+ SparkConf sparkConf =
+ baseConf().set("spark.plugins", GRAVITINO_PLUGIN + "," +
DISCOVERY_PLUGIN);
+
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
initialize(sparkConf));
+
+ assertTrue(exception.getMessage().contains("must be listed before"));
+ assertFalse(sparkConf.contains(CATALOG_PREFIX + "catalog_a"));
+ }
+
+ @Test
+ void testNoConfiguredUriDoesNotChangeSparkConf() {
+ SparkConf sparkConf =
+ new SparkConf(false).set("spark.plugins", DISCOVERY_PLUGIN + "," +
GRAVITINO_PLUGIN);
+
+ driver().initialize(sparkConf, providerClasses("fake",
FailingProvider.class));
+
+
assertFalse(sparkConf.contains(StaticSQLConf.SPARK_SESSION_EXTENSIONS().key()));
+ }
+
+ @Test
+ void testGeneratedConfigurationAndPrecedence() {
+ SparkConf sparkConf =
+ baseConf()
+ .set("spark.sql.gravitino.fakeREST.catalogProperties.impl",
"global-impl")
+ .set("spark.sql.gravitino.fakeREST.catalogProperties.uri",
"global-uri")
+ .set("spark.sql.gravitino.fakeREST.catalogProperties.extra",
"global-extra")
+ .set(CATALOG_PREFIX + "catalog_a.uri", "user-uri")
+ .set(
+ StaticSQLConf.SPARK_SESSION_EXTENSIONS().key(),
+ "example.UserExtension,java.lang.Runnable");
+
+ initialize(sparkConf);
+
+ assertEquals(String.class.getName(), sparkConf.get(CATALOG_PREFIX +
"catalog_a"));
+ assertEquals("rest", sparkConf.get(CATALOG_PREFIX + "catalog_a.impl"));
+ assertEquals("user-uri", sparkConf.get(CATALOG_PREFIX + "catalog_a.uri"));
+ assertEquals("catalog_a", sparkConf.get(CATALOG_PREFIX +
"catalog_a.parent"));
+ assertEquals("global-extra", sparkConf.get(CATALOG_PREFIX +
"catalog_a.extra"));
+ assertEquals(
+ "java.lang.Runnable,example.UserExtension",
+ sparkConf.get(StaticSQLConf.SPARK_SESSION_EXTENSIONS().key()));
+ }
+
+ @Test
+ void testUserOwnedCatalogDoesNotReachPolicy() {
+ SparkConf sparkConf =
+ baseConf()
+ .set(CATALOG_PREFIX + "catalog_a", "example.UserCatalog")
+ .set(
+
GravitinoLakehouseRESTDiscoveryDriverPlugin.REGISTRATION_POLICY_CONFIG,
+ TrackingPolicy.class.getName());
+
+ initialize(sparkConf);
+
+ assertEquals(0, TrackingPolicy.invocationCount);
+ assertEquals("example.UserCatalog", sparkConf.get(CATALOG_PREFIX +
"catalog_a"));
+ assertFalse(sparkConf.contains(CATALOG_PREFIX + "catalog_a.parent"));
+ }
+
+ @Test
+ void testPolicyFiltersAndRenamesCatalogs() {
+ SparkConf sparkConf =
+ baseConf()
+ .set(
+
GravitinoLakehouseRESTDiscoveryDriverPlugin.REGISTRATION_POLICY_CONFIG,
+ RenamePolicy.class.getName());
+
+ driver().initialize(sparkConf, providerClasses("fake",
MultipleCatalogProvider.class));
+
+ assertFalse(sparkConf.contains(CATALOG_PREFIX + "catalog_a"));
+ assertEquals(String.class.getName(), sparkConf.get(CATALOG_PREFIX +
"renamed_b"));
+ assertEquals("catalog_b", sparkConf.get(CATALOG_PREFIX +
"renamed_b.parent"));
+ }
+
+ @Test
+ void testDuplicatePolicyOutputDoesNotPartiallyModifySparkConf() {
+ SparkConf sparkConf =
+ baseConf()
+ .set(
+
GravitinoLakehouseRESTDiscoveryDriverPlugin.REGISTRATION_POLICY_CONFIG,
+ DuplicatePolicy.class.getName());
+
+ IllegalArgumentException exception =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ driver()
+ .initialize(sparkConf, providerClasses("fake",
MultipleCatalogProvider.class)));
+
+ assertTrue(exception.getMessage().contains("duplicate name"));
+ assertFalse(sparkConf.contains(CATALOG_PREFIX + "duplicate"));
+
assertFalse(sparkConf.contains(StaticSQLConf.SPARK_SESSION_EXTENSIONS().key()));
+ }
+
+ @Test
+ void testInvalidPolicyOutputFails() {
+ SparkConf sparkConf =
+ baseConf()
+ .set(
+
GravitinoLakehouseRESTDiscoveryDriverPlugin.REGISTRATION_POLICY_CONFIG,
+ InvalidNamePolicy.class.getName());
+
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
initialize(sparkConf));
+
+ assertTrue(exception.getMessage().contains("invalid Spark identifier"));
+ assertFalse(sparkConf.contains(CATALOG_PREFIX + "invalid-name"));
+ }
+
+ @Test
+ void testInvalidDiscoveredNameIsSkippedUnderDefaultPolicy() {
+ SparkConf sparkConf = baseConf();
+
+ driver().initialize(sparkConf, providerClasses("fake",
InvalidNameProvider.class));
+
+ assertFalse(sparkConf.contains(CATALOG_PREFIX + "my-catalog"));
+ assertEquals(String.class.getName(), sparkConf.get(CATALOG_PREFIX +
"catalog_a"));
+ assertEquals("catalog_a", sparkConf.get(CATALOG_PREFIX +
"catalog_a.parent"));
+ assertEquals(
+ Runnable.class.getName(),
sparkConf.get(StaticSQLConf.SPARK_SESSION_EXTENSIONS().key()));
+ }
+
+ @Test
+ void testBlankPolicyOutputFails() {
+ SparkConf sparkConf =
+ baseConf()
+ .set(
+
GravitinoLakehouseRESTDiscoveryDriverPlugin.REGISTRATION_POLICY_CONFIG,
+ BlankNamePolicy.class.getName());
+
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
initialize(sparkConf));
+
+ assertTrue(exception.getMessage().contains("returned a blank catalog
name"));
+ }
+
+ @Test
+ void testConfiguredFormatWithoutProviderFails() {
+ SparkConf sparkConf = baseConf();
+
+ IllegalArgumentException exception =
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> driver().initialize(sparkConf, Collections.emptyMap()));
+
+ assertTrue(exception.getMessage().contains("No lakehouse REST catalog
provider"));
+ }
+
+ @Test
+ void testBuiltinProviderClassNames() {
+ Map<String, String> providerClassNames =
BuiltinRESTCatalogProviders.providerClassNames();
+
+ assertEquals(1, providerClassNames.size());
+ assertEquals(
+
"org.apache.gravitino.spark.connector.plugin.restcatalog.lance.LanceRESTCatalogProvider",
+ providerClassNames.get("lance"));
+ }
+
+ @Test
+ void testMissingProviderClassFails() {
+ IllegalArgumentException exception =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ driver()
+ .initialize(
+ baseConf(),
+ Collections.singletonMap("fake",
"example.MissingRESTCatalogProvider")));
+
+ assertTrue(exception.getMessage().contains("Failed to instantiate"));
+ assertTrue(exception.getCause() instanceof ClassNotFoundException);
+ }
+
+ @Test
+ void testProviderClassMustImplementProviderInterface() {
+ IllegalArgumentException exception =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ driver()
+ .initialize(
+ baseConf(), Collections.singletonMap("fake",
String.class.getName())));
+
+ assertTrue(exception.getMessage().contains("does not implement"));
+
assertTrue(exception.getMessage().contains(LakehouseRESTCatalogProvider.class.getName()));
+ }
+
+ @Test
+ void testMisspelledUriKeysAreReported() {
+ SparkConf sparkConf =
+ new SparkConf(false)
+ .set("spark.plugins", DISCOVERY_PLUGIN)
+ .set("spark.sql.gravitino.lanceRest.uri", "rest-uri")
+ .set("spark.sql.gravitino.REST.uri", "rest-uri")
+ .set("spark.sql.gravitino.uri", "http://127.0.0.1:8090";)
+
.set("spark.sql.gravitino.lanceREST.catalogProperties.namespaceDelimiter", "#");
+
+ List<String> similarKeys =
+ GravitinoLakehouseRESTDiscoveryDriverPlugin.similarUriKeys(sparkConf);
+
+ assertEquals(2, similarKeys.size());
+ assertTrue(similarKeys.contains("spark.sql.gravitino.lanceRest.uri"));
+ assertTrue(similarKeys.contains("spark.sql.gravitino.REST.uri"));
+ }
+
+ @Test
+ void testMatchingUriKeyIsNotReportedAsMisspelled() {
+ assertTrue(
+ GravitinoLakehouseRESTDiscoveryDriverPlugin.similarUriKeys(
+ new SparkConf(false).set("spark.sql.gravitino.lanceREST.uri",
"rest-uri"))
+ .isEmpty());
+ }
+
+ @Test
+ void testMissingCatalogClassFails() {
+ SparkConf sparkConf = baseConf();
+
+ IllegalArgumentException exception =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ driver()
+ .initialize(
+ sparkConf, providerClasses("fake",
MissingCatalogClassProvider.class)));
+
+ assertTrue(exception.getMessage().contains("Required runtime class"));
+
assertTrue(exception.getMessage().contains(MissingCatalogClassProvider.CATALOG_CLASS));
+ assertTrue(exception.getCause() instanceof ClassNotFoundException);
+ assertFalse(sparkConf.contains(CATALOG_PREFIX + "catalog_a"));
+ }
+
+ @Test
+ void testMissingSparkExtensionClassFails() {
+ IllegalArgumentException exception =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ driver()
+ .initialize(
+ baseConf(), providerClasses("fake",
MissingExtensionProvider.class)));
+
+ assertTrue(exception.getMessage().contains("Required runtime class"));
+
assertTrue(exception.getMessage().contains(MissingExtensionProvider.EXTENSION_CLASS));
+ }
+
+ @Test
+ void testBlankCatalogClassFails() {
+ IllegalStateException exception =
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ driver()
+ .initialize(
+ baseConf(), providerClasses("fake",
BlankCatalogClassProvider.class)));
+
+ assertTrue(exception.getMessage().contains("returned a blank runtime
class"));
+ }
+
+ @Test
+ void testNullSparkExtensionsFails() {
+ IllegalStateException exception =
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ driver()
+ .initialize(baseConf(), providerClasses("fake",
NullExtensionsProvider.class)));
+
+ assertTrue(exception.getMessage().contains("returned null Spark
extensions"));
+ }
+
+ @Test
+ void testProviderListingFailureReportsFormatAndUri() {
+ IllegalStateException exception =
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ driver()
+ .initialize(baseConf(), providerClasses("fake",
FailingListProvider.class)));
+
+ assertTrue(exception.getMessage().contains("Failed to list fake REST
catalogs from rest-uri"));
+ assertTrue(exception.getCause().getMessage().contains("listing is
unavailable"));
+ }
+
+ @Test
+ void testBlankPolicyClassNameFails() {
+ SparkConf sparkConf =
+
baseConf().set(GravitinoLakehouseRESTDiscoveryDriverPlugin.REGISTRATION_POLICY_CONFIG,
" ");
+
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
initialize(sparkConf));
+
+ assertTrue(
+ exception
+ .getMessage()
+ .contains(
+
GravitinoLakehouseRESTDiscoveryDriverPlugin.REGISTRATION_POLICY_CONFIG
+ + " must not be blank"));
+ }
+
+ @Test
+ void testMissingPolicyClassFails() {
+ SparkConf sparkConf =
+ baseConf()
+ .set(
+
GravitinoLakehouseRESTDiscoveryDriverPlugin.REGISTRATION_POLICY_CONFIG,
+ "example.MissingCatalogRegistrationPolicy");
+
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
initialize(sparkConf));
+
+ assertTrue(exception.getMessage().contains("Failed to instantiate"));
+ assertTrue(
+ exception
+ .getMessage()
+
.contains(GravitinoLakehouseRESTDiscoveryDriverPlugin.REGISTRATION_POLICY_CONFIG));
+ assertTrue(exception.getCause() instanceof ClassNotFoundException);
+ }
+
+ @Test
+ void testPolicyClassMustImplementPolicyInterface() {
+ SparkConf sparkConf =
+ baseConf()
+ .set(
+
GravitinoLakehouseRESTDiscoveryDriverPlugin.REGISTRATION_POLICY_CONFIG,
+ String.class.getName());
+
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
initialize(sparkConf));
+
+ assertTrue(exception.getMessage().contains("does not implement"));
+
assertTrue(exception.getMessage().contains(CatalogRegistrationPolicy.class.getName()));
+ }
+
+ @Test
+ void testPolicyWithoutNoArgConstructorFails() {
+ SparkConf sparkConf =
+ baseConf()
+ .set(
+
GravitinoLakehouseRESTDiscoveryDriverPlugin.REGISTRATION_POLICY_CONFIG,
+ NoArgConstructorMissingPolicy.class.getName());
+
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
initialize(sparkConf));
+
+ assertTrue(exception.getMessage().contains("Failed to instantiate"));
+ assertTrue(exception.getCause() instanceof NoSuchMethodException);
+ }
+
+ @Test
+ void testPolicyReturningUserOwnedNameFails() {
+ SparkConf sparkConf =
+ baseConf()
+ .set(CATALOG_PREFIX + "renamed_b", "example.UserCatalog")
+ .set(
+
GravitinoLakehouseRESTDiscoveryDriverPlugin.REGISTRATION_POLICY_CONFIG,
+ RenamePolicy.class.getName());
+
+ IllegalArgumentException exception =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ driver()
+ .initialize(sparkConf, providerClasses("fake",
MultipleCatalogProvider.class)));
+
+ assertTrue(exception.getMessage().contains("which is configured by the
user"));
+ assertEquals("example.UserCatalog", sparkConf.get(CATALOG_PREFIX +
"renamed_b"));
+ }
+
+ private static SparkConf baseConf() {
+ return new SparkConf(false).set("spark.plugins",
DISCOVERY_PLUGIN).set(URI_CONFIG, "rest-uri");
+ }
+
+ private static GravitinoLakehouseRESTDiscoveryDriverPlugin driver() {
+ return new GravitinoLakehouseRESTDiscoveryDriverPlugin();
+ }
+
+ private static void initialize(SparkConf sparkConf) {
+ driver().initialize(sparkConf, providerClasses("fake",
FakeProvider.class));
+ }
+
+ private static Map<String, String> providerClasses(
+ String format, Class<? extends LakehouseRESTCatalogProvider>
providerClass) {
+ return Collections.singletonMap(format, providerClass.getName());
+ }
+
+ static class FakeProvider implements LakehouseRESTCatalogProvider {
+
+ /** Creates the fake provider. */
+ public FakeProvider() {}
+
+ @Override
+ public String format() {
+ return "fake";
+ }
+
+ @Override
+ public List<String> listCatalogs(String uri, Map<String, String>
catalogProperties) {
+ return Collections.singletonList("catalog_a");
+ }
+
+ @Override
+ public String catalogClassName() {
+ return String.class.getName();
+ }
+
+ @Override
+ public Map<String, String> generatedCatalogProperties(
+ String uri, String discoveredCatalogName) {
+ return ImmutableMap.of("impl", "rest", "uri", uri, "parent",
discoveredCatalogName);
+ }
+
+ @Override
+ public String[] sparkExtensions() {
+ return new String[] {Runnable.class.getName()};
+ }
+ }
+
+ static class MultipleCatalogProvider extends FakeProvider {
+
+ /** Creates the provider that advertises multiple catalogs. */
+ public MultipleCatalogProvider() {}
+
+ @Override
+ public List<String> listCatalogs(String uri, Map<String, String>
catalogProperties) {
+ return Arrays.asList("catalog_a", "catalog_b");
+ }
+ }
+
+ static class FailingProvider implements LakehouseRESTCatalogProvider {
+
+ static {
+ if (Boolean.parseBoolean("true")) {
+ throw new AssertionError("Provider must not be loaded without a
configured URI");
+ }
+ }
+
+ /** Creates the provider that must never be loaded. */
+ public FailingProvider() {}
+
+ @Override
+ public String format() {
+ throw new AssertionError("Provider must not be loaded without a
configured URI");
+ }
+
+ @Override
+ public List<String> listCatalogs(String uri, Map<String, String>
catalogProperties) {
+ throw new AssertionError("Provider must not be called without a
configured URI");
+ }
+
+ @Override
+ public String catalogClassName() {
+ throw new AssertionError("Provider must not be called without a
configured URI");
+ }
+
+ @Override
+ public Map<String, String> generatedCatalogProperties(
+ String uri, String discoveredCatalogName) {
+ throw new AssertionError("Provider must not be called without a
configured URI");
+ }
+
+ @Override
+ public String[] sparkExtensions() {
+ throw new AssertionError("Provider must not be called without a
configured URI");
+ }
+ }
+
+ /** Policy used to verify user-owned catalogs are filtered before policy
invocation. */
+ public static class TrackingPolicy implements CatalogRegistrationPolicy {
+ private static int invocationCount;
+
+ /** Creates the policy. */
+ public TrackingPolicy() {}
+
+ @Override
+ public boolean shouldRegister(String format, String catalogName) {
+ invocationCount++;
+ return true;
+ }
+ }
+
+ /** Policy used to filter one catalog and rename another. */
+ public static class RenamePolicy implements CatalogRegistrationPolicy {
+
+ /** Creates the policy. */
+ public RenamePolicy() {}
+
+ @Override
+ public boolean shouldRegister(String format, String catalogName) {
+ return "catalog_b".equals(catalogName);
+ }
+
+ @Override
+ public String registeredCatalogName(String format, String catalogName) {
+ return "renamed_b";
+ }
+ }
+
+ /** Policy used to produce a duplicate registration name. */
+ public static class DuplicatePolicy implements CatalogRegistrationPolicy {
+
+ /** Creates the policy. */
+ public DuplicatePolicy() {}
+
+ @Override
+ public boolean shouldRegister(String format, String catalogName) {
+ return true;
+ }
+
+ @Override
+ public String registeredCatalogName(String format, String catalogName) {
+ return "duplicate";
+ }
+ }
+
+ /** Policy used to produce an invalid Spark identifier. */
+ public static class InvalidNamePolicy implements CatalogRegistrationPolicy {
+
+ /** Creates the policy. */
+ public InvalidNamePolicy() {}
+
+ @Override
+ public boolean shouldRegister(String format, String catalogName) {
+ return true;
+ }
+
+ @Override
+ public String registeredCatalogName(String format, String catalogName) {
+ return "invalid-name";
+ }
+ }
+
+ /** Policy that cannot be instantiated reflectively. */
+ public static class NoArgConstructorMissingPolicy implements
CatalogRegistrationPolicy {
+
+ /**
+ * Creates the policy.
+ *
+ * @param unused an argument that prevents reflective instantiation.
+ */
+ public NoArgConstructorMissingPolicy(String unused) {}
+
+ @Override
+ public boolean shouldRegister(String format, String catalogName) {
+ return true;
+ }
+ }
+
+ static class MissingCatalogClassProvider extends FakeProvider {
+
+ static final String CATALOG_CLASS = "example.MissingNamespaceSparkCatalog";
+
+ /** Creates the provider that advertises an absent catalog class. */
+ public MissingCatalogClassProvider() {}
+
+ @Override
+ public String catalogClassName() {
+ return CATALOG_CLASS;
+ }
+ }
+
+ static class BlankCatalogClassProvider extends FakeProvider {
+
+ /** Creates the provider that advertises a blank catalog class. */
+ public BlankCatalogClassProvider() {}
+
+ @Override
+ public String catalogClassName() {
+ return " ";
+ }
+ }
+
+ static class MissingExtensionProvider extends FakeProvider {
+
+ static final String EXTENSION_CLASS =
"example.MissingSparkSessionExtensions";
+
+ /** Creates the provider that advertises an absent Spark extension. */
+ public MissingExtensionProvider() {}
+
+ @Override
+ public String[] sparkExtensions() {
+ return new String[] {EXTENSION_CLASS};
+ }
+ }
+
+ static class NullExtensionsProvider extends FakeProvider {
+
+ /** Creates the provider that returns null Spark extensions. */
+ public NullExtensionsProvider() {}
+
+ @Override
+ public String[] sparkExtensions() {
+ return null;
+ }
+ }
+
+ static class InvalidNameProvider extends FakeProvider {
+
+ /** Creates the provider that advertises a name Spark cannot reference
unquoted. */
+ public InvalidNameProvider() {}
+
+ @Override
+ public List<String> listCatalogs(String uri, Map<String, String>
catalogProperties) {
+ return Arrays.asList("my-catalog", "catalog_a");
+ }
+ }
+
+ /** Policy used to produce a blank registration name. */
+ public static class BlankNamePolicy implements CatalogRegistrationPolicy {
+
+ /** Creates the policy. */
+ public BlankNamePolicy() {}
+
+ @Override
+ public boolean shouldRegister(String format, String catalogName) {
+ return true;
+ }
+
+ @Override
+ public String registeredCatalogName(String format, String catalogName) {
+ return " ";
+ }
+ }
+
+ static class FailingListProvider extends FakeProvider {
+
+ /** Creates the provider whose catalog listing fails. */
+ public FailingListProvider() {}
+
+ @Override
+ public List<String> listCatalogs(String uri, Map<String, String>
catalogProperties) {
+ throw new IllegalStateException("listing is unavailable");
+ }
+ }
+}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/plugin/restcatalog/lance/TestLanceRESTCatalogProvider.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/plugin/restcatalog/lance/TestLanceRESTCatalogProvider.java
new file mode 100644
index 0000000000..f334a26f5f
--- /dev/null
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/plugin/restcatalog/lance/TestLanceRESTCatalogProvider.java
@@ -0,0 +1,196 @@
+/*
+ * 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.spark.connector.plugin.restcatalog.lance;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class TestLanceRESTCatalogProvider {
+
+ private HttpServer server;
+ private String serverUri;
+
+ @BeforeEach
+ void startServer() throws IOException {
+ server = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
+ serverUri = "http://localhost:"; + server.getAddress().getPort() +
"/lance/";
+ }
+
+ @AfterEach
+ void stopServer() {
+ server.stop(0);
+ }
+
+ @Test
+ void testListsAllCatalogPages() {
+ AtomicInteger requests = new AtomicInteger();
+ List<String> queries = new CopyOnWriteArrayList<>();
+ server.createContext(
+ "/lance/v1/namespace/$/list",
+ exchange -> {
+ int request = requests.getAndIncrement();
+ queries.add(String.valueOf(exchange.getRequestURI().getQuery()));
+ if (request == 0) {
+ respond(exchange, 200,
"{\"namespaces\":[\"catalog_b\"],\"page_token\":\"next\"}");
+ } else {
+ respond(exchange, 200, "{\"namespaces\":[\"catalog_a\"]}");
+ }
+ });
+ server.start();
+
+ List<String> catalogs =
+ new LanceRESTCatalogProvider().listCatalogs(serverUri,
Collections.emptyMap());
+
+ assertEquals(2, requests.get());
+ assertEquals(Arrays.asList("catalog_b", "catalog_a"), catalogs);
+ assertEquals(2, queries.size());
+ assertFalse(queries.get(0).contains("page_token"));
+ assertTrue(queries.get(1).contains("page_token=next"));
+ }
+
+ @Test
+ void testUsesConfiguredNamespaceDelimiter() {
+ List<String> rawQueries = new CopyOnWriteArrayList<>();
+ server.createContext(
+ "/lance/v1/namespace/$/list",
+ exchange -> {
+
rawQueries.add(String.valueOf(exchange.getRequestURI().getRawQuery()));
+ respond(exchange, 200, "{\"namespaces\":[\"catalog_a\"]}");
+ });
+ server.start();
+
+ List<String> catalogs =
+ new LanceRESTCatalogProvider()
+ .listCatalogs(
+ serverUri,
+ Collections.singletonMap(
+ LanceRESTCatalogProvider.NAMESPACE_DELIMITER_PROPERTY,
"#"));
+
+ assertEquals(Collections.singletonList("catalog_a"), catalogs);
+ assertEquals(1, rawQueries.size());
+ assertTrue(rawQueries.get(0).contains("delimiter=%23"));
+ }
+
+ @Test
+ void testErrorResponseReportsStatusAndBody() {
+ server.createContext(
+ "/lance/v1/namespace/$/list",
+ exchange -> respond(exchange, 404, "{\"error\":\"namespace not
found\"}"));
+ server.start();
+
+ IllegalStateException exception =
+ assertThrows(
+ IllegalStateException.class,
+ () -> new LanceRESTCatalogProvider().listCatalogs(serverUri,
Collections.emptyMap()));
+
+ assertTrue(exception.getMessage().contains("HTTP 404"));
+ assertTrue(exception.getMessage().contains("namespace not found"));
+ }
+
+ @Test
+ void testEmptyErrorResponseStillReportsStatus() {
+ server.createContext("/lance/v1/namespace/$/list", exchange ->
respond(exchange, 401, ""));
+ server.start();
+
+ IllegalStateException exception =
+ assertThrows(
+ IllegalStateException.class,
+ () -> new LanceRESTCatalogProvider().listCatalogs(serverUri,
Collections.emptyMap()));
+
+ assertTrue(exception.getMessage().contains("HTTP 401"));
+ assertTrue(exception.getMessage().contains(serverUri));
+ }
+
+ @Test
+ void testRejectsBlankNamespaceDelimiter() {
+ IllegalArgumentException exception =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new LanceRESTCatalogProvider()
+ .listCatalogs(
+ serverUri,
+ Collections.singletonMap(
+
LanceRESTCatalogProvider.NAMESPACE_DELIMITER_PROPERTY, " ")));
+
+ assertTrue(exception.getMessage().contains("delimiter must not be blank"));
+ }
+
+ @Test
+ void testRejectsRepeatedPageToken() {
+ server.createContext(
+ "/lance/v1/namespace/$/list",
+ exchange ->
+ respond(exchange, 200,
"{\"namespaces\":[\"catalog_a\"],\"page_token\":\"repeat\"}"));
+ server.start();
+
+ IllegalStateException exception =
+ assertThrows(
+ IllegalStateException.class,
+ () -> new LanceRESTCatalogProvider().listCatalogs(serverUri,
Collections.emptyMap()));
+
+ assertTrue(exception.getMessage().contains("repeated page token"));
+ }
+
+ @Test
+ void testGeneratedCatalogConfiguration() {
+ LanceRESTCatalogProvider provider = new LanceRESTCatalogProvider();
+
+ Map<String, String> properties =
+ provider.generatedCatalogProperties("http://localhost:9101/lance";,
"catalog_a");
+
+ assertEquals("lance", provider.format());
+ assertEquals("org.lance.spark.LanceNamespaceSparkCatalog",
provider.catalogClassName());
+ assertEquals("rest", properties.get("impl"));
+ assertEquals("http://localhost:9101/lance";, properties.get("uri"));
+ assertEquals("catalog_a", properties.get("parent"));
+ assertEquals(3, properties.size());
+ assertEquals(
+ "org.lance.spark.extensions.LanceSparkSessionExtensions",
provider.sparkExtensions()[0]);
+ }
+
+ private static void respond(HttpExchange exchange, int statusCode, String
body)
+ throws IOException {
+ byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().set("Content-Type", "application/json");
+ exchange.sendResponseHeaders(statusCode, bytes.length);
+ try (OutputStream output = exchange.getResponseBody()) {
+ output.write(bytes);
+ }
+ }
+}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/utils/TestConnectorUtil.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/utils/TestConnectorUtil.java
index 3225034086..ab482fd113 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/utils/TestConnectorUtil.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/utils/TestConnectorUtil.java
@@ -47,5 +47,10 @@ public class TestConnectorUtil {
extensions = new String[] {"a", "a", "b", "c"};
result = ConnectorUtil.removeDuplicateSparkExtensions(extensions, null);
Assertions.assertEquals(result, "a,b,c");
+
+ extensions = new String[] {"a", " b ", ""};
+ addedExtensions = " a, b, ,c ";
+ result = ConnectorUtil.removeDuplicateSparkExtensions(extensions,
addedExtensions.split(COMMA));
+ Assertions.assertEquals(result, "a,b,c");
}
}
diff --git a/spark-connector/v3.5/spark/build.gradle.kts
b/spark-connector/v3.5/spark/build.gradle.kts
index 04075bc391..704a7a3177 100644
--- a/spark-connector/v3.5/spark/build.gradle.kts
+++ b/spark-connector/v3.5/spark/build.gradle.kts
@@ -113,6 +113,7 @@ dependencies {
compileOnly("org.scala-lang.modules:scala-java8-compat_$scalaVersion:$scalaJava8CompatVersion")
compileOnly(project(":clients:client-java-runtime", configuration =
"shadow"))
compileOnly("org.apache.iceberg:iceberg-spark-runtime-${sparkMajorVersion}_$scalaVersion:$icebergVersion")
+ compileOnly(libs.lance.namespace.apache.client)
compileOnly(libs.aws.glue)
if (scalaVersion == "2.12") {
compileOnly("org.apache.paimon:paimon-spark-$sparkMajorVersion:$paimonVersion")
{
@@ -212,6 +213,7 @@ dependencies {
}
testImplementation(libs.junit.jupiter.api)
testImplementation(libs.junit.jupiter.params)
+ testImplementation(libs.lance.namespace.apache.client)
testImplementation(libs.mysql.driver)
testImplementation(libs.postgresql.driver)
testImplementation(libs.testcontainers)
diff --git a/spark-connector/v4.0/spark/build.gradle.kts
b/spark-connector/v4.0/spark/build.gradle.kts
index 47ec80bbf6..8a9364add3 100644
--- a/spark-connector/v4.0/spark/build.gradle.kts
+++ b/spark-connector/v4.0/spark/build.gradle.kts
@@ -146,6 +146,7 @@ dependencies {
compileOnly("org.apache.spark:spark-sql_$scalaVersion:$sparkVersion")
compileOnly(project(":clients:client-java-runtime", configuration =
"shadow"))
compileOnly("org.apache.iceberg:iceberg-spark-runtime-${sparkMajorVersion}_$scalaVersion:$icebergVersion")
+ compileOnly(libs.lance.namespace.apache.client)
compileOnly(libs.aws.glue)
annotationProcessor(libs.lombok)
@@ -242,6 +243,7 @@ dependencies {
}
testImplementation(libs.junit.jupiter.api)
testImplementation(libs.junit.jupiter.params)
+ testImplementation(libs.lance.namespace.apache.client)
testImplementation(libs.mysql.driver)
testImplementation(libs.postgresql.driver)
testImplementation(libs.testcontainers)
