This is an automated email from the ASF dual-hosted git repository. diqiu50 pushed a commit to branch trino-irc-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit b3eef41feafe4d1f936a674cc4f5f86919478def Author: diqiu50 <[email protected]> AuthorDate: Tue Aug 25 17:46:11 2026 +0800 improvement(spark-connector): Route lakehouse-iceberg catalogs through the Iceberg REST server Mirrors the Trino connector's IRC routing: hive/jdbc backed lakehouse-iceberg catalogs are now automatically routed through Gravitino's Iceberg REST server when the server exposes a discoverable endpoint for the metalake, gaining vended-credentials support instead of static long-lived secrets. Manual override and REST client auth pass-through are configurable via spark.sql.gravitino.iceberg.rest-uri and spark.sql.gravitino.iceberg.rest.. --- docs/spark-connector/spark-catalog-iceberg.md | 40 +++++ docs/spark-connector/spark-connector.md | 2 + .../spark/connector/GravitinoSparkConfig.java | 9 ++ .../connector/catalog/GravitinoCatalogManager.java | 45 +++++- .../connector/iceberg/GravitinoIcebergCatalog.java | 73 ++++++++- .../iceberg/IcebergPropertiesConstants.java | 17 ++ .../iceberg/IcebergPropertiesConverter.java | 176 +++++++++++++++++++++ .../connector/plugin/GravitinoDriverPlugin.java | 1 + .../connector/glue/TestGravitinoGlueCatalog.java | 2 +- .../iceberg/TestIcebergPropertiesConverter.java | 136 ++++++++++++++++ .../connector/integration/test/SparkEnvIT.java | 59 ++++--- .../iceberg/SparkIcebergCatalogRestRoutingIT.java | 76 +++++++++ .../SparkIcebergCatalogRestRoutingIT33.java | 62 ++++++++ .../SparkIcebergCatalogRestRoutingIT34.java | 25 +++ .../SparkIcebergCatalogRestRoutingIT35.java | 25 +++ 15 files changed, 722 insertions(+), 26 deletions(-) diff --git a/docs/spark-connector/spark-catalog-iceberg.md b/docs/spark-connector/spark-catalog-iceberg.md index fd3210e4eb..ee8a5080a1 100644 --- a/docs/spark-connector/spark-catalog-iceberg.md +++ b/docs/spark-connector/spark-catalog-iceberg.md @@ -148,6 +148,46 @@ 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** +instead of the static, long-lived secrets (JDBC password, S3/OSS access keys) that would otherwise be +placed in the Spark Iceberg connector configuration. See [Credential vending](../security/credential-vending.md) +for how to enable credential vending on the Gravitino server. + +The endpoint is discovered once, when a catalog is first initialized in a Spark session; it is not +re-checked afterward. If the server does not expose a discoverable endpoint (for example, the `iceberg-rest` +auxiliary service is disabled, or not configured with `catalog-config-provider=dynamic-config-provider`), the +catalog falls back to the existing hive/jdbc translation described above. + +To force a specific endpoint instead of relying on auto-discovery, set: + +```properties +spark.sql.gravitino.iceberg.rest-uri http://<gravitino-host>:9001/iceberg +``` + +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. + +Because vended credentials are only consumed by Iceberg's native `FileIO` implementations, make sure the +warehouse storage jars listed under [Storage](#storage) below are on the Spark classpath; the connector +derives `io-impl` automatically from the warehouse location's scheme (`s3`/`s3a`/`s3n`, `gs`, +`abfs`/`abfss`/`wasb`/`wasbs`) unless `io-impl` is already set explicitly on the catalog. + ## Storage Spark connector could convert storage properties in the Gravitino catalog to Spark Iceberg connector automatically, No extra configuration is needed for `S3`, `ADLS`, `OSS`, `GCS`. diff --git a/docs/spark-connector/spark-connector.md b/docs/spark-connector/spark-connector.md index 5a3d8562c1..8f24b1d9b3 100644 --- a/docs/spark-connector/spark-connector.md +++ b/docs/spark-connector/spark-connector.md @@ -34,6 +34,8 @@ The Apache Gravitino Spark connector leverages the Spark DataSourceV2 interface | spark.sql.gravitino.enableIcebergSupport | string | `false` | Set to `true` to use Iceberg catalog. | No | | spark.sql.gravitino.enablePaimonSupport | string | `false` | Set to `true` to use Paimon catalog. | No | | spark.sql.gravitino.client. | string | (none) | The configuration key prefix for the Gravitino client config. | No | +| spark.sql.gravitino.iceberg.rest-uri | string | (none) | Overrides the auto-discovered Gravitino Iceberg REST server endpoint. See [Iceberg catalog](spark-catalog-iceberg.md#routing-through-the-gravitino-iceberg-rest-server). | No | +| spark.sql.gravitino.iceberg.rest. | string | (none) | The configuration key prefix for the Iceberg REST client config (e.g. `rest.auth.type`), applied when a catalog is routed through the Gravitino Iceberg REST server. | No | To configure the Gravitino client, use properties prefixed with `spark.sql.gravitino.client.`. These properties will be passed to the Gravitino client after removing the `spark.sql.` prefix. diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/GravitinoSparkConfig.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/GravitinoSparkConfig.java index be108c9af1..78f8de661e 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/GravitinoSparkConfig.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/GravitinoSparkConfig.java @@ -32,6 +32,15 @@ public class GravitinoSparkConfig { GRAVITINO_PREFIX + "enablePaimonSupport"; public static final String GRAVITINO_CLIENT_CONFIG_PREFIX = GRAVITINO_PREFIX + "client."; + // Manually overrides the Gravitino Iceberg REST server endpoint that hive/jdbc backed + // lakehouse-iceberg catalogs are routed through; takes precedence over auto-discovery. + public static final String GRAVITINO_ICEBERG_REST_URI = GRAVITINO_PREFIX + "iceberg.rest-uri"; + + // Pass-through prefix for the Iceberg REST client config (e.g. rest.auth.type, + // rest.auth.basic.username), applied when a catalog is routed through the Iceberg REST server. + public static final String GRAVITINO_ICEBERG_REST_CONFIG_PREFIX = + GRAVITINO_PREFIX + "iceberg.rest."; + public static final String GRAVITINO_AUTH_TYPE = GRAVITINO_PREFIX + AuthProperties.GRAVITINO_CLIENT_AUTH_TYPE; public static final String GRAVITINO_OAUTH2_URI = diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/GravitinoCatalogManager.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/GravitinoCatalogManager.java index a484972656..0d92b96ee8 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/GravitinoCatalogManager.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/GravitinoCatalogManager.java @@ -24,6 +24,7 @@ import com.google.common.base.Preconditions; import com.google.common.base.Supplier; import java.util.Arrays; import java.util.Map; +import java.util.Optional; import org.apache.gravitino.Catalog; import org.apache.gravitino.client.GravitinoClient; import org.slf4j.Logger; @@ -37,17 +38,24 @@ public class GravitinoCatalogManager { private volatile boolean isClosed = false; private final Cache<String, Catalog> gravitinoCatalogs; private final GravitinoClient gravitinoClient; + private final String metalakeName; - private GravitinoCatalogManager(Supplier<GravitinoClient> clientBuilder) { + // Resolved lazily on first access and cached for the life of this manager; Spark catalogs are + // initialized once, so there is no refresh path if the Iceberg REST endpoint changes later. + private volatile Optional<String> icebergRestUri; + + private GravitinoCatalogManager(String metalakeName, Supplier<GravitinoClient> clientBuilder) { + this.metalakeName = metalakeName; this.gravitinoClient = clientBuilder.get(); // Will not evict catalog by default this.gravitinoCatalogs = Caffeine.newBuilder().build(); } - public static GravitinoCatalogManager create(Supplier<GravitinoClient> clientBuilder) { + public static GravitinoCatalogManager create( + String metalakeName, Supplier<GravitinoClient> clientBuilder) { Preconditions.checkState( gravitinoCatalogManager == null, "Should not create duplicate GravitinoCatalogManager"); - gravitinoCatalogManager = new GravitinoCatalogManager(clientBuilder); + gravitinoCatalogManager = new GravitinoCatalogManager(metalakeName, clientBuilder); return gravitinoCatalogManager; } @@ -86,6 +94,37 @@ public class GravitinoCatalogManager { return gravitinoCatalogs.asMap(); } + /** + * Resolves the Gravitino Iceberg REST server endpoint for this manager's metalake, if the server + * exposes one. The lookup is performed once and the result, including a lookup failure, is cached + * for the lifetime of this manager. + * + * @return the discovered Iceberg REST endpoint, or empty if none is available + */ + public Optional<String> getIcebergRestUri() { + if (icebergRestUri == null) { + synchronized (this) { + if (icebergRestUri == null) { + icebergRestUri = resolveIcebergRestUri(); + } + } + } + return icebergRestUri; + } + + private Optional<String> resolveIcebergRestUri() { + try { + return gravitinoClient.icebergRestServiceUri(metalakeName); + } catch (Exception e) { + LOG.debug( + "No Iceberg REST server endpoint is available for metalake {}, " + + "falling back to native catalog backend routing.", + metalakeName, + e); + return Optional.empty(); + } + } + private Catalog loadCatalog(String catalogName) { Catalog catalog = gravitinoClient.loadCatalog(catalogName); Preconditions.checkArgument( diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/GravitinoIcebergCatalog.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/GravitinoIcebergCatalog.java index 0fa99d3ab7..cb0615dfad 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/GravitinoIcebergCatalog.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/GravitinoIcebergCatalog.java @@ -21,21 +21,29 @@ package org.apache.gravitino.spark.connector.iceberg; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; +import java.util.HashMap; +import java.util.Locale; import java.util.Map; +import java.util.Optional; +import java.util.stream.Collectors; +import java.util.stream.Stream; import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergPropertiesUtils; import org.apache.gravitino.credential.CredentialPropertyUtils; import org.apache.gravitino.rel.Table; +import org.apache.gravitino.spark.connector.GravitinoSparkConfig; import org.apache.gravitino.spark.connector.PropertiesConverter; import org.apache.gravitino.spark.connector.SparkTransformConverter; import org.apache.gravitino.spark.connector.SparkTypeConverter; import org.apache.gravitino.spark.connector.catalog.BaseCatalog; +import org.apache.gravitino.spark.connector.catalog.GravitinoCatalogManager; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.spark.SparkCatalog; import org.apache.iceberg.spark.procedures.SparkProcedures; import org.apache.iceberg.spark.source.HasIcebergCatalog; import org.apache.iceberg.spark.source.SparkTable; +import org.apache.spark.sql.SparkSession; import org.apache.spark.sql.catalyst.analysis.NoSuchFunctionException; import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException; import org.apache.spark.sql.catalyst.analysis.NoSuchProcedureException; @@ -72,15 +80,72 @@ public class GravitinoIcebergCatalog extends BaseCatalog } } String catalogBackendName = IcebergPropertiesUtils.getCatalogBackendName(properties); - Map<String, String> all = - getPropertiesConverter().toSparkCatalogProperties(options, properties); - CredentialPropertyUtils.applyIcebergCredentials( - CredentialPropertyUtils.getCredentials(gravitinoCatalogClient), all); + Optional<String> icebergRestUri = resolveIcebergRestUri(properties); + Map<String, String> all; + if (icebergRestUri.isPresent()) { + all = buildIcebergRestSparkCatalogProperties(name, options, properties, icebergRestUri.get()); + } 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. + */ + private Optional<String> resolveIcebergRestUri(Map<String, String> properties) { + 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 manualUri = + SparkSession.active().conf().get(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_URI, null); + if (StringUtils.isNotBlank(manualUri)) { + return Optional.of(manualUri); + } + return GravitinoCatalogManager.get().getIcebergRestUri(); + } + + private Map<String, String> buildIcebergRestSparkCatalogProperties( + String gravitinoCatalogName, + CaseInsensitiveStringMap options, + Map<String, String> properties, + String restUri) { + IcebergPropertiesConverter converter = (IcebergPropertiesConverter) getPropertiesConverter(); + Map<String, String> all = + new HashMap<>( + converter.buildIcebergRestProperties( + gravitinoCatalogName, restUri, properties, getIcebergRestClientConfig())); + if (options != null) { + all.putAll(options); + } + return all; + } + + private Map<String, String> getIcebergRestClientConfig() { + return Stream.of( + SparkSession.active() + .sparkContext() + .conf() + .getAllWithPrefix(GravitinoSparkConfig.GRAVITINO_ICEBERG_REST_CONFIG_PREFIX)) + .collect(Collectors.toMap(t -> t._1, t -> t._2, (oldVal, newVal) -> newVal)); + } + @Override protected org.apache.spark.sql.connector.catalog.Table createSparkTable( Identifier identifier, diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConstants.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConstants.java index 3d6ea418b4..71957d79da 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConstants.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConstants.java @@ -81,5 +81,22 @@ public class IcebergPropertiesConstants { static final String GRAVITINO_ICEBERG_CATALOG_BACKEND_NAME = IcebergConstants.CATALOG_BACKEND_NAME; + // Only used when routing a hive/jdbc backed catalog through the Gravitino Iceberg REST server; + // not part of the Gravitino <-> Iceberg property mapping in IcebergPropertiesUtils. + static final String ICEBERG_REST_CATALOG_PREFIX = "prefix"; + + static final String ICEBERG_IO_IMPL = IcebergConstants.IO_IMPL; + static final String ICEBERG_S3_ENDPOINT = IcebergConstants.ICEBERG_S3_ENDPOINT; + static final String ICEBERG_S3_PATH_STYLE_ACCESS = IcebergConstants.ICEBERG_S3_PATH_STYLE_ACCESS; + static final String ICEBERG_AWS_S3_REGION = IcebergConstants.AWS_S3_REGION; + static final String ICEBERG_OSS_ENDPOINT = IcebergConstants.ICEBERG_OSS_ENDPOINT; + + static final String ICEBERG_S3_FILE_IO_IMPL = "org.apache.iceberg.aws.s3.S3FileIO"; + static final String ICEBERG_GCS_FILE_IO_IMPL = "org.apache.iceberg.gcp.gcs.GCSFileIO"; + static final String ICEBERG_ADLS_FILE_IO_IMPL = "org.apache.iceberg.azure.adlsv2.ADLSFileIO"; + + static final String ICEBERG_ACCESS_DELEGATION = IcebergConstants.ICEBERG_ACCESS_DELEGATION; + static final String ICEBERG_ACCESS_DELEGATION_VENDED_CREDENTIALS = "vended-credentials"; + private IcebergPropertiesConstants() {} } diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConverter.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConverter.java index 9f7011aed5..e9e2a8d118 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConverter.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/iceberg/IcebergPropertiesConverter.java @@ -20,17 +20,33 @@ package org.apache.gravitino.spark.connector.iceberg; import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; import java.util.HashMap; +import java.util.List; +import java.util.Locale; import java.util.Map; import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergPropertiesUtils; +import org.apache.gravitino.credential.CredentialConstants; import org.apache.gravitino.spark.connector.PropertiesConverter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** Transform Apache Iceberg catalog properties between Apache Spark and Apache Gravitino. */ public class IcebergPropertiesConverter implements PropertiesConverter { + private static final Logger LOG = LoggerFactory.getLogger(IcebergPropertiesConverter.class); + + /** Routing keys the connector always derives itself; they cannot be overridden. */ + private static final List<String> RESERVED_REST_PROPERTIES = + ImmutableList.of( + IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE, + IcebergPropertiesConstants.ICEBERG_CATALOG_URI, + IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE, + IcebergPropertiesConstants.ICEBERG_REST_CATALOG_PREFIX); + public static class IcebergPropertiesConverterHolder { private static final IcebergPropertiesConverter INSTANCE = new IcebergPropertiesConverter(); } @@ -68,6 +84,166 @@ public class IcebergPropertiesConverter implements PropertiesConverter { 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); + + 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); + all.put(IcebergPropertiesConstants.ICEBERG_CATALOG_CACHE_ENABLED, "FALSE"); + return all; + } + + 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.info( + "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; + } + + private 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": + return IcebergPropertiesConstants.ICEBERG_GCS_FILE_IO_IMPL; + case "abfs": + case "abfss": + case "wasb": + case "wasbs": + return IcebergPropertiesConstants.ICEBERG_ADLS_FILE_IO_IMPL; + default: + return null; + } + } + + /** + * Warns when a warehouse scheme has no native Iceberg FileIO but the catalog vends credentials. + * Vended credentials are only consumed by Iceberg's native FileIO implementations, so table + * access would otherwise fail at read time with a storage authentication error far from its + * cause, unless {@code io-impl} is set explicitly on the catalog. + */ + private void warnOnSchemeWithoutNativeFileIo( + Map<String, String> gravitinoProperties, String warehouse) { + if (StringUtils.isBlank(warehouse) || !warehouse.contains("://")) { + // hdfs and file legitimately have no native FileIO here, and neither vends credentials. + return; + } + if (StringUtils.isBlank(gravitinoProperties.get(CredentialConstants.CREDENTIAL_PROVIDERS))) { + return; + } + LOG.warn( + "Warehouse '{}' has no native Iceberg FileIO for the credentials vended by '{}' to be " + + "applied to; table access may fail to authenticate unless 'io-impl' is set " + + "explicitly. Schemes with a native FileIO: s3/s3a/s3n, gs, abfs/abfss/wasb/wasbs.", + warehouse, + gravitinoProperties.get(CredentialConstants.CREDENTIAL_PROVIDERS)); + } + + private void copyIfPresent(Map<String, String> source, String key, Map<String, String> target) { + String value = source.get(key); + if (StringUtils.isNotBlank(value)) { + target.put(key, value); + } + } + @Override public Map<String, String> toGravitinoTableProperties(Map<String, String> properties) { return new HashMap<>(properties); diff --git a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/GravitinoDriverPlugin.java b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/GravitinoDriverPlugin.java index ea0e03aa9d..59636ec0ef 100644 --- a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/GravitinoDriverPlugin.java +++ b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/GravitinoDriverPlugin.java @@ -112,6 +112,7 @@ public class GravitinoDriverPlugin implements DriverPlugin { this.catalogManager = GravitinoCatalogManager.create( + metalake, () -> createGravitinoClient( gravitinoUri, metalake, conf, sc.sparkUser(), gravitinoClientConfig)); diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGravitinoGlueCatalog.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGravitinoGlueCatalog.java index af980f7309..c30265f089 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGravitinoGlueCatalog.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGravitinoGlueCatalog.java @@ -66,7 +66,7 @@ public class TestGravitinoGlueCatalog { // GravitinoGlueCatalog extends BaseCatalog which calls GravitinoCatalogManager.get() // in its constructor, so we must initialize the manager first. GravitinoClient mockClient = mock(GravitinoClient.class); - GravitinoCatalogManager.create(() -> mockClient); + GravitinoCatalogManager.create("test_metalake", () -> mockClient); } @AfterAll diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergPropertiesConverter.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergPropertiesConverter.java index 643a977cdb..e59fe216da 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergPropertiesConverter.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/iceberg/TestIcebergPropertiesConverter.java @@ -140,4 +140,140 @@ public class TestIcebergPropertiesConverter { "custom-warehouse"), properties); } + + @Test + void testIcebergRestPropertiesFromHiveBackendExcludesStaticCredentials() { + Map<String, String> gravitinoProperties = + ImmutableMap.of( + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND, + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE, + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_URI, + "thrift://hive-metastore:9083", + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE, + "s3://bucket/warehouse", + "s3-access-key-id", + "AKIDEXAMPLE", + "s3-secret-access-key", + "secret"); + + Map<String, String> properties = + icebergPropertiesConverter.buildIcebergRestProperties( + "my_catalog", "http://gravitino:9001/iceberg";, gravitinoProperties, ImmutableMap.of()); + + Assertions.assertEquals( + "rest", properties.get(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE)); + Assertions.assertEquals( + "http://gravitino:9001/iceberg";, + properties.get(IcebergPropertiesConstants.ICEBERG_CATALOG_URI)); + Assertions.assertEquals( + "my_catalog", properties.get(IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE)); + Assertions.assertEquals("my_catalog", properties.get("prefix")); + Assertions.assertEquals( + "vended-credentials", properties.get("header.X-Iceberg-Access-Delegation")); + Assertions.assertEquals( + "org.apache.iceberg.aws.s3.S3FileIO", properties.get(IcebergConstants.IO_IMPL)); + + // The REST protocol vends its own credentials, so no static secret should be present. + Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_S3_ACCESS_KEY_ID)); + Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_S3_SECRET_ACCESS_KEY)); + Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_JDBC_USER)); + Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_JDBC_PASSWORD)); + } + + @Test + void testIcebergRestPropertiesFromJdbcBackendExcludesStaticCredentials() { + Map<String, String> gravitinoProperties = + ImmutableMap.of( + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND, + IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_JDBC, + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_URI, + "jdbc:postgresql://db:5432/iceberg", + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_JDBC_USER, + "user", + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_JDBC_PASSWORD, + "passwd"); + + Map<String, String> properties = + icebergPropertiesConverter.buildIcebergRestProperties( + "my_catalog", "http://gravitino:9001/iceberg";, gravitinoProperties, ImmutableMap.of()); + + Assertions.assertEquals( + "rest", properties.get(IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE)); + Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_JDBC_USER)); + Assertions.assertFalse(properties.containsKey(IcebergConstants.ICEBERG_JDBC_PASSWORD)); + } + + @Test + void testIcebergRestPropertiesRespectsExplicitIoImpl() { + Map<String, String> gravitinoProperties = + ImmutableMap.of( + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND, + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE, + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE, + "s3://bucket/warehouse", + IcebergConstants.IO_IMPL, + "com.example.CustomFileIO"); + + Map<String, String> properties = + icebergPropertiesConverter.buildIcebergRestProperties( + "my_catalog", "http://gravitino:9001/iceberg";, gravitinoProperties, ImmutableMap.of()); + + Assertions.assertEquals("com.example.CustomFileIO", properties.get(IcebergConstants.IO_IMPL)); + } + + @Test + void testIcebergRestPropertiesWarehouseSchemeWithoutNativeFileIo() { + Map<String, String> gravitinoProperties = + ImmutableMap.of( + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND, + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE, + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE, + "hdfs://namenode/warehouse"); + + Map<String, String> properties = + icebergPropertiesConverter.buildIcebergRestProperties( + "my_catalog", "http://gravitino:9001/iceberg";, gravitinoProperties, ImmutableMap.of()); + + Assertions.assertFalse(properties.containsKey(IcebergConstants.IO_IMPL)); + } + + @Test + void testIcebergRestPropertiesBypassAndReservedKeyOverride() { + Map<String, String> gravitinoProperties = + ImmutableMap.of( + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND, + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE, + "spark.bypass.some-custom-property", + "custom-value", + // Attempting to override a reserved routing key via spark.bypass must be ignored. + "spark.bypass." + IcebergPropertiesConstants.ICEBERG_CATALOG_URI, + "http://attacker-controlled/iceberg";); + + Map<String, String> properties = + icebergPropertiesConverter.buildIcebergRestProperties( + "my_catalog", "http://gravitino:9001/iceberg";, gravitinoProperties, ImmutableMap.of()); + + Assertions.assertEquals("custom-value", properties.get("some-custom-property")); + Assertions.assertEquals( + "http://gravitino:9001/iceberg";, + properties.get(IcebergPropertiesConstants.ICEBERG_CATALOG_URI)); + } + + @Test + void testIcebergRestPropertiesRestClientConfigPassthrough() { + Map<String, String> gravitinoProperties = + ImmutableMap.of( + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND, + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND_HIVE); + + Map<String, String> properties = + icebergPropertiesConverter.buildIcebergRestProperties( + "my_catalog", + "http://gravitino:9001/iceberg";, + gravitinoProperties, + ImmutableMap.of("rest.auth.type", "basic", "rest.auth.basic.username", "admin")); + + Assertions.assertEquals("basic", properties.get("rest.auth.type")); + Assertions.assertEquals("admin", properties.get("rest.auth.basic.username")); + } } diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java index fcd2c2d819..8344ccc21b 100644 --- a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkEnvIT.java @@ -28,6 +28,7 @@ import java.util.HashMap; import java.util.Map; import org.apache.gravitino.Catalog; import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; import org.apache.gravitino.client.GravitinoMetalake; import org.apache.gravitino.function.FunctionDefinition; import org.apache.gravitino.function.FunctionDefinitions; @@ -84,6 +85,16 @@ public abstract class SparkEnvIT extends SparkUtilIT { return true; } + /** + * Whether the Iceberg REST auxiliary service should run with {@code dynamic-config-provider} + * (routing requests to whichever Gravitino catalog the request's {@code prefix} names) instead of + * the default single-catalog {@code static-config-provider}. Tests that verify the Spark + * connector's Iceberg REST auto-discovery/routing need this. + */ + protected boolean useDynamicIcebergRestConfigProvider() { + return false; + } + /** Returns the Gravitino {@link Catalog} for the catalog under test. */ protected Catalog getGravitinoCatalog() { return client.loadMetalake(metalakeName).loadCatalog(getCatalogName()); @@ -232,24 +243,36 @@ public abstract class SparkEnvIT extends SparkUtilIT { private void initIcebergRestServiceEnv() { super.ignoreIcebergAuxRestService = false; Map<String, String> icebergRestServiceConfigs = new HashMap<>(); - icebergRestServiceConfigs.put( - "gravitino." - + icebergRestServiceName - + "." - + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND, - IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_HIVE); - icebergRestServiceConfigs.put( - "gravitino." - + icebergRestServiceName - + "." - + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_URI, - hiveMetastoreUri); - icebergRestServiceConfigs.put( - "gravitino." - + icebergRestServiceName - + "." - + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE, - warehouse); + if (useDynamicIcebergRestConfigProvider()) { + icebergRestServiceConfigs.put( + "gravitino." + + icebergRestServiceName + + "." + + IcebergConstants.ICEBERG_REST_CATALOG_CONFIG_PROVIDER, + IcebergConstants.DYNAMIC_ICEBERG_CATALOG_CONFIG_PROVIDER_NAME); + icebergRestServiceConfigs.put( + "gravitino." + icebergRestServiceName + "." + IcebergConstants.GRAVITINO_METALAKE, + metalakeName); + } else { + icebergRestServiceConfigs.put( + "gravitino." + + icebergRestServiceName + + "." + + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND, + IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_HIVE); + icebergRestServiceConfigs.put( + "gravitino." + + icebergRestServiceName + + "." + + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_URI, + hiveMetastoreUri); + icebergRestServiceConfigs.put( + "gravitino." + + icebergRestServiceName + + "." + + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE, + warehouse); + } registerCustomConfigs(icebergRestServiceConfigs); } diff --git a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT.java b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT.java new file mode 100644 index 0000000000..80b726429b --- /dev/null +++ b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT.java @@ -0,0 +1,76 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.spark.connector.integration.test.iceberg; + +import com.google.common.collect.Maps; +import java.util.Map; +import org.apache.gravitino.spark.connector.iceberg.GravitinoIcebergCatalog; +import org.apache.gravitino.spark.connector.iceberg.IcebergPropertiesConstants; +import org.apache.iceberg.catalog.Catalog; +import org.apache.spark.sql.connector.catalog.CatalogPlugin; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIf; + +/** + * This class configures the Gravitino Iceberg catalog with a Hive backend, but runs the Iceberg + * REST auxiliary service in {@code dynamic-config-provider} mode so the Spark connector + * auto-discovers it and routes the catalog through the Iceberg REST protocol instead of talking to + * the Hive metastore directly. All test cases inherited from {@link SparkIcebergCatalogIT} exercise + * this REST-routed path end to end. + */ +@Tag("gravitino-docker-test") +// Spark connector uses a low Iceberg version, couldn't work with Iceberg REST server with high +// Iceberg version in embedded mode. +@DisabledIf("org.apache.gravitino.integration.test.util.ITUtils#isEmbedded") +public abstract class SparkIcebergCatalogRestRoutingIT extends SparkIcebergCatalogIT { + + @Override + protected boolean useDynamicIcebergRestConfigProvider() { + return true; + } + + @Override + protected Map<String, String> getCatalogConfigs() { + Map<String, String> catalogProperties = Maps.newHashMap(); + catalogProperties.put( + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_BACKEND, + IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_HIVE); + catalogProperties.put( + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE, warehouse); + catalogProperties.put( + IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_URI, hiveMetastoreUri); + return catalogProperties; + } + + @Test + void testCatalogIsRoutedThroughIcebergRestServer() { + CatalogPlugin catalogPlugin = + getSparkSession().sessionState().catalogManager().catalog(getCatalogName()); + Assertions.assertInstanceOf(GravitinoIcebergCatalog.class, catalogPlugin); + + Catalog icebergCatalog = ((GravitinoIcebergCatalog) catalogPlugin).icebergCatalog(); + Assertions.assertEquals( + "org.apache.iceberg.rest.RESTCatalog", + icebergCatalog.getClass().getName(), + "The hive-backed catalog should have been routed through the discovered Iceberg REST " + + "server, instead of Iceberg's native HiveCatalog."); + } +} diff --git a/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT33.java b/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT33.java new file mode 100644 index 0000000000..181cf6a0bc --- /dev/null +++ b/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT33.java @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.gravitino.spark.connector.integration.test.iceberg; + +import org.apache.gravitino.spark.connector.integration.test.util.SparkMetadataColumnInfo; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructField; +import org.junit.jupiter.api.condition.DisabledIf; + +/** + * Spark 3.3 integration tests for Iceberg REST routing. + * + * <p>Spark 3.3 pins Iceberg to 1.8.x ({@code iceberg4spark33}); Gravitino's lakehouse Iceberg stack + * is 1.11. {@link DisabledIf} skips embedded mode to avoid classpath conflicts. Deploy-mode tests + * run a 1.8.x client against Gravitino's 1.11 REST server and cover basic catalog CRUD only; they + * do not exercise Iceberg 1.11 REST changes such as {@code file-scan-tasks} scan planning or the + * configurable multi-level namespace separator contract. + */ +@DisabledIf("org.apache.gravitino.integration.test.util.ITUtils#isEmbedded") +public class SparkIcebergCatalogRestRoutingIT33 extends SparkIcebergCatalogRestRoutingIT { + @Override + protected boolean supportsFunction() { + // Spark 3.3 does not support function operations + return false; + } + + /** + * Spark 3.3 uses Iceberg 1.8.x; its {@code SparkTable#metadataColumns()} exposes five columns, + * not the seven row-lineage columns added in Iceberg 1.11. + */ + @Override + protected SparkMetadataColumnInfo[] getIcebergMetadataColumns() { + return new SparkMetadataColumnInfo[] { + new SparkMetadataColumnInfo("_spec_id", DataTypes.IntegerType, false), + new SparkMetadataColumnInfo( + "_partition", + DataTypes.createStructType( + new StructField[] {DataTypes.createStructField("name", DataTypes.StringType, true)}), + true), + new SparkMetadataColumnInfo("_file", DataTypes.StringType, false), + new SparkMetadataColumnInfo("_pos", DataTypes.LongType, false), + new SparkMetadataColumnInfo("_deleted", DataTypes.BooleanType, false) + }; + } +} diff --git a/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT34.java b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT34.java new file mode 100644 index 0000000000..ae23429b09 --- /dev/null +++ b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT34.java @@ -0,0 +1,25 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.gravitino.spark.connector.integration.test.iceberg; + +import org.junit.jupiter.api.condition.DisabledIf; + +@DisabledIf("org.apache.gravitino.integration.test.util.ITUtils#isEmbedded") +public class SparkIcebergCatalogRestRoutingIT34 extends SparkIcebergCatalogRestRoutingIT {} diff --git a/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT35.java b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT35.java new file mode 100644 index 0000000000..7a091bc0c5 --- /dev/null +++ b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogRestRoutingIT35.java @@ -0,0 +1,25 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.gravitino.spark.connector.integration.test.iceberg; + +import org.junit.jupiter.api.condition.DisabledIf; + +@DisabledIf("org.apache.gravitino.integration.test.util.ITUtils#isEmbedded") +public class SparkIcebergCatalogRestRoutingIT35 extends SparkIcebergCatalogRestRoutingIT {}
