This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new c504421dbc [Cherry-pick to branch-1.3] [#13301] improvement
(trino-connector): Make gravitino.metalake optional (#13302) (#13319)
c504421dbc is described below
commit c504421dbc4aa9ca9f0e46a3db29bd80018a9616
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Sep 18 21:20:11 2026 +0800
[Cherry-pick to branch-1.3] [#13301] improvement (trino-connector): Make
gravitino.metalake optional (#13302) (#13319)
**Cherry-pick Information:**
- Original commit: c8e6d3b1f158a78f3a095b6864d7295baaec1aac
- Target branch: `branch-1.3`
- Status: ⚠️ **Has conflicts - manual resolution required**
---------
Co-authored-by: Yuhui <[email protected]>
Co-authored-by: Claude Opus 5 <[email protected]>
---
docs/trino-connector/catalog-iceberg.md | 7 +-
docs/trino-connector/configuration.md | 60 +++---
docs/trino-connector/installation.md | 2 +-
docs/trino-connector/supported-catalog.md | 38 +++-
.../testsets/jdbc-mysql/00008_alter_catalog.txt | 8 +-
.../gravitino/trino/connector/GravitinoConfig.java | 103 +++++++---
.../trino/connector/GravitinoConnectorFactory.java | 14 +-
.../connector/catalog/CatalogConnectorManager.java | 103 +++++++---
.../AlterCatalogStoredProcedure.java | 42 +++--
.../CreateCatalogStoredProcedure.java | 40 ++--
.../DropCatalogStoredProcedure.java | 28 +--
.../storedprocedure/GravitinoStoredProcedure.java | 29 +++
.../GravitinoStoredProcedureFactory.java | 16 +-
.../system/table/GravitinoSystemTableCatalog.java | 29 +--
.../table/GravitinoSystemTableCatalogStatus.java | 9 +-
.../system/table/GravitinoSystemTableFactory.java | 8 +-
.../table/GravitinoSystemTableLoadStatus.java | 19 +-
.../trino/connector/TestGravitinoConfig.java | 157 ++++++++++++---
.../TestGravitinoConnectorFactoryStart.java | 11 ++
.../catalog/TestCatalogConnectorManager.java | 210 ++++++++++++++++++++-
.../TestAlterCatalogStoredProcedure.java | 108 +++++++++++
.../TestCreateCatalogStoredProcedure.java | 108 +++++++++++
.../TestDropCatalogStoredProcedure.java | 64 ++++++-
.../TestGravitinoStoredProcedureFactory.java | 55 ++++++
.../table/TestGravitinoSystemStatusTables.java | 71 ++++++-
25 files changed, 1130 insertions(+), 209 deletions(-)
diff --git a/docs/trino-connector/catalog-iceberg.md
b/docs/trino-connector/catalog-iceberg.md
index ccfae7bf9e..e3721dc354 100644
--- a/docs/trino-connector/catalog-iceberg.md
+++ b/docs/trino-connector/catalog-iceberg.md
@@ -135,9 +135,10 @@ token cannot be exchanged, so it would carry no identity
to the IRC. Set
- One IRC serves exactly one metalake, fixed at startup by
`gravitino.iceberg-rest.gravitino-metalake`. The Gravitino server only
reports the IRC's endpoint
- for that metalake. In multi-metalake mode
(`gravitino.use-single-metalake=false`), a non-REST
- Iceberg catalog in another metalake therefore requires a metalake-scoped
manual URI or remains
- unregistered while REST routing is enabled.
+ for that metalake. When several metalakes are loaded
(`gravitino.catalog-name-with-metalake=true` or `gravitino.metalake` unset), a
non-REST
+ Iceberg catalog in another metalake therefore requires a manual URI
(`gravitino.iceberg.rest-uri`
+ as the default, overridden per metalake by
`gravitino.iceberg.rest-uri.<metalake_name>`) or
+ remains unregistered while REST routing is enabled.
- A catalog created with `catalog-backend=rest` keeps pointing at its own
configured `uri` and is
not re-routed, since it already reaches an Iceberg REST catalog directly.
- A deployment that does not run the IRC must set
diff --git a/docs/trino-connector/configuration.md
b/docs/trino-connector/configuration.md
index 449969e289..c682782e98 100644
--- a/docs/trino-connector/configuration.md
+++ b/docs/trino-connector/configuration.md
@@ -7,31 +7,32 @@ license: "This software is licensed under the Apache License
version 2."
## Introduction
-| Property | Type | Default Value
| Description
| Required |
-|---------------------------------------------|---------|-----------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------|
-| connector.name | string | (none)
| The `connector.name` defines the type of Trino connector, this value is
always 'gravitino'.
| Yes |
-| gravitino.metalake | string | (none)
| The `gravitino.metalake` defines which metalake in Gravitino server the
Trino connector uses. Trino connector should set it at start, the value of
`gravitino.metalake` needs to be a valid name, Trino connector can detect and
load the metalake with catalogs, schemas and tables once created and keep in
sync. | Yes |
-| gravitino.uri | string |
http://localhost:8090 | The `gravitino.uri` defines the connection URL of the
Gravitino server, the default value is `http://localhost:8090`. Trino connector
can detect and connect to Gravitino server once it is ready, no need to start
Gravitino server beforehand.
| No |
-| trino.jdbc.user | string | admin
| The jdbc user name of current Trino.
| NO |
-| trino.jdbc.password | string | (none)
| The jdbc password of current Trino.
| NO |
-| trino.jdbc.ssl.enabled | boolean | (derived)
| Whether the internal JDBC connection to the Trino coordinator uses TLS. If
not set, it is derived from the scheme of the Trino `discovery.uri`, so a
coordinator whose `discovery.uri` is `https://...` needs no explicit setting.
| No |
-| trino.jdbc.ssl.truststore.path | string | (none)
| Path of the truststore holding the Trino coordinator certificate. If
omitted, the default JVM truststore is used. Requires TLS, which is enabled
automatically for an HTTPS `discovery.uri` or explicitly with
`trino.jdbc.ssl.enabled=true`, and a `trino.jdbc.ssl.verification` other than
`NONE`. | No |
-| trino.jdbc.ssl.truststore.password | string | (none)
| Password of the truststore configured by `trino.jdbc.ssl.truststore.path`.
Requires TLS and `trino.jdbc.ssl.truststore.path`, otherwise the connector
fails to start.
| No |
-| trino.jdbc.ssl.truststore.type | string | (none)
| Type of the truststore, for example `JKS` or `PKCS12`. If omitted, the
default JVM truststore type is used. Requires TLS and
`trino.jdbc.ssl.truststore.path`, otherwise the connector fails to start.
| No |
-| trino.jdbc.ssl.keystore.path | string | (none)
| Path of the keystore holding the client certificate presented to the
coordinator, for coordinators that require mutual TLS. Requires TLS, which is
enabled automatically for an HTTPS `discovery.uri` or explicitly with
`trino.jdbc.ssl.enabled=true`, and a `trino.jdbc.ssl.verification` other than
`NONE`. See the note on mutual TLS below. | No |
-| trino.jdbc.ssl.keystore.password | string | (none)
| Password of the keystore configured by `trino.jdbc.ssl.keystore.path`.
Requires TLS and `trino.jdbc.ssl.keystore.path`, otherwise the connector fails
to start.
| No |
-| trino.jdbc.ssl.keystore.type | string | (none)
| Type of the keystore, for example `JKS` or `PKCS12`. If omitted, the default
JVM keystore type is used. Requires TLS and `trino.jdbc.ssl.keystore.path`,
otherwise the connector fails to start.
| No |
-| trino.jdbc.ssl.verification | string | FULL
| Certificate verification mode of the internal JDBC connection: `FULL`, `CA`
or `NONE`. Any value other than `FULL` requires TLS, which may be derived from
an HTTPS `discovery.uri`. `NONE` disables certificate verification entirely and
should only be used for troubleshooting.
| No |
-| trino.jdbc.roles | string | (none)
| Session roles applied to the internal JDBC connection, for example
`system:sysadmin`. Required by deployments that only allow `CREATE CATALOG`
with a privileged role.
| No |
-| trino.jdbc.properties. | string | (none)
| The configuration key prefix for raw Trino JDBC driver properties, see
[Connecting to a TLS-enabled
coordinator](#connecting-to-a-tls-enabled-coordinator).
|
No |
-| gravitino.metadata.refresh-interval-seconds | integer | 10
| The `gravitino.metadata.refresh-interval-seconds` defines the interval in
seconds to refresh metadata from Gravitino server, the default value is 10
seconds.
| No |
-| gravitino.trino.skip-version-validation | boolean | false
| The `gravitino.trino.skip-version-validation` defines whether to skip Trino
version validation. Gravitino supports Trino versions between 435 and 478. If
this option is `true`, unsupported Trino versions can still be used, but
compatibility is not guaranteed.
| No |
-| gravitino.client. | string | (none)
| The configuration key prefix for the Gravitino client config.
| No |
-| gravitino.trino.skip-catalog-patterns | string | (none)
| The `gravitino.trino.skip-catalog-patterns` defines a comma-separated list
of catalog name regex patterns that should be excluded from loading. For
example, `test_.*, .*_tmp` excludes all catalogs starting with `test_` or
ending with `_tmp`.
| No |
-| gravitino.use-single-metalake | boolean | true
| If `true`, only one metalake is used and catalogs are identified by
`<catalog_name>`. If `false`, multi-metalake mode is enabled and catalogs are
identified by `<metalake_name>.<catalog_name>`.
| No |
-| gravitino.iceberg.rest-routing-enabled | boolean | true
| Whether non-REST `lakehouse-iceberg` catalogs must be routed through the
Gravitino Iceberg REST server. When enabled, a catalog is not registered until
discovery succeeds or `gravitino.iceberg.rest-uri` is configured. Set this to
`false` to retain legacy `catalog-backend` translation and skip discovery. | No
|
-| gravitino.iceberg.rest-uri | string | (none)
| The endpoint of the Gravitino Iceberg REST server (IRC). It is discovered
automatically from the Gravitino server for this connector's metalake; set this
only to override the discovered value. When available, eligible
`lakehouse-iceberg` catalogs are loaded through IRC, enabling credential
vending. | No |
-| gravitino.iceberg.rest-catalog. | string | (none)
| Prefix for properties passed to the internal Trino Iceberg REST catalog. The
prefix is rewritten to `iceberg.rest-catalog.`. The `uri`, `warehouse`, and
`prefix` keys are reserved and derived by the connector.
| No |
+| Property | Type | Default Value
| Description
[...]
+|---------------------------------------------|---------|-----------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
+| connector.name | string | (none)
| The `connector.name` defines the type of Trino connector, this value is
always 'gravitino'.
[...]
+| gravitino.metalake | string | (none)
| The `gravitino.metalake` defines which metalake in Gravitino server the
Trino connector uses. The value needs to be a valid name, Trino connector can
detect and load the metalake with catalogs, schemas and tables once created and
keep in sync. When unset, the catalogs of every metalake are loaded, the system
tables report every metalake, and the `METALAKE` argument of the stored
procedures is required. [...]
+| gravitino.uri | string |
http://localhost:8090 | The `gravitino.uri` defines the connection URL of the
Gravitino server, the default value is `http://localhost:8090`. Trino connector
can detect and connect to Gravitino server once it is ready, no need to start
Gravitino server beforehand.
[...]
+| trino.jdbc.user | string | admin
| The jdbc user name of current Trino.
[...]
+| trino.jdbc.password | string | (none)
| The jdbc password of current Trino.
[...]
+| trino.jdbc.ssl.enabled | boolean | (derived)
| Whether the internal JDBC connection to the Trino coordinator uses TLS. If
not set, it is derived from the scheme of the Trino `discovery.uri`, so a
coordinator whose `discovery.uri` is `https://...` needs no explicit setting.
[...]
+| trino.jdbc.ssl.truststore.path | string | (none)
| Path of the truststore holding the Trino coordinator certificate. If
omitted, the default JVM truststore is used. Requires TLS, which is enabled
automatically for an HTTPS `discovery.uri` or explicitly with
`trino.jdbc.ssl.enabled=true`, and a `trino.jdbc.ssl.verification` other than
`NONE`.
[...]
+| trino.jdbc.ssl.truststore.password | string | (none)
| Password of the truststore configured by `trino.jdbc.ssl.truststore.path`.
Requires TLS and `trino.jdbc.ssl.truststore.path`, otherwise the connector
fails to start.
[...]
+| trino.jdbc.ssl.truststore.type | string | (none)
| Type of the truststore, for example `JKS` or `PKCS12`. If omitted, the
default JVM truststore type is used. Requires TLS and
`trino.jdbc.ssl.truststore.path`, otherwise the connector fails to start.
[...]
+| trino.jdbc.ssl.keystore.path | string | (none)
| Path of the keystore holding the client certificate presented to the
coordinator, for coordinators that require mutual TLS. Requires TLS, which is
enabled automatically for an HTTPS `discovery.uri` or explicitly with
`trino.jdbc.ssl.enabled=true`, and a `trino.jdbc.ssl.verification` other than
`NONE`. See the note on mutual TLS below.
[...]
+| trino.jdbc.ssl.keystore.password | string | (none)
| Password of the keystore configured by `trino.jdbc.ssl.keystore.path`.
Requires TLS and `trino.jdbc.ssl.keystore.path`, otherwise the connector fails
to start.
[...]
+| trino.jdbc.ssl.keystore.type | string | (none)
| Type of the keystore, for example `JKS` or `PKCS12`. If omitted, the default
JVM keystore type is used. Requires TLS and `trino.jdbc.ssl.keystore.path`,
otherwise the connector fails to start.
[...]
+| trino.jdbc.ssl.verification | string | FULL
| Certificate verification mode of the internal JDBC connection: `FULL`, `CA`
or `NONE`. Any value other than `FULL` requires TLS, which may be derived from
an HTTPS `discovery.uri`. `NONE` disables certificate verification entirely and
should only be used for troubleshooting.
[...]
+| trino.jdbc.roles | string | (none)
| Session roles applied to the internal JDBC connection, for example
`system:sysadmin`. Required by deployments that only allow `CREATE CATALOG`
with a privileged role.
[...]
+| trino.jdbc.properties. | string | (none)
| The configuration key prefix for raw Trino JDBC driver properties, see
[Connecting to a TLS-enabled
coordinator](#connecting-to-a-tls-enabled-coordinator).
[...]
+| gravitino.metadata.refresh-interval-seconds | integer | 10
| The `gravitino.metadata.refresh-interval-seconds` defines the interval in
seconds to refresh metadata from Gravitino server, the default value is 10
seconds.
[...]
+| gravitino.trino.skip-version-validation | boolean | false
| The `gravitino.trino.skip-version-validation` defines whether to skip Trino
version validation. Gravitino supports Trino versions between 435 and 478. If
this option is `true`, unsupported Trino versions can still be used, but
compatibility is not guaranteed.
[...]
+| gravitino.client. | string | (none)
| The configuration key prefix for the Gravitino client config.
[...]
+| gravitino.trino.skip-catalog-patterns | string | (none)
| The `gravitino.trino.skip-catalog-patterns` defines a comma-separated list
of catalog name regex patterns that should be excluded from loading. For
example, `test_.*, .*_tmp` excludes all catalogs starting with `test_` or
ending with `_tmp`.
[...]
+| gravitino.catalog-name-with-metalake | boolean | false
| If `true`, multi-metalake mode is enabled: every metalake is loaded and
catalogs are identified by `<metalake_name>.<catalog_name>`. If `false`,
catalogs are identified by `<catalog_name>`; when several metalakes are loaded
and two of them have a catalog with the same name, only one of them is
registered, in the order the server lists the metalakes, and the other is
reported as `FAILED` in `gravitino.system [...]
+| gravitino.use-single-metalake | boolean | true
| **Deprecated**, use `gravitino.catalog-name-with-metalake` instead. `false`
is equivalent to `gravitino.catalog-name-with-metalake=true`. Ignored when
`gravitino.catalog-name-with-metalake` is set.
[...]
+| gravitino.iceberg.rest-routing-enabled | boolean | true
| Whether non-REST `lakehouse-iceberg` catalogs must be routed through the
Gravitino Iceberg REST server. When enabled, a catalog is not registered until
discovery succeeds or `gravitino.iceberg.rest-uri` is configured. Set this to
`false` to retain legacy `catalog-backend` translation and skip discovery.
[...]
+| gravitino.iceberg.rest-uri | string | (none)
| The endpoint of the Gravitino Iceberg REST server (IRC). It is discovered
automatically from the Gravitino server for each loaded metalake; set this only
to override the discovered value. When available, eligible `lakehouse-iceberg`
catalogs are loaded through IRC, enabling credential vending.
[...]
+| gravitino.iceberg.rest-catalog. | string | (none)
| Prefix for properties passed to the internal Trino Iceberg REST catalog. The
prefix is rewritten to `iceberg.rest-catalog.`. The `uri`, `warehouse`, and
`prefix` keys are reserved and derived by the connector.
[...]
To configure the Gravitino client, use properties prefixed with
`gravitino.client.`. These properties will directly passed to the Gravitino
client.
@@ -43,12 +44,11 @@ legacy `catalog-backend` translation. Otherwise, non-REST
`lakehouse-iceberg` ca
unregistered until discovery succeeds. See
[Iceberg catalog](./catalog-iceberg.md#how-trino-reaches-the-catalog).
-Multi-metalake mode (`gravitino.use-single-metalake=false`) is supported on
Trino connector versions 435-445 and 469-478. On versions 446-468, a warning is
logged and the connector initializes, but the mode is not fully supported and
some operations may fail.
+Qualified catalog names (`gravitino.catalog-name-with-metalake=true`) are
supported on Trino connector versions 435-445 and 469-478. On versions 446-468,
a warning is logged and the connector initializes, but the mode is not fully
supported and some operations may fail.
-**Note:** In multi-metalake mode, `gravitino.iceberg.rest-uri` is only honored
when scoped to a
-metalake, as `gravitino.iceberg.rest-uri.<metalake_name>` — the unscoped form
is ignored, since a
-single Iceberg REST server serves exactly one metalake and applying it to
every metalake would
-misroute the others. The unscoped form remains valid in single-metalake mode.
+**Note:** `gravitino.iceberg.rest-uri` is the default endpoint for every
loaded metalake. Since a
+single Iceberg REST server serves exactly one metalake, override it per
metalake with
+`gravitino.iceberg.rest-uri.<metalake_name>` when several metalakes are loaded.
## Connecting to a TLS-enabled coordinator
diff --git a/docs/trino-connector/installation.md
b/docs/trino-connector/installation.md
index 9543261510..edb1824b51 100644
--- a/docs/trino-connector/installation.md
+++ b/docs/trino-connector/installation.md
@@ -145,7 +145,7 @@ gravitino.metalake=test
```
- The `gravitino.name` defines which Gravitino Trino connector is used. It
must be `gravitino`.
-- The `gravitino.metalake` defines which metalake are used. It should exist in
the Gravitino server.
+- The `gravitino.metalake` defines which metalake is used. It should exist in
the Gravitino server. It is optional: when omitted, the catalogs of every
metalake are loaded, see [configuration](./configuration.md).
- The `gravitino.uri` defines the connection information about Gravitino
server. Make sure your container can access the Gravitino server.
Full configurations for Apache Gravitino Trino connector can be seen
[here](configuration.md)
diff --git a/docs/trino-connector/supported-catalog.md
b/docs/trino-connector/supported-catalog.md
index 0c2d547eeb..762a5d5bbf 100644
--- a/docs/trino-connector/supported-catalog.md
+++ b/docs/trino-connector/supported-catalog.md
@@ -24,13 +24,14 @@ User can also use the system table `catalog` to describe
all the catalogs.
Create catalog:
```sql
-create_catalog(CATALOG varchar, PROVIDER varchar, PROPERTIES MAP(VARCHAR,
VARCHAR), IGNORE_EXIST boolean);
+create_catalog(CATALOG varchar, PROVIDER varchar, PROPERTIES MAP(VARCHAR,
VARCHAR), IGNORE_EXIST boolean, METALAKE varchar);
```
- CATALOG: The catalog name to be created.
- PROVIDER: The catalog provider. Supported values: `hive`,
`lakehouse-iceberg`, `jdbc-mysql`, `jdbc-postgresql`, `glue`.
- PROPERTIES: The properties of the catalog.
- IGNORE_EXIST: The flag to ignore the error if the catalog already exists.
It's optional, the default value is `false`.
+- METALAKE: The metalake to create the catalog in. It's optional, the default
value is the configured `gravitino.metalake`; it is required when
`gravitino.metalake` is unset.
The type of catalog properties reference:
- [Hive catalog](../apache-hive-catalog.md#catalog-properties)
@@ -43,30 +44,47 @@ The type of catalog properties reference:
Drop catalog:
```sql
-drop_catalog(CATALOG varchar, IGNORE_NOT_EXIST boolean);
+drop_catalog(CATALOG varchar, IGNORE_NOT_EXIST boolean, METALAKE varchar);
```
- CATALOG: The catalog name to be deleted.
- IGNORE_NOT_EXIST: The flag to ignore the error if the catalog does not
exist. It's optional, the default value is `false`.
+- METALAKE: The metalake the catalog belongs to. It's optional, the default
value is the configured `gravitino.metalake`; it is required when
`gravitino.metalake` is unset.
Alter catalog:
```sql
-alter_catalog(CATALOG varchar, SET_PROPERTIES MAP(VARCHAR, VARCHAR),
REMOVE_PROPERTIES ARRY[VARCHAR]);
+alter_catalog(CATALOG varchar, SET_PROPERTIES MAP(VARCHAR, VARCHAR),
REMOVE_PROPERTIES ARRY[VARCHAR], METALAKE varchar);
```
- CATALOG: The catalog name to be altered.
- SET_PROPERTIES: The properties to be set.
- REMOVE_PROPERTIES: The properties to be removed.
+- METALAKE: The metalake the catalog belongs to. It's optional, the default
value is the configured `gravitino.metalake`; it is required when
`gravitino.metalake` is unset.
+
+A metalake other than the configured one can only be targeted when every
metalake is loaded
+(`gravitino.metalake` unset or `gravitino.catalog-name-with-metalake=true`).
With unqualified
+catalog names, the procedures look the catalog up by name and metalake, so a
catalog of another
+metalake holding the same Trino catalog name is not affected.
These stored procedures are under the `gravitino` connector and the `system`
schema.
-So you need to use the following SQL to call them in the `trino-cli`:
+So you need to use the following SQL to call them in the `trino-cli`, passing
the metalake by name
+when `gravitino.metalake` is unset:
+```sql
+call gravitino.system.create_catalog(
+ catalog => 'gt_hive',
+ provider => 'hive',
+ properties => map(array['metastore.uris'],
array['thrift://trino-ci-hive:9083']),
+ metalake => 'test'
+);
+```
Describe catalogs:
-The system table `gravitino.system.catalog` is used to describe all the
catalogs.
+The system table `gravitino.system.catalog` is used to describe all the
catalogs of the configured
+metalake, or of every metalake when `gravitino.metalake` is unset.
```sql
select * from gravitino.system.catalog;
@@ -75,11 +93,14 @@ select * from gravitino.system.catalog;
The result is like:
```test
- name | provider |
properties
---------------+----------+-------------------------------------------------------------------------------------------------------------
- gt_hive | hive |
{gravitino.bypass.hive.metastore.client.capability.check=false,
metastore.uris=thrift://trino-ci-hive:9083}
+ name | provider |
properties | metalake
+--------------+----------+-------------------------------------------------------------------------------------------------------------+----------
+ gt_hive | hive |
{gravitino.bypass.hive.metastore.client.capability.check=false,
metastore.uris=thrift://trino-ci-hive:9083} | test
```
+The `metalake` column tells apart catalogs of different metalakes that share a
name when
+`gravitino.metalake` is unset.
+
Check catalog registration status:
`gravitino.system.catalog` lists the relational catalogs the Gravitino server
knows about, minus any
@@ -254,6 +275,7 @@ Registration happens in the background, so a catalog that
fails to register simp
| A catalog is missing from `SHOW CATALOGS` | Query
`gravitino.system.catalog_status` and read `status` and `last_error`, then
follow the rows below |
| `status = FAILED`, `last_error` mentions `Access Denied` | The
`trino.jdbc.user` lacks a Trino system role permitted to run `CREATE CATALOG`
|
| `status = FAILED`, `last_error` mentions a configuration property | A
`trino.bypass.` property is not accepted by the underlying Trino connector
|
+| `status = FAILED`, `last_error` mentions `already registered by metalake` |
Another metalake owns the same Trino catalog name. Rename the catalog or set
`gravitino.catalog-name-with-metalake=true` |
| `status = UNSUPPORTED` | The
catalog is not relational, or its provider is outside the supported list.
`last_error` names the supported providers |
| `status = SKIPPED` | The
catalog matches `gravitino.trino.skip-catalog-patterns`
|
| The catalog has no row in `catalog_status` at all | The
load loop never reached it. Check `gravitino.system.load_status`
|
diff --git
a/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00008_alter_catalog.txt
b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00008_alter_catalog.txt
index 501fa0d8aa..608e8a55cb 100644
---
a/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00008_alter_catalog.txt
+++
b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00008_alter_catalog.txt
@@ -1,17 +1,17 @@
CALL
-"gt_mysql_xxx1","jdbc-mysql","{""in-use"":""true"",""jdbc-driver"":""com.mysql.cj.jdbc.Driver"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false""}"
+"gt_mysql_xxx1","jdbc-mysql","{""in-use"":""true"",""jdbc-driver"":""com.mysql.cj.jdbc.Driver"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false""}","test"
CALL
-"gt_mysql_xxx1","jdbc-mysql","{""in-use"":""true"",""jdbc-driver"":""com.mysql.cj.jdbc.Driver"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""test_key"":""test_value"",""trino.bypass.join-pushdown.strategy"":""EAGER""}"
+"gt_mysql_xxx1","jdbc-mysql","{""in-use"":""true"",""jdbc-driver"":""com.mysql.cj.jdbc.Driver"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""test_key"":""test_value"",""trino.bypass.join-pushdown.strategy"":""EAGER""}","test"
CALL
-"gt_mysql_xxx1","jdbc-mysql","{""in-use"":""true"",""jdbc-driver"":""com.mysql.cj.jdbc.Driver"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""test_key"":""test_value""}"
+"gt_mysql_xxx1","jdbc-mysql","{""in-use"":""true"",""jdbc-driver"":""com.mysql.cj.jdbc.Driver"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""test_key"":""test_value""}","test"
CALL
-"gt_mysql_xxx1","jdbc-mysql","{""in-use"":""true"",""jdbc-driver"":""com.mysql.cj.jdbc.Driver"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""trino.bypass.join-pushdown.strategy"":""EAGER""}"
+"gt_mysql_xxx1","jdbc-mysql","{""in-use"":""true"",""jdbc-driver"":""com.mysql.cj.jdbc.Driver"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""trino.bypass.join-pushdown.strategy"":""EAGER""}","test"
CALL
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConfig.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConfig.java
index 107cde4b82..4113101418 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConfig.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConfig.java
@@ -104,9 +104,10 @@ public class GravitinoConfig {
private static final ConfigEntry GRAVITINO_METALAKE =
new ConfigEntry(
"gravitino.metalake",
- "The name of the metalake (top-level namespace) to connect to",
+ "The name of the metalake (top-level namespace) to connect to. "
+ + "When unset, the catalogs of every metalake are loaded.",
"",
- true);
+ false);
private static final ConfigEntry GRAVITINO_USER =
new ConfigEntry(
@@ -116,25 +117,39 @@ public class GravitinoConfig {
false);
/**
- * @deprecated Please use {@code gravitino.use-single-metalake} instead.
+ * @deprecated Please use {@code gravitino.catalog-name-with-metalake}
instead.
*/
@Deprecated
@SuppressWarnings("UnusedVariable")
private static final ConfigEntry GRAVITINO_SIMPLIFY_CATALOG_NAMES =
new ConfigEntry(
"gravitino.simplify-catalog-names",
- "Deprecated: omits the metalake prefix from catalog names. Use
gravitino.use-single-metalake instead.",
+ "Deprecated: omits the metalake prefix from catalog names. Use
gravitino.catalog-name-with-metalake instead.",
"true",
false);
+ /**
+ * Legacy switch, read only when {@code
gravitino.catalog-name-with-metalake} is unset.
+ *
+ * @deprecated Please use {@code gravitino.catalog-name-with-metalake}
instead.
+ */
+ @Deprecated
private static final ConfigEntry GRAVITINO_SINGLE_METALAKE_MODE =
new ConfigEntry(
"gravitino.use-single-metalake",
- "If true, only one metalake is supported in this connector; identify
the catalog by <catalog_name>. "
- + "If false, multiple metalakes are supported; identify the
catalog by <metalake_name>.<catalog_name>.",
+ "Deprecated: if false, identify the catalog by
<metalake_name>.<catalog_name> and load "
+ + "every metalake. Use gravitino.catalog-name-with-metalake
instead.",
"true",
false);
+ private static final ConfigEntry GRAVITINO_CATALOG_NAME_WITH_METALAKE =
+ new ConfigEntry(
+ "gravitino.catalog-name-with-metalake",
+ "If true, identify the catalog by <metalake_name>.<catalog_name> and
load every metalake. "
+ + "If false, identify the catalog by <catalog_name>.",
+ "false",
+ false);
+
private static final ConfigEntry GRAVITINO_CLOUD_REGION_CODE =
new ConfigEntry(
"gravitino.cloud.region-code",
@@ -347,9 +362,10 @@ public class GravitinoConfig {
}
/**
- * Retrieves the metalake name for used.
+ * Retrieves the configured metalake name.
*
- * @return the metalake name for used
+ * @return the trimmed metalake name, or an empty string when {@code
gravitino.metalake} is unset;
+ * see {@link #hasMetalake()}
*/
public String getMetalake() {
// Trimmed so a stray leading/trailing space in the catalog properties
file does not make this
@@ -358,6 +374,25 @@ public class GravitinoConfig {
return config.getOrDefault(GRAVITINO_METALAKE.key,
GRAVITINO_METALAKE.defaultValue).trim();
}
+ /**
+ * Whether a metalake is configured.
+ *
+ * @return true if {@code gravitino.metalake} is set to a non-blank value
+ */
+ public boolean hasMetalake() {
+ return !getMetalake().isEmpty();
+ }
+
+ /**
+ * Whether the catalogs of every metalake are loaded. This is the case when
no metalake is
+ * configured, or when catalog names carry the metalake.
+ *
+ * @return true if all metalakes are loaded
+ */
+ public boolean loadAllMetalakes() {
+ return !hasMetalake() || catalogNameWithMetalake();
+ }
+
/**
* Retrieves the username for simple authentication.
*
@@ -379,14 +414,33 @@ public class GravitinoConfig {
}
/**
- * Retrieves the single metalake mode.
+ * Whether Trino catalog names carry the metalake, as {@code
"<metalake>.<catalog>"}. The
+ * deprecated {@code gravitino.use-single-metalake=false} is honored when
the new key is unset.
*
- * @return the single metalake mode
+ * @return true if catalog names are qualified with the metalake
*/
- public boolean singleMetalakeMode() {
- return Boolean.parseBoolean(
- config.getOrDefault(
- GRAVITINO_SINGLE_METALAKE_MODE.key,
GRAVITINO_SINGLE_METALAKE_MODE.defaultValue));
+ public boolean catalogNameWithMetalake() {
+ String value = config.get(GRAVITINO_CATALOG_NAME_WITH_METALAKE.key);
+ if (value != null) {
+ return parseBooleanConfig(GRAVITINO_CATALOG_NAME_WITH_METALAKE.key,
value.trim());
+ }
+ return !parseBooleanConfig(
+ GRAVITINO_SINGLE_METALAKE_MODE.key,
+ config
+ .getOrDefault(
+ GRAVITINO_SINGLE_METALAKE_MODE.key,
GRAVITINO_SINGLE_METALAKE_MODE.defaultValue)
+ .trim());
+ }
+
+ /**
+ * Whether the deprecated {@code gravitino.use-single-metalake} key is
present and still decides
+ * the catalog naming, so that its use can be reported.
+ *
+ * @return true if the deprecated key is set and the replacing key is not
+ */
+ public boolean usesDeprecatedSingleMetalakeKey() {
+ return config.containsKey(GRAVITINO_SINGLE_METALAKE_MODE.key)
+ && !config.containsKey(GRAVITINO_CATALOG_NAME_WITH_METALAKE.key);
}
boolean isDynamicConnector() {
@@ -647,15 +701,17 @@ public class GravitinoConfig {
stringList.add(String.format("\"%s\"='%s'", entry.getKey(), value));
}
}
- // copy the configuration by the prefix of GRAVITINO_CLIENT_CONFIG_PREFIX
and
- // GRAVITINO_ICEBERG_REST_CATALOG_CONFIG_PREFIX
+ // Copy configuration with prefixes that are not represented by exact
entries in
+ // CONFIG_DEFINITIONS. In particular, scoped Iceberg REST URIs must reach
dynamic catalogs so
+ // that workers use the same per-metalake endpoint as the coordinator.
config.entrySet().stream()
.filter(
entry ->
(entry.getKey().startsWith(GRAVITINO_CLIENT_CONFIG_PREFIX.key)
|| entry
.getKey()
-
.startsWith(GRAVITINO_ICEBERG_REST_CATALOG_CONFIG_PREFIX.key))
+
.startsWith(GRAVITINO_ICEBERG_REST_CATALOG_CONFIG_PREFIX.key)
+ ||
entry.getKey().startsWith(GRAVITINO_ICEBERG_REST_URI.key + "."))
&&
!GravitinoConnectorFactory.isSecuritySensitivePropertyName(entry.getKey()))
.forEach(
entry ->
@@ -802,10 +858,8 @@ public class GravitinoConfig {
* Unlike the discovered endpoint, this is plain local file configuration
and is therefore
* identical and valid on every node — coordinator and workers alike.
*
- * <p>{@code gravitino.iceberg.rest-uri.<metalake>} is checked first. The
unscoped {@code
- * gravitino.iceberg.rest-uri} is honored only in single-metalake mode,
where it is unambiguous;
- * in multi-metalake mode it is ignored, since a single Iceberg REST server
serves exactly one
- * metalake and applying it to every metalake would misroute the others.
+ * <p>{@code gravitino.iceberg.rest-uri.<metalake>} is checked first and
overrides the unscoped
+ * {@code gravitino.iceberg.rest-uri}, which is the default for every
metalake.
*
* @param metalake the metalake to resolve the override for
* @return the manually configured Iceberg REST server endpoint, or an empty
string when unset
@@ -815,11 +869,8 @@ public class GravitinoConfig {
if (StringUtils.isNotBlank(scopedValue)) {
return scopedValue;
}
- if (singleMetalakeMode()) {
- return config.getOrDefault(
- GRAVITINO_ICEBERG_REST_URI.key,
GRAVITINO_ICEBERG_REST_URI.defaultValue);
- }
- return GRAVITINO_ICEBERG_REST_URI.defaultValue;
+ return config.getOrDefault(
+ GRAVITINO_ICEBERG_REST_URI.key,
GRAVITINO_ICEBERG_REST_URI.defaultValue);
}
/**
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java
index 739a7f1d50..f5d7266d20 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java
@@ -24,7 +24,6 @@ import static
org.apache.gravitino.trino.connector.GravitinoErrorCode.GRAVITINO_
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import com.google.common.base.Strings;
import io.airlift.log.Logger;
import io.trino.spi.HostAddress;
import io.trino.spi.TrinoException;
@@ -179,13 +178,10 @@ public class GravitinoConnectorFactory implements
ConnectorFactory {
} else {
// The static connector is an instance of GravitinoSystemConnector. It
is loaded by Trino
// using the connector configuration.
- String metalake = config.getMetalake();
- if (Strings.isNullOrEmpty(metalake)) {
- throw new TrinoException(
- GravitinoErrorCode.GRAVITINO_METALAKE_NOT_EXISTS, "No gravitino
metalake selected");
- }
// Built per entry catalog, like the stored procedures: both are scoped
to this catalog's
- // metalake even though the underlying manager is shared.
+ // metalake even though the underlying manager is shared. Without a
configured metalake
+ // they cover every metalake the manager loads.
+ String metalake = config.hasMetalake() ? config.getMetalake() : null;
GravitinoStoredProcedureFactory gravitinoStoredProcedureFactory =
new GravitinoStoredProcedureFactory(catalogConnectorManager,
metalake);
GravitinoSystemTableFactory systemTableFactory =
@@ -224,10 +220,10 @@ public class GravitinoConnectorFactory implements
ConnectorFactory {
trinoVersion = parseTrinoSpiVersion(spiVersion);
// check catalog name with metalake are supported in this trino version
- if (!config.singleMetalakeMode() && !supportCatalogNameWithMetalake()) {
+ if (config.catalogNameWithMetalake() && !supportCatalogNameWithMetalake())
{
LOG.warn(
"The trino-connector-%s-%s does not fully support catalog name with
metalake. "
- + "The DROP CATALOG operation may not work correctly in
multi-metalake mode.",
+ + "The DROP CATALOG operation may not work correctly when
catalog names carry the metalake.",
getMinSupportTrinoSpiVersion(), getMaxSupportTrinoSpiVersion());
}
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java
index fdc369c559..d7fed43f55 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java
@@ -26,6 +26,7 @@ import io.trino.spi.TrinoException;
import io.trino.spi.connector.ConnectorContext;
import java.util.ArrayList;
import java.util.HashSet;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -75,9 +76,11 @@ public class CatalogConnectorManager {
private final ConcurrentHashMap<String, CatalogConnectorContext>
catalogConnectors =
new ConcurrentHashMap<>();
- // The registration state of every catalog seen by the load loop, keyed by
the Trino catalog
- // name. Written only by the load loop thread, read by query threads through
the system tables
- // and by stored procedure threads through describeRegistrationFailure().
+ // The registration state of every catalog seen by the load loop, keyed by
the metalake and the
+ // Trino catalog name (see stateKey), since two metalakes can report the
same Trino catalog name
+ // when every metalake is loaded with unqualified names. Written only by the
load loop thread,
+ // read by query threads through the system tables and by stored procedure
threads through
+ // describeRegistrationFailure().
private final ConcurrentHashMap<String, CatalogRegistrationState>
catalogStates =
new ConcurrentHashMap<>();
@@ -183,6 +186,14 @@ public class CatalogConnectorManager {
// Parsed eagerly so a misconfigured value fails startup instead of
surfacing every poll as an
// unrelated "Load Metalake failed" error.
config.isIcebergRestRoutingEnabled();
+ if (config.usesDeprecatedSingleMetalakeKey()) {
+ LOG.warn(
+ "gravitino.use-single-metalake is deprecated, use
gravitino.catalog-name-with-metalake "
+ + "instead.");
+ }
+ if (!config.hasMetalake()) {
+ LOG.info("gravitino.metalake is not set, the catalogs of every metalake
are loaded.");
+ }
}
/**
@@ -228,8 +239,12 @@ public class CatalogConnectorManager {
}
trinoReachable = true;
- Set<String> usedMetalakes = new HashSet<>();
- if (config.singleMetalakeMode()) {
+ // Iterated in the server's listing order, so that with unqualified
catalog names a clash
+ // between two metalakes first seen in the same cycle follows that
order. Once a metalake
+ // has registered a Trino name it keeps it across cycles (see the
ownership check in
+ // loadCatalogs), whatever the order becomes later.
+ Set<String> usedMetalakes = new LinkedHashSet<>();
+ if (!config.loadAllMetalakes()) {
usedMetalakes.add(targetMetalake);
metalakes.computeIfAbsent(targetMetalake, this::retrieveMetalake);
} else {
@@ -307,9 +322,7 @@ public class CatalogConnectorManager {
catalogStates
.values()
.removeIf(
- state ->
- !usedMetalakes.contains(state.getMetalake())
- &&
!catalogConnectors.containsKey(state.getTrinoCatalogName()));
+ state -> !usedMetalakes.contains(state.getMetalake()) &&
!hasLiveConnector(state));
metalakeErrors.keySet().removeIf(metalakeName ->
!usedMetalakes.contains(metalakeName));
metalakes.keySet().removeIf(metalakeName ->
!usedMetalakes.contains(metalakeName));
}
@@ -517,13 +530,31 @@ public class CatalogConnectorManager {
state ->
state.getMetalake().equals(metalakeName)
&& !presentTrinoNames.contains(state.getTrinoCatalogName())
- &&
!catalogConnectors.containsKey(state.getTrinoCatalogName()));
+ && !hasLiveConnector(state));
// Load new catalogs belows to the metalake.
for (String catalogName : catalogNames) {
String trinoCatalogName = getTrinoCatalogName(metalakeName, catalogName);
+ // With unqualified names a catalog of another metalake may already hold
this Trino name.
+ // The first one registered keeps it; this one is reported and left
alone, so that it is
+ // neither loaded over the existing connector nor treated as a refresh
of it.
+ CatalogConnectorContext existing =
catalogConnectors.get(trinoCatalogName);
+ if (existing != null &&
!existing.getCatalog().getMetalake().equals(metalakeName)) {
+ recordCatalogState(
+ CatalogRegistrationState.failed(
+ metalakeName,
+ catalogName,
+ trinoCatalogName,
+ null,
+ String.format(
+ "Trino catalog name %s is already registered by metalake
%s. Rename the "
+ + "catalog or set
gravitino.catalog-name-with-metalake=true",
+ trinoCatalogName, existing.getCatalog().getMetalake())),
+ null);
+ continue;
+ }
// Known before the catalog is even loaded, since it only depends on the
name.
- boolean alreadyRegistered =
catalogConnectors.containsKey(trinoCatalogName);
+ boolean alreadyRegistered = existing != null;
// Tracked outside the try so that a failure can still report the
provider it knows about.
String provider = null;
try {
@@ -598,7 +629,7 @@ public class CatalogConnectorManager {
CatalogRegistrationState[] seen = new CatalogRegistrationState[1];
CatalogRegistrationState state =
catalogStates.compute(
- newState.getTrinoCatalogName(),
+ stateKey(newState.getMetalake(), newState.getTrinoCatalogName()),
(name, previous) -> {
seen[0] = previous;
return newState.withHistoryOf(previous);
@@ -738,6 +769,21 @@ public class CatalogConnectorManager {
return catalogConnectors.get(catalogName);
}
+ /**
+ * Retrieves the catalog connector context of a Gravitino catalog. With
unqualified catalog names
+ * the Trino name may be held by a catalog of another metalake, which is not
the one asked for.
+ *
+ * @param metalake the metalake the catalog belongs to
+ * @param catalogName the name of the catalog in Gravitino
+ * @return the catalog connector context, or null if this catalog is not
registered
+ */
+ @Nullable
+ public CatalogConnectorContext getCatalogConnector(String metalake, String
catalogName) {
+ CatalogConnectorContext context =
+ catalogConnectors.get(getTrinoCatalogName(metalake, catalogName));
+ return context != null &&
context.getCatalog().getMetalake().equals(metalake) ? context : null;
+ }
+
/**
* Checks if a catalog connector exists for the specified catalog name.
*
@@ -779,7 +825,7 @@ public class CatalogConnectorManager {
* @return the Trino catalog name
*/
public String getTrinoCatalogName(String metalake, String catalog) {
- return config.singleMetalakeMode()
+ return !config.catalogNameWithMetalake()
? catalog
: trinoCatalogNameHandler.getCatalogName(metalake, catalog);
}
@@ -796,13 +842,11 @@ public class CatalogConnectorManager {
/**
* Retrieves a snapshot of the registration state of every Gravitino catalog
seen by the load
- * loop. Package-private: production callers always know which metalake they
report on and use
- * {@link #getCatalogRegistrationStates(String)}; this exists for tests that
assert on the whole
- * set of tracked catalogs.
+ * loop.
*
* @return the registration states
*/
- List<CatalogRegistrationState> getCatalogRegistrationStates() {
+ public List<CatalogRegistrationState> getCatalogRegistrationStates() {
return List.copyOf(catalogStates.values());
}
@@ -810,15 +854,29 @@ public class CatalogConnectorManager {
* Retrieves a snapshot of the registration state of every Gravitino catalog
seen by the load loop
* that belongs to the given metalake.
*
- * @param metalake the metalake to filter by
+ * @param metalake the metalake to filter by, or null for every metalake
* @return the registration states belonging to that metalake
*/
- public List<CatalogRegistrationState> getCatalogRegistrationStates(String
metalake) {
+ public List<CatalogRegistrationState> getCatalogRegistrationStates(@Nullable
String metalake) {
+ if (metalake == null) {
+ return getCatalogRegistrationStates();
+ }
return catalogStates.values().stream()
.filter(state -> state.getMetalake().equals(metalake))
.toList();
}
+ private static String stateKey(String metalake, String trinoCatalogName) {
+ return metalake + "." + trinoCatalogName;
+ }
+
+ // Whether the catalog this state describes still has its connector in
Trino. The connector under
+ // the same Trino name may belong to another metalake, which says nothing
about this catalog.
+ private boolean hasLiveConnector(CatalogRegistrationState state) {
+ CatalogConnectorContext live =
catalogConnectors.get(state.getTrinoCatalogName());
+ return live != null &&
live.getCatalog().getMetalake().equals(state.getMetalake());
+ }
+
/**
* Checks whether the Trino server answered the last time the load loop
probed it. No catalog can
* be registered while it does not.
@@ -866,7 +924,7 @@ public class CatalogConnectorManager {
* @return a human readable explanation
*/
public String describeRegistrationFailure(String metalake, String
trinoCatalogName) {
- CatalogRegistrationState state = catalogStates.get(trinoCatalogName);
+ CatalogRegistrationState state = catalogStates.get(stateKey(metalake,
trinoCatalogName));
if (state != null && state.getLastError() != null) {
return String.format("%s: %s", state.getStatus(), state.getLastError());
}
@@ -913,12 +971,15 @@ public class CatalogConnectorManager {
String catalogConfig = config.getCatalogConfig();
GravitinoCatalog catalog = GravitinoCatalog.fromJson(catalogConfig);
- if (this.config.singleMetalakeMode()
+ if (!this.config.catalogNameWithMetalake()
&& StringUtils.isNotBlank(targetMetalake)
&& !targetMetalake.equals(catalog.getMetalake())) {
throw new TrinoException(
GravitinoErrorCode.GRAVITINO_UNSUPPORTED_OPERATION,
- "Multiple metalakes are not supported");
+ String.format(
+ "Catalog %s belongs to metalake %s but this connector is
configured for metalake "
+ + "%s; set gravitino.catalog-name-with-metalake=true to
serve several metalakes",
+ catalog.getName(), catalog.getMetalake(), targetMetalake));
}
GravitinoMetalake metalake =
metalakes.computeIfAbsent(catalog.getMetalake(),
this::retrieveMetalake);
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/AlterCatalogStoredProcedure.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/AlterCatalogStoredProcedure.java
index 43f53933e9..f0dcf7cf0e 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/AlterCatalogStoredProcedure.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/AlterCatalogStoredProcedure.java
@@ -33,6 +33,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.CatalogChange;
import org.apache.gravitino.NameIdentifier;
@@ -53,29 +54,30 @@ public class AlterCatalogStoredProcedure extends
GravitinoStoredProcedure {
private static final Logger LOG =
Logger.get(AlterCatalogStoredProcedure.class);
private final CatalogConnectorManager catalogConnectorManager;
- private final String metalake;
+ @Nullable private final String configuredMetalake;
/**
* Constructs a new AlterCatalogStoredProcedure.
*
* @param catalogConnectorManager the catalog connector manager
- * @param metalake the metalake name
+ * @param configuredMetalake the metalake name, or null when the connector
is not configured with
+ * one
*/
public AlterCatalogStoredProcedure(
- CatalogConnectorManager catalogConnectorManager, String metalake) {
+ CatalogConnectorManager catalogConnectorManager, @Nullable String
configuredMetalake) {
this.catalogConnectorManager = catalogConnectorManager;
- this.metalake = metalake;
+ this.configuredMetalake = configuredMetalake;
}
@Override
public Procedure createStoredProcedure() throws NoSuchMethodException,
IllegalAccessException {
- // call gravitino.system.alter_catalog(catalogName, set_properties,
remove_properties
+ // call gravitino.system.alter_catalog(catalogName, set_properties,
remove_properties, metalake)
MethodHandle dropCatalog =
MethodHandles.lookup()
.unreflect(
AlterCatalogStoredProcedure.class.getMethod(
- "alterCatalog", String.class, Map.class, List.class))
+ "alterCatalog", String.class, Map.class, List.class,
String.class))
.bindTo(this);
List<Procedure.Argument> arguments =
List.of(
@@ -87,7 +89,8 @@ public class AlterCatalogStoredProcedure extends
GravitinoStoredProcedure {
new ArrayType(VARCHAR),
false,
ArrayBlock.fromElementBlock(
- 0, Optional.empty(), new int[1],
VARCHAR.createBlockBuilder(null, 1).build())));
+ 0, Optional.empty(), new int[1],
VARCHAR.createBlockBuilder(null, 1).build())),
+ new Procedure.Argument(METALAKE_ARGUMENT, VARCHAR, false, null));
return new Procedure(
GravitinoSystemTable.SYSTEM_TABLE_SCHEMA_NAME, "alter_catalog",
arguments, dropCatalog);
}
@@ -98,14 +101,27 @@ public class AlterCatalogStoredProcedure extends
GravitinoStoredProcedure {
* @param catalogName the name of the catalog to alter
* @param setProperties the properties to set
* @param removeProperties the properties to remove
+ * @param metalakeArgument the metalake the catalog belongs to, null to use
the configured one
* @throws TrinoException if the catalog does not exist or the operation
fails
*/
public void alterCatalog(
- String catalogName, Map<String, String> setProperties, List<String>
removeProperties) {
+ String catalogName,
+ Map<String, String> setProperties,
+ List<String> removeProperties,
+ @Nullable String metalakeArgument) {
+ String metalake = resolveMetalake(configuredMetalake, metalakeArgument);
try {
CatalogConnectorContext catalogConnectorContext =
- catalogConnectorManager.getCatalogConnector(
- catalogConnectorManager.getTrinoCatalogName(metalake,
catalogName));
+ catalogConnectorManager.getCatalogConnector(metalake, catalogName);
+ if (catalogConnectorContext == null) {
+ throw new TrinoException(
+ GravitinoErrorCode.GRAVITINO_CATALOG_NOT_EXISTS,
+ String.format(
+ "Catalog %s is not registered in Trino. %s",
+ NameIdentifier.of(metalake, catalogName),
+ catalogConnectorManager.describeRegistrationFailure(
+ metalake,
catalogConnectorManager.getTrinoCatalogName(metalake, catalogName))));
+ }
GravitinoCatalog oldCatalog = catalogConnectorContext.getCatalog();
List<CatalogChange> changes = new ArrayList<>();
@@ -136,15 +152,15 @@ public class AlterCatalogStoredProcedure extends
GravitinoStoredProcedure {
.alterCatalog(catalogName, changes.toArray(changes.toArray(new
CatalogChange[0])));
catalogConnectorManager.loadMetalakeSync();
- String trinoCatalogName =
catalogConnectorManager.getTrinoCatalogName(metalake, catalogName);
- catalogConnectorContext =
catalogConnectorManager.getCatalogConnector(trinoCatalogName);
+ catalogConnectorContext =
catalogConnectorManager.getCatalogConnector(metalake, catalogName);
if (catalogConnectorContext == null
|| catalogConnectorContext.getCatalog().getLastModifiedTime()
== oldCatalog.getLastModifiedTime()) {
throw new TrinoException(
GravitinoErrorCode.GRAVITINO_OPERATION_FAILED,
"Update catalog failed due to the reloading process fails. "
- +
catalogConnectorManager.describeRegistrationFailure(metalake,
trinoCatalogName));
+ + catalogConnectorManager.describeRegistrationFailure(
+ metalake,
catalogConnectorManager.getTrinoCatalogName(metalake, catalogName)));
}
LOG.info("Alter catalog %s in metalake %s successfully.", catalogName,
metalake);
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/CreateCatalogStoredProcedure.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/CreateCatalogStoredProcedure.java
index 2bc7180d8a..405ef0ce73 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/CreateCatalogStoredProcedure.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/CreateCatalogStoredProcedure.java
@@ -30,6 +30,7 @@ import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.util.List;
import java.util.Map;
+import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
@@ -49,28 +50,34 @@ public class CreateCatalogStoredProcedure extends
GravitinoStoredProcedure {
private static final Logger LOG =
Logger.get(CreateCatalogStoredProcedure.class);
private final CatalogConnectorManager catalogConnectorManager;
- private final String metalake;
+ @Nullable private final String configuredMetalake;
/**
* Constructs a new CreateCatalogStoredProcedure.
*
* @param catalogConnectorManager the catalog connector manager
- * @param metalake the metalake name
+ * @param configuredMetalake the metalake name, or null when the connector
is not configured with
+ * one
*/
public CreateCatalogStoredProcedure(
- CatalogConnectorManager catalogConnectorManager, String metalake) {
+ CatalogConnectorManager catalogConnectorManager, @Nullable String
configuredMetalake) {
this.catalogConnectorManager = catalogConnectorManager;
- this.metalake = metalake;
+ this.configuredMetalake = configuredMetalake;
}
@Override
public Procedure createStoredProcedure() throws NoSuchMethodException,
IllegalAccessException {
- // call gravitino.system.create_catalog(catalog, provider, properties,
ignore_exist)
+ // call gravitino.system.create_catalog(catalog, provider, properties,
ignore_exist, metalake)
MethodHandle createCatalog =
MethodHandles.lookup()
.unreflect(
CreateCatalogStoredProcedure.class.getMethod(
- "createCatalog", String.class, String.class, Map.class,
boolean.class))
+ "createCatalog",
+ String.class,
+ String.class,
+ Map.class,
+ boolean.class,
+ String.class))
.bindTo(this);
List<Procedure.Argument> arguments =
@@ -79,7 +86,8 @@ public class CreateCatalogStoredProcedure extends
GravitinoStoredProcedure {
new Procedure.Argument("PROVIDER", VARCHAR),
new Procedure.Argument(
"PROPERTIES", new MapType(VARCHAR, VARCHAR, new
TypeOperators())),
- new Procedure.Argument("IGNORE_EXIST", BOOLEAN, false, false));
+ new Procedure.Argument("IGNORE_EXIST", BOOLEAN, false, false),
+ new Procedure.Argument(METALAKE_ARGUMENT, VARCHAR, false, null));
return new Procedure(
GravitinoSystemTable.SYSTEM_TABLE_SCHEMA_NAME, "create_catalog",
arguments, createCatalog);
@@ -92,13 +100,17 @@ public class CreateCatalogStoredProcedure extends
GravitinoStoredProcedure {
* @param provider the provider of the catalog
* @param properties the properties of the catalog
* @param ignoreExist whether to ignore if the catalog already exists
+ * @param metalakeArgument the metalake to create the catalog in, null to
use the configured one
* @throws TrinoException if the catalog already exists and ignoreExist is
false
*/
public void createCatalog(
- String catalogName, String provider, Map<String, String> properties,
boolean ignoreExist) {
- boolean exists =
- catalogConnectorManager.catalogConnectorExist(
- catalogConnectorManager.getTrinoCatalogName(metalake,
catalogName));
+ String catalogName,
+ String provider,
+ Map<String, String> properties,
+ boolean ignoreExist,
+ @Nullable String metalakeArgument) {
+ String metalake = resolveMetalake(configuredMetalake, metalakeArgument);
+ boolean exists = catalogConnectorManager.getCatalogConnector(metalake,
catalogName) != null;
if (exists) {
if (!ignoreExist) {
throw new TrinoException(
@@ -115,12 +127,12 @@ public class CreateCatalogStoredProcedure extends
GravitinoStoredProcedure {
catalogName, Catalog.Type.RELATIONAL, provider, "Trino created",
properties);
catalogConnectorManager.loadMetalakeSync();
- String trinoCatalogName =
catalogConnectorManager.getTrinoCatalogName(metalake, catalogName);
- if (!catalogConnectorManager.catalogConnectorExist(trinoCatalogName)) {
+ if (catalogConnectorManager.getCatalogConnector(metalake, catalogName)
== null) {
throw new TrinoException(
GravitinoErrorCode.GRAVITINO_OPERATION_FAILED,
"Create catalog failed due to the loading process fails. "
- +
catalogConnectorManager.describeRegistrationFailure(metalake,
trinoCatalogName));
+ + catalogConnectorManager.describeRegistrationFailure(
+ metalake,
catalogConnectorManager.getTrinoCatalogName(metalake, catalogName)));
}
LOG.info("Create catalog %s in metalake %s successfully.", catalogName,
metalake);
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/DropCatalogStoredProcedure.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/DropCatalogStoredProcedure.java
index eea2fd7815..a813a35df1 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/DropCatalogStoredProcedure.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/DropCatalogStoredProcedure.java
@@ -27,6 +27,7 @@ import io.trino.spi.procedure.Procedure;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.util.List;
+import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.exceptions.NoSuchMetalakeException;
@@ -45,33 +46,35 @@ public class DropCatalogStoredProcedure extends
GravitinoStoredProcedure {
private static final Logger LOG =
Logger.get(DropCatalogStoredProcedure.class);
private final CatalogConnectorManager catalogConnectorManager;
- private final String metalake;
+ @Nullable private final String configuredMetalake;
/**
* Constructs a new DropCatalogStoredProcedure.
*
* @param catalogConnectorManager the catalog connector manager
- * @param metalake the metalake name
+ * @param configuredMetalake the metalake name, or null when the connector
is not configured with
+ * one
*/
public DropCatalogStoredProcedure(
- CatalogConnectorManager catalogConnectorManager, String metalake) {
+ CatalogConnectorManager catalogConnectorManager, @Nullable String
configuredMetalake) {
this.catalogConnectorManager = catalogConnectorManager;
- this.metalake = metalake;
+ this.configuredMetalake = configuredMetalake;
}
@Override
public Procedure createStoredProcedure() throws NoSuchMethodException,
IllegalAccessException {
- // call gravitino.system.drop_catalog(catalog, ignore_not_exist)
+ // call gravitino.system.drop_catalog(catalog, ignore_not_exist, metalake)
MethodHandle dropCatalog =
MethodHandles.lookup()
.unreflect(
DropCatalogStoredProcedure.class.getMethod(
- "dropCatalog", String.class, boolean.class))
+ "dropCatalog", String.class, boolean.class, String.class))
.bindTo(this);
List<Procedure.Argument> arguments =
List.of(
new Procedure.Argument("CATALOG", VARCHAR),
- new Procedure.Argument("IGNORE_NOT_EXIST", BOOLEAN, false, false));
+ new Procedure.Argument("IGNORE_NOT_EXIST", BOOLEAN, false, false),
+ new Procedure.Argument(METALAKE_ARGUMENT, VARCHAR, false, null));
return new Procedure(
GravitinoSystemTable.SYSTEM_TABLE_SCHEMA_NAME, "drop_catalog",
arguments, dropCatalog);
}
@@ -84,13 +87,15 @@ public class DropCatalogStoredProcedure extends
GravitinoStoredProcedure {
*
* @param catalogName the name of the catalog to drop
* @param ignoreNotExist whether to ignore if the catalog does not exist
+ * @param metalakeArgument the metalake to drop the catalog from, null to
use the configured one
* @throws TrinoException if the catalog does not exist and ignoreNotExist
is false
*/
- public void dropCatalog(String catalogName, boolean ignoreNotExist) {
+ public void dropCatalog(
+ String catalogName, boolean ignoreNotExist, @Nullable String
metalakeArgument) {
+ String metalake = resolveMetalake(configuredMetalake, metalakeArgument);
try {
CatalogConnectorContext catalogConnector =
- catalogConnectorManager.getCatalogConnector(
- catalogConnectorManager.getTrinoCatalogName(metalake,
catalogName));
+ catalogConnectorManager.getCatalogConnector(metalake, catalogName);
if (catalogConnector == null) {
boolean dropped =
catalogConnectorManager.getMetalake(metalake).dropCatalog(catalogName, true);
@@ -126,8 +131,7 @@ public class DropCatalogStoredProcedure extends
GravitinoStoredProcedure {
catalogConnectorManager.loadMetalakeSync();
- if (catalogConnectorManager.catalogConnectorExist(
- catalogConnectorManager.getTrinoCatalogName(metalake, catalogName)))
{
+ if (catalogConnectorManager.getCatalogConnector(metalake, catalogName)
!= null) {
throw new TrinoException(
GravitinoErrorCode.GRAVITINO_OPERATION_FAILED,
"Drop catalog failed due to the reloading process fails");
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/GravitinoStoredProcedure.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/GravitinoStoredProcedure.java
index 487cecf4ce..045ce9f5cf 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/GravitinoStoredProcedure.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/GravitinoStoredProcedure.java
@@ -18,11 +18,18 @@
*/
package org.apache.gravitino.trino.connector.system.storedprocedure;
+import io.trino.spi.TrinoException;
import io.trino.spi.procedure.Procedure;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.trino.connector.GravitinoErrorCode;
/** Gravitino System stored procedure interfaces */
public abstract class GravitinoStoredProcedure {
+ /** The name of the optional argument selecting the metalake a procedure
operates on. */
+ protected static final String METALAKE_ARGUMENT = "METALAKE";
+
/**
* Return the definition of the stored procedure.
*
@@ -30,4 +37,26 @@ public abstract class GravitinoStoredProcedure {
* @throws Exception if creation of the procedure fails
*/
public abstract Procedure createStoredProcedure() throws Exception;
+
+ /**
+ * Resolves the metalake a procedure call operates on: the METALAKE argument
when given, otherwise
+ * the metalake the connector is configured with.
+ *
+ * @param configuredMetalake the metalake from {@code gravitino.metalake},
null when unset
+ * @param argument the METALAKE argument of the call, null when not passed
+ * @return the metalake name
+ * @throws TrinoException if neither is available
+ */
+ protected static String resolveMetalake(
+ @Nullable String configuredMetalake, @Nullable String argument) {
+ if (StringUtils.isNotBlank(argument)) {
+ return argument.trim();
+ }
+ if (configuredMetalake != null) {
+ return configuredMetalake;
+ }
+ throw new TrinoException(
+ GravitinoErrorCode.GRAVITINO_MISSING_CONFIG,
+ "No metalake specified: pass the METALAKE argument or set
gravitino.metalake");
+ }
}
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/GravitinoStoredProcedureFactory.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/GravitinoStoredProcedureFactory.java
index 7b920b6155..dae7c70c85 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/GravitinoStoredProcedureFactory.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/GravitinoStoredProcedureFactory.java
@@ -24,6 +24,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
+import javax.annotation.Nullable;
import org.apache.gravitino.trino.connector.GravitinoErrorCode;
import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager;
@@ -31,7 +32,7 @@ import
org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager;
public class GravitinoStoredProcedureFactory {
private final CatalogConnectorManager catalogConnectorManager;
- private final String metalake;
+ @Nullable private final String configuredMetalake;
/** List of all registered Gravitino stored procedures */
public final List<GravitinoStoredProcedure> procedures = new ArrayList<>();
@@ -40,21 +41,22 @@ public class GravitinoStoredProcedureFactory {
* Constructs a new GravitinoStoredProcedureFactory.
*
* @param catalogConnectorManager the catalog connector manager
- * @param metalake the metalake name
+ * @param configuredMetalake the metalake name, or null when the connector
is not configured with
+ * one
*/
public GravitinoStoredProcedureFactory(
- CatalogConnectorManager catalogConnectorManager, String metalake) {
+ CatalogConnectorManager catalogConnectorManager, @Nullable String
configuredMetalake) {
this.catalogConnectorManager = catalogConnectorManager;
- this.metalake = metalake;
+ this.configuredMetalake = configuredMetalake;
registerStoredProcedure();
}
/** Register all the stored procedures * */
private void registerStoredProcedure() {
- procedures.add(new CreateCatalogStoredProcedure(catalogConnectorManager,
metalake));
- procedures.add(new DropCatalogStoredProcedure(catalogConnectorManager,
metalake));
- procedures.add(new AlterCatalogStoredProcedure(catalogConnectorManager,
metalake));
+ procedures.add(new CreateCatalogStoredProcedure(catalogConnectorManager,
configuredMetalake));
+ procedures.add(new DropCatalogStoredProcedure(catalogConnectorManager,
configuredMetalake));
+ procedures.add(new AlterCatalogStoredProcedure(catalogConnectorManager,
configuredMetalake));
}
/**
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableCatalog.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableCatalog.java
index 55e4376ee9..07e8550ff2 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableCatalog.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableCatalog.java
@@ -32,6 +32,7 @@ import io.trino.spi.connector.SchemaTableName;
import java.util.ArrayList;
import java.util.List;
import java.util.TreeMap;
+import javax.annotation.Nullable;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.client.GravitinoMetalake;
import org.apache.gravitino.trino.connector.GravitinoErrorCode;
@@ -53,19 +54,20 @@ public class GravitinoSystemTableCatalog extends
GravitinoSystemTable {
List.of(
ColumnMetadata.builder().setName("name").setType(VARCHAR).build(),
ColumnMetadata.builder().setName("provider").setType(VARCHAR).build(),
-
ColumnMetadata.builder().setName("properties").setType(VARCHAR).build()));
+
ColumnMetadata.builder().setName("properties").setType(VARCHAR).build(),
+
ColumnMetadata.builder().setName("metalake").setType(VARCHAR).build()));
private final CatalogConnectorManager catalogConnectorManager;
- private final String metalake;
+ @Nullable private final String metalake;
/**
* Constructs a new GravitinoSystemTableCatalog.
*
* @param catalogConnectorManager the manager for catalog connectors
- * @param metalake the metalake to report on
+ * @param metalake the metalake to report on, or null for every metalake
*/
public GravitinoSystemTableCatalog(
- CatalogConnectorManager catalogConnectorManager, String metalake) {
+ CatalogConnectorManager catalogConnectorManager, @Nullable String
metalake) {
this.catalogConnectorManager = catalogConnectorManager;
this.metalake = metalake;
}
@@ -75,17 +77,19 @@ public class GravitinoSystemTableCatalog extends
GravitinoSystemTable {
List<GravitinoCatalog> gravitinoCatalogs = new ArrayList<>();
// retrieve catalogs form the Gravitino server with the configuration
metalakes,
// the catalogConnectorManager does not manager catalogs in worker nodes
- // Only the metalake this connector is configured with: the manager is
shared by every entry
- // catalog in this Trino.
+ // The manager is shared by every entry catalog in this Trino, so narrow
to the metalake this
+ // connector is configured with; without one, report every metalake.
catalogConnectorManager.getUsedMetalakes().stream()
- .filter(metalake::equals)
+ .filter(metalakeName -> metalake == null ||
metalake.equals(metalakeName))
.forEach(
(metalakeName) -> {
- GravitinoMetalake metalake =
catalogConnectorManager.getMetalake(metalakeName);
- Catalog[] catalogs = metalake.listCatalogsInfo();
+ GravitinoMetalake gravitinoMetalake =
+ catalogConnectorManager.getMetalake(metalakeName);
+ Catalog[] catalogs = gravitinoMetalake.listCatalogsInfo();
for (Catalog catalog : catalogs) {
// Must match against the same Trino-qualified name the load
loop skips against
- // (quoted "metalake.catalog" in multi-metalake mode), or a
skip pattern written
+ // (quoted "metalake.catalog" when catalog names carry the
metalake), or a skip
+ // pattern written
// against the qualified name never matches here.
String trinoCatalogName =
catalogConnectorManager.getTrinoCatalogName(metalakeName,
catalog.name());
@@ -102,6 +106,7 @@ public class GravitinoSystemTableCatalog extends
GravitinoSystemTable {
BlockBuilder nameColumnBuilder = VARCHAR.createBlockBuilder(null, size);
BlockBuilder providerColumnBuilder = VARCHAR.createBlockBuilder(null,
size);
BlockBuilder propertyColumnBuilder = VARCHAR.createBlockBuilder(null,
size);
+ BlockBuilder metalakeColumnBuilder = VARCHAR.createBlockBuilder(null,
size);
for (GravitinoCatalog catalog : gravitinoCatalogs) {
Preconditions.checkArgument(catalog != null, "catalog should not be
null");
@@ -116,12 +121,14 @@ public class GravitinoSystemTableCatalog extends
GravitinoSystemTable {
throw new TrinoException(
GravitinoErrorCode.GRAVITINO_ILLEGAL_ARGUMENT, "Invalid property
format", e); //
}
+ VARCHAR.writeString(metalakeColumnBuilder, catalog.getMetalake());
}
return new Page(
size,
nameColumnBuilder.build(),
providerColumnBuilder.build(),
- propertyColumnBuilder.build());
+ propertyColumnBuilder.build(),
+ metalakeColumnBuilder.build());
}
@Override
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableCatalogStatus.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableCatalogStatus.java
index de4ed697c7..9f3e4d9e3d 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableCatalogStatus.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableCatalogStatus.java
@@ -27,6 +27,7 @@ import io.trino.spi.connector.ColumnMetadata;
import io.trino.spi.connector.ConnectorTableMetadata;
import io.trino.spi.connector.SchemaTableName;
import java.util.List;
+import javax.annotation.Nullable;
import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager;
import org.apache.gravitino.trino.connector.catalog.CatalogRegistrationState;
@@ -57,16 +58,16 @@ public class GravitinoSystemTableCatalogStatus extends
GravitinoSystemTable {
ColumnMetadata.builder().setName("failure_count").setType(BIGINT).build()));
private final CatalogConnectorManager catalogConnectorManager;
- private final String metalake;
+ @Nullable private final String metalake;
/**
* Constructs a new GravitinoSystemTableCatalogStatus.
*
* @param catalogConnectorManager the manager for catalog connectors
- * @param metalake the metalake to report on
+ * @param metalake the metalake to report on, or null for every metalake
*/
public GravitinoSystemTableCatalogStatus(
- CatalogConnectorManager catalogConnectorManager, String metalake) {
+ CatalogConnectorManager catalogConnectorManager, @Nullable String
metalake) {
this.catalogConnectorManager = catalogConnectorManager;
this.metalake = metalake;
}
@@ -76,7 +77,7 @@ public class GravitinoSystemTableCatalogStatus extends
GravitinoSystemTable {
// Take a snapshot first, the load loop writes these states concurrently
and the column
// builders must all end up with the same number of positions.
// The load loop is shared by every entry catalog in this Trino, so report
only the metalake
- // this connector is configured with.
+ // this connector is configured with, or every metalake when none is
configured.
List<CatalogRegistrationState> states =
catalogConnectorManager.getCatalogRegistrationStates(metalake);
int size = states.size();
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableFactory.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableFactory.java
index 4b41c8d623..8ee9254832 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableFactory.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableFactory.java
@@ -25,13 +25,14 @@ import io.trino.spi.connector.SchemaTableName;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import javax.annotation.Nullable;
import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager;
/** This class managed all the system tables */
public class GravitinoSystemTableFactory {
private final CatalogConnectorManager catalogConnectorManager;
- private final String metalake;
+ @Nullable private final String metalake;
// Per instance, not static: the tables are bound to one
CatalogConnectorManager, and only the
// manager on the coordinator runs the load loop that fills in the
registration state. A shared
@@ -43,10 +44,11 @@ public class GravitinoSystemTableFactory {
*
* @param catalogConnectorManager the manager for catalog connectors
* @param metalake the metalake this connector is configured with; the
tables only report on it,
- * so that two entry catalogs pointed at different metalakes do not
report each other's state
+ * so that two entry catalogs pointed at different metalakes do not
report each other's state.
+ * Null when no metalake is configured, the tables then report on every
metalake
*/
public GravitinoSystemTableFactory(
- CatalogConnectorManager catalogConnectorManager, String metalake) {
+ CatalogConnectorManager catalogConnectorManager, @Nullable String
metalake) {
this.catalogConnectorManager = catalogConnectorManager;
this.metalake = metalake;
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableLoadStatus.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableLoadStatus.java
index df87741e35..34e11fb30b 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableLoadStatus.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableLoadStatus.java
@@ -33,6 +33,7 @@ import io.trino.spi.connector.SchemaTableName;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
+import javax.annotation.Nullable;
import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager;
import
org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager.LoadOutcome;
@@ -64,16 +65,16 @@ public class GravitinoSystemTableLoadStatus extends
GravitinoSystemTable {
ColumnMetadata.builder().setName("metalake_errors").setType(VARCHAR).build()));
private final CatalogConnectorManager catalogConnectorManager;
- private final String metalake;
+ @Nullable private final String metalake;
/**
* Constructs a new GravitinoSystemTableLoadStatus.
*
* @param catalogConnectorManager the manager for catalog connectors
- * @param metalake the metalake to report errors for
+ * @param metalake the metalake to report errors for, or null for every
metalake
*/
public GravitinoSystemTableLoadStatus(
- CatalogConnectorManager catalogConnectorManager, String metalake) {
+ CatalogConnectorManager catalogConnectorManager, @Nullable String
metalake) {
this.catalogConnectorManager = catalogConnectorManager;
this.metalake = metalake;
}
@@ -99,10 +100,16 @@ public class GravitinoSystemTableLoadStatus extends
GravitinoSystemTable {
writeNullableString(lastErrorColumnBuilder, loadOutcome.getLastError());
// The load loop itself is shared by every entry catalog, so the columns
above are global.
- // Only the per metalake errors are narrowed to the metalake this
connector reports on.
+ // Only the per metalake errors are narrowed to the metalake this
connector reports on; without
+ // a configured metalake every metalake's error is shown.
Map<String, String> allErrors = loadOutcome.getMetalakeErrors();
- Map<String, String> metalakeErrors =
- allErrors.containsKey(metalake) ? Map.of(metalake,
allErrors.get(metalake)) : Map.of();
+ Map<String, String> metalakeErrors;
+ if (metalake == null) {
+ metalakeErrors = allErrors;
+ } else {
+ metalakeErrors =
+ allErrors.containsKey(metalake) ? Map.of(metalake,
allErrors.get(metalake)) : Map.of();
+ }
if (metalakeErrors.isEmpty()) {
metalakeErrorsColumnBuilder.appendNull();
} else {
diff --git
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConfig.java
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConfig.java
index 0c55d5a26e..78a8431da6 100644
---
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConfig.java
+++
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConfig.java
@@ -18,7 +18,7 @@
*/
package org.apache.gravitino.trino.connector;
-import static
org.apache.gravitino.trino.connector.GravitinoErrorCode.GRAVITINO_MISSING_CONFIG;
+import static
org.apache.gravitino.trino.connector.GravitinoErrorCode.GRAVITINO_ILLEGAL_ARGUMENT;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -57,17 +57,92 @@ public class TestGravitinoConfig {
}
@Test
- public void testMissingConfig() {
+ public void testMetalakeIsOptional() {
String gravitinoUrl = "http://127.0.0.1:8000";;
- ImmutableMap<String, String> configMap = ImmutableMap.of("gravitino.uri",
gravitinoUrl);
- try {
- GravitinoConfig config = new GravitinoConfig(configMap);
- assertEquals(gravitinoUrl, config.getURI());
- } catch (TrinoException e) {
- if (!GRAVITINO_MISSING_CONFIG.toErrorCode().equals(e.getErrorCode())) {
- throw e;
- }
- }
+ GravitinoConfig config = new
GravitinoConfig(ImmutableMap.of("gravitino.uri", gravitinoUrl));
+
+ assertEquals(gravitinoUrl, config.getURI());
+ assertFalse(config.hasMetalake());
+ assertEquals("", config.getMetalake());
+ // A blank value is the same as an unset one.
+ assertFalse(new GravitinoConfig(ImmutableMap.of("gravitino.metalake", "
")).hasMetalake());
+ assertTrue(new GravitinoConfig(ImmutableMap.of("gravitino.metalake",
"test")).hasMetalake());
+ }
+
+ @Test
+ public void testLoadAllMetalakes() {
+ // Without a metalake every metalake is loaded, whatever the naming mode
says.
+ assertTrue(new GravitinoConfig(ImmutableMap.of()).loadAllMetalakes());
+ assertTrue(
+ new
GravitinoConfig(ImmutableMap.of("gravitino.catalog-name-with-metalake", "true"))
+ .loadAllMetalakes());
+ // With a metalake only the qualified naming mode loads every metalake.
+ assertFalse(
+ new GravitinoConfig(ImmutableMap.of("gravitino.metalake",
"test")).loadAllMetalakes());
+ assertTrue(
+ new GravitinoConfig(
+ ImmutableMap.of(
+ "gravitino.metalake", "test",
"gravitino.catalog-name-with-metalake", "true"))
+ .loadAllMetalakes());
+ }
+
+ @Test
+ public void testCatalogNameWithMetalake() {
+ assertFalse(new
GravitinoConfig(ImmutableMap.of()).catalogNameWithMetalake());
+ assertTrue(
+ new
GravitinoConfig(ImmutableMap.of("gravitino.catalog-name-with-metalake", "true"))
+ .catalogNameWithMetalake());
+ // The deprecated key still works when the new one is unset ...
+ assertTrue(
+ new GravitinoConfig(ImmutableMap.of("gravitino.use-single-metalake",
"false"))
+ .catalogNameWithMetalake());
+ assertFalse(
+ new GravitinoConfig(ImmutableMap.of("gravitino.use-single-metalake",
"true"))
+ .catalogNameWithMetalake());
+ // ... and is ignored once the new one is set.
+ assertFalse(
+ new GravitinoConfig(
+ ImmutableMap.of(
+ "gravitino.use-single-metalake", "false",
+ "gravitino.catalog-name-with-metalake", "false"))
+ .catalogNameWithMetalake());
+
+ assertTrue(
+ new GravitinoConfig(
+ ImmutableMap.of(
+ "gravitino.use-single-metalake", "true",
+ "gravitino.catalog-name-with-metalake", "true"))
+ .catalogNameWithMetalake());
+
+ TrinoException error =
+ assertThrows(
+ TrinoException.class,
+ () ->
+ new
GravitinoConfig(ImmutableMap.of("gravitino.catalog-name-with-metalake", "yes"))
+ .catalogNameWithMetalake());
+ assertEquals(GRAVITINO_ILLEGAL_ARGUMENT.toErrorCode(),
error.getErrorCode());
+ // The deprecated key is parsed just as strictly: a typo must not flip the
naming mode.
+ error =
+ assertThrows(
+ TrinoException.class,
+ () ->
+ new
GravitinoConfig(ImmutableMap.of("gravitino.use-single-metalake", "yes"))
+ .catalogNameWithMetalake());
+ assertEquals(GRAVITINO_ILLEGAL_ARGUMENT.toErrorCode(),
error.getErrorCode());
+ }
+
+ @Test
+ public void testUsesDeprecatedSingleMetalakeKey() {
+ assertFalse(new
GravitinoConfig(ImmutableMap.of()).usesDeprecatedSingleMetalakeKey());
+ assertTrue(
+ new GravitinoConfig(ImmutableMap.of("gravitino.use-single-metalake",
"true"))
+ .usesDeprecatedSingleMetalakeKey());
+ assertFalse(
+ new GravitinoConfig(
+ ImmutableMap.of(
+ "gravitino.use-single-metalake", "false",
+ "gravitino.catalog-name-with-metalake", "true"))
+ .usesDeprecatedSingleMetalakeKey());
}
@Test
@@ -179,6 +254,28 @@ public class TestGravitinoConfig {
assertTrue(catalogConfig.contains("\"gravitino.user\"='admin'"));
}
+ @Test
+ public void testToCatalogConfigIncludesScopedIcebergRestUris() {
+ GravitinoConfig config =
+ new GravitinoConfig(
+ ImmutableMap.of(
+ "gravitino.iceberg.rest-uri",
+ "http://default-irc:9001/iceberg";,
+ "gravitino.iceberg.rest-uri.prod",
+ "http://prod-irc:9001/iceberg";,
+ "gravitino.iceberg.rest-uri.dev",
+ "http://dev-irc:9001/iceberg";));
+
+ String catalogConfig = config.toCatalogConfig();
+ assertTrue(
+
catalogConfig.contains("\"gravitino.iceberg.rest-uri\"='http://default-irc:9001/iceberg'"));
+ assertTrue(
+ catalogConfig.contains(
+
"\"gravitino.iceberg.rest-uri.prod\"='http://prod-irc:9001/iceberg'"));
+ assertTrue(
+
catalogConfig.contains("\"gravitino.iceberg.rest-uri.dev\"='http://dev-irc:9001/iceberg'"));
+ }
+
@Test
public void testTrinoJdbcConfigDefaults() {
GravitinoConfig config =
@@ -355,8 +452,9 @@ public class TestGravitinoConfig {
"gravitino.iceberg.rest-catalog.security", "OAUTH2",
"gravitino.iceberg.rest-catalog.oauth2.credential",
"client_id:client_secret"));
- // The unscoped URI is honored as-is in single-metalake mode.
+ // The unscoped URI is the default for every metalake.
assertEquals("http://127.0.0.1:9001/iceberg";,
config.getManualIcebergRestUri("user_001"));
+ assertEquals("http://127.0.0.1:9001/iceberg";,
config.getManualIcebergRestUri("user_002"));
Map<String, String> restCatalogConfig =
config.getIcebergRestCatalogConfig();
assertEquals(2, restCatalogConfig.size());
@@ -439,24 +537,25 @@ public class TestGravitinoConfig {
}
@Test
- public void testIcebergRestConfigScopedToMetalakeInMultiMetalakeMode() {
- ImmutableMap<String, String> configMap =
- ImmutableMap.of(
- "gravitino.metalake",
- "metalake_a",
- "gravitino.use-single-metalake",
- "false",
- "gravitino.iceberg.rest-uri",
- "http://unscoped:9001/iceberg";,
- "gravitino.iceberg.rest-uri.metalake_a",
- "http://metalake-a:9001/iceberg";);
- GravitinoConfig config = new GravitinoConfig(configMap);
-
- // The scoped key wins for the metalake it names.
+ public void testScopedIcebergRestUriOverridesUnscopedPerMetalake() {
+ GravitinoConfig config =
+ new GravitinoConfig(
+ ImmutableMap.of(
+ "gravitino.catalog-name-with-metalake",
+ "true",
+ "gravitino.iceberg.rest-uri",
+ "http://unscoped:9001/iceberg";,
+ "gravitino.iceberg.rest-uri.metalake_a",
+ "http://metalake-a:9001/iceberg";));
+
+ // The scoped key wins for the metalake it names, the others fall back to
the unscoped one,
+ // whether or not gravitino.metalake is configured.
assertEquals("http://metalake-a:9001/iceberg";,
config.getManualIcebergRestUri("metalake_a"));
- // The unscoped key is ignored in multi-metalake mode, since it would
otherwise misroute every
- // metalake other than the one the Iceberg REST server actually serves.
- assertEquals("", config.getManualIcebergRestUri("metalake_b"));
+ assertEquals("http://unscoped:9001/iceberg";,
config.getManualIcebergRestUri("metalake_b"));
+ assertEquals(
+ "",
+ new GravitinoConfig(ImmutableMap.of("gravitino.metalake",
"metalake_a"))
+ .getManualIcebergRestUri("metalake_a"));
}
@Test
diff --git
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorFactoryStart.java
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorFactoryStart.java
index 7123554efe..6500bc8200 100644
---
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorFactoryStart.java
+++
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnectorFactoryStart.java
@@ -108,6 +108,17 @@ public class TestGravitinoConnectorFactoryStart {
assertTrue(factory.isCatalogConnectorManagerStartTriggered());
}
+ @Test
+ public void testStaticConnectorStartsWithoutMetalake() {
+ CoordinatorFactory factory = newFactory();
+ Map<String, String> config = staticConfig();
+ config.remove("gravitino.metalake");
+
+ // Without a metalake the entry catalog serves every metalake instead of
refusing to start.
+ assertNotNull(factory.create("gravitino", config, mockContext()));
+ assertTrue(factory.isCatalogConnectorManagerStartTriggered());
+ }
+
@Test
public void testDynamicConnectorDoesNotStartTheManager() {
CoordinatorFactory factory = newFactory();
diff --git
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogConnectorManager.java
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogConnectorManager.java
index acf389e7c0..8bea849ee3 100644
---
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogConnectorManager.java
+++
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogConnectorManager.java
@@ -26,6 +26,7 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -113,7 +114,7 @@ public class TestCatalogConnectorManager {
"test1", otherConnectorConfig, mockContext()));
assertEquals(
GravitinoErrorCode.GRAVITINO_UNSUPPORTED_OPERATION.toErrorCode(),
error.getErrorCode());
- assertTrue(error.getMessage().contains("Multiple metalakes are not
supported"));
+ assertTrue(error.getMessage().contains("belongs to metalake test2"),
error.getMessage());
}
@Test
@@ -683,6 +684,185 @@ public class TestCatalogConnectorManager {
assertTrue(manager.getUsedMetalakes().isEmpty());
}
+ @Test
+ public void testUnsetMetalakeLoadsEveryMetalake() throws Exception {
+ LoadFixture fixture = new LoadFixture().withoutMetalake();
+ fixture.withCatalogs(mockCatalog("memory", "memory",
Catalog.Type.RELATIONAL));
+ fixture.withSecondMetalake("dev", mockCatalog("sandbox", "memory",
Catalog.Type.RELATIONAL));
+
+ // The default naming mode, which loads only the configured metalake when
there is one.
+ CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of());
+ manager.loadMetalakeSync();
+
+ verify(fixture.client, times(1)).listMetalakes();
+ verify(fixture.client, never()).loadMetalake(any());
+ // Unqualified names: the naming mode is independent from which metalakes
are loaded.
+ verify(fixture.catalogRegister, times(1)).registerCatalog(eq("memory"),
any());
+ verify(fixture.catalogRegister, times(1)).registerCatalog(eq("sandbox"),
any());
+ assertEquals(ImmutableSet.of("test", "dev"),
ImmutableSet.copyOf(manager.getUsedMetalakes()));
+ assertEquals(2, manager.getCatalogRegistrationStates().size());
+ assertEquals(2, manager.getCatalogRegistrationStates(null).size());
+ assertEquals(1, manager.getCatalogRegistrationStates("dev").size());
+ assertNull(manager.getLoadOutcome().getLastError());
+ }
+
+ @Test
+ public void testConfiguredMetalakeStillLoadsOnlyItself() throws Exception {
+ LoadFixture fixture = new LoadFixture();
+ fixture.withCatalogs(mockCatalog("memory", "memory",
Catalog.Type.RELATIONAL));
+ fixture.withSecondMetalake("dev", mockCatalog("sandbox", "memory",
Catalog.Type.RELATIONAL));
+
+ CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of());
+ manager.loadMetalakeSync();
+
+ verify(fixture.client, never()).listMetalakes();
+ verify(fixture.catalogRegister, never()).registerCatalog(eq("sandbox"),
any());
+ assertEquals(ImmutableSet.of("test"),
ImmutableSet.copyOf(manager.getUsedMetalakes()));
+ }
+
+ @Test
+ public void testSameCatalogNameInAnotherMetalakeIsReportedNotRegistered()
throws Exception {
+ LoadFixture fixture = new LoadFixture().withoutMetalake();
+ Catalog catalog = mockCatalog("memory", "memory", Catalog.Type.RELATIONAL);
+ fixture.withCatalogs(catalog);
+ fixture.withSecondMetalake("dev", mockCatalog("memory", "memory",
Catalog.Type.RELATIONAL));
+
+ CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of());
+ // Built from the same definition the server reports, so that the next
cycle sees it as up to
+ // date. Built outside the answer below: touching the catalog mock while a
stubbing is in
+ // progress would break that stubbing.
+ GravitinoCatalog registered = new GravitinoCatalog("test", catalog);
+ // What Trino does while CREATE CATALOG runs: the first registration
creates the connector, so
+ // the manager knows which metalake owns the Trino catalog name from then
on.
+ Mockito.doAnswer(
+ invocation -> {
+ CatalogConnectorContext context =
+ manager.createCatalogConnectorContext(
+ "memory",
+ createConnectorConfig(catalogConfigJson("test",
"memory")),
+ mockContext());
+ when(context.getCatalog()).thenReturn(registered);
+ return null;
+ })
+ .when(fixture.catalogRegister)
+ .registerCatalog(eq("memory"), any());
+ manager.loadMetalakeSync();
+
+ // Registered once, by the metalake listed first.
+ verify(fixture.catalogRegister, times(1)).registerCatalog(eq("memory"),
any());
+ verify(fixture.catalogRegister, never()).unregisterCatalog(any());
+ assertEquals(2, manager.getCatalogRegistrationStates().size());
+ CatalogRegistrationState winner =
manager.getCatalogRegistrationStates("test").get(0);
+ assertEquals(CatalogRegistrationState.Status.REGISTERED,
winner.getStatus());
+ CatalogRegistrationState loser =
manager.getCatalogRegistrationStates("dev").get(0);
+ assertEquals(CatalogRegistrationState.Status.FAILED, loser.getStatus());
+ assertEquals("memory", loser.getTrinoCatalogName());
+ assertTrue(
+ loser.getLastError().contains("already registered by metalake test"),
loser.getLastError());
+ assertTrue(manager.describeRegistrationFailure("dev",
"memory").contains("already registered"));
+ // The other metalake's failure does not fail the run as a whole.
+ assertTrue(manager.getMetalakeErrors().isEmpty());
+
+ // The next cycle neither refreshes the first over the second nor
unregisters it.
+ manager.loadMetalakeSync();
+ verify(fixture.catalogRegister, times(1)).registerCatalog(eq("memory"),
any());
+ verify(fixture.catalogRegister, never()).unregisterCatalog(any());
+ assertEquals(
+ CatalogRegistrationState.Status.REGISTERED,
+ manager.getCatalogRegistrationStates("test").get(0).getStatus());
+ assertEquals(
+ CatalogRegistrationState.Status.FAILED,
+ manager.getCatalogRegistrationStates("dev").get(0).getStatus());
+ }
+
+ @Test
+ public void testLoserStateIsPrunedWhenItsCatalogIsDeleted() throws Exception
{
+ LoadFixture fixture = new LoadFixture().withoutMetalake();
+ Catalog catalog = mockCatalog("memory", "memory", Catalog.Type.RELATIONAL);
+ fixture.withCatalogs(catalog);
+ GravitinoMetalake dev =
+ fixture.withSecondMetalake("dev", mockCatalog("memory", "memory",
Catalog.Type.RELATIONAL));
+ CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of());
+ String[] owner = {"test"};
+ registerOwnerOnCreate(fixture, manager, owner, catalog);
+ manager.loadMetalakeSync();
+ assertEquals(
+ CatalogRegistrationState.Status.FAILED,
+ manager.getCatalogRegistrationStates("dev").get(0).getStatus());
+
+ // The losing catalog is deleted in Gravitino. The winner still holds the
Trino name, which
+ // must not keep the loser's row alive.
+ Mockito.doReturn(new String[0]).when(dev).listCatalogs();
+ manager.loadMetalakeSync();
+
+ assertTrue(manager.getCatalogRegistrationStates("dev").isEmpty());
+ assertEquals(
+ CatalogRegistrationState.Status.REGISTERED,
+ manager.getCatalogRegistrationStates("test").get(0).getStatus());
+ verify(fixture.catalogRegister, never()).unregisterCatalog(any());
+
+ // The same holds when the losing metalake disappears altogether.
+ Mockito.doReturn(new GravitinoMetalake[] {fixture.metalake})
+ .when(fixture.client)
+ .listMetalakes();
+ manager.loadMetalakeSync();
+ assertTrue(manager.getCatalogRegistrationStates("dev").isEmpty());
+ assertEquals(1, manager.getCatalogRegistrationStates().size());
+ verify(fixture.catalogRegister, never()).unregisterCatalog(any());
+ }
+
+ @Test
+ public void testLoserTakesOverWhenWinnerIsDropped() throws Exception {
+ LoadFixture fixture = new LoadFixture().withoutMetalake();
+ Catalog catalog = mockCatalog("memory", "memory", Catalog.Type.RELATIONAL);
+ fixture.withCatalogs(catalog);
+ fixture.withSecondMetalake("dev", catalog);
+ CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of());
+ String[] owner = {"test"};
+ registerOwnerOnCreate(fixture, manager, owner, catalog);
+ manager.loadMetalakeSync();
+ assertNotNull(manager.getCatalogConnector("test", "memory"));
+ assertNull(manager.getCatalogConnector("dev", "memory"));
+
+ // The winning catalog is deleted in Gravitino, which frees the Trino name
for the loser.
+ Mockito.doReturn(new String[0]).when(fixture.metalake).listCatalogs();
+ owner[0] = "dev";
+ manager.loadMetalakeSync();
+
+ verify(fixture.catalogRegister, times(1)).unregisterCatalog("memory");
+ verify(fixture.catalogRegister, times(2)).registerCatalog(eq("memory"),
any());
+ assertTrue(manager.getCatalogRegistrationStates("test").isEmpty());
+ assertEquals(
+ CatalogRegistrationState.Status.REGISTERED,
+ manager.getCatalogRegistrationStates("dev").get(0).getStatus());
+ assertNull(manager.getCatalogConnector("test", "memory"));
+ assertNotNull(manager.getCatalogConnector("dev", "memory"));
+ }
+
+ // What Trino does while CREATE CATALOG runs: registering creates the
connector, whose owning
+ // metalake is whatever owner[0] names at that time.
+ private static void registerOwnerOnCreate(
+ LoadFixture fixture, CatalogConnectorManager manager, String[] owner,
Catalog catalog) {
+ Mockito.doAnswer(
+ invocation -> {
+ String metalake = owner[0];
+ // Built before the stubbing: touching the catalog mock while a
stubbing is in
+ // progress would break that stubbing.
+ GravitinoCatalog registered = new GravitinoCatalog(metalake,
catalog);
+ CatalogConnectorContext context =
+ manager.createCatalogConnectorContext(
+ "memory",
+ createConnectorConfig(catalogConfigJson(metalake,
"memory")),
+ mockContext());
+ GravitinoMetalake handle = manager.getMetalake(metalake);
+ when(context.getCatalog()).thenReturn(registered);
+ when(context.getMetalake()).thenReturn(handle);
+ return null;
+ })
+ .when(fixture.catalogRegister)
+ .registerCatalog(eq("memory"), any());
+ }
+
@Test
public void testCatalogStatesAreFilteredPerMetalake() throws Exception {
LoadFixture fixture = new LoadFixture();
@@ -1276,8 +1456,7 @@ public class TestCatalogConnectorManager {
CatalogConnectorFactory catalogFactory, ImmutableMap<String, String>
configMap) {
CatalogRegister catalogRegister = mock(CatalogRegister.class);
- boolean singleMetalakeMode =
- configMap.getOrDefault("gravitino.use-single-metalake",
"true").equals("true");
+ boolean singleMetalakeMode = !new
GravitinoConfig(configMap).catalogNameWithMetalake();
CatalogConnectorManager manager =
new CatalogConnectorManager(
catalogRegister,
@@ -1355,6 +1534,7 @@ public class TestCatalogConnectorManager {
private final GravitinoAdminClient client =
mock(GravitinoAdminClient.class);
private final GravitinoMetalake metalake = mock(GravitinoMetalake.class);
private final CatalogConnectorFactory catalogFactory =
mock(CatalogConnectorFactory.class);
+ private boolean metalakeConfigured = true;
LoadFixture() throws Exception {
when(catalogRegister.isTrinoReachable()).thenReturn(true);
@@ -1369,6 +1549,11 @@ public class TestCatalogConnectorManager {
when(builder.build()).thenReturn(mock(CatalogConnectorContext.class));
}
+ LoadFixture withoutMetalake() {
+ metalakeConfigured = false;
+ return this;
+ }
+
void withCatalogs(Catalog... catalogs) {
String[] names = new String[catalogs.length];
for (int i = 0; i < catalogs.length; i++) {
@@ -1395,10 +1580,27 @@ public class TestCatalogConnectorManager {
GravitinoConfig config(Map<String, String> extraConfig) {
Map<String, String> defaults = new HashMap<>();
defaults.put("gravitino.uri", "http://127.0.0.1:8090";);
- defaults.put("gravitino.metalake", "test");
+ if (metalakeConfigured) {
+ defaults.put("gravitino.metalake", "test");
+ }
defaults.put("gravitino.use-single-metalake", "true");
defaults.putAll(extraConfig);
return new GravitinoConfig(ImmutableMap.copyOf(defaults));
}
+
+ /** A second metalake the mock client lists after the fixture's own. */
+ GravitinoMetalake withSecondMetalake(String name, Catalog... catalogs) {
+ GravitinoMetalake second = mock(GravitinoMetalake.class);
+ when(second.name()).thenReturn(name);
+ String[] names = new String[catalogs.length];
+ for (int i = 0; i < catalogs.length; i++) {
+ names[i] = catalogs[i].name();
+ Mockito.doReturn(catalogs[i]).when(second).loadCatalog(names[i]);
+ }
+ Mockito.doReturn(names).when(second).listCatalogs();
+ Mockito.doReturn(second).when(client).loadMetalake(name);
+ Mockito.doReturn(new GravitinoMetalake[] {metalake,
second}).when(client).listMetalakes();
+ return second;
+ }
}
}
diff --git
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/storedprocedure/TestAlterCatalogStoredProcedure.java
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/storedprocedure/TestAlterCatalogStoredProcedure.java
new file mode 100644
index 0000000000..5ec6fe7133
--- /dev/null
+++
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/storedprocedure/TestAlterCatalogStoredProcedure.java
@@ -0,0 +1,108 @@
+/*
+ * 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.trino.connector.system.storedprocedure;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableMap;
+import io.trino.spi.TrinoException;
+import java.util.List;
+import org.apache.gravitino.client.GravitinoMetalake;
+import org.apache.gravitino.trino.connector.GravitinoErrorCode;
+import org.apache.gravitino.trino.connector.catalog.CatalogConnectorContext;
+import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager;
+import org.apache.gravitino.trino.connector.metadata.GravitinoCatalog;
+import org.junit.jupiter.api.Test;
+
+/** Unit tests for {@link AlterCatalogStoredProcedure} around the METALAKE
argument. */
+public class TestAlterCatalogStoredProcedure {
+
+ private static final String METALAKE = "test_metalake";
+ private static final String CATALOG = "test_catalog";
+
+ @Test
+ public void testMetalakeArgumentOverridesConfiguredMetalake() {
+ CatalogConnectorManager manager = mock(CatalogConnectorManager.class);
+ CatalogConnectorContext context = mock(CatalogConnectorContext.class);
+ GravitinoMetalake other = mock(GravitinoMetalake.class);
+ when(manager.getTrinoCatalogName(anyString(),
eq(CATALOG))).thenReturn(CATALOG);
+ when(manager.getCatalogConnector("other", CATALOG)).thenReturn(context);
+ when(context.getMetalake()).thenReturn(other);
+ when(context.getCatalog())
+ .thenReturn(new GravitinoCatalog("other", "memory", CATALOG,
ImmutableMap.of(), 1L))
+ .thenReturn(
+ new GravitinoCatalog("other", "memory", CATALOG,
ImmutableMap.of("k", "v"), 2L));
+
+ AlterCatalogStoredProcedure procedure = new
AlterCatalogStoredProcedure(manager, METALAKE);
+
+ assertDoesNotThrow(
+ () -> procedure.alterCatalog(CATALOG, ImmutableMap.of("k", "v"),
List.of(), "other"));
+
+ verify(other, times(1)).alterCatalog(eq(CATALOG), any());
+ verify(manager, never()).getCatalogConnector(METALAKE, CATALOG);
+ }
+
+ @Test
+ public void testMetalakeArgumentRequiredWithoutConfiguredMetalake() {
+ CatalogConnectorManager manager = mock(CatalogConnectorManager.class);
+ AlterCatalogStoredProcedure procedure = new
AlterCatalogStoredProcedure(manager, null);
+
+ TrinoException error =
+ assertThrows(
+ TrinoException.class,
+ () -> procedure.alterCatalog(CATALOG, ImmutableMap.of("k", "v"),
List.of(), null));
+ assertEquals(GravitinoErrorCode.GRAVITINO_MISSING_CONFIG.toErrorCode(),
error.getErrorCode());
+ verify(manager, never()).getCatalogConnector(anyString(), anyString());
+ }
+
+ @Test
+ public void testTrinoNameHeldByAnotherMetalakeIsNotAltered() {
+ // With unqualified names another metalake's catalog may hold the Trino
name. It is not the
+ // catalog asked for, so nothing is altered and the caller learns why it
is not registered.
+ CatalogConnectorManager manager = mock(CatalogConnectorManager.class);
+ CatalogConnectorContext otherContext = mock(CatalogConnectorContext.class);
+ GravitinoMetalake otherMetalake = mock(GravitinoMetalake.class);
+ when(manager.getTrinoCatalogName(anyString(),
eq(CATALOG))).thenReturn(CATALOG);
+ when(manager.getCatalogConnector(CATALOG)).thenReturn(otherContext);
+ when(manager.getCatalogConnector(METALAKE, CATALOG)).thenReturn(null);
+ when(otherContext.getMetalake()).thenReturn(otherMetalake);
+ when(manager.describeRegistrationFailure(METALAKE, CATALOG))
+ .thenReturn("FAILED: already registered by metalake other");
+
+ AlterCatalogStoredProcedure procedure = new
AlterCatalogStoredProcedure(manager, METALAKE);
+
+ TrinoException error =
+ assertThrows(
+ TrinoException.class,
+ () -> procedure.alterCatalog(CATALOG, ImmutableMap.of("k", "v"),
List.of(), null));
+ assertTrue(error.getMessage().contains("already registered"),
error.getMessage());
+ verify(otherMetalake, never()).alterCatalog(anyString(), any());
+ }
+}
diff --git
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/storedprocedure/TestCreateCatalogStoredProcedure.java
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/storedprocedure/TestCreateCatalogStoredProcedure.java
new file mode 100644
index 0000000000..9a451f386a
--- /dev/null
+++
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/storedprocedure/TestCreateCatalogStoredProcedure.java
@@ -0,0 +1,108 @@
+/*
+ * 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.trino.connector.system.storedprocedure;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableMap;
+import io.trino.spi.TrinoException;
+import java.util.Map;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.client.GravitinoMetalake;
+import org.apache.gravitino.trino.connector.GravitinoErrorCode;
+import org.apache.gravitino.trino.connector.catalog.CatalogConnectorContext;
+import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager;
+import org.junit.jupiter.api.Test;
+
+/** Unit tests for {@link CreateCatalogStoredProcedure} around the METALAKE
argument. */
+public class TestCreateCatalogStoredProcedure {
+
+ private static final String METALAKE = "test_metalake";
+ private static final String CATALOG = "test_catalog";
+ private static final Map<String, String> PROPERTIES = ImmutableMap.of("k",
"v");
+
+ @Test
+ public void testMetalakeArgumentOverridesConfiguredMetalake() {
+ CatalogConnectorManager manager = mock(CatalogConnectorManager.class);
+ GravitinoMetalake other = mock(GravitinoMetalake.class);
+ when(manager.getTrinoCatalogName(anyString(),
eq(CATALOG))).thenReturn(CATALOG);
+ when(manager.getMetalake("other")).thenReturn(other);
+ // Registered by the load loop once created.
+ when(manager.getCatalogConnector("other", CATALOG))
+ .thenReturn(null)
+ .thenReturn(mock(CatalogConnectorContext.class));
+
+ CreateCatalogStoredProcedure procedure = new
CreateCatalogStoredProcedure(manager, METALAKE);
+
+ assertDoesNotThrow(
+ () -> procedure.createCatalog(CATALOG, "memory", PROPERTIES, false,
"other"));
+
+ verify(other, times(1))
+ .createCatalog(CATALOG, Catalog.Type.RELATIONAL, "memory", "Trino
created", PROPERTIES);
+ verify(manager, never()).getMetalake(METALAKE);
+ }
+
+ @Test
+ public void testMetalakeArgumentRequiredWithoutConfiguredMetalake() {
+ CatalogConnectorManager manager = mock(CatalogConnectorManager.class);
+ CreateCatalogStoredProcedure procedure = new
CreateCatalogStoredProcedure(manager, null);
+
+ TrinoException error =
+ assertThrows(
+ TrinoException.class,
+ () -> procedure.createCatalog(CATALOG, "memory", PROPERTIES,
false, null));
+ assertEquals(GravitinoErrorCode.GRAVITINO_MISSING_CONFIG.toErrorCode(),
error.getErrorCode());
+ verify(manager, never()).getMetalake(anyString());
+ }
+
+ @Test
+ public void testTrinoNameHeldByAnotherMetalakeIsNotAnExistingCatalog() {
+ // With unqualified names another metalake's catalog may hold the Trino
name; that is not the
+ // catalog being created, so it is created on the server rather than
reported as existing.
+ CatalogConnectorManager manager = mock(CatalogConnectorManager.class);
+ GravitinoMetalake requested = mock(GravitinoMetalake.class);
+ when(manager.getTrinoCatalogName(anyString(),
eq(CATALOG))).thenReturn(CATALOG);
+
when(manager.getCatalogConnector(CATALOG)).thenReturn(mock(CatalogConnectorContext.class));
+ when(manager.getCatalogConnector(METALAKE, CATALOG)).thenReturn(null);
+ when(manager.getMetalake(METALAKE)).thenReturn(requested);
+ when(manager.describeRegistrationFailure(METALAKE, CATALOG))
+ .thenReturn("FAILED: already registered by metalake other");
+
+ CreateCatalogStoredProcedure procedure = new
CreateCatalogStoredProcedure(manager, METALAKE);
+
+ // Created on the server, but the name stays with the other metalake so
registration fails.
+ TrinoException error =
+ assertThrows(
+ TrinoException.class,
+ () -> procedure.createCatalog(CATALOG, "memory", PROPERTIES,
false, null));
+ verify(requested, times(1)).createCatalog(eq(CATALOG), any(),
eq("memory"), any(), any());
+ assertTrue(error.getMessage().contains("already registered"),
error.getMessage());
+ }
+}
diff --git
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/storedprocedure/TestDropCatalogStoredProcedure.java
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/storedprocedure/TestDropCatalogStoredProcedure.java
index c4b770c16e..cffd2d5a9f 100644
---
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/storedprocedure/TestDropCatalogStoredProcedure.java
+++
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/storedprocedure/TestDropCatalogStoredProcedure.java
@@ -22,7 +22,9 @@ import static
org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
@@ -32,6 +34,7 @@ import static org.mockito.Mockito.when;
import io.trino.spi.TrinoException;
import org.apache.gravitino.client.GravitinoMetalake;
import org.apache.gravitino.trino.connector.GravitinoErrorCode;
+import org.apache.gravitino.trino.connector.catalog.CatalogConnectorContext;
import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager;
import org.junit.jupiter.api.Test;
@@ -45,6 +48,61 @@ public class TestDropCatalogStoredProcedure {
private static final String CATALOG = "test_catalog";
private static final String TRINO_CATALOG = "test_metalake.test_catalog";
+ @Test
+ public void testMetalakeArgumentOverridesConfiguredMetalake() {
+ CatalogConnectorManager manager = mock(CatalogConnectorManager.class);
+ GravitinoMetalake other = mock(GravitinoMetalake.class);
+ when(manager.getTrinoCatalogName("other", CATALOG)).thenReturn("other." +
CATALOG);
+ when(manager.getCatalogConnector("other." + CATALOG)).thenReturn(null);
+ when(manager.getMetalake("other")).thenReturn(other);
+ when(other.dropCatalog(CATALOG, true)).thenReturn(true);
+
+ DropCatalogStoredProcedure procedure = new
DropCatalogStoredProcedure(manager, METALAKE);
+
+ assertDoesNotThrow(() -> procedure.dropCatalog(CATALOG, false, "other"));
+
+ verify(other, times(1)).dropCatalog(CATALOG, true);
+ verify(manager, never()).getMetalake(METALAKE);
+ }
+
+ @Test
+ public void testTrinoNameHeldByAnotherMetalakeIsNotDropped() {
+ // With unqualified names another metalake's catalog may hold the Trino
name. The lookup is by
+ // metalake, so the procedure falls back to the server for the metalake it
was asked for and
+ // the other metalake is never touched.
+ CatalogConnectorManager manager = mock(CatalogConnectorManager.class);
+ CatalogConnectorContext otherContext = mock(CatalogConnectorContext.class);
+ GravitinoMetalake otherMetalake = mock(GravitinoMetalake.class);
+ GravitinoMetalake requested = mock(GravitinoMetalake.class);
+ when(manager.getTrinoCatalogName(anyString(),
eq(CATALOG))).thenReturn(CATALOG);
+ when(manager.getCatalogConnector(CATALOG)).thenReturn(otherContext);
+ when(manager.getCatalogConnector("other",
CATALOG)).thenReturn(otherContext);
+ when(otherContext.getMetalake()).thenReturn(otherMetalake);
+ when(manager.getMetalake(METALAKE)).thenReturn(requested);
+ when(requested.dropCatalog(CATALOG, true)).thenReturn(true);
+
+ DropCatalogStoredProcedure procedure = new
DropCatalogStoredProcedure(manager, METALAKE);
+
+ assertDoesNotThrow(() -> procedure.dropCatalog(CATALOG, false, null));
+
+ verify(requested, times(1)).dropCatalog(CATALOG, true);
+ verify(otherMetalake, never()).dropCatalog(anyString(), anyBoolean());
+ }
+
+ @Test
+ public void testMetalakeArgumentRequiredWithoutConfiguredMetalake() {
+ CatalogConnectorManager manager = mock(CatalogConnectorManager.class);
+ DropCatalogStoredProcedure procedure = new
DropCatalogStoredProcedure(manager, null);
+
+ TrinoException error =
+ assertThrows(TrinoException.class, () ->
procedure.dropCatalog(CATALOG, false, null));
+ assertEquals(GravitinoErrorCode.GRAVITINO_MISSING_CONFIG.toErrorCode(),
error.getErrorCode());
+ assertTrue(error.getMessage().contains("METALAKE"), error.getMessage());
+ // Blank counts as not passed.
+ assertThrows(TrinoException.class, () -> procedure.dropCatalog(CATALOG,
false, " "));
+ verify(manager, never()).getMetalake(anyString());
+ }
+
@Test
public void
testDropCatalogFallsBackToServerWhenLocalCacheMissesAndServerHasIt() {
// The local connector cache does not have the catalog, but the Gravitino
server does.
@@ -59,7 +117,7 @@ public class TestDropCatalogStoredProcedure {
DropCatalogStoredProcedure procedure = new
DropCatalogStoredProcedure(manager, METALAKE);
- assertDoesNotThrow(() -> procedure.dropCatalog(CATALOG, false));
+ assertDoesNotThrow(() -> procedure.dropCatalog(CATALOG, false, null));
verify(metalake, times(1)).dropCatalog(CATALOG, true);
}
@@ -83,7 +141,7 @@ public class TestDropCatalogStoredProcedure {
// The contract we verify is: the user-visible message still tells them
the catalog does not
// exist, and the root cause carries the GRAVITINO_CATALOG_NOT_EXISTS
error code.
TrinoException error =
- assertThrows(TrinoException.class, () ->
procedure.dropCatalog(CATALOG, false));
+ assertThrows(TrinoException.class, () ->
procedure.dropCatalog(CATALOG, false, null));
assertTrue(
error.getMessage().contains("not exists"),
() -> "Expected message to contain 'not exists' but was: " +
error.getMessage());
@@ -108,7 +166,7 @@ public class TestDropCatalogStoredProcedure {
DropCatalogStoredProcedure procedure = new
DropCatalogStoredProcedure(manager, METALAKE);
- assertDoesNotThrow(() -> procedure.dropCatalog(CATALOG, true));
+ assertDoesNotThrow(() -> procedure.dropCatalog(CATALOG, true, null));
// We DID consult the server, but no exception was thrown.
verify(metalake, times(1)).dropCatalog(CATALOG, true);
diff --git
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/storedprocedure/TestGravitinoStoredProcedureFactory.java
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/storedprocedure/TestGravitinoStoredProcedureFactory.java
new file mode 100644
index 0000000000..f1e35d57a7
--- /dev/null
+++
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/storedprocedure/TestGravitinoStoredProcedureFactory.java
@@ -0,0 +1,55 @@
+/*
+ * 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.trino.connector.system.storedprocedure;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Mockito.mock;
+
+import io.trino.spi.procedure.Procedure;
+import java.util.List;
+import java.util.Set;
+import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager;
+import org.junit.jupiter.api.Test;
+
+public class TestGravitinoStoredProcedureFactory {
+
+ @Test
+ public void testProceduresDeclareOptionalMetalakeArgument() {
+ // Trino validates a procedure's arguments against its method handle when
the catalog loads,
+ // so a mismatch surfaces as a failure to load the entry catalog rather
than in a call.
+ for (String metalake : new String[] {"test", null}) {
+ GravitinoStoredProcedureFactory factory =
+ new
GravitinoStoredProcedureFactory(mock(CatalogConnectorManager.class), metalake);
+ Set<Procedure> procedures = factory.getStoredProcedures();
+
+ assertEquals(
+ Set.of("create_catalog", "drop_catalog", "alter_catalog"),
+
Set.of(procedures.stream().map(Procedure::getName).toArray(String[]::new)));
+ for (Procedure procedure : procedures) {
+ List<Procedure.Argument> arguments = procedure.getArguments();
+ Procedure.Argument last = arguments.get(arguments.size() - 1);
+ assertEquals("METALAKE", last.getName(), procedure.getName());
+ assertFalse(last.isRequired(), procedure.getName());
+ assertNull(last.getDefaultValue(), procedure.getName());
+ }
+ }
+ }
+}
diff --git
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/table/TestGravitinoSystemStatusTables.java
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/table/TestGravitinoSystemStatusTables.java
index c8984861e6..0fea4b1992 100644
---
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/table/TestGravitinoSystemStatusTables.java
+++
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/table/TestGravitinoSystemStatusTables.java
@@ -85,6 +85,71 @@ public class TestGravitinoSystemStatusTables {
assertEquals("memory", varchar(page, 0));
}
+ @Test
+ public void testCatalogTableReportsEveryMetalakeWithoutConfiguredMetalake() {
+ CatalogConnectorManager manager = mock(CatalogConnectorManager.class);
+ when(manager.getUsedMetalakes()).thenReturn(Set.of("prod", "dev"));
+ for (String metalakeName : List.of("prod", "dev")) {
+ GravitinoMetalake metalake = mock(GravitinoMetalake.class);
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.name()).thenReturn(metalakeName + "_memory");
+ when(catalog.provider()).thenReturn("memory");
+ when(catalog.type()).thenReturn(Catalog.Type.RELATIONAL);
+ when(catalog.properties()).thenReturn(Map.of());
+ Audit audit = mock(Audit.class);
+ when(audit.createTime()).thenReturn(Instant.now());
+ when(catalog.auditInfo()).thenReturn(audit);
+ when(metalake.listCatalogsInfo()).thenReturn(new Catalog[] {catalog});
+ when(manager.getMetalake(metalakeName)).thenReturn(metalake);
+ when(manager.getTrinoCatalogName(metalakeName, metalakeName + "_memory"))
+ .thenReturn(metalakeName + "_memory");
+ }
+
+ Page page = new GravitinoSystemTableCatalog(manager, null).loadPageData();
+
+ assertEquals(2, page.getPositionCount());
+ assertEquals(
+ Set.of("prod_memory", "dev_memory"), Set.of(varchar(page, 0),
varchar(page, 0, 1)));
+ // The metalake column tells same-named catalogs of different metalakes
apart.
+ for (int position = 0; position < 2; position++) {
+ String name = varchar(page, 0, position);
+ assertEquals(name.substring(0, name.indexOf('_')), varchar(page, 3,
position));
+ }
+ }
+
+ @Test
+ public void
testCatalogStatusTableReportsEveryMetalakeWithoutConfiguredMetalake() {
+ CatalogConnectorManager manager = mock(CatalogConnectorManager.class);
+ List<CatalogRegistrationState> states =
+ List.of(
+ CatalogRegistrationState.failed("prod", "memory", "memory", null,
"boom"),
+ CatalogRegistrationState.failed("dev", "memory", "memory", null,
"boom"));
+ when(manager.getCatalogRegistrationStates(null)).thenReturn(states);
+
+ Page page = new GravitinoSystemTableCatalogStatus(manager,
null).loadPageData();
+
+ assertEquals(2, page.getPositionCount());
+ assertEquals(Set.of("prod", "dev"), Set.of(varchar(page, 0), varchar(page,
0, 1)));
+ }
+
+ @Test
+ public void
testLoadStatusTableReportsEveryMetalakeErrorWithoutConfiguredMetalake() {
+ CatalogConnectorManager manager = mock(CatalogConnectorManager.class);
+ when(manager.getLastLoadAttemptTimeMs()).thenReturn(2000L);
+ when(manager.getLoadOutcome())
+ .thenReturn(
+ new CatalogConnectorManager.LoadOutcome(
+ true,
+ 0L,
+ "1 of 2 metalakes failed",
+ 1L,
+ Map.of("test", "Connection refused", "dev", "Access Denied")));
+
+ Page page = new GravitinoSystemTableLoadStatus(manager,
null).loadPageData();
+
+ assertEquals("{\"dev\":\"Access Denied\",\"test\":\"Connection
refused\"}", varchar(page, 5));
+ }
+
@Test
public void testCatalogStatusTableRendersRegisteredCatalog() {
GravitinoCatalog catalog =
@@ -313,7 +378,11 @@ public class TestGravitinoSystemStatusTables {
}
private static String varchar(Page page, int channel) {
+ return varchar(page, channel, 0);
+ }
+
+ private static String varchar(Page page, int channel, int position) {
Block block = page.getBlock(channel);
- return VARCHAR.getSlice(block, 0).toStringUtf8();
+ return VARCHAR.getSlice(block, position).toStringUtf8();
}
}
