This is an automated email from the ASF dual-hosted git repository. diqiu50 pushed a commit to branch feat/trino-optional-metalake in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit b1c7bd221b1fc3c615dedcc5fbe6f49d54a18f02 Author: yuhui <[email protected]> AuthorDate: Fri Sep 18 12:23:21 2026 +0800 [#13301] feat(trino-connector): Make gravitino.metalake optional Co-Authored-By: Claude Opus 5 <[email protected]> --- docs/trino-connector/catalog-iceberg.md | 7 +- docs/trino-connector/configuration.md | 16 +- 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 | 95 +++++++--- .../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 | 135 ++++++++++--- .../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, 1081 insertions(+), 184 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 cf8e562c31..fa89529206 100644 --- a/docs/trino-connector/configuration.md +++ b/docs/trino-connector/configuration.md @@ -10,7 +10,7 @@ license: "This software is licensed under the Apache License version 2." | 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.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. | No [...] | 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 | @@ -28,9 +28,10 @@ license: "This software is licensed under the Apache License version 2." | 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 440 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.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. | 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-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. | 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 | 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 440-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 440-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 1bf0568086..5f5de24cf7 100644 --- a/docs/trino-connector/installation.md +++ b/docs/trino-connector/installation.md @@ -144,7 +144,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 8c395ff93a..2c9ffc204f 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-password"":""******"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""jdbc-user"":""trino""}" +"gt_mysql_xxx1","jdbc-mysql","{""in-use"":""true"",""jdbc-driver"":""com.mysql.cj.jdbc.Driver"",""jdbc-password"":""******"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""jdbc-user"":""trino""}","test" CALL -"gt_mysql_xxx1","jdbc-mysql","{""in-use"":""true"",""jdbc-driver"":""com.mysql.cj.jdbc.Driver"",""jdbc-password"":""******"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""jdbc-user"":""trino"",""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-password"":""******"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""jdbc-user"":""trino"",""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-password"":""******"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""jdbc-user"":""trino"",""test_key"":""test_value""}" +"gt_mysql_xxx1","jdbc-mysql","{""in-use"":""true"",""jdbc-driver"":""com.mysql.cj.jdbc.Driver"",""jdbc-password"":""******"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""jdbc-user"":""trino"",""test_key"":""test_value""}","test" CALL -"gt_mysql_xxx1","jdbc-mysql","{""in-use"":""true"",""jdbc-driver"":""com.mysql.cj.jdbc.Driver"",""jdbc-password"":""******"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""jdbc-user"":""trino"",""trino.bypass.join-pushdown.strategy"":""EAGER""}" +"gt_mysql_xxx1","jdbc-mysql","{""in-use"":""true"",""jdbc-driver"":""com.mysql.cj.jdbc.Driver"",""jdbc-password"":""******"",""jdbc-url"":""jdbc:mysql://%/?useSSL=false"",""jdbc-user"":""trino"",""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 dcf663546d..5c26b79863 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() { @@ -802,10 +856,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 +867,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 d59ff46ede..ebf7f2e378 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 5b9daab632..fb393a78eb 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 @@ -27,6 +27,7 @@ import io.trino.spi.connector.ConnectorContext; import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Objects; @@ -76,9 +77,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<>(); @@ -184,6 +187,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."); + } } /** @@ -229,8 +240,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 { @@ -308,9 +323,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)); } @@ -518,13 +531,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 { @@ -604,7 +635,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); @@ -744,6 +775,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. * @@ -785,7 +831,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); } @@ -802,13 +848,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()); } @@ -816,15 +860,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. @@ -872,7 +930,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()); } @@ -919,12 +977,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..8b8a98dfc8 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 @@ -355,8 +430,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 +515,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 b6a0df05f1..c4d08237d9 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 @@ -116,7 +116,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 @@ -686,6 +686,186 @@ 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, ImmutableMap.of()); + // 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, ImmutableMap.of()); + 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(); @@ -1344,8 +1524,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, @@ -1426,6 +1605,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); @@ -1440,6 +1620,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++) { @@ -1466,10 +1651,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(); } }
