This is an automated email from the ASF dual-hosted git repository. diqiu50 pushed a commit to branch dell-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 6214c3469346115527f9bfea48cbdc4a4e81b8fd Author: diqiu50 <[email protected]> AuthorDate: Sat Aug 22 21:38:49 2026 +0800 [Cherry-pick to branch-1.3] [#12554] improvement(trino-connector): Address review findings on the Iceberg REST routing Surface the missing rest-uri at connector startup instead of only when a catalog load fails inside the refresh loop, where the error is swallowed. Correct the claim that the REST path is the only one supporting credential vending: it is the only one supporting temporary credentials, while static keys are still injected on the backend path by applyCredentials. Derive s3.path-style-access, which S3-compatible stores such as MinIO need, and warn when a warehouse scheme has no Trino native file system but the catalog vends credentials. Log ignored reserved keys and the ambiguity of routing by catalog name alone in multi-metalake mode. Match Gravitino's case-insensitive handling of catalog-backend, and set the Iceberg REST config from an instance initializer so two @BeforeAll methods on the same test class no longer race. (cherry picked from commit a1b67fbd7f56f9522f64f62182696da130beaf93) --- docs/trino-connector/catalog-iceberg.md | 22 ++- docs/trino-connector/configuration.md | 4 +- .../docker-script/init/trino/init.sh | 5 +- .../test/container/TrinoITContainers.java | 4 + .../gravitino/integration/test/util/BaseIT.java | 3 +- .../integration/test/TrinoConnectorIT.java | 24 +-- .../gravitino/trino/connector/GravitinoConfig.java | 26 ++++ .../iceberg/IcebergCatalogPropertyConverter.java | 173 +++++++++++++++------ .../catalog/iceberg/IcebergConnectorAdapter.java | 16 +- .../trino/connector/TestGravitinoConfig.java | 12 ++ .../TestIcebergCatalogPropertyConverter.java | 163 ++++++++++++++++++- 11 files changed, 383 insertions(+), 69 deletions(-) diff --git a/docs/trino-connector/catalog-iceberg.md b/docs/trino-connector/catalog-iceberg.md index 223b597fe9..dd00d8cab1 100644 --- a/docs/trino-connector/catalog-iceberg.md +++ b/docs/trino-connector/catalog-iceberg.md @@ -50,9 +50,20 @@ them, and one catalog could not serve two Trino clusters that reach the IRC by d The connector derives everything else from the catalog itself. The Gravitino catalog name is passed as both `iceberg.rest-catalog.warehouse` and `iceberg.rest-catalog.prefix` — the Iceberg client selects the catalog twice over, first as the query parameter of the `GET /v1/config` call that -discovers it, then as the path segment of every request after that. The Trino native file system -(`fs.native-s3.enabled` and `s3.region`, or the GCS and Azure equivalents) is derived from the -catalog's `warehouse` scheme and storage properties. +discovers it, then as the path segment of every request after that. + +The Trino native file system is derived from the catalog's `warehouse` scheme, because vended +credentials are only consumed by Trino's native file systems: + +| Warehouse scheme | Derived properties | +|:------------------------------------------|:------------------------------------------------------------------------------------------------------| +| `s3://`, `s3a://`, `s3n://` | `fs.native-s3.enabled`, plus `s3.region`, `s3.endpoint` and `s3.path-style-access` where the catalog defines `s3-region`, `s3-endpoint` and `s3-path-style-access` | +| `gs://` | `fs.native-gcs.enabled` | +| `abfs://`, `abfss://`, `wasb://`, `wasbs://` | `fs.native-azure.enabled` | +| anything else (`hdfs://`, `file://`, `oss://`) | none — only `fs.hadoop.enabled` | + +A scheme in the last row has no Trino native file system, so a catalog that vends credentials for it +cannot have them applied; the connector logs a warning when it detects that combination. On the Gravitino side, the IRC must run with the dynamic config provider so it can serve catalogs defined in Gravitino: @@ -81,6 +92,11 @@ gravitino.iceberg.rest-catalog.oauth2.scope=email Omitting this block against an authenticated IRC surfaces as an authentication error rather than a missing-configuration error, which is easy to misdiagnose. +Four keys are reserved: `iceberg.rest-catalog.uri`, `.warehouse`, `.prefix` and +`iceberg.catalog.type`. The connector always derives these itself, so setting them through either +`gravitino.iceberg.rest-catalog.` or a catalog's `trino.bypass.` has no effect — the connector logs +when it ignores one. + When `gravitino.client.session.forwardUser=true`, the connector also sets `iceberg.rest-catalog.session=USER` so that each query carries the end user's identity to the IRC, keeping per-user credential vending and per-user authorization intact. Set diff --git a/docs/trino-connector/configuration.md b/docs/trino-connector/configuration.md index 1d756be8de..6d225a6cec 100644 --- a/docs/trino-connector/configuration.md +++ b/docs/trino-connector/configuration.md @@ -30,8 +30,8 @@ license: "This software is licensed under the Apache License version 2." | gravitino.trino.skip-catalog-patterns | string | (none) | The `gravitino.trino.skip-catalog-patterns` defines a comma-separated list of catalog name regex patterns that should be excluded from loading. For example, `test_.*, .*_tmp` excludes all catalogs starting with `test_` or ending with `_tmp`. | No | | gravitino.use-single-metalake | boolean | true | If `true`, only one metalake is used and catalogs are identified by `<catalog_name>`. If `false`, multi-metalake mode is enabled and catalogs are identified by `<metalake_name>.<catalog_name>`. | No | | gravitino.iceberg.rest-enabled | boolean | true | If `true`, `lakehouse-iceberg` catalogs are loaded through the Gravitino Iceberg REST server (IRC) instead of being translated into Trino's `jdbc` or `hive_metastore` Iceberg catalog type. This is what makes credential vending work. Requires `gravitino.iceberg.rest-uri`. Set to `false` for a deployment that does not run the IRC. | No | -| gravitino.iceberg.rest-uri | string | (none) | The endpoint of the Gravitino Iceberg REST server, for example `http://gravitino-host:9001/iceberg`. Required when `gravitino.iceberg.rest-enabled` is `true` and the metalake contains `lakehouse-iceberg` catalogs. | No | -| gravitino.iceberg.rest-catalog. | string | (none) | The configuration key prefix for properties passed through to the internal Trino Iceberg REST catalog. The prefix is rewritten to `iceberg.rest-catalog.`, so `gravitino.iceberg.rest-catalog.security=OAUTH2` becomes `iceberg.rest-catalog.security=OAUTH2`. | No | +| gravitino.iceberg.rest-uri | string | (none) | The endpoint of the Gravitino Iceberg REST server, for example `http://gravitino-host:9001/iceberg`. Required when `gravitino.iceberg.rest-enabled` is `true` and the metalake contains `lakehouse-iceberg` catalogs whose `catalog-backend` is not `rest`. | No | +| gravitino.iceberg.rest-catalog. | string | (none) | The configuration key prefix for properties passed through to the internal Trino Iceberg REST catalog. The prefix is rewritten to `iceberg.rest-catalog.`, so `gravitino.iceberg.rest-catalog.security=OAUTH2` becomes `iceberg.rest-catalog.security=OAUTH2`. The `uri`, `warehouse` and `prefix` keys are reserved and always 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. diff --git a/integration-test-common/docker-script/init/trino/init.sh b/integration-test-common/docker-script/init/trino/init.sh index d713708e52..a65ec5ab90 100644 --- a/integration-test-common/docker-script/init/trino/init.sh +++ b/integration-test-common/docker-script/init/trino/init.sh @@ -34,8 +34,9 @@ cp /usr/lib/trino/plugin/mysql/*mysql-connector-j-*.jar /usr/lib/trino/plugin/ic sed -i "s/GRAVITINO_HOST_IP:GRAVITINO_HOST_PORT/${GRAVITINO_HOST_IP}:${GRAVITINO_HOST_PORT}/g" /etc/trino/catalog/gravitino.properties # Update `gravitino.metalake = GRAVITINO_METALAKE_NAME` in the `conf/catalog/gravitino.properties` sed -i "s/GRAVITINO_METALAKE_NAME/${GRAVITINO_METALAKE_NAME}/g" /etc/trino/catalog/gravitino.properties -# Update `gravitino.iceberg.rest-uri` in the `conf/catalog/gravitino.properties`. It uses its own -# placeholders so that the substitution above cannot leave a half-replaced host behind. +# Update `gravitino.iceberg.rest-uri` in the `conf/catalog/gravitino.properties`. The Iceberg REST +# server runs on the same host as the Gravitino server but on a different port, so it needs its own +# placeholder pair — reusing GRAVITINO_HOST_IP:GRAVITINO_HOST_PORT would substitute the server port. sed -i "s/GRAVITINO_ICEBERG_REST_HOST:GRAVITINO_ICEBERG_REST_PORT/${GRAVITINO_HOST_IP}:${GRAVITINO_ICEBERG_REST_PORT}/g" /etc/trino/catalog/gravitino.properties # Update `node.id=NODE_ID` in the `/conf/node.properties` sed -i "s/NODE_ID/${RANDOM}-${RANDOM}-${RANDOM}-${RANDOM}-${RANDOM}/g" /etc/trino/node.properties diff --git a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TrinoITContainers.java b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TrinoITContainers.java index 14fe768c0b..572b948d55 100644 --- a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TrinoITContainers.java +++ b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TrinoITContainers.java @@ -84,6 +84,10 @@ public class TrinoITContainers implements AutoCloseable { env.put("HIVE_RUNTIME_VERSION", hiveRuntimeVersion); if (icebergRestServerPort > 0) { env.put("GRAVITINO_ICEBERG_REST_PORT", String.valueOf(icebergRestServerPort)); + } else { + LOG.info( + "No Iceberg REST server port supplied; Trino falls back to the compose default. " + + "Catalogs routed through the Iceberg REST server will not work in this container."); } env.put("TRINO_CONNECTOR_TEST", String.valueOf(isTrinoConnectorTest)); if (System.getProperty("gravitino.log.path") != null) { diff --git a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java index f3a0724ef0..0c247a22f1 100644 --- a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java +++ b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java @@ -688,7 +688,8 @@ public class BaseIT { } /** - * Returns the port the Iceberg REST auxiliary service listens on. + * Returns the port the Iceberg REST auxiliary service listens on. Only valid once {@link + * #startIntegrationTest()} has run, since it reads the started server's config. * * @return the Iceberg REST service port */ diff --git a/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoConnectorIT.java b/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoConnectorIT.java index d0796e044f..1be5ef43bc 100644 --- a/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoConnectorIT.java +++ b/trino-connector/integration-test/src/test/java/org/apache/gravitino/trino/connector/integration/test/TrinoConnectorIT.java @@ -23,7 +23,6 @@ import static org.testcontainers.shaded.org.awaitility.Awaitility.await; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Maps; import com.google.common.collect.Sets; -import java.net.URI; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -94,18 +93,19 @@ public class TrinoConnectorIT extends BaseIT { private static GravitinoMetalake metalake; private static Catalog catalog; - @BeforeAll - @Override - public void startIntegrationTest() throws Exception { + // Set from an instance initializer rather than an overridden @BeforeAll: BaseIT is PER_CLASS, so + // the instance exists before any lifecycle method runs, and JUnit only guarantees that a + // superclass @BeforeAll precedes a subclass one — two @BeforeAll methods declared on this same + // class would have no defined order relative to each other. + { // The Trino connector loads every lakehouse-iceberg catalog through the Iceberg REST server, // so the auxiliary service has to run and serve this test's metalake. - ignoreIcebergAuxRestService = false; - customConfigs.put( - GRAVITINO_ICEBERG_REST_PREFIX + IcebergConstants.ICEBERG_REST_CATALOG_CONFIG_PROVIDER, - IcebergConstants.DYNAMIC_ICEBERG_CATALOG_CONFIG_PROVIDER_NAME); - customConfigs.put( - GRAVITINO_ICEBERG_REST_PREFIX + IcebergConstants.GRAVITINO_METALAKE, metalakeName); - super.startIntegrationTest(); + enableIcebergAuxRestService( + ImmutableMap.of( + GRAVITINO_ICEBERG_REST_PREFIX + IcebergConstants.ICEBERG_REST_CATALOG_CONFIG_PROVIDER, + IcebergConstants.DYNAMIC_ICEBERG_CATALOG_CONFIG_PROVIDER_NAME, + GRAVITINO_ICEBERG_REST_PREFIX + IcebergConstants.GRAVITINO_METALAKE, + metalakeName)); } @BeforeAll @@ -131,7 +131,7 @@ public class TrinoConnectorIT extends BaseIT { trinoConfDir, System.getenv("GRAVITINO_ROOT_DIR") + "/trino-connector/build/libs", getGravitinoServerPort(), - URI.create(getIcebergRestServiceUri()).getPort(), + getIcebergRestServicePort(), metalakeName); Assertions.assertTrue( containerSuite.getTrinoContainer().checkSyncCatalogFromGravitino(5, catalogName), 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 b6a518584a..50af268066 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 @@ -35,10 +35,14 @@ import java.util.regex.Pattern; import java.util.stream.Collectors; import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.trino.connector.security.GravitinoAuthProvider; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** Gravitino config. */ public class GravitinoConfig { + private static final Logger LOG = LoggerFactory.getLogger(GravitinoConfig.class); + // Trino config keys /** The Trino discovery URI. */ private static final String TRINO_DISCOVERY_URI = "discovery.uri"; @@ -321,6 +325,28 @@ public class GravitinoConfig { "Config `gravitino.trino.skip-catalog-patterns` is invalid because it contains an illegal regular expression", e); } + warnOnMissingIcebergRestUri(); + } + + /** + * Warns at startup when the Iceberg REST routing is on but has no endpoint. Without this, the + * failure only surfaces once a lakehouse-iceberg catalog is loaded, and that error is swallowed + * by the catalog refresh loop, leaving the user with an unexplained missing catalog. + */ + private void warnOnMissingIcebergRestUri() { + // Only the statically configured connector warns; the dynamic per-catalog ones would repeat it. + if (isDynamicConnector() + || !isIcebergRestEnabled() + || StringUtils.isNotBlank(getIcebergRestUri())) { + return; + } + LOG.warn( + "'{}' is enabled but '{}' is not set, so every lakehouse-iceberg catalog will fail to " + + "load. Set the Iceberg REST server endpoint, or set '{}=false' to load Iceberg " + + "catalogs through their catalog backend instead.", + GRAVITINO_ICEBERG_REST_ENABLED.key, + GRAVITINO_ICEBERG_REST_URI.key, + GRAVITINO_ICEBERG_REST_ENABLED.key); } /** diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergCatalogPropertyConverter.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergCatalogPropertyConverter.java index 8073f1e1c2..dd8a492e1d 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergCatalogPropertyConverter.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergCatalogPropertyConverter.java @@ -22,6 +22,7 @@ package org.apache.gravitino.trino.connector.catalog.iceberg; import com.google.common.collect.Sets; import io.trino.spi.TrinoException; import java.util.HashMap; +import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Set; @@ -30,6 +31,7 @@ import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergPropertiesUtils; import org.apache.gravitino.credential.AzureAccountKeyCredential; import org.apache.gravitino.credential.Credential; +import org.apache.gravitino.credential.CredentialConstants; import org.apache.gravitino.credential.JdbcCredential; import org.apache.gravitino.credential.S3SecretKeyCredential; import org.apache.gravitino.storage.S3Properties; @@ -37,6 +39,8 @@ import org.apache.gravitino.trino.connector.GravitinoConfig; import org.apache.gravitino.trino.connector.GravitinoErrorCode; import org.apache.gravitino.trino.connector.catalog.CatalogPropertyConverter; import org.apache.gravitino.trino.connector.metadata.GravitinoCatalog; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * A property converter for Iceberg catalogs that handles the conversion between Trino and Gravitino @@ -45,6 +49,8 @@ import org.apache.gravitino.trino.connector.metadata.GravitinoCatalog; */ public class IcebergCatalogPropertyConverter extends CatalogPropertyConverter { + private static final Logger LOG = LoggerFactory.getLogger(IcebergCatalogPropertyConverter.class); + private static final Set<String> JDBC_BACKEND_REQUIRED_PROPERTIES = Set.of("jdbc-driver", "uri"); private static final Set<String> HIVE_BACKEND_REQUIRED_PROPERTIES = Set.of("uri"); @@ -65,6 +71,15 @@ public class IcebergCatalogPropertyConverter extends CatalogPropertyConverter { private static final String TRINO_FS_NATIVE_AZURE_ENABLED = "fs.native-azure.enabled"; private static final String TRINO_S3_REGION = "s3.region"; private static final String TRINO_S3_ENDPOINT = "s3.endpoint"; + private static final String TRINO_S3_PATH_STYLE_ACCESS = "s3.path-style-access"; + + /** Routing keys the connector always derives itself; they cannot be overridden. */ + private static final List<String> RESERVED_REST_PROPERTIES = + List.of( + TRINO_ICEBERG_CATALOG_TYPE, + TRINO_ICEBERG_REST_URI, + TRINO_ICEBERG_REST_WAREHOUSE, + TRINO_ICEBERG_REST_PREFIX); /** * Injects credentials from credential vending into the Iceberg catalog config. Applies JDBC @@ -100,7 +115,9 @@ public class IcebergCatalogPropertyConverter extends CatalogPropertyConverter { throw new TrinoException( GravitinoErrorCode.GRAVITINO_MISSING_REQUIRED_PROPERTY, "Missing required property 'catalog-backend'"); - switch (backend) { + // Gravitino accepts the backend as an enum value, and its own consumers upper-case it before + // resolving, so the value reaching here may be in any case. + switch (backend.toLowerCase(Locale.ROOT)) { case "hive": stringStringMap = buildHiveBackendProperties(properties); break; @@ -121,6 +138,73 @@ public class IcebergCatalogPropertyConverter extends CatalogPropertyConverter { return config; } + /** + * Builds the Trino Iceberg connector config that reaches this catalog through the Gravitino + * Iceberg REST server, regardless of the catalog backend Gravitino uses to store its metadata. + * + * <p>This is the only path on which <em>temporary</em> credentials work — {@code s3-token} and + * its GCS and ADLS counterparts. The Iceberg REST protocol issues a fresh one per table access, + * while Trino's jdbc and hive_metastore Iceberg catalog types have nowhere to put the session + * token of an STS credential. Static-key credentials are unaffected: they are still injected on + * the backend path by {@link #applyCredentials}. + * + * @param catalog the Gravitino catalog to load + * @param gravitinoConfig the connector configuration holding the Iceberg REST server endpoint + * @return the Trino Iceberg connector config + * @throws TrinoException if {@code gravitino.iceberg.rest-uri} is not configured + */ + public Map<String, String> buildIcebergRestProperties( + GravitinoCatalog catalog, GravitinoConfig gravitinoConfig) { + String restUri = gravitinoConfig.getIcebergRestUri(); + if (StringUtils.isBlank(restUri)) { + throw new TrinoException( + GravitinoErrorCode.GRAVITINO_MISSING_CONFIG, + "Missing required config 'gravitino.iceberg.rest-uri'. Set it to the Gravitino Iceberg " + + "REST server endpoint, for example http://localhost:9001/iceberg, or set " + + "'gravitino.iceberg.rest-enabled=false' to load Iceberg catalogs through their " + + "catalog backend instead."); + } + + Map<String, String> config = new HashMap<>(); + // The order of put operations determines the priority of parameters. + config.putAll(buildStorageProperties(catalog.getProperties())); + config.put(TRINO_ICEBERG_REST_VENDED_CREDENTIALS, "true"); + if (gravitinoConfig.isForwardUser()) { + config.put(TRINO_ICEBERG_REST_SESSION, "USER"); + } + // The catalog's own trino.bypass properties override the defaults above, so that a Trino + // release renaming one of them can be worked around without a connector change. + config.putAll(super.gravitinoToEngineProperties(catalog.getProperties())); + // The IRC's own authentication is a cluster-level operational setting, so it takes precedence + // over anything set on a single catalog. + config.putAll(gravitinoConfig.getIcebergRestCatalogConfig()); + + warnOnReservedOverrides(catalog, config); + + config.put(TRINO_ICEBERG_CATALOG_TYPE, TRINO_ICEBERG_CATALOG_TYPE_REST); + config.put(TRINO_ICEBERG_REST_URI, restUri); + // The Iceberg client selects the catalog twice over: `warehouse` is the query parameter of the + // initial GET /v1/config that discovers the catalog, and `prefix` is the path segment of every + // request after it. Setting only the latter leaves the config call without a catalog name. + // The server echoes `prefix` back as a config default, so setting it here is belt-and-braces. + config.put(TRINO_ICEBERG_REST_WAREHOUSE, catalog.getName()); + config.put(TRINO_ICEBERG_REST_PREFIX, catalog.getName()); + + if (!gravitinoConfig.singleMetalakeMode()) { + // The IRC serves exactly one metalake, and the prefix carries only the catalog name, so two + // metalakes holding a same-named catalog both resolve to whichever one the IRC was started + // with. + LOG.warn( + "Catalog '{}' in metalake '{}' is routed to the Iceberg REST server using the catalog " + + "name alone. In multi-metalake mode this is ambiguous: the Iceberg REST server " + + "serves a single metalake, so a same-named catalog in another metalake resolves to " + + "the same prefix.", + catalog.getName(), + catalog.getMetalake()); + } + return config; + } + private Map<String, String> buildHiveBackendProperties(Map<String, String> properties) { Set<String> missingProperty = Sets.difference(HIVE_BACKEND_REQUIRED_PROPERTIES, properties.keySet()); @@ -167,51 +251,16 @@ public class IcebergCatalogPropertyConverter extends CatalogPropertyConverter { return jdbcProperties; } - /** - * Builds the Trino Iceberg connector config that reaches this catalog through the Gravitino - * Iceberg REST server, regardless of the catalog backend Gravitino uses to store its metadata. - * - * <p>This is the only path on which credential vending works: the Iceberg REST protocol issues a - * fresh temporary credential per table access, while Trino's jdbc and hive_metastore Iceberg - * catalog types have nowhere to put the session token of an STS credential. - * - * @param catalog the Gravitino catalog to load - * @param gravitinoConfig the connector configuration holding the Iceberg REST server endpoint - * @return the Trino Iceberg connector config - */ - public Map<String, String> buildIcebergRestProperties( - GravitinoCatalog catalog, GravitinoConfig gravitinoConfig) { - String restUri = gravitinoConfig.getIcebergRestUri(); - if (StringUtils.isBlank(restUri)) { - throw new TrinoException( - GravitinoErrorCode.GRAVITINO_MISSING_CONFIG, - "Missing required config 'gravitino.iceberg.rest-uri'. Set it to the Gravitino Iceberg " - + "REST server endpoint, for example http://localhost:9001/iceberg, or set " - + "'gravitino.iceberg.rest-enabled=false' to load Iceberg catalogs through their " - + "catalog backend instead."); - } - - Map<String, String> config = new HashMap<>(); - // The order of put operations determines the priority of parameters. - config.putAll(buildStorageProperties(catalog.getProperties())); - config.put(TRINO_ICEBERG_REST_VENDED_CREDENTIALS, "true"); - if (gravitinoConfig.isForwardUser()) { - config.put(TRINO_ICEBERG_REST_SESSION, "USER"); + private void warnOnReservedOverrides(GravitinoCatalog catalog, Map<String, String> config) { + for (String reserved : RESERVED_REST_PROPERTIES) { + if (config.containsKey(reserved)) { + LOG.info( + "Property '{}' set on catalog '{}' is ignored; the connector always derives it when " + + "routing through the Iceberg REST server.", + reserved, + catalog.getName()); + } } - // The catalog's own trino.bypass properties override the defaults above, so that a Trino - // release renaming one of them can be worked around without a connector change. - config.putAll(super.gravitinoToEngineProperties(catalog.getProperties())); - // The Iceberg REST server endpoint and its authentication are cluster-level operational - // settings, so they take precedence over anything set on a single catalog. - config.putAll(gravitinoConfig.getIcebergRestCatalogConfig()); - config.put(TRINO_ICEBERG_CATALOG_TYPE, TRINO_ICEBERG_CATALOG_TYPE_REST); - config.put(TRINO_ICEBERG_REST_URI, restUri); - // The Iceberg client selects the catalog twice over: `warehouse` is the query parameter of the - // initial GET /v1/config that discovers the catalog, and `prefix` is the path segment of every - // request after it. Setting only the latter leaves the config call without a catalog name. - config.put(TRINO_ICEBERG_REST_WAREHOUSE, catalog.getName()); - config.put(TRINO_ICEBERG_REST_PREFIX, catalog.getName()); - return config; } /** @@ -221,6 +270,8 @@ public class IcebergCatalogPropertyConverter extends CatalogPropertyConverter { */ private Map<String, String> buildStorageProperties(Map<String, String> properties) { Map<String, String> storageProperties = new HashMap<>(); + // Always available as the fallback; where a native file system is also enabled below, that one + // takes precedence for its own scheme. storageProperties.put(TRINO_FS_HADOOP_ENABLED, "true"); String warehouse = properties.get(IcebergConstants.WAREHOUSE); @@ -238,6 +289,12 @@ public class IcebergCatalogPropertyConverter extends CatalogPropertyConverter { properties, S3Properties.GRAVITINO_S3_REGION, storageProperties, TRINO_S3_REGION); copyProperty( properties, S3Properties.GRAVITINO_S3_ENDPOINT, storageProperties, TRINO_S3_ENDPOINT); + // Required by S3-compatible stores such as MinIO and Ceph, which a custom endpoint implies. + copyProperty( + properties, + S3Properties.GRAVITINO_S3_PATH_STYLE_ACCESS, + storageProperties, + TRINO_S3_PATH_STYLE_ACCESS); break; case "gs": storageProperties.put(TRINO_FS_NATIVE_GCS_ENABLED, "true"); @@ -249,12 +306,36 @@ public class IcebergCatalogPropertyConverter extends CatalogPropertyConverter { storageProperties.put(TRINO_FS_NATIVE_AZURE_ENABLED, "true"); break; default: - // hdfs, file and any scheme without a Trino native file system stay on fs.hadoop.enabled. + warnOnSchemeWithoutNativeFileSystem(properties, warehouse, scheme); break; } return storageProperties; } + /** + * Warns when a warehouse scheme has no Trino native file system but the catalog vends + * credentials. The Hadoop file system cannot consume a vended credential, so table access would + * fail at read time with a storage authentication error far from its cause. + */ + private void warnOnSchemeWithoutNativeFileSystem( + Map<String, String> properties, String warehouse, String scheme) { + // hdfs and file legitimately run on the Hadoop file system, and a warehouse without a scheme + // is a plain path. None of them vend credentials. + if (!warehouse.contains("://") || "hdfs".equals(scheme) || "file".equals(scheme)) { + return; + } + if (StringUtils.isBlank(properties.get(CredentialConstants.CREDENTIAL_PROVIDERS))) { + return; + } + LOG.warn( + "Warehouse scheme '{}' has no Trino native file system, so the credentials vended by " + + "'{}' cannot be applied. Table access falls back to the Hadoop file system and may " + + "fail to authenticate. Schemes with a native file system: s3/s3a/s3n, gs, " + + "abfs/abfss/wasb/wasbs.", + scheme, + properties.get(CredentialConstants.CREDENTIAL_PROVIDERS)); + } + private void copyProperty( Map<String, String> source, String sourceKey, Map<String, String> target, String targetKey) { String value = source.get(sourceKey); diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergConnectorAdapter.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergConnectorAdapter.java index 102c15d0f5..1480dc13d4 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergConnectorAdapter.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/iceberg/IcebergConnectorAdapter.java @@ -30,6 +30,8 @@ import org.apache.gravitino.trino.connector.GravitinoConfig; import org.apache.gravitino.trino.connector.catalog.CatalogConnectorAdapter; import org.apache.gravitino.trino.connector.catalog.CatalogConnectorMetadataAdapter; import org.apache.gravitino.trino.connector.metadata.GravitinoCatalog; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Transforming Apache Iceberg connector configuration and components into Apache Gravitino @@ -37,6 +39,8 @@ import org.apache.gravitino.trino.connector.metadata.GravitinoCatalog; */ public class IcebergConnectorAdapter implements CatalogConnectorAdapter { + private static final Logger LOG = LoggerFactory.getLogger(IcebergConnectorAdapter.class); + private static final String CONNECTOR_ICEBERG = "iceberg"; private static final String REST_CATALOG_BACKEND = "rest"; @@ -61,11 +65,19 @@ public class IcebergConnectorAdapter implements CatalogConnectorAdapter { GravitinoCatalog catalog, Credential[] credentials) throws Exception { // The catalog backend describes how Gravitino stores the metadata; it does not decide how // Trino reaches the data. Unless the routing is disabled, every catalog is loaded through the - // Gravitino Iceberg REST server, which is the only path that supports credential vending. + // Gravitino Iceberg REST server, the only path that supports temporary credentials. // A catalog that already has a REST backend keeps pointing at its own configured endpoint. if (config.isIcebergRestEnabled() - && !REST_CATALOG_BACKEND.equals( + && !REST_CATALOG_BACKEND.equalsIgnoreCase( catalog.getProperty(IcebergConstants.CATALOG_BACKEND, null))) { + // `credentials` is intentionally unused here: with vended credentials enabled, Trino obtains + // a fresh one per table access over the REST protocol, rather than the catalog-level + // snapshot applyCredentials installs. + LOG.debug( + "Routing catalog '{}' through the Iceberg REST server; its {} catalog-level credential(s)" + + " are not applied because the REST protocol vends one per table access.", + catalog.getName(), + credentials.length); return catalogConverter.buildIcebergRestProperties(catalog, config); } 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 c31c594be0..357280cf90 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 @@ -353,6 +353,18 @@ public class TestGravitinoConfig { assertTrue(catalogConfig.contains("\"gravitino.iceberg.rest-catalog.security\"='OAUTH2'")); } + @Test + public void testToCatalogConfigPropagatesIcebergRestEnabled() { + // The switch rides the exact-key loop rather than the prefix filter; if it fails to propagate, + // the coordinator and the workers build different configs for the same catalog. + GravitinoConfig config = + new GravitinoConfig( + ImmutableMap.of( + "gravitino.metalake", "user_001", "gravitino.iceberg.rest-enabled", "false")); + + assertTrue(config.toCatalogConfig().contains("\"gravitino.iceberg.rest-enabled\"='false'")); + } + private static boolean skipCatalog(String catalogName, GravitinoConfig config) { for (Pattern pattern : config.getSkipCatalogPatterns()) { if (pattern.matcher(catalogName).matches()) { diff --git a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/iceberg/TestIcebergCatalogPropertyConverter.java b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/iceberg/TestIcebergCatalogPropertyConverter.java index fc3571c8e8..f09e4d7469 100644 --- a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/iceberg/TestIcebergCatalogPropertyConverter.java +++ b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/iceberg/TestIcebergCatalogPropertyConverter.java @@ -27,6 +27,7 @@ import org.apache.gravitino.Catalog; import org.apache.gravitino.catalog.property.PropertyConverter; import org.apache.gravitino.credential.Credential; import org.apache.gravitino.credential.JdbcCredential; +import org.apache.gravitino.credential.S3SecretKeyCredential; import org.apache.gravitino.trino.connector.GravitinoConfig; import org.apache.gravitino.trino.connector.metadata.GravitinoCatalog; import org.apache.gravitino.trino.connector.metadata.TestGravitinoCatalog; @@ -451,11 +452,171 @@ public class TestIcebergCatalogPropertyConverter { private static Map<String, String> buildConnectorConfig( String catalogName, Map<String, String> properties, GravitinoConfig gravitinoConfig) throws Exception { + return buildConnectorConfig(catalogName, properties, gravitinoConfig, new Credential[0]); + } + + @Test + public void testIcebergRestPathIgnoresCatalogLevelCredentials() throws Exception { + Map<String, String> properties = + ImmutableMap.<String, String>builder() + .put("catalog-backend", "jdbc") + .put("uri", "jdbc:postgresql://localhost:5432/iceberg") + .put("jdbc-driver", "org.postgresql.Driver") + .put("warehouse", "s3://bucket/warehouse/") + .build(); + Credential[] credentials = { + new JdbcCredential("root", "ds123"), new S3SecretKeyCredential("AKIAEXAMPLE", "secret") + }; + + Map<String, String> config = + buildConnectorConfig( + "catalog1", properties, icebergRestEnabledConfig(ImmutableMap.of()), credentials); + + // The REST protocol vends a fresh credential per table access, so the catalog-level snapshot + // must not be pinned into the connector config. + Assertions.assertNull(config.get("iceberg.jdbc-catalog.connection-user")); + Assertions.assertNull(config.get("iceberg.jdbc-catalog.connection-password")); + Assertions.assertNull(config.get("hive.s3.aws-access-key")); + Assertions.assertNull(config.get("hive.s3.aws-secret-key")); + Assertions.assertEquals("true", config.get("iceberg.rest-catalog.vended-credentials-enabled")); + + // The backend path still applies them. + Map<String, String> legacyConfig = + buildConnectorConfig("catalog1", properties, icebergRestDisabledConfig(), credentials); + Assertions.assertEquals("root", legacyConfig.get("iceberg.jdbc-catalog.connection-user")); + Assertions.assertEquals("AKIAEXAMPLE", legacyConfig.get("hive.s3.aws-access-key")); + } + + @Test + public void testIcebergRestReservedPropertiesResistCatalogOverride() throws Exception { + Map<String, String> properties = + ImmutableMap.<String, String>builder() + .put("catalog-backend", "jdbc") + .put("warehouse", "s3://bucket/warehouse/") + .put("trino.bypass.iceberg.catalog.type", "jdbc") + .put("trino.bypass.iceberg.rest-catalog.uri", "http://elsewhere:9001/iceberg";) + .put("trino.bypass.iceberg.rest-catalog.prefix", "other_catalog") + .put("trino.bypass.iceberg.rest-catalog.warehouse", "other_catalog") + .build(); + + Map<String, String> config = + buildConnectorConfig("catalog1", properties, icebergRestEnabledConfig(ImmutableMap.of())); + + // A catalog property must never redirect the connector at another endpoint or catalog. + Assertions.assertEquals("rest", config.get("iceberg.catalog.type")); + Assertions.assertEquals( + "http://localhost:9001/iceberg";, config.get("iceberg.rest-catalog.uri")); + Assertions.assertEquals("catalog1", config.get("iceberg.rest-catalog.prefix")); + Assertions.assertEquals("catalog1", config.get("iceberg.rest-catalog.warehouse")); + } + + @Test + public void testConnectorRestCatalogConfigOverridesCatalogBypass() throws Exception { + Map<String, String> properties = + ImmutableMap.<String, String>builder() + .put("catalog-backend", "jdbc") + .put("warehouse", "s3://bucket/warehouse/") + .put("trino.bypass.iceberg.rest-catalog.security", "NONE") + .build(); + + Map<String, String> config = + buildConnectorConfig( + "catalog1", + properties, + icebergRestEnabledConfig( + ImmutableMap.of("gravitino.iceberg.rest-catalog.security", "OAUTH2"))); + + // Cluster-level operational settings outrank per-catalog properties. + Assertions.assertEquals("OAUTH2", config.get("iceberg.rest-catalog.security")); + } + + @Test + public void testBuildConnectorPropertiesStorageDetectionEdgeCases() throws Exception { + // s3a/s3n aliases and an uppercase scheme all resolve to the native S3 file system. + for (String warehouse : new String[] {"s3a://bucket/wh", "s3n://bucket/wh", "S3://bucket/wh"}) { + Map<String, String> config = + buildConnectorConfig( + "catalog1", + ImmutableMap.of("catalog-backend", "jdbc", "warehouse", warehouse), + icebergRestEnabledConfig(ImmutableMap.of())); + Assertions.assertEquals( + "true", config.get("fs.native-s3.enabled"), "warehouse: " + warehouse); + } + + // A warehouse without a scheme is a plain path and stays on the Hadoop file system. + Map<String, String> schemeless = + buildConnectorConfig( + "catalog1", + ImmutableMap.of("catalog-backend", "jdbc", "warehouse", "gt_iceberg_rest"), + icebergRestEnabledConfig(ImmutableMap.of())); + Assertions.assertEquals("true", schemeless.get("fs.hadoop.enabled")); + Assertions.assertNull(schemeless.get("fs.native-s3.enabled")); + + // No warehouse at all. + Map<String, String> noWarehouse = + buildConnectorConfig( + "catalog1", + ImmutableMap.of("catalog-backend", "jdbc"), + icebergRestEnabledConfig(ImmutableMap.of())); + Assertions.assertEquals("true", noWarehouse.get("fs.hadoop.enabled")); + Assertions.assertNull(noWarehouse.get("fs.native-s3.enabled")); + Assertions.assertNull(noWarehouse.get("fs.native-gcs.enabled")); + Assertions.assertNull(noWarehouse.get("fs.native-azure.enabled")); + + // oss:// has no Trino native file system, so nothing is enabled beyond Hadoop. + Map<String, String> oss = + buildConnectorConfig( + "catalog1", + ImmutableMap.of( + "catalog-backend", "jdbc", + "warehouse", "oss://bucket/wh", + "credential-providers", "oss-token"), + icebergRestEnabledConfig(ImmutableMap.of())); + Assertions.assertEquals("true", oss.get("fs.hadoop.enabled")); + Assertions.assertNull(oss.get("fs.native-s3.enabled")); + } + + @Test + public void testBuildConnectorPropertiesCopiesS3PathStyleAccess() throws Exception { + Map<String, String> config = + buildConnectorConfig( + "catalog1", + ImmutableMap.of( + "catalog-backend", "jdbc", + "warehouse", "s3://bucket/wh", + "s3-endpoint", "http://minio:9000";, + "s3-path-style-access", "true"), + icebergRestEnabledConfig(ImmutableMap.of())); + + // S3-compatible stores such as MinIO need path-style addressing to resolve the bucket. + Assertions.assertEquals("true", config.get("s3.path-style-access")); + Assertions.assertEquals("http://minio:9000";, config.get("s3.endpoint")); + } + + @Test + public void testRestBackendIsNotReRoutedRegardlessOfCase() throws Exception { + Map<String, String> config = + buildConnectorConfig( + "catalog1", + ImmutableMap.of("catalog-backend", "REST", "uri", "http://other-irc:9001/iceberg";), + icebergRestEnabledConfig(ImmutableMap.of())); + + Assertions.assertEquals( + "http://other-irc:9001/iceberg";, config.get("iceberg.rest-catalog.uri")); + Assertions.assertNull(config.get("iceberg.rest-catalog.prefix")); + } + + private static Map<String, String> buildConnectorConfig( + String catalogName, + Map<String, String> properties, + GravitinoConfig gravitinoConfig, + Credential[] credentials) + throws Exception { Catalog mockCatalog = TestGravitinoCatalog.mockCatalog( catalogName, "lakehouse-iceberg", "test catalog", Catalog.Type.RELATIONAL, properties); return new IcebergConnectorAdapter(gravitinoConfig) - .buildInternalConnectorConfig(new GravitinoCatalog("test", mockCatalog), new Credential[0]); + .buildInternalConnectorConfig(new GravitinoCatalog("test", mockCatalog), credentials); } private static GravitinoConfig icebergRestDisabledConfig() {
