Copilot commented on code in PR #12710:
URL: https://github.com/apache/gravitino/pull/12710#discussion_r3912061541
##########
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:
The default for spark.sql.gravitino.iceberg.rest-routing-enabled is shown as
`true`, but the implementation treats the unset case differently: when unset it
does best-effort routing (falls back to legacy if discovery fails / no
endpoint), while `true` explicitly requires routing and fails initialization
when no endpoint is available. Update the table default/description to reflect
the three-state behavior (unset vs true vs false).
##########
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);
+ }
Review Comment:
buildStorageProperties() calls warnOnSchemeWithoutNativeFileIo() whenever
the warehouse scheme has no built-in native FileIO, even if the catalog already
set io-impl explicitly. That produces misleading warnings in the exact case the
message says is safe (explicit io-impl). Only warn when neither a derived
FileIO nor an explicit io-impl is present.
##########
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);
+ }
Review Comment:
reapplyReservedRestProperties() logs a WARN every time it is called after
the initial derivation, because warnOnReservedRestPropertyOverrides() checks
only for key presence. In the auto-routed path, reserved keys are always
present on the second call (after merging Spark options), so this becomes noisy
and can mask real warnings. Consider warning only when a reserved key’s value
is actually being overridden (i.e., differs from the derived value) before
rewriting it.
--
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]
