diqiu50 commented on code in PR #12710:
URL: https://github.com/apache/gravitino/pull/12710#discussion_r3910197534
##########
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:
Yes. that in the Trino connector
--
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]
