This is an automated email from the ASF dual-hosted git repository.
yuqi1129 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 4c45a86181 [#11089] feat(lance): enable auxiliary-mode metadata
authentication for Lance REST (#11113)
4c45a86181 is described below
commit 4c45a86181cfb6aca565a9f3bfa44d84bd94a7e1
Author: Shane <[email protected]>
AuthorDate: Wed Aug 5 20:23:39 2026 +0800
[#11089] feat(lance): enable auxiliary-mode metadata authentication for
Lance REST (#11113)
### What changes were proposed in this pull request?
This PR improves Lance REST when it works with Gravitino-backed
namespaces.
It adds:
- An aux-mode catalog fetcher that uses Gravitino's internal
CatalogDispatcher.
- A standalone-mode HTTP catalog fetcher that supports simple and OAuth2
GravitinoClient auth.
- Lance REST authentication context and HK2 method interception.
- Authentication expressions for Lance namespace and table REST
endpoints.
- Unit tests for fetcher selection, standalone auth config validation,
endpoint annotations, and Lance path id metadata extraction.
### Why are the changes needed?
When Lance REST runs as a Gravitino auxiliary service, calling Gravitino
through GravitinoClient causes duplicate authentication. Lance REST
should use internal Gravitino services in aux mode, the same direction
as Iceberg REST Catalog.
Fix: #11089
### Does this PR introduce _any_ user-facing change?
Yes.
For standalone Lance REST, the following optional properties are added:
- `gravitino.lance-rest.gravitino-auth-type`
- `gravitino.lance-rest.gravitino-simple.user-name`
- `gravitino.lance-rest.gravitino-oauth2.server-uri`
- `gravitino.lance-rest.gravitino-oauth2.credential`
- `gravitino.lance-rest.gravitino-oauth2.token-path`
- `gravitino.lance-rest.gravitino-oauth2.scope`
### How was this patch tested?
```bash
./gradlew :lance:lance-common:test :lance:lance-rest-server:test -PskipITs
./gradlew :lance:lance-common:spotlessApply
:lance:lance-rest-server:spotlessApply
git diff --check
---------
Co-authored-by: TRAE CLI <[email protected]>
---
conf/gravitino.conf.template | 5 +-
docs/lance-rest-server-chart.md | 2 +-
docs/lance-rest-service.md | 27 +-
.../lance/common/ops/NamespaceWrapper.java | 15 +
.../GravitinoLanceNameSpaceOperations.java | 42 +-
.../gravitino/GravitinoLanceNamespaceWrapper.java | 434 +++++++++++++++++++--
.../gravitino/GravitinoLanceTableOperations.java | 22 +-
.../TestGravitinoLanceNamespaceWrapper.java | 235 +++++++++++
.../apache/gravitino/lance/LanceRESTService.java | 17 +-
.../lance/service/LanceServiceIdentityFilter.java | 67 ++++
.../gravitino/TestGravitinoLanceModeParsing.java | 2 +-
.../TestGravitinoLanceTableOperations.java | 8 +-
.../service/TestLanceServiceIdentityFilter.java | 57 +++
13 files changed, 846 insertions(+), 87 deletions(-)
diff --git a/conf/gravitino.conf.template b/conf/gravitino.conf.template
index 4517c07137..4d936dda34 100644
--- a/conf/gravitino.conf.template
+++ b/conf/gravitino.conf.template
@@ -122,7 +122,8 @@ gravitino.lance-rest.httpPort = 9101
# THE CONFIGURATION FOR Lance namespace backend
# The backend Lance namespace for Lance REST service, it's recommended to use
Gravitino
gravitino.lance-rest.namespace-backend = gravitino
-# The uri of the Lance REST service gravitino namespace backend
-gravitino.lance-rest.gravitino-uri = http://localhost:8090
+# The uri of the Lance REST service gravitino namespace backend.
+# This is not required when Lance REST runs as an auxiliary service embedded
in Gravitino server.
+# gravitino.lance-rest.gravitino-uri = http://localhost:8090
# The metalake name used for Lance REST service gravitino namespace backend,
please create the metalake first before using it, and configure the metalake
name here.
# gravitino.lance-rest.gravitino-metalake = metalake
diff --git a/docs/lance-rest-server-chart.md b/docs/lance-rest-server-chart.md
index cfa4880417..617bb660bf 100644
--- a/docs/lance-rest-server-chart.md
+++ b/docs/lance-rest-server-chart.md
@@ -78,7 +78,7 @@ helm upgrade --install gravitino-lance
oci://registry-1.docker.io/apache/graviti
--create-namespace \
--set lanceRest.gravitinoUri=http://gravitino:8090 \
--set lanceRest.gravitinoMetalake=your-metalake \
- --set replicas=2 \
+ --set replicaCount=2 \
--set resources.requests.memory="4Gi" \
--set resources.requests.cpu="2"
```
diff --git a/docs/lance-rest-service.md b/docs/lance-rest-service.md
index bef31016cd..d997df9903 100644
--- a/docs/lance-rest-service.md
+++ b/docs/lance-rest-service.md
@@ -43,7 +43,7 @@ The Lance REST service acts as a bridge between Lance
datasets and applications:
│ Service │
└────────┬────────┘
│
- ▼ Gravitino Client API
+ ▼ Gravitino internal API or HTTP client
┌─────────────────┐
│ Gravitino Server │
│(Metadata Backend)│
@@ -130,8 +130,8 @@ To enable the Lance REST service within Gravitino server,
configure the followin
| `gravitino.lance-rest.classpath` | Classpath for Lance REST
service, relative to Gravitino home directory | lance-rest-server/libs |
Yes | 1.1.0 |
| `gravitino.lance-rest.httpPort` | Port number for Lance REST
service | 9101 |
No | 1.1.0 |
| `gravitino.lance-rest.host` | Hostname for Lance REST service
| 0.0.0.0 | No
| 1.1.0 |
-| `gravitino.lance-rest.namespace-backend` | Namespace metadata backend (only
`gravitino` is supported) | gravitino | Yes
| 1.1.0 |
-| `gravitino.lance-rest.gravitino-uri` | Gravitino server URI (required
when namespace-backend is `gravitino`) | http://localhost:8090 | Yes
| 1.1.0 |
+| `gravitino.lance-rest.namespace-backend` | Namespace metadata backend
(currently only `gravitino` is supported) | gravitino |
Yes | 1.1.0 |
+| `gravitino.lance-rest.gravitino-uri` | Gravitino server URI. Not
required in auxiliary mode. | http://localhost:8090 |
No | 1.1.0 |
| `gravitino.lance-rest.gravitino-metalake` | Gravitino metalake name
(required when namespace-backend is `gravitino`) | (none)
| Yes | 1.1.0 |
**Authentication to the Gravitino Server**
@@ -160,10 +160,12 @@ gravitino.auxService.names = lance-rest
gravitino.lance-rest.httpPort = 9101
gravitino.lance-rest.host = 0.0.0.0
gravitino.lance-rest.namespace-backend = gravitino
-gravitino.lance-rest.gravitino-uri = http://localhost:8090
gravitino.lance-rest.gravitino-metalake = my_metalake
```
+In auxiliary mode, `gravitino.lance-rest.gravitino-uri` is not required
because the Lance
+REST service uses Gravitino's internal API.
+
### Run Standalone
To run Lance REST service independently without Gravitino server (You need to
start Gravitino server first):
@@ -174,16 +176,17 @@ To run Lance REST service independently without Gravitino
server (You need to st
Configure the service by editing
`{GRAVITINO_HOME}/conf/gravitino-lance-rest-server.conf` or passing
command-line arguments:
-| Configuration Property | Description |
Default Value | Required | Since Version |
-|-------------------------------------------|-----------------------------|-----------------------|----------|---------------|
-| `gravitino.lance-rest.namespace-backend` | Namespace metadata backend |
gravitino | Yes | 1.1.0 |
-| `gravitino.lance-rest.gravitino-uri` | Gravitino server URI |
http://localhost:8090 | Yes | 1.1.0 |
-| `gravitino.lance-rest.gravitino-metalake` | Gravitino metalake name |
(none) | Yes | 1.1.0 |
-| `gravitino.lance-rest.httpPort` | Service port number |
9101 | No | 1.1.0 |
-| `gravitino.lance-rest.host` | Service hostname |
0.0.0.0 | No | 1.1.0 |
+| Configuration Property | Description |
Default Value | Required | Since Version |
+|-------------------------------------------|----------------------------|-----------------------|----------|---------------|
+| `gravitino.lance-rest.namespace-backend` | Namespace metadata backend |
gravitino | Yes | 1.1.0 |
+| `gravitino.lance-rest.gravitino-uri` | Gravitino server URI |
http://localhost:8090 | Yes | 1.1.0 |
+| `gravitino.lance-rest.gravitino-metalake` | Gravitino metalake name |
(none) | Yes | 1.1.0 |
+| `gravitino.lance-rest.httpPort` | Service port number |
9101 | No | 1.1.0 |
+| `gravitino.lance-rest.host` | Service hostname |
0.0.0.0 | No | 1.1.0 |
:::tip
-In most cases, you only need to configure
`gravitino.lance-rest.gravitino-metalake` and other properties can use their
default values.
+In standalone deployments, you only need to configure
`gravitino.lance-rest.gravitino-metalake`,
+`gravitino.lance-rest.gravitino-uri`, and other properties can use their
default values.
:::
diff --git
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/NamespaceWrapper.java
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/NamespaceWrapper.java
index 5973fa4f52..b466cb7455 100644
---
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/NamespaceWrapper.java
+++
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/NamespaceWrapper.java
@@ -24,13 +24,19 @@ public abstract class NamespaceWrapper {
public static final String NAMESPACE_DELIMITER_DEFAULT = "$";
private final LanceConfig config;
+ private final boolean auxMode;
private volatile boolean initialized = false;
private LanceNamespaceOperations namespaceOps;
private LanceTableOperations tableOps;
public NamespaceWrapper(LanceConfig config) {
+ this(config, false);
+ }
+
+ public NamespaceWrapper(LanceConfig config, boolean auxMode) {
this.config = config;
+ this.auxMode = auxMode;
}
protected abstract void initialize();
@@ -61,6 +67,15 @@ public abstract class NamespaceWrapper {
return config;
}
+ /**
+ * Whether Lance REST runs as an auxiliary service embedded in the Gravitino
server.
+ *
+ * @return {@code true} when running in auxiliary mode, {@code false} in
standalone mode.
+ */
+ public boolean isAuxMode() {
+ return auxMode;
+ }
+
private void initAll() {
initialize();
namespaceOps = newNamespaceOps();
diff --git
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceNameSpaceOperations.java
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceNameSpaceOperations.java
index a51fc15d49..a5fae9ecd4 100644
---
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceNameSpaceOperations.java
+++
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceNameSpaceOperations.java
@@ -40,11 +40,11 @@ import org.apache.gravitino.CatalogChange;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
import org.apache.gravitino.SchemaChange;
-import org.apache.gravitino.client.GravitinoClient;
import org.apache.gravitino.exceptions.CatalogInUseException;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
import org.apache.gravitino.exceptions.NonEmptyCatalogException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
import org.apache.gravitino.exceptions.NonEmptySchemaException;
import org.apache.gravitino.lance.common.ops.LanceNamespaceOperations;
import org.lance.namespace.errors.InvalidInputException;
@@ -60,7 +60,6 @@ import org.lance.namespace.model.ListTablesResponse;
public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperations {
private final GravitinoLanceNamespaceWrapper namespaceWrapper;
- private final GravitinoClient client;
// lance-namespace 0.4.5 switched mode/behavior fields to plain strings in
request models.
// Keep local enums as normalized internal states for type-safe branching.
@@ -82,7 +81,6 @@ public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperatio
public GravitinoLanceNameSpaceOperations(GravitinoLanceNamespaceWrapper
namespaceWrapper) {
this.namespaceWrapper = namespaceWrapper;
- this.client = namespaceWrapper.getClient();
}
@Override
@@ -96,7 +94,7 @@ public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperatio
switch (nsId.levels()) {
case 0:
namespaces =
- Arrays.stream(client.listCatalogsInfo())
+ Arrays.stream(namespaceWrapper.listCatalogsInfo())
.filter(namespaceWrapper::isLakehouseCatalog)
.map(Catalog::name)
.collect(Collectors.toList());
@@ -104,7 +102,7 @@ public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperatio
case 1:
Catalog catalog =
namespaceWrapper.loadAndValidateLakehouseCatalog(nsId.levelAtListPos(0));
- namespaces = Lists.newArrayList(catalog.asSchemas().listSchemas());
+ namespaces = Lists.newArrayList(namespaceWrapper.listSchemas(catalog));
break;
case 2:
@@ -142,7 +140,7 @@ public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperatio
break;
case 2:
String schemaName = nsId.levelAtListPos(1);
- Schema schema = catalog.asSchemas().loadSchema(schemaName);
+ Schema schema = namespaceWrapper.loadSchema(catalog, schemaName);
Optional.ofNullable(schema.properties()).ifPresent(properties::putAll);
break;
default:
@@ -218,7 +216,7 @@ public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperatio
Catalog catalog =
namespaceWrapper.loadAndValidateLakehouseCatalog(nsId.levelAtListPos(0));
if (nsId.levels() == 2) {
String schemaName = nsId.levelAtListPos(1);
- if (!catalog.asSchemas().schemaExists(schemaName)) {
+ if (!namespaceWrapper.schemaExists(catalog, schemaName)) {
throw new NamespaceNotFoundException(
"Schema not found: " + schemaName,
CommonUtil.formatCurrentStackTrace(), schemaName);
}
@@ -231,11 +229,11 @@ public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperatio
Catalog catalog;
try {
- catalog = client.loadCatalog(catalogName);
+ catalog = namespaceWrapper.loadCatalog(catalogName);
} catch (NoSuchCatalogException e) {
// Catalog does not exist, create it
Catalog createdCatalog =
- client.createCatalog(
+ namespaceWrapper.createCatalog(
catalogName,
Catalog.Type.RELATIONAL,
"lakehouse-generic",
@@ -273,7 +271,7 @@ public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperatio
CatalogChange::setProperty,
CatalogChange::removeProperty,
CatalogChange[]::new);
- Catalog alteredCatalog = client.alterCatalog(catalogName, changes);
+ Catalog alteredCatalog = namespaceWrapper.alterCatalog(catalogName,
changes);
Optional.ofNullable(alteredCatalog.properties()).ifPresent(response::setProperties);
return response;
default:
@@ -294,10 +292,11 @@ public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperatio
Schema schema;
try {
- schema = loadedCatalog.asSchemas().loadSchema(schemaName);
+ schema = namespaceWrapper.loadSchema(loadedCatalog, schemaName);
} catch (NoSuchSchemaException e) {
// Schema does not exist, create it
- Schema createdSchema =
loadedCatalog.asSchemas().createSchema(schemaName, null, properties);
+ Schema createdSchema =
+ namespaceWrapper.createSchema(loadedCatalog, schemaName, null,
properties);
response.setProperties(
createdSchema.properties() == null ? Maps.newHashMap() :
createdSchema.properties());
return response;
@@ -322,7 +321,7 @@ public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperatio
SchemaChange::setProperty,
SchemaChange::removeProperty,
SchemaChange[]::new);
- Schema alteredSchema =
loadedCatalog.asSchemas().alterSchema(schemaName, changes);
+ Schema alteredSchema = namespaceWrapper.alterSchema(loadedCatalog,
schemaName, changes);
Optional.ofNullable(alteredSchema.properties()).ifPresent(response::setProperties);
return response;
default:
@@ -333,7 +332,7 @@ public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperatio
private DropNamespaceResponse dropCatalog(
String catalogName, DropMode mode, DropBehavior behavior) {
try {
- boolean dropped = client.dropCatalog(catalogName, behavior ==
DropBehavior.CASCADE);
+ boolean dropped = namespaceWrapper.dropCatalog(catalogName, behavior ==
DropBehavior.CASCADE);
if (dropped) {
return new DropNamespaceResponse();
} else {
@@ -351,6 +350,11 @@ public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperatio
String.format("Catalog %s is not empty", catalogName),
CommonUtil.formatCurrentStackTrace(),
catalogName);
+ } catch (NonEmptyEntityException e) {
+ throw new InvalidInputException(
+ String.format("Catalog %s is not empty", catalogName),
+ CommonUtil.formatCurrentStackTrace(),
+ catalogName);
} catch (CatalogInUseException e) {
throw new InvalidInputException(
String.format("Catalog %s is in use", catalogName),
@@ -363,10 +367,10 @@ public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperatio
String catalogName, String schemaName, DropMode mode, DropBehavior
behavior) {
try {
boolean dropped =
- client
- .loadCatalog(catalogName)
- .asSchemas()
- .dropSchema(schemaName, behavior == DropBehavior.CASCADE);
+ namespaceWrapper.dropSchema(
+ namespaceWrapper.loadAndValidateLakehouseCatalog(catalogName),
+ schemaName,
+ behavior == DropBehavior.CASCADE);
if (dropped) {
return new DropNamespaceResponse();
} else {
@@ -442,7 +446,7 @@ public class GravitinoLanceNameSpaceOperations implements
LanceNamespaceOperatio
Catalog catalog =
namespaceWrapper.loadAndValidateLakehouseCatalog(catalogName);
String schemaName = nsId.levelAtListPos(1);
List<String> tables =
-
Arrays.stream(catalog.asTableCatalog().listTables(Namespace.of(schemaName)))
+
Arrays.stream(namespaceWrapper.asTableCatalog(catalog).listTables(Namespace.of(schemaName)))
.map(ident -> ident.name())
.sorted()
.collect(Collectors.toList());
diff --git
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceNamespaceWrapper.java
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceNamespaceWrapper.java
index 058dbc311f..001f993f7c 100644
---
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceNamespaceWrapper.java
+++
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceNamespaceWrapper.java
@@ -23,20 +23,48 @@ import static
org.apache.gravitino.lance.common.config.LanceConfig.NAMESPACE_BAC
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+import java.io.Closeable;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
+import java.util.Optional;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.Catalog;
+import org.apache.gravitino.CatalogChange;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.Schema;
+import org.apache.gravitino.SchemaChange;
import org.apache.gravitino.auth.AuthProperties;
+import org.apache.gravitino.catalog.CatalogDispatcher;
+import org.apache.gravitino.catalog.SchemaDispatcher;
+import org.apache.gravitino.catalog.TableDispatcher;
import org.apache.gravitino.client.DefaultOAuth2TokenProvider;
import org.apache.gravitino.client.GravitinoClient;
import org.apache.gravitino.client.GravitinoClient.ClientBuilder;
import org.apache.gravitino.config.ConfigEntry;
+import org.apache.gravitino.exceptions.CatalogAlreadyExistsException;
+import org.apache.gravitino.exceptions.CatalogInUseException;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
+import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NoSuchSchemaException;
+import org.apache.gravitino.exceptions.NoSuchTableException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
+import org.apache.gravitino.exceptions.TableAlreadyExistsException;
import org.apache.gravitino.lance.common.config.LanceConfig;
import org.apache.gravitino.lance.common.ops.LanceNamespaceOperations;
import org.apache.gravitino.lance.common.ops.LanceTableOperations;
import org.apache.gravitino.lance.common.ops.NamespaceWrapper;
+import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.rel.TableCatalog;
+import org.apache.gravitino.rel.TableChange;
+import org.apache.gravitino.rel.expressions.distributions.Distribution;
+import org.apache.gravitino.rel.expressions.sorts.SortOrder;
+import org.apache.gravitino.rel.expressions.transforms.Transform;
+import org.apache.gravitino.rel.indexes.Index;
+import org.apache.gravitino.utils.NameIdentifierUtil;
import org.lance.namespace.errors.NamespaceNotFoundException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,8 +72,9 @@ import org.slf4j.LoggerFactory;
public class GravitinoLanceNamespaceWrapper extends NamespaceWrapper {
private static final Logger LOG =
LoggerFactory.getLogger(GravitinoLanceNamespaceWrapper.class);
- private GravitinoClient client;
+ private String metalakeName;
+ private CatalogOperator catalogOperator;
private LanceNamespaceOperations namespaceOperations;
private LanceTableOperations tableOperations;
@@ -54,41 +83,18 @@ public class GravitinoLanceNamespaceWrapper extends
NamespaceWrapper {
super(null);
}
- public GravitinoLanceNamespaceWrapper(LanceConfig config) {
- super(config);
- }
-
- public GravitinoClient getClient() {
- return client;
+ public GravitinoLanceNamespaceWrapper(LanceConfig config, boolean auxMode) {
+ super(config, auxMode);
}
@Override
protected void initialize() {
- String uri = config().get(NAMESPACE_BACKEND_URI);
- String metalakeName = config().get(METALAKE_NAME);
+ metalakeName = config().get(METALAKE_NAME);
Preconditions.checkArgument(
StringUtils.isNotBlank(metalakeName),
"Metalake name must be provided for Lance Gravitino namespace
backend");
- // Extract client configuration properties (e.g., connection pool settings)
- Map<String, String> clientProperties = new HashMap<>();
- config()
- .getAllConfig()
- .forEach(
- (key, value) -> {
- if (key.startsWith("gravitino.client.")) {
- clientProperties.put(key, value);
- LOG.info("Applying client config: {} = {}", key, value);
- }
- });
-
- this.client = createGravitinoClient(uri, metalakeName, clientProperties,
config());
-
- LOG.info(
- "GravitinoClient initialized with auth type {} and {} client
properties for metalake: {}",
- config().getGravitinoAuthType(),
- clientProperties.size(),
- metalakeName);
+ this.catalogOperator = createCatalogOperator(metalakeName);
this.namespaceOperations = new GravitinoLanceNameSpaceOperations(this);
this.tableOperations = new GravitinoLanceTableOperations(this);
@@ -106,15 +112,42 @@ public class GravitinoLanceNamespaceWrapper extends
NamespaceWrapper {
@Override
public void close() {
- if (client != null) {
+ if (catalogOperator != null) {
try {
- client.close();
+ catalogOperator.close();
} catch (Exception e) {
- LOG.warn("Error closing Gravitino client", e);
+ LOG.warn("Error closing Lance catalog operator", e);
}
}
}
+ Catalog[] listCatalogsInfo() throws NoSuchMetalakeException {
+ return catalogOperator.listCatalogsInfo();
+ }
+
+ Catalog loadCatalog(String catalogName) throws NoSuchCatalogException {
+ return catalogOperator.loadCatalog(catalogName);
+ }
+
+ Catalog createCatalog(
+ String catalogName,
+ Catalog.Type type,
+ String provider,
+ String comment,
+ Map<String, String> properties)
+ throws NoSuchMetalakeException, CatalogAlreadyExistsException {
+ return catalogOperator.createCatalog(catalogName, type, provider, comment,
properties);
+ }
+
+ Catalog alterCatalog(String catalogName, CatalogChange... changes) throws
NoSuchCatalogException {
+ return catalogOperator.alterCatalog(catalogName, changes);
+ }
+
+ boolean dropCatalog(String catalogName, boolean force)
+ throws NonEmptyEntityException, CatalogInUseException {
+ return catalogOperator.dropCatalog(catalogName, force);
+ }
+
public boolean isLakehouseCatalog(Catalog catalog) {
return catalog.type().equals(Catalog.Type.RELATIONAL)
&& "lakehouse-generic".equals(catalog.provider());
@@ -123,7 +156,7 @@ public class GravitinoLanceNamespaceWrapper extends
NamespaceWrapper {
public Catalog loadAndValidateLakehouseCatalog(String catalogName) {
Catalog catalog;
try {
- catalog = client.loadCatalog(catalogName);
+ catalog = loadCatalog(catalogName);
} catch (NoSuchCatalogException e) {
throw new NamespaceNotFoundException(
"Catalog not found: " + catalogName,
CommonUtil.formatCurrentStackTrace(), catalogName);
@@ -137,6 +170,343 @@ public class GravitinoLanceNamespaceWrapper extends
NamespaceWrapper {
return catalog;
}
+ String[] listSchemas(Catalog catalog) throws NoSuchCatalogException {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return
Arrays.stream(schemaDispatcher.listSchemas(Namespace.of(metalakeName,
catalog.name())))
+ .map(NameIdentifier::name)
+ .toArray(String[]::new);
+ }
+
+ return catalog.asSchemas().listSchemas();
+ }
+
+ boolean schemaExists(Catalog catalog, String schemaName) {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return schemaDispatcher.schemaExists(schemaIdent(catalog.name(),
schemaName));
+ }
+
+ return catalog.asSchemas().schemaExists(schemaName);
+ }
+
+ Schema loadSchema(Catalog catalog, String schemaName) {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return schemaDispatcher.loadSchema(schemaIdent(catalog.name(),
schemaName));
+ }
+
+ return catalog.asSchemas().loadSchema(schemaName);
+ }
+
+ Schema createSchema(
+ Catalog catalog, String schemaName, String comment, Map<String, String>
properties) {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return schemaDispatcher.createSchema(
+ schemaIdent(catalog.name(), schemaName), comment, properties);
+ }
+
+ return catalog.asSchemas().createSchema(schemaName, comment, properties);
+ }
+
+ Schema alterSchema(Catalog catalog, String schemaName, SchemaChange...
changes) {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return schemaDispatcher.alterSchema(schemaIdent(catalog.name(),
schemaName), changes);
+ }
+
+ return catalog.asSchemas().alterSchema(schemaName, changes);
+ }
+
+ boolean dropSchema(Catalog catalog, String schemaName, boolean cascade) {
+ SchemaDispatcher schemaDispatcher = currentSchemaDispatcher();
+ if (schemaDispatcher != null) {
+ return schemaDispatcher.dropSchema(schemaIdent(catalog.name(),
schemaName), cascade);
+ }
+
+ return catalog.asSchemas().dropSchema(schemaName, cascade);
+ }
+
+ TableCatalog asTableCatalog(Catalog catalog) {
+ TableDispatcher tableDispatcher = currentTableDispatcher();
+ if (tableDispatcher != null) {
+ return new InternalTableCatalogAdapter(catalog.name(), tableDispatcher);
+ }
+
+ return catalog.asTableCatalog();
+ }
+
+ private NameIdentifier schemaIdent(String catalogName, String schemaName) {
+ return NameIdentifierUtil.ofSchema(metalakeName, catalogName, schemaName);
+ }
+
+ private NameIdentifier tableIdent(String catalogName, NameIdentifier ident) {
+ return NameIdentifierUtil.ofTable(
+ metalakeName, catalogName, ident.namespace().level(0), ident.name());
+ }
+
+ private Namespace tableNamespace(String catalogName, Namespace namespace) {
+ return Namespace.of(metalakeName, catalogName, namespace.level(0));
+ }
+
+ private SchemaDispatcher currentSchemaDispatcher() {
+ if (!isAuxMode()) {
+ return null;
+ }
+
+ return GravitinoEnv.getInstance().schemaDispatcher();
+ }
+
+ private TableDispatcher currentTableDispatcher() {
+ if (!isAuxMode()) {
+ return null;
+ }
+
+ return GravitinoEnv.getInstance().tableDispatcher();
+ }
+
+ @VisibleForTesting
+ CatalogOperator createCatalogOperator(String metalakeName) {
+ return isAuxMode()
+ ? new InternalCatalogOperator(metalakeName)
+ : new HttpCatalogOperator(
+ config().get(NAMESPACE_BACKEND_URI), metalakeName, config(),
extractClientProperties());
+ }
+
+ @VisibleForTesting
+ void setCatalogOperator(CatalogOperator catalogOperator) {
+ this.catalogOperator = catalogOperator;
+ }
+
+ private Map<String, String> extractClientProperties() {
+ Map<String, String> clientProperties = new HashMap<>();
+ config()
+ .getAllConfig()
+ .forEach(
+ (key, value) -> {
+ if (key.startsWith("gravitino.client.")) {
+ clientProperties.put(key, value);
+ LOG.debug("Applying Gravitino client config key: {}", key);
+ }
+ });
+ return clientProperties;
+ }
+
+ interface CatalogOperator extends Closeable {
+
+ Catalog[] listCatalogsInfo() throws NoSuchMetalakeException;
+
+ Catalog loadCatalog(String catalogName) throws NoSuchCatalogException;
+
+ Catalog createCatalog(
+ String catalogName,
+ Catalog.Type type,
+ String provider,
+ String comment,
+ Map<String, String> properties)
+ throws NoSuchMetalakeException, CatalogAlreadyExistsException;
+
+ Catalog alterCatalog(String catalogName, CatalogChange... changes)
+ throws NoSuchCatalogException;
+
+ boolean dropCatalog(String catalogName, boolean force)
+ throws NonEmptyEntityException, CatalogInUseException;
+
+ @Override
+ default void close() {}
+ }
+
+ private class InternalTableCatalogAdapter implements TableCatalog {
+ private final String catalogName;
+ private final TableDispatcher dispatcher;
+
+ private InternalTableCatalogAdapter(String catalogName, TableDispatcher
dispatcher) {
+ this.catalogName = catalogName;
+ this.dispatcher = dispatcher;
+ }
+
+ @Override
+ public NameIdentifier[] listTables(Namespace namespace) throws
NoSuchSchemaException {
+ return dispatcher.listTables(tableNamespace(catalogName, namespace));
+ }
+
+ @Override
+ public Table loadTable(NameIdentifier ident) throws NoSuchTableException {
+ return dispatcher.loadTable(tableIdent(catalogName, ident));
+ }
+
+ @Override
+ public Table createTable(
+ NameIdentifier ident,
+ Column[] columns,
+ String comment,
+ Map<String, String> properties,
+ Transform[] partitions,
+ Distribution distribution,
+ SortOrder[] sortOrders,
+ Index[] indexes)
+ throws NoSuchSchemaException, TableAlreadyExistsException {
+ return dispatcher.createTable(
+ tableIdent(catalogName, ident),
+ columns,
+ comment,
+ properties,
+ partitions,
+ distribution,
+ sortOrders,
+ indexes);
+ }
+
+ @Override
+ public Table alterTable(NameIdentifier ident, TableChange... changes)
+ throws NoSuchTableException, IllegalArgumentException {
+ return dispatcher.alterTable(tableIdent(catalogName, ident), changes);
+ }
+
+ @Override
+ public boolean dropTable(NameIdentifier ident) {
+ return dispatcher.dropTable(tableIdent(catalogName, ident));
+ }
+
+ @Override
+ public boolean purgeTable(NameIdentifier ident) throws
UnsupportedOperationException {
+ return dispatcher.purgeTable(tableIdent(catalogName, ident));
+ }
+
+ @Override
+ public boolean tableExists(NameIdentifier ident) {
+ return dispatcher.tableExists(tableIdent(catalogName, ident));
+ }
+ }
+
+ private static class InternalCatalogOperator implements CatalogOperator {
+ private final String metalakeName;
+ private final CatalogDispatcher catalogDispatcher;
+
+ private InternalCatalogOperator(String metalakeName) {
+ this.metalakeName = metalakeName;
+ CatalogDispatcher dispatcher =
GravitinoEnv.getInstance().catalogDispatcher();
+ Preconditions.checkState(
+ dispatcher != null,
+ "CatalogDispatcher is not available. Internal catalog operator
requires Gravitino server mode.");
+ this.catalogDispatcher = dispatcher;
+ }
+
+ @Override
+ public Catalog[] listCatalogsInfo() throws NoSuchMetalakeException {
+ return catalogDispatcher.listCatalogsInfo(Namespace.of(metalakeName));
+ }
+
+ @Override
+ public Catalog loadCatalog(String catalogName) throws
NoSuchCatalogException {
+ return
catalogDispatcher.loadCatalog(NameIdentifierUtil.ofCatalog(metalakeName,
catalogName));
+ }
+
+ @Override
+ public Catalog createCatalog(
+ String catalogName,
+ Catalog.Type type,
+ String provider,
+ String comment,
+ Map<String, String> properties)
+ throws NoSuchMetalakeException, CatalogAlreadyExistsException {
+ return catalogDispatcher.createCatalog(
+ NameIdentifierUtil.ofCatalog(metalakeName, catalogName),
+ type,
+ provider,
+ comment,
+ properties);
+ }
+
+ @Override
+ public Catalog alterCatalog(String catalogName, CatalogChange... changes)
+ throws NoSuchCatalogException {
+ return catalogDispatcher.alterCatalog(
+ NameIdentifierUtil.ofCatalog(metalakeName, catalogName), changes);
+ }
+
+ @Override
+ public boolean dropCatalog(String catalogName, boolean force)
+ throws NonEmptyEntityException, CatalogInUseException {
+ return catalogDispatcher.dropCatalog(
+ NameIdentifierUtil.ofCatalog(metalakeName, catalogName), force);
+ }
+ }
+
+ private static class HttpCatalogOperator implements CatalogOperator {
+ private final String uri;
+ private final String metalakeName;
+ private final LanceConfig config;
+ private final Map<String, String> clientProperties;
+ private volatile GravitinoClient client;
+
+ private HttpCatalogOperator(
+ String uri, String metalakeName, LanceConfig config, Map<String,
String> clientProperties) {
+ this.uri = uri;
+ this.metalakeName = metalakeName;
+ this.config = config;
+ this.clientProperties = clientProperties;
+ }
+
+ @Override
+ public Catalog[] listCatalogsInfo() throws NoSuchMetalakeException {
+ return getClient().listCatalogsInfo();
+ }
+
+ @Override
+ public Catalog loadCatalog(String catalogName) throws
NoSuchCatalogException {
+ return getClient().loadCatalog(catalogName);
+ }
+
+ @Override
+ public Catalog createCatalog(
+ String catalogName,
+ Catalog.Type type,
+ String provider,
+ String comment,
+ Map<String, String> properties)
+ throws NoSuchMetalakeException, CatalogAlreadyExistsException {
+ return getClient().createCatalog(catalogName, type, provider, comment,
properties);
+ }
+
+ @Override
+ public Catalog alterCatalog(String catalogName, CatalogChange... changes)
+ throws NoSuchCatalogException {
+ return getClient().alterCatalog(catalogName, changes);
+ }
+
+ @Override
+ public boolean dropCatalog(String catalogName, boolean force)
+ throws NonEmptyEntityException, CatalogInUseException {
+ return getClient().dropCatalog(catalogName, force);
+ }
+
+ @Override
+ public void close() {
+ Optional.ofNullable(client).ifPresent(GravitinoClient::close);
+ }
+
+ private GravitinoClient getClient() {
+ if (client != null) {
+ return client;
+ }
+
+ synchronized (this) {
+ if (client == null) {
+ client = createGravitinoClient(uri, metalakeName, clientProperties,
config);
+ LOG.info(
+ "GravitinoClient initialized with auth type {} and {} client
properties for metalake: {}",
+ config.getGravitinoAuthType(),
+ clientProperties.size(),
+ metalakeName);
+ }
+ }
+
+ return client;
+ }
+ }
+
static GravitinoClient createGravitinoClient(
String uri, String metalake, Map<String, String> clientProperties,
LanceConfig config) {
return newClientBuilder(uri, metalake, clientProperties, config).build();
diff --git
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceTableOperations.java
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceTableOperations.java
index 50ba09df30..6a1fcd6e11 100644
---
a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceTableOperations.java
+++
b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceTableOperations.java
@@ -129,7 +129,7 @@ public class GravitinoLanceTableOperations implements
LanceTableOperations {
Table table;
try {
- table = catalog.asTableCatalog().loadTable(tableIdentifier);
+ table =
namespaceWrapper.asTableCatalog(catalog).loadTable(tableIdentifier);
} catch (NoSuchTableException e) {
throw new TableNotFoundException(
"Table not found: " + tableId, CommonUtil.formatCurrentStackTrace(),
tableId);
@@ -194,8 +194,8 @@ public class GravitinoLanceTableOperations implements
LanceTableOperations {
// Single call - mode is handled server-side
Table t =
- catalog
- .asTableCatalog()
+ namespaceWrapper
+ .asTableCatalog(catalog)
.createTable(
tableIdentifier, columns.toArray(new Column[0]), null,
createTableProperties);
Map<String, String> properties = t.properties();
@@ -252,8 +252,8 @@ public class GravitinoLanceTableOperations implements
LanceTableOperations {
// Single call - mode is handled server-side
Table t =
- catalog
- .asTableCatalog()
+ namespaceWrapper
+ .asTableCatalog(catalog)
.createTable(tableIdentifier, new Column[] {}, null,
copiedTableProperties);
RegisterTableResponse response = new RegisterTableResponse();
@@ -275,7 +275,7 @@ public class GravitinoLanceTableOperations implements
LanceTableOperations {
NameIdentifier.of(nsId.levelAtListPos(1), nsId.levelAtListPos(2));
Table t;
try {
- t = catalog.asTableCatalog().loadTable(tableIdentifier);
+ t = namespaceWrapper.asTableCatalog(catalog).loadTable(tableIdentifier);
} catch (NoSuchTableException e) {
throw new TableNotFoundException(
"Table not found: " + tableId, CommonUtil.formatCurrentStackTrace(),
tableId);
@@ -296,7 +296,7 @@ public class GravitinoLanceTableOperations implements
LanceTableOperations {
}
// External tables: dropTable removes catalog metadata only, preserving
Lance data.
- boolean result = catalog.asTableCatalog().dropTable(tableIdentifier);
+ boolean result =
namespaceWrapper.asTableCatalog(catalog).dropTable(tableIdentifier);
if (!result) {
throw new TableNotFoundException(
"Table not found: " + tableId, CommonUtil.formatCurrentStackTrace(),
tableId);
@@ -321,7 +321,7 @@ public class GravitinoLanceTableOperations implements
LanceTableOperations {
NameIdentifier tableIdentifier =
NameIdentifier.of(nsId.levelAtListPos(1), nsId.levelAtListPos(2));
- return catalog.asTableCatalog().tableExists(tableIdentifier);
+ return
namespaceWrapper.asTableCatalog(catalog).tableExists(tableIdentifier);
}
@Override
@@ -338,13 +338,13 @@ public class GravitinoLanceTableOperations implements
LanceTableOperations {
Table table;
try {
- table = catalog.asTableCatalog().loadTable(tableIdentifier);
+ table =
namespaceWrapper.asTableCatalog(catalog).loadTable(tableIdentifier);
} catch (NoSuchTableException e) {
throw new TableNotFoundException(
"Table not found: " + tableId, CommonUtil.formatCurrentStackTrace(),
tableId);
}
- boolean deleted = catalog.asTableCatalog().purgeTable(tableIdentifier);
+ boolean deleted =
namespaceWrapper.asTableCatalog(catalog).purgeTable(tableIdentifier);
if (!deleted) {
throw new TableNotFoundException(
"Table not found: " + tableId, CommonUtil.formatCurrentStackTrace(),
tableId);
@@ -376,7 +376,7 @@ public class GravitinoLanceTableOperations implements
LanceTableOperations {
}
TableChange[] changes = handler.buildGravitinoTableChange(request);
- Table table = catalog.asTableCatalog().alterTable(tableIdentifier,
changes);
+ Table table =
namespaceWrapper.asTableCatalog(catalog).alterTable(tableIdentifier, changes);
return handler.handle(table, request);
}
diff --git
a/lance/lance-common/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceNamespaceWrapper.java
b/lance/lance-common/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceNamespaceWrapper.java
index 6eb3fe2c74..ee8fd98c93 100644
---
a/lance/lance-common/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceNamespaceWrapper.java
+++
b/lance/lance-common/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceNamespaceWrapper.java
@@ -19,14 +19,38 @@
package org.apache.gravitino.lance.common.ops.gravitino;
import com.google.common.collect.ImmutableMap;
+import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.CatalogChange;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.Schema;
+import org.apache.gravitino.catalog.CatalogDispatcher;
+import org.apache.gravitino.catalog.SchemaDispatcher;
+import org.apache.gravitino.catalog.TableDispatcher;
import org.apache.gravitino.lance.common.config.LanceConfig;
+import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.rel.TableCatalog;
+import org.apache.gravitino.utils.NameIdentifierUtil;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
public class TestGravitinoLanceNamespaceWrapper {
+ @AfterEach
+ public void tearDown() throws IllegalAccessException {
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "catalogDispatcher",
null, true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "catalogManager", null,
true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "schemaDispatcher",
null, true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "tableDispatcher", null,
true);
+ }
+
@Test
public void testClientPropertiesExtraction() {
// Test that client properties are correctly extracted from LanceConfig
@@ -229,4 +253,215 @@ public class TestGravitinoLanceNamespaceWrapper {
Assertions.assertEquals("10000",
allConfig.get("gravitino.client.connectionTimeoutMs"));
Assertions.assertEquals("60000",
allConfig.get("gravitino.client.socketTimeoutMs"));
}
+
+ @Test
+ public void testCreateCatalogOperatorUsesHttpClientInStandaloneMode() {
+ LanceConfig lanceConfig =
+ new LanceConfig(
+ ImmutableMap.of(
+ LanceConfig.METALAKE_NAME.getKey(), "test_metalake",
+ LanceConfig.NAMESPACE_BACKEND_URI.getKey(),
"http://localhost:8090";));
+ GravitinoLanceNamespaceWrapper wrapper = new
GravitinoLanceNamespaceWrapper(lanceConfig, false);
+
+ GravitinoLanceNamespaceWrapper.CatalogOperator operator =
+ wrapper.createCatalogOperator("test_metalake");
+
+ Assertions.assertEquals("HttpCatalogOperator",
operator.getClass().getSimpleName());
+ Assertions.assertDoesNotThrow(operator::close);
+ }
+
+ @Test
+ public void testLoadAndValidateLakehouseCatalogUsesCatalogOperator() {
+ GravitinoLanceNamespaceWrapper wrapper = new
GravitinoLanceNamespaceWrapper();
+ Catalog expectedCatalog = createCatalogProxy(Catalog.Type.RELATIONAL,
"lakehouse-generic");
+ wrapper.setCatalogOperator(
+ new GravitinoLanceNamespaceWrapper.CatalogOperator() {
+ @Override
+ public Catalog[] listCatalogsInfo() {
+ return new Catalog[0];
+ }
+
+ @Override
+ public Catalog loadCatalog(String catalogName) {
+ return expectedCatalog;
+ }
+
+ @Override
+ public Catalog createCatalog(
+ String catalogName,
+ Catalog.Type type,
+ String provider,
+ String comment,
+ Map<String, String> properties) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Catalog alterCatalog(String catalogName, CatalogChange...
changes) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean dropCatalog(String catalogName, boolean force) {
+ throw new UnsupportedOperationException();
+ }
+ });
+
+ Assertions.assertSame(expectedCatalog,
wrapper.loadAndValidateLakehouseCatalog("test_catalog"));
+ }
+
+ @Test
+ public void testLoadSchemaUsesSchemaDispatcherInAuxMode() throws Exception {
+ Schema expectedSchema = createSchemaProxy();
+ AtomicReference<NameIdentifier> loadedSchemaIdent = new
AtomicReference<>();
+ SchemaDispatcher schemaDispatcher =
+ (SchemaDispatcher)
+ Proxy.newProxyInstance(
+ SchemaDispatcher.class.getClassLoader(),
+ new Class<?>[] {SchemaDispatcher.class},
+ (proxy, method, args) -> {
+ if ("loadSchema".equals(method.getName())) {
+ loadedSchemaIdent.set((NameIdentifier) args[0]);
+ return expectedSchema;
+ }
+
+ Class<?> returnType = method.getReturnType();
+ if (returnType.equals(boolean.class)) {
+ return false;
+ }
+ if (returnType.equals(int.class)) {
+ return 0;
+ }
+ return null;
+ });
+ FieldUtils.writeField(
+ GravitinoEnv.getInstance(),
+ "catalogDispatcher",
+ Proxy.newProxyInstance(
+ CatalogDispatcher.class.getClassLoader(),
+ new Class<?>[] {CatalogDispatcher.class},
+ (proxy, method, args) -> null),
+ true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "schemaDispatcher",
schemaDispatcher, true);
+
+ LanceConfig lanceConfig =
+ new LanceConfig(ImmutableMap.of(LanceConfig.METALAKE_NAME.getKey(),
"test_metalake"));
+ GravitinoLanceNamespaceWrapper wrapper = new
GravitinoLanceNamespaceWrapper(lanceConfig, true);
+ wrapper.asNamespaceOps();
+
+ Schema actualSchema =
+ wrapper.loadSchema(
+ createCatalogProxy(Catalog.Type.RELATIONAL, "lakehouse-generic"),
"test_schema");
+
+ Assertions.assertSame(expectedSchema, actualSchema);
+ Assertions.assertEquals(
+ NameIdentifierUtil.ofSchema("test_metalake", "test_catalog",
"test_schema"),
+ loadedSchemaIdent.get());
+ }
+
+ @Test
+ public void testAsTableCatalogUsesTableDispatcherInAuxMode() throws
Exception {
+ Table expectedTable = createTableProxy();
+ AtomicReference<NameIdentifier> loadedTableIdent = new AtomicReference<>();
+ AtomicReference<Namespace> listedNamespace = new AtomicReference<>();
+ TableDispatcher tableDispatcher =
+ (TableDispatcher)
+ Proxy.newProxyInstance(
+ TableDispatcher.class.getClassLoader(),
+ new Class<?>[] {TableDispatcher.class},
+ (proxy, method, args) -> {
+ if ("loadTable".equals(method.getName())) {
+ loadedTableIdent.set((NameIdentifier) args[0]);
+ return expectedTable;
+ }
+ if ("listTables".equals(method.getName())) {
+ listedNamespace.set((Namespace) args[0]);
+ return new NameIdentifier[]
{NameIdentifier.of("test_schema", "test_table")};
+ }
+
+ Class<?> returnType = method.getReturnType();
+ if (returnType.equals(boolean.class)) {
+ return false;
+ }
+ if (returnType.equals(int.class)) {
+ return 0;
+ }
+ return null;
+ });
+ FieldUtils.writeField(
+ GravitinoEnv.getInstance(),
+ "catalogDispatcher",
+ Proxy.newProxyInstance(
+ CatalogDispatcher.class.getClassLoader(),
+ new Class<?>[] {CatalogDispatcher.class},
+ (proxy, method, args) -> null),
+ true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "tableDispatcher",
tableDispatcher, true);
+
+ LanceConfig lanceConfig =
+ new LanceConfig(ImmutableMap.of(LanceConfig.METALAKE_NAME.getKey(),
"test_metalake"));
+ GravitinoLanceNamespaceWrapper wrapper = new
GravitinoLanceNamespaceWrapper(lanceConfig, true);
+ wrapper.asTableOps();
+
+ TableCatalog tableCatalog =
+ wrapper.asTableCatalog(createCatalogProxy(Catalog.Type.RELATIONAL,
"lakehouse-generic"));
+
+ Assertions.assertSame(
+ expectedTable, tableCatalog.loadTable(NameIdentifier.of("test_schema",
"test_table")));
+ Assertions.assertArrayEquals(
+ new NameIdentifier[] {NameIdentifier.of("test_schema", "test_table")},
+ tableCatalog.listTables(Namespace.of("test_schema")));
+ Assertions.assertEquals(
+ NameIdentifierUtil.ofTable("test_metalake", "test_catalog",
"test_schema", "test_table"),
+ loadedTableIdent.get());
+ Assertions.assertEquals(
+ Namespace.of("test_metalake", "test_catalog", "test_schema"),
listedNamespace.get());
+ }
+
+ private Schema createSchemaProxy() {
+ return (Schema)
+ Proxy.newProxyInstance(
+ Schema.class.getClassLoader(),
+ new Class<?>[] {Schema.class},
+ (proxy, method, args) -> null);
+ }
+
+ private Table createTableProxy() {
+ return (Table)
+ Proxy.newProxyInstance(
+ Table.class.getClassLoader(),
+ new Class<?>[] {Table.class},
+ (proxy, method, args) -> null);
+ }
+
+ private Catalog createCatalogProxy(Catalog.Type type, String provider) {
+ return (Catalog)
+ Proxy.newProxyInstance(
+ Catalog.class.getClassLoader(),
+ new Class<?>[] {Catalog.class},
+ (proxy, method, args) -> {
+ switch (method.getName()) {
+ case "type":
+ return type;
+ case "provider":
+ return provider;
+ case "name":
+ return "test_catalog";
+ case "properties":
+ return null;
+ default:
+ Class<?> returnType = method.getReturnType();
+ if (returnType.equals(boolean.class)) {
+ return false;
+ }
+ if (returnType.equals(int.class)) {
+ return 0;
+ }
+ if (returnType.equals(long.class)) {
+ return 0L;
+ }
+ return null;
+ }
+ });
+ }
}
diff --git
a/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/LanceRESTService.java
b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/LanceRESTService.java
index 3375dd9060..00a6241384 100644
---
a/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/LanceRESTService.java
+++
b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/LanceRESTService.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.lance;
import static
org.apache.gravitino.lance.common.config.LanceConfig.NAMESPACE_BACKEND;
import java.lang.reflect.Constructor;
+import java.util.HashMap;
import java.util.Map;
import javax.servlet.Servlet;
import org.apache.gravitino.GravitinoEnv;
@@ -29,6 +30,7 @@ import org.apache.gravitino.lance.common.config.LanceConfig;
import org.apache.gravitino.lance.common.ops.LanceNamespaceBackend;
import org.apache.gravitino.lance.common.ops.NamespaceWrapper;
import org.apache.gravitino.lance.service.LanceHealthCheckPathMatcher;
+import org.apache.gravitino.lance.service.LanceServiceIdentityFilter;
import org.apache.gravitino.listener.EventBus;
import org.apache.gravitino.listener.api.event.EventSource;
import org.apache.gravitino.metrics.MetricsSystem;
@@ -65,7 +67,7 @@ public class LanceRESTService implements
GravitinoAuxiliaryService {
@Override
public void serviceInit(Map<String, String> properties, boolean auxMode) {
- LanceConfig lanceConfig = new LanceConfig(properties);
+ LanceConfig lanceConfig = new LanceConfig(new HashMap<>(properties));
JettyServerConfig serverConfig = JettyServerConfig.fromConfig(lanceConfig);
server = new LanceJettyServer();
@@ -74,7 +76,7 @@ public class LanceRESTService implements
GravitinoAuxiliaryService {
EventBus eventBus = GravitinoEnv.getInstance().eventBus();
server.initialize(serverConfig, SERVICE_NAME, false);
- this.lanceNamespace = loadNamespaceImpl(lanceConfig);
+ this.lanceNamespace = loadNamespaceImpl(lanceConfig, auxMode);
ResourceConfig resourceConfig = new ResourceConfig();
resourceConfig.register(JacksonFeature.class);
@@ -101,6 +103,11 @@ public class LanceRESTService implements
GravitinoAuxiliaryService {
LANCE_SPEC);
server.addCustomFilters(LANCE_SPEC);
server.addSystemFilters(LANCE_SPEC);
+ if (auxMode) {
+ server.addFilter(
+ new
LanceServiceIdentityFilter(lanceConfig.get(LanceConfig.GRAVITINO_SIMPLE_USERNAME)),
+ LANCE_SPEC);
+ }
// Root-level aliases for health checks to improve compatibility with
various monitoring
// systems that expect a /health endpoint.
@@ -138,15 +145,15 @@ public class LanceRESTService implements
GravitinoAuxiliaryService {
}
}
- private NamespaceWrapper loadNamespaceImpl(LanceConfig lanceConfig) {
+ private NamespaceWrapper loadNamespaceImpl(LanceConfig lanceConfig, boolean
auxMode) {
String backendType = lanceConfig.get(NAMESPACE_BACKEND);
LanceNamespaceBackend lanceNamespaceBackend =
LanceNamespaceBackend.fromType(backendType);
try {
Constructor<? extends NamespaceWrapper> constructor =
-
lanceNamespaceBackend.getWrapperClass().getConstructor(LanceConfig.class);
+
lanceNamespaceBackend.getWrapperClass().getConstructor(LanceConfig.class,
boolean.class);
- return constructor.newInstance(lanceConfig);
+ return constructor.newInstance(lanceConfig, auxMode);
} catch (Exception e) {
LOG.error("Error loading namespace implementation for backend type: {}",
backendType, e);
throw new RuntimeException("Failed to load namespace implementation", e);
diff --git
a/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/LanceServiceIdentityFilter.java
b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/LanceServiceIdentityFilter.java
new file mode 100644
index 0000000000..9f8430582d
--- /dev/null
+++
b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/LanceServiceIdentityFilter.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.lance.service;
+
+import java.io.IOException;
+import javax.servlet.Filter;
+import javax.servlet.FilterChain;
+import javax.servlet.FilterConfig;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import org.apache.gravitino.UserPrincipal;
+import org.apache.gravitino.utils.PrincipalUtils;
+
+/** Executes Lance REST requests as the configured service identity in
auxiliary mode. */
+public class LanceServiceIdentityFilter implements Filter {
+
+ private final UserPrincipal servicePrincipal;
+
+ /**
+ * Creates a service identity filter.
+ *
+ * @param userName the configured Lance REST service user name
+ */
+ public LanceServiceIdentityFilter(String userName) {
+ this.servicePrincipal = new UserPrincipal(userName);
+ }
+
+ @Override
+ public void init(FilterConfig filterConfig) {}
+
+ @Override
+ public void doFilter(ServletRequest request, ServletResponse response,
FilterChain chain)
+ throws IOException, ServletException {
+ try {
+ PrincipalUtils.doAs(
+ servicePrincipal,
+ () -> {
+ chain.doFilter(request, response);
+ return null;
+ });
+ } catch (IOException | ServletException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new ServletException("Failed to execute as the Lance REST service
identity", e);
+ }
+ }
+
+ @Override
+ public void destroy() {}
+}
diff --git
a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceModeParsing.java
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceModeParsing.java
index d27799c1f8..2c142e55b3 100644
---
a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceModeParsing.java
+++
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceModeParsing.java
@@ -228,7 +228,7 @@ class TestGravitinoLanceModeParsing {
Mockito.mock(GravitinoLanceNamespaceWrapper.class);
Catalog catalog = Mockito.mock(Catalog.class);
when(namespaceWrapper.loadAndValidateLakehouseCatalog("catalog")).thenReturn(catalog);
- when(catalog.asTableCatalog()).thenReturn(tableCatalog);
+ when(namespaceWrapper.asTableCatalog(catalog)).thenReturn(tableCatalog);
return new GravitinoLanceTableOperations(namespaceWrapper);
}
diff --git
a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestGravitinoLanceTableOperations.java
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceTableOperations.java
similarity index 94%
rename from
lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestGravitinoLanceTableOperations.java
rename to
lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceTableOperations.java
index 84618a0932..a291e80ce1 100644
---
a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestGravitinoLanceTableOperations.java
+++
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/common/ops/gravitino/TestGravitinoLanceTableOperations.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.gravitino.lance.service.rest;
+package org.apache.gravitino.lance.common.ops.gravitino;
import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_TABLE_VERSION;
@@ -26,10 +26,8 @@ import java.util.List;
import java.util.Map;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
-import
org.apache.gravitino.lance.common.ops.gravitino.GravitinoLanceNamespaceWrapper;
import
org.apache.gravitino.lance.common.ops.gravitino.GravitinoLanceTableAlterHandler.AlterColumnsGravitinoLance;
import
org.apache.gravitino.lance.common.ops.gravitino.GravitinoLanceTableAlterHandler.DropColumns;
-import
org.apache.gravitino.lance.common.ops.gravitino.GravitinoLanceTableOperations;
import org.apache.gravitino.rel.Table;
import org.apache.gravitino.rel.TableCatalog;
import org.apache.gravitino.rel.TableChange;
@@ -110,10 +108,12 @@ class TestGravitinoLanceTableOperations {
.thenReturn(managedTable);
Catalog catalog = Mockito.mock(Catalog.class);
- Mockito.when(catalog.asTableCatalog()).thenReturn(tableCatalog);
GravitinoLanceNamespaceWrapper wrapper =
Mockito.mock(GravitinoLanceNamespaceWrapper.class);
Mockito.when(wrapper.loadAndValidateLakehouseCatalog(Mockito.anyString())).thenReturn(catalog);
+ // In auxiliary mode the catalog is accessed through the namespace
wrapper's dispatcher rather
+ // than catalog.asTableCatalog() directly, so stub the wrapper routing
accordingly.
+ Mockito.when(wrapper.asTableCatalog(catalog)).thenReturn(tableCatalog);
GravitinoLanceTableOperations ops = new
GravitinoLanceTableOperations(wrapper);
diff --git
a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/TestLanceServiceIdentityFilter.java
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/TestLanceServiceIdentityFilter.java
new file mode 100644
index 0000000000..774056e91c
--- /dev/null
+++
b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/TestLanceServiceIdentityFilter.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.lance.service;
+
+import static org.mockito.Mockito.mock;
+
+import java.util.concurrent.atomic.AtomicReference;
+import javax.servlet.FilterChain;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import org.apache.gravitino.UserPrincipal;
+import org.apache.gravitino.auth.AuthConstants;
+import org.apache.gravitino.utils.PrincipalUtils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestLanceServiceIdentityFilter {
+
+ @Test
+ public void testBindsConfiguredServiceIdentity() throws Exception {
+ String userName = "lance_rest_service_user";
+ LanceServiceIdentityFilter filter = new
LanceServiceIdentityFilter(userName);
+ ServletRequest request = mock(ServletRequest.class);
+ ServletResponse response = mock(ServletResponse.class);
+ AtomicReference<String> userInChain = new AtomicReference<>();
+ FilterChain chain =
+ (servletRequest, servletResponse) ->
userInChain.set(PrincipalUtils.getCurrentUserName());
+
+ Assertions.assertEquals(AuthConstants.ANONYMOUS_USER,
PrincipalUtils.getCurrentUserName());
+
+ PrincipalUtils.doAs(
+ new UserPrincipal("request_user"),
+ () -> {
+ filter.doFilter(request, response, chain);
+ return null;
+ });
+
+ Assertions.assertEquals(userName, userInChain.get());
+ Assertions.assertEquals(AuthConstants.ANONYMOUS_USER,
PrincipalUtils.getCurrentUserName());
+ }
+}
