diqiu50 commented on code in PR #12710:
URL: https://github.com/apache/gravitino/pull/12710#discussion_r3910350394
##########
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:
Blank means unset, which falls through to the documented default
--
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]
