yuqi1129 commented on code in PR #12710:
URL: https://github.com/apache/gravitino/pull/12710#discussion_r3904062310
##########
docs/spark-connector/spark-catalog-iceberg.md:
##########
@@ -148,13 +148,111 @@ Gravitino catalog property names with the prefix
`spark.bypass.` are passed to S
Iceberg catalog property `cache-enabled` is setting to `false` internally and
not allowed to change.
:::
+## Routing Through the Gravitino Iceberg REST Server
+
+If the Gravitino server exposes an [Iceberg REST
catalog](../iceberg-rest-service.md) (IRC) endpoint for the
+current metalake, the Spark connector automatically routes `hive` and `jdbc`
backed Iceberg catalogs through
+that endpoint instead of talking to the Hive metastore or JDBC database
directly. This has no effect on
+catalogs whose `catalog-backend` is already `rest` or `custom`.
+
+Routing through the IRC server is the only way to receive short-lived,
per-table **vended credentials**
+that Iceberg's native REST protocol refreshes automatically. The non-REST path
can still inject a single
+vended credential fetched once at catalog initialization (see
+[Credential vending](../security/credential-vending.md)), but it is not
refreshed per table access.
+
+REST routing is enabled by default. The endpoint is discovered once per Spark
application — for the life
+of the Gravitino Spark plugin, not per `SparkSession` — and is not re-checked
afterward.
+
+- If no discoverable endpoint is found (for example, the `iceberg-rest`
auxiliary service is disabled or
+ not configured with `catalog-config-provider=dynamic-config-provider`) and
routing was left at its
+ default, the connector falls back to the native Hive/JDBC backend and logs a
warning. Set
+ `spark.sql.gravitino.iceberg.rest-routing-enabled=true` explicitly to
require Iceberg REST routing and
+ fail catalog initialization instead.
+- A catalog whose warehouse uses a scheme with a native Iceberg FileIO
(`s3://`, `gs://`, `abfs://`, etc.)
+ must have [credential vending](../security/credential-vending.md) configured
(`credential-providers`)
+ before it can be routed: routing replaces any static storage credentials for
that FileIO with vended
+ ones, and without credential vending the catalog would lose storage access.
Catalog initialization
+ fails with an actionable error if this is not configured. Set
`rest-routing-enabled=false` for that
+ catalog to keep using the legacy Hive/JDBC backend instead.
+
+To force a specific endpoint instead of relying on auto-discovery, set:
+
+```properties
+spark.sql.gravitino.iceberg.rest-uri http://<gravitino-host>:9001/iceberg
+```
+
+To retain the legacy Hive/JDBC translation and skip endpoint discovery,
disable routing explicitly:
+
+```properties
+spark.sql.gravitino.iceberg.rest-routing-enabled false
+```
+
+If Gravitino requires authentication on the IRC endpoint, pass the Iceberg
REST client's own auth
+properties using the `spark.sql.gravitino.iceberg.rest.` prefix. For example,
for Basic authentication:
+
+```properties
+spark.sql.gravitino.iceberg.rest.rest.auth.type basic
+spark.sql.gravitino.iceberg.rest.rest.auth.basic.username <username>
+spark.sql.gravitino.iceberg.rest.rest.auth.basic.password <password>
+```
+
+See [Connect Spark to Iceberg REST](../iceberg-rest-engine/spark.md) for the
full set of supported
+`rest.auth.*` properties and how to configure them when connecting directly to
the IRC endpoint.
+
+When the Gravitino client uses OAuth2, the connector reuses its OAuth2 client
configuration for IRC by
+default. This avoids duplicating configuration when both endpoints accept the
same client identity. The
+equivalent explicit setting is:
+
+```properties
+spark.sql.gravitino.iceberg.reuseOAuth2 true
Review Comment:
Why not just use the format `key=value` here?
##########
.github/workflows/spark-integration-test-action.yml:
##########
@@ -25,7 +25,7 @@ jobs:
start-runner:
name: JDK${{ inputs.java-version }}-${{ inputs.test-mode }}-Scala${{
inputs.scala-version }}
runs-on: ubuntu-latest
- timeout-minutes: 60
+ timeout-minutes: 120
Review Comment:
Why do we increase the time with such significance?
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConverter.java:
##########
@@ -68,6 +84,186 @@ public Map<String, String>
toSparkCatalogProperties(Map<String, String> properti
return all;
}
+ /**
+ * Builds Spark Iceberg catalog properties that route requests through the
Gravitino Iceberg REST
+ * server, regardless of the catalog's actual storage backend (hive/jdbc).
This is the only path
+ * on which temporary credentials work: the Iceberg REST protocol vends a
fresh credential per
+ * table access, so static secrets are intentionally not carried over from
{@code
+ * gravitinoProperties}. Static-key credentials for the non-REST path are
unaffected; they are
+ * still injected via {@link
org.apache.gravitino.credential.CredentialPropertyUtils}.
+ *
+ * @param gravitinoCatalogName the Gravitino catalog name, used as both
{@code warehouse} (for the
+ * initial catalog discovery request) and {@code prefix} (for every
subsequent request path
+ * segment) so the REST server resolves the same Gravitino catalog
+ * @param restUri the Iceberg REST server endpoint to route through,
resolved by the caller
+ * @param gravitinoProperties the Gravitino catalog properties
+ * @param icebergRestClientConfig operator-level Iceberg REST client config
(e.g. {@code
+ * rest.auth.type}), applied after catalog-level {@code spark.bypass.}
overrides since it is a
+ * cluster-wide operational setting
+ * @return the Spark Iceberg catalog properties
+ */
+ Map<String, String> buildIcebergRestProperties(
+ String gravitinoCatalogName,
+ String restUri,
+ Map<String, String> gravitinoProperties,
+ Map<String, String> icebergRestClientConfig) {
+ Preconditions.checkArgument(StringUtils.isNotBlank(restUri), "restUri
should not be blank");
+
+ Map<String, String> all = new HashMap<>();
+ // Later put/putAll calls override earlier ones for the same key; this is
call-order
+ // precedence, unrelated to HashMap's (unspecified) iteration order.
+ all.putAll(buildStorageProperties(gravitinoProperties));
+ all.put(
+ IcebergPropertiesConstants.ICEBERG_ACCESS_DELEGATION,
+
IcebergPropertiesConstants.ICEBERG_ACCESS_DELEGATION_VENDED_CREDENTIALS);
+ // The catalog's own spark.bypass properties override the defaults above,
so a renamed
+ // Iceberg client property can be worked around without a connector change.
+ all.putAll(extractSparkBypassProperties(gravitinoProperties));
+ all.putAll(icebergRestClientConfig);
+
+ reapplyReservedRestProperties(gravitinoCatalogName, restUri, all);
+ all.put(IcebergPropertiesConstants.ICEBERG_CATALOG_CACHE_ENABLED, "FALSE");
+ return all;
+ }
+
+ /**
+ * Re-derives the reserved routing keys ({@code type}/{@code uri}/{@code
warehouse}/{@code
+ * prefix}) on {@code all}, so that a source merged in after {@link
#buildIcebergRestProperties}
+ * (e.g. Spark catalog {@code options}) can never redirect a routed catalog.
+ *
+ * @param gravitinoCatalogName the Gravitino catalog name
+ * @param restUri the Iceberg REST server endpoint to route through
+ * @param all the properties to re-derive the reserved keys on, mutated in
place
+ */
+ void reapplyReservedRestProperties(
+ String gravitinoCatalogName, String restUri, Map<String, String> all) {
+ warnOnReservedRestPropertyOverrides(gravitinoCatalogName, all);
+ all.put(
+ IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
+ IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_REST);
+ all.put(IcebergPropertiesConstants.ICEBERG_CATALOG_URI, restUri);
+ all.put(IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
gravitinoCatalogName);
+ all.put(IcebergPropertiesConstants.ICEBERG_REST_CATALOG_PREFIX,
gravitinoCatalogName);
+ }
+
+ private Map<String, String> extractSparkBypassProperties(Map<String, String>
properties) {
+ Map<String, String> bypass = new HashMap<>();
+ if (properties != null) {
+ properties.forEach(
+ (k, v) -> {
+ if (k.startsWith(SPARK_PROPERTY_PREFIX)) {
+ bypass.put(k.substring(SPARK_PROPERTY_PREFIX.length()), v);
+ }
+ });
+ }
+ return bypass;
+ }
+
+ private void warnOnReservedRestPropertyOverrides(
+ String gravitinoCatalogName, Map<String, String> config) {
+ for (String reserved : RESERVED_REST_PROPERTIES) {
+ if (config.containsKey(reserved)) {
+ LOG.warn(
+ "Property '{}' set on catalog '{}' is ignored; the connector
always derives it when "
+ + "routing through the Iceberg REST server.",
+ reserved,
+ gravitinoCatalogName);
+ }
+ }
+ }
+
+ /**
+ * Derives the storage-related Iceberg client config from the catalog's
warehouse location. Only
+ * non-secret settings (native FileIO impl, custom endpoint, region,
path-style access) are
+ * carried over; static access keys and JDBC credentials are deliberately
excluded since the REST
+ * path vends its own temporary credentials.
+ */
+ private Map<String, String> buildStorageProperties(Map<String, String>
gravitinoProperties) {
+ Map<String, String> icebergProperties =
+ IcebergPropertiesUtils.toIcebergCatalogProperties(gravitinoProperties);
+ Map<String, String> storageProperties = new HashMap<>();
+ copyIfPresent(icebergProperties,
IcebergPropertiesConstants.ICEBERG_IO_IMPL, storageProperties);
+
+ String warehouse = gravitinoProperties.get(IcebergConstants.WAREHOUSE);
+ String fileIoImpl = deriveFileIoImpl(warehouse);
+ if (fileIoImpl != null) {
+
storageProperties.putIfAbsent(IcebergPropertiesConstants.ICEBERG_IO_IMPL,
fileIoImpl);
+ } else {
+ warnOnSchemeWithoutNativeFileIo(gravitinoProperties, warehouse);
+ }
+
+ copyIfPresent(
+ icebergProperties, IcebergPropertiesConstants.ICEBERG_S3_ENDPOINT,
storageProperties);
+ copyIfPresent(
+ icebergProperties, IcebergPropertiesConstants.ICEBERG_AWS_S3_REGION,
storageProperties);
+ copyIfPresent(
+ icebergProperties,
+ IcebergPropertiesConstants.ICEBERG_S3_PATH_STYLE_ACCESS,
+ storageProperties);
+ copyIfPresent(
+ icebergProperties, IcebergPropertiesConstants.ICEBERG_OSS_ENDPOINT,
storageProperties);
+ return storageProperties;
+ }
+
+ /**
+ * Derives the native Iceberg FileIO implementation for a warehouse
location, or {@code null} if
+ * the scheme has none (e.g. {@code hdfs://}, {@code file://}).
Package-private so the routing
+ * decision can check whether a warehouse has a native FileIO without
duplicating the scheme list.
+ */
+ static String deriveFileIoImpl(String warehouse) {
+ if (StringUtils.isBlank(warehouse) || !warehouse.contains("://")) {
+ return null;
+ }
+ String scheme = StringUtils.substringBefore(warehouse,
"://").toLowerCase(Locale.ROOT);
+ switch (scheme) {
+ case "s3":
+ case "s3a":
+ case "s3n":
+ return IcebergPropertiesConstants.ICEBERG_S3_FILE_IO_IMPL;
+ case "gs":
Review Comment:
I recalled that a similar code block already existed in another PR from you.
Is that true?
##########
docs/spark-connector/spark-connector.md:
##########
@@ -41,6 +41,10 @@ The Apache Gravitino Spark connector leverages the Spark
DataSourceV2 interface
| spark.sql.gravitino.clientCacheMaxSize | int | `100` | The
maximum number of Gravitino clients cached, one per identity.
| No |
| spark.sql.gravitino.clientCacheTtlSec | long | `3600` | Evicts a
cached Gravitino client this many seconds after it was last used.
| No |
| spark.sql.gravitino.catalogCacheTtlSec | long | `300` | Evicts a
cached catalog this many seconds after it was loaded.
| No |
+| spark.sql.gravitino.iceberg.rest-routing-enabled | boolean | `true` |
Whether `hive` and `jdbc` backed Iceberg catalogs must be routed through the
Gravitino Iceberg REST server. Set to `false` to retain legacy native backend
translation. | No |
Review Comment:
format the table
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/GravitinoIcebergCatalog.java:
##########
@@ -72,15 +88,166 @@ protected TableCatalog createAndInitSparkCatalog(
}
}
String catalogBackendName =
IcebergPropertiesUtils.getCatalogBackendName(properties);
- Map<String, String> all =
- getPropertiesConverter().toSparkCatalogProperties(options, properties);
- CredentialPropertyUtils.applyIcebergCredentials(
- CredentialPropertyUtils.getCredentials(gravitinoCatalogClient), all);
+ SparkConf sparkConf = SparkSession.active().sparkContext().conf();
+ Optional<String> icebergRestUri =
+ resolveIcebergRestUri(
+ properties,
+ key -> sparkConf.get(key, null),
+ () -> GravitinoCatalogManager.get().getIcebergRestUri());
+ Map<String, String> all;
+ if (icebergRestUri.isPresent()) {
+ all =
+ buildAutoRoutedIcebergRestProperties(
+ name, options, properties, icebergRestUri.get(), sparkConf);
+ } else {
+ all = getPropertiesConverter().toSparkCatalogProperties(options,
properties);
+ CredentialPropertyUtils.applyIcebergCredentials(
+ CredentialPropertyUtils.getCredentials(gravitinoCatalogClient), all);
+ }
TableCatalog icebergCatalog = new SparkCatalog();
icebergCatalog.initialize(catalogBackendName, new
CaseInsensitiveStringMap(all));
return icebergCatalog;
}
+ /**
+ * Resolves the Iceberg REST server endpoint to route this catalog through,
if any. Only hive/jdbc
+ * backed catalogs are eligible; a catalog already configured with {@code
catalog-backend=rest} or
+ * {@code custom} is left untouched, and so is a catalog with no {@code
catalog-backend} property
+ * at all.
+ *
+ * <p>An eligible catalog whose warehouse has a native Iceberg FileIO
(s3/gs/abfs-family schemes)
+ * fails immediately unless {@code credential-providers} is configured:
routing replaces any
+ * static storage credentials for that FileIO with vended ones, so such a
catalog would silently
+ * lose storage access once routed. A warehouse scheme with no native FileIO
(e.g. {@code
+ * hdfs://}, {@code file://}) carries no such risk and is unaffected.
+ */
+ static Optional<String> resolveIcebergRestUri(
+ Map<String, String> properties,
+ UnaryOperator<String> sessionConfig,
+ Supplier<Optional<String>> endpointDiscovery) {
+ String backend = properties.get(IcebergConstants.CATALOG_BACKEND);
+ if (backend == null) {
+ return Optional.empty();
+ }
+ String normalizedBackend = backend.toLowerCase(Locale.ROOT);
+ boolean eligible =
+
IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE.equals(normalizedBackend)
+ ||
IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_JDBC.equals(
+ normalizedBackend);
+ if (!eligible) {
+ return Optional.empty();
+ }
+
+ String routingEnabled =
+
sessionConfig.apply(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_ROUTING_ENABLED);
+ if (StringUtils.isNotBlank(routingEnabled)
Review Comment:
What if `routingEnabled` is blank? I think the logic here will not function
well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
