This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 5b0c5ef1b5 [Cherry-pick to branch-1.3] [#11185] feat(spark-connector):
Add Glue catalog support for Spark connector (#11186) (#11428)
5b0c5ef1b5 is described below
commit 5b0c5ef1b5aeed6afa1f54b407a86dec709746af
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Jun 5 11:29:36 2026 +0800
[Cherry-pick to branch-1.3] [#11185] feat(spark-connector): Add Glue
catalog support for Spark connector (#11186) (#11428)
**Cherry-pick Information:**
- Original commit: c49183662b7f3770ea84df336a481b162bbea6b1
- Target branch: `branch-1.3`
- Status: ✅ Clean cherry-pick (no conflicts)
Co-authored-by: Yuhui <[email protected]>
---
.../gravitino/catalog/glue/GlueConstants.java | 3 +
.../catalog/glue/TestAwsGlueCatalogOperations.java | 4 +-
gradle/libs.versions.toml | 1 +
.../gravitino/integration/test/util/BaseIT.java | 14 +-
spark-connector/spark-common/build.gradle.kts | 49 ++
.../spark/connector/catalog/BaseCatalog.java | 65 ++-
.../connector/glue/GluePropertiesConverter.java | 223 ++++++++
.../spark/connector/glue/GravitinoGlueCatalog.java | 234 +++++++++
.../glue/GravitinoGlueCredentialsProvider.java | 74 +++
.../connector/hive/HivePropertiesConverter.java | 21 +-
.../connector/version/CatalogNameAdaptor.java | 58 ++-
.../glue/TestGluePropertiesConverter.java | 274 ++++++++++
.../connector/glue/TestGravitinoGlueCatalog.java | 217 ++++++++
.../glue/TestGravitinoGlueCredentialsProvider.java | 51 ++
.../connector/integration/test/SparkCommonIT.java | 13 +-
.../connector/integration/test/SparkEnvIT.java | 4 +-
.../integration/test/glue/SparkGlueCatalogIT.java | 573 +++++++++++++++++++++
.../integration/test/glue/SparkGlueEnvIT.java | 363 +++++++++++++
.../integration/test/util/SparkUtilIT.java | 56 +-
spark-connector/v3.3/spark/build.gradle.kts | 19 +
.../glue/GravitinoGlueCatalogSpark33.java | 22 +
.../test/glue/SparkAwsGlueCatalogIT33.java | 37 ++
spark-connector/v3.4/spark/build.gradle.kts | 19 +
.../glue/GravitinoGlueCatalogSpark34.java | 38 ++
.../test/glue/SparkAwsGlueCatalogIT34.java | 37 ++
spark-connector/v3.5/spark/build.gradle.kts | 19 +
.../glue/GravitinoGlueCatalogSpark35.java | 34 ++
.../test/glue/SparkAwsGlueCatalogIT35.java | 37 ++
28 files changed, 2482 insertions(+), 77 deletions(-)
diff --git
a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueConstants.java
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/glue/GlueConstants.java
similarity index 97%
rename from
catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueConstants.java
rename to
catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/glue/GlueConstants.java
index 24cc744a31..76de4483d5 100644
---
a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueConstants.java
+++
b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/glue/GlueConstants.java
@@ -82,6 +82,9 @@ public final class GlueConstants {
*/
public static final String TABLE_FORMAT = "table-format";
+ /** Table-format value indicating an Iceberg table stored in Glue {@code
Table.parameters()}. */
+ public static final String TABLE_FORMAT_ICEBERG = "ICEBERG";
+
/**
* Table file format (lowercase, e.g. "parquet", "orc", "textfile") used
when creating Hive-format
* tables via Trino. Maps to input-format/output-format/serde-lib.
diff --git
a/catalogs/catalog-glue/src/test/java/org/apache/gravitino/catalog/glue/TestAwsGlueCatalogOperations.java
b/catalogs/catalog-glue/src/test/java/org/apache/gravitino/catalog/glue/TestAwsGlueCatalogOperations.java
index aac40a92ac..2169eb418e 100644
---
a/catalogs/catalog-glue/src/test/java/org/apache/gravitino/catalog/glue/TestAwsGlueCatalogOperations.java
+++
b/catalogs/catalog-glue/src/test/java/org/apache/gravitino/catalog/glue/TestAwsGlueCatalogOperations.java
@@ -110,7 +110,7 @@ class TestAwsGlueCatalogOperations {
}
@Test
- void testRenameTableIsUnsupported() {
+ void testRenameIsUnsupported() {
String tableName = "rename_test_" + System.currentTimeMillis();
NameIdentifier ident = NameIdentifier.of(NS.level(0), NS.level(1),
TEST_SCHEMA, tableName);
@@ -119,7 +119,7 @@ class TestAwsGlueCatalogOperations {
ident,
new Column[] {col},
"test table",
- Map.of(GlueConstants.FORMAT, "parquet"),
+ Map.of(),
Transforms.EMPTY_TRANSFORM,
Distributions.NONE,
SortOrders.NONE,
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 7d3f7a00ae..a28060485e 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -162,6 +162,7 @@ ql-expression = "4.0.3"
[libraries]
aspectj-aspectjrt = { group = "org.aspectj", name = "aspectjrt", version.ref =
"aspectj" }
+aws-dynamodb = { group = "software.amazon.awssdk", name = "dynamodb",
version.ref = "awssdk" }
aws-glue = { group = "software.amazon.awssdk", name = "glue", version.ref =
"awssdk" }
aws-iam = { group = "software.amazon.awssdk", name = "iam", version.ref =
"awssdk" }
aws-policy = { group = "software.amazon.awssdk", name = "iam-policy-builder",
version.ref = "awssdk" }
diff --git
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java
index 5f328fb023..8c9cbfa510 100644
---
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java
+++
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/util/BaseIT.java
@@ -332,8 +332,13 @@ public class BaseIT {
}
}
- @BeforeAll
- public void startIntegrationTest() throws Exception {
+ /**
+ * Starts the Gravitino server (embedded or external) and initialises the
admin client. Extracted
+ * from {@link #startIntegrationTest()} so that subclasses can call it
directly without triggering
+ * virtual-dispatch through overriding {@code @BeforeAll} overrides (e.g.
{@code
+ * SparkEnvIT.startIntegrationTest()} which is intentionally empty).
+ */
+ protected void startServer() throws Exception {
testMode =
System.getProperty(ITUtils.TEST_MODE) == null
? ITUtils.EMBEDDED_TEST_MODE
@@ -464,6 +469,11 @@ public class BaseIT {
}
}
+ @BeforeAll
+ public void startIntegrationTest() throws Exception {
+ startServer();
+ }
+
@AfterAll
public void stopIntegrationTest() throws IOException, InterruptedException {
if (testMode != null && testMode.equals(ITUtils.EMBEDDED_TEST_MODE) &&
miniGravitino != null) {
diff --git a/spark-connector/spark-common/build.gradle.kts
b/spark-connector/spark-common/build.gradle.kts
index afd0324df8..7211d6e3bb 100644
--- a/spark-connector/spark-common/build.gradle.kts
+++ b/spark-connector/spark-common/build.gradle.kts
@@ -16,6 +16,8 @@
* specific language governing permissions and limitations
* under the License.
*/
+import java.net.URI
+
plugins {
`maven-publish`
id("java")
@@ -51,6 +53,7 @@ dependencies {
compileOnly(project(":clients:client-java-runtime", configuration =
"shadow"))
compileOnly("org.apache.iceberg:iceberg-spark-runtime-${sparkMajorVersion}_$scalaVersion:$icebergVersion")
+ compileOnly(libs.aws.glue)
compileOnly("org.apache.kyuubi:kyuubi-spark-connector-hive_$scalaVersion:$kyuubiVersion")
compileOnly("org.apache.spark:spark-catalyst_$scalaVersion:$sparkVersion")
compileOnly("org.apache.spark:spark-core_$scalaVersion:$sparkVersion")
@@ -87,6 +90,12 @@ dependencies {
testImplementation(project(":server-common")) {
exclude("org.apache.logging.log4j")
}
+ testImplementation(project(":catalogs:catalog-glue")) {
+ exclude("org.apache.logging.log4j")
+ }
+ // Iceberg's GlueCatalog (in iceberg-spark-runtime) references AWS Glue SDK
classes at runtime.
+ // catalog-glue uses `implementation` (not `api`) so the SDK is not
transitively exposed here.
+ testImplementation(libs.aws.glue)
testImplementation(project(":integration-test-common", "testArtifacts"))
testImplementation(libs.hive2.common) {
@@ -124,6 +133,8 @@ dependencies {
testImplementation(libs.nimbus.jose.jwt)
testImplementation(libs.mysql.driver)
testImplementation(libs.postgresql.driver)
+ testImplementation(libs.mockito.core)
+ testImplementation(libs.mockito.inline)
testImplementation(libs.testcontainers)
testImplementation("org.apache.iceberg:iceberg-core:$icebergVersion")
@@ -150,6 +161,44 @@ dependencies {
testRuntimeOnly(libs.junit.jupiter.engine)
}
+// Jars for Spark's IsolatedClientLoader: patched Hive 2.3.10 + Glue
datacatalog client.
+// aws-java-sdk-glue 1.12.31 requires
PropertyNamingStrategy$PascalCaseStrategy which was
+// removed in Jackson 2.12. Jackson included here so the isolated classloader
uses a compatible
+// version instead of the app-level Jackson bundled with Spark (2.14+).
+// Only set when AWS_ACCESS_KEY_ID is present so version-specific modules can
skip the download
+// when Glue tests are not enabled.
+val glueHiveJarsDir: String? =
+ if (System.getenv("AWS_ACCESS_KEY_ID") != null)
"$buildDir/tmp/glue-hive-jars" else null
+extra["glueHiveJarsDir"] = glueHiveJarsDir
+val glueLibsApiUrl =
+
"https://api.github.com/repos/datastrato/spark-hive-glue-libs/contents/spark3/glue-3.4.0";
+
+val downloadGlueHiveJars by
+tasks.registering {
+ glueHiveJarsDir?.let { outputs.dir(it) }
+ doLast {
+ val outputDir = file(glueHiveJarsDir ?: return@doLast)
+ outputDir.mkdirs()
+ val response = URI(glueLibsApiUrl).toURL().readText()
+
+ @Suppress("UNCHECKED_CAST")
+ val entries = groovy.json.JsonSlurper().parseText(response) as
List<Map<String, Any>>
+ entries
+ .filter { (it["name"] as String).endsWith(".jar") }
+ .forEach { entry ->
+ val jarName = entry["name"] as String
+ val downloadUrl = entry["download_url"] as String
+ val dest = outputDir.resolve(jarName)
+ if (!dest.exists()) {
+ logger.lifecycle("Downloading $jarName ...")
+ URI(downloadUrl).toURL().openStream().use { input ->
+ dest.outputStream().use { output -> input.copyTo(output) }
+ }
+ }
+ }
+ }
+}
+
tasks.test {
val skipITs = project.hasProperty("skipITs")
if (skipITs) {
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
index 2f26beae32..6f93518ad5 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java
@@ -197,34 +197,9 @@ public abstract class BaseCatalog implements TableCatalog,
SupportsNamespaces, F
public Table createTable(
Identifier ident, StructType schema, Transform[] transforms, Map<String,
String> properties)
throws TableAlreadyExistsException, NoSuchNamespaceException {
- NameIdentifier gravitinoIdentifier = NameIdentifier.of(getDatabase(ident),
ident.name());
- org.apache.gravitino.rel.Column[] gravitinoColumns =
- Arrays.stream(schema.fields())
- .map(structField -> createGravitinoColumn(structField))
- .toArray(org.apache.gravitino.rel.Column[]::new);
-
- Map<String, String> gravitinoProperties =
- propertiesConverter.toGravitinoTableProperties(properties);
- // Spark store comment in properties, we should retrieve it and pass to
Gravitino explicitly.
- String comment = gravitinoProperties.remove(ConnectorConstants.COMMENT);
-
- DistributionAndSortOrdersInfo distributionAndSortOrdersInfo =
-
sparkTransformConverter.toGravitinoDistributionAndSortOrders(transforms);
- org.apache.gravitino.rel.expressions.transforms.Transform[] partitionings =
- sparkTransformConverter.toGravitinoPartitionings(transforms);
-
try {
org.apache.gravitino.rel.Table gravitinoTable =
- gravitinoCatalogClient
- .asTableCatalog()
- .createTable(
- gravitinoIdentifier,
- gravitinoColumns,
- comment,
- gravitinoProperties,
- partitionings,
- distributionAndSortOrdersInfo.getDistribution(),
- distributionAndSortOrdersInfo.getSortOrders());
+ createGravitinoTable(ident, schema, transforms, properties);
org.apache.spark.sql.connector.catalog.Table sparkTable =
loadSparkTable(ident);
return createSparkTable(
ident,
@@ -241,6 +216,42 @@ public abstract class BaseCatalog implements TableCatalog,
SupportsNamespaces, F
}
}
+ /**
+ * Creates a table in Gravitino and returns the Gravitino table object.
Subclasses may call this
+ * to create via Gravitino without triggering the default {@code
loadSparkTable()} that follows in
+ * {@link #createTable}.
+ */
+ protected org.apache.gravitino.rel.Table createGravitinoTable(
+ Identifier ident, StructType schema, Transform[] transforms, Map<String,
String> properties)
+ throws NoSuchSchemaException,
org.apache.gravitino.exceptions.TableAlreadyExistsException {
+ NameIdentifier gravitinoIdentifier = NameIdentifier.of(getDatabase(ident),
ident.name());
+ org.apache.gravitino.rel.Column[] gravitinoColumns =
+ Arrays.stream(schema.fields())
+ .map(structField -> createGravitinoColumn(structField))
+ .toArray(org.apache.gravitino.rel.Column[]::new);
+
+ Map<String, String> gravitinoProperties =
+ propertiesConverter.toGravitinoTableProperties(properties);
+ // Spark store comment in properties, we should retrieve it and pass to
Gravitino explicitly.
+ String comment = gravitinoProperties.remove(ConnectorConstants.COMMENT);
+
+ DistributionAndSortOrdersInfo distributionAndSortOrdersInfo =
+
sparkTransformConverter.toGravitinoDistributionAndSortOrders(transforms);
+ org.apache.gravitino.rel.expressions.transforms.Transform[] partitionings =
+ sparkTransformConverter.toGravitinoPartitionings(transforms);
+
+ return gravitinoCatalogClient
+ .asTableCatalog()
+ .createTable(
+ gravitinoIdentifier,
+ gravitinoColumns,
+ comment,
+ gravitinoProperties,
+ partitionings,
+ distributionAndSortOrdersInfo.getDistribution(),
+ distributionAndSortOrdersInfo.getSortOrders());
+ }
+
@Override
public Table loadTable(Identifier ident) throws NoSuchTableException {
org.apache.gravitino.rel.Table gravitinoTable;
@@ -655,7 +666,7 @@ public abstract class BaseCatalog implements TableCatalog,
SupportsNamespaces, F
throw new NoSuchFunctionException(ident);
}
- private Table loadSparkTable(Identifier ident) {
+ protected Table loadSparkTable(Identifier ident) {
try {
return sparkCatalog.loadTable(ident);
} catch (NoSuchTableException e) {
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GluePropertiesConverter.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GluePropertiesConverter.java
new file mode 100644
index 0000000000..a24e5f315f
--- /dev/null
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GluePropertiesConverter.java
@@ -0,0 +1,223 @@
+/*
+ * 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.glue;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.catalog.glue.GlueConstants;
+import org.apache.gravitino.spark.connector.PropertiesConverter;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.aws.glue.GlueCatalog;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Transform AWS Glue catalog properties between Apache Spark and Apache
Gravitino.
+ *
+ * <p>This converter handles the property mapping for:
+ *
+ * <ul>
+ * <li>Non-Iceberg tables: pass through AWS credentials and region to
HiveTableCatalog
+ * <li>Iceberg tables: map Gravitino Glue properties to Iceberg's
GlueCatalog configuration
+ * </ul>
+ */
+public class GluePropertiesConverter implements PropertiesConverter {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(GluePropertiesConverter.class);
+
+ /** Iceberg GlueCatalog implementation class name, set as {@code
catalog-impl}. */
+ public static final String GLUE_CATALOG_IMPL = GlueCatalog.class.getName();
+
+ /** Iceberg property key for the AWS Glue catalog ID (account-level
catalog). */
+ public static final String GLUE_ID = "glue.id";
+
+ /** Iceberg property key for a custom AWS Glue service endpoint URL. */
+ public static final String GLUE_ENDPOINT = "glue.endpoint";
+
+ /** Iceberg property key for the AWS region used by the Glue client. */
+ public static final String CLIENT_REGION = "client.region";
+
+ /**
+ * Iceberg property key for the credentials provider class. Set to {@link
+ * #GRAVITINO_GLUE_CREDENTIALS_PROVIDER} when static credentials are
configured.
+ */
+ public static final String CLIENT_CREDENTIALS_PROVIDER =
"client.credentials-provider";
+
+ /**
+ * Fully-qualified class name of {@link GravitinoGlueCredentialsProvider}.
Implements {@code
+ * AwsCredentialsProvider} with a {@code create(Map)} factory so that
Iceberg's {@code
+ * AwsClientProperties} can instantiate it dynamically via {@code
client.credentials-provider}.
+ */
+ public static final String GRAVITINO_GLUE_CREDENTIALS_PROVIDER =
+
"org.apache.gravitino.spark.connector.glue.GravitinoGlueCredentialsProvider";
+
+ /** Gravitino catalog property key for the AWS access key ID. */
+ public static final String AWS_ACCESS_KEY_ID = "aws-access-key-id";
+
+ /** Gravitino catalog property key for the AWS secret access key. */
+ public static final String AWS_SECRET_ACCESS_KEY = "aws-secret-access-key";
+
+ /** Gravitino catalog property key for the AWS region. */
+ public static final String AWS_REGION = "aws-region";
+
+ /** Gravitino catalog property key for the AWS Glue catalog ID. */
+ public static final String AWS_GLUE_CATALOG_ID = "aws-glue-catalog-id";
+
+ /** Gravitino catalog property key for a custom AWS Glue endpoint URL. */
+ public static final String AWS_GLUE_ENDPOINT = "aws-glue-endpoint";
+
+ private static final String SPARK_PROVIDER_ICEBERG = "iceberg";
+
+ /**
+ * Hive config key that activates the AWS Glue Data Catalog as the Hive
metastore client.
+ *
+ * <p>Note: the patched Hive 2.3.10 (HIVE-12679) uses the ConfVar {@code
+ * IMETASTORE_CLIENT_FACTORY_CLASS} whose string key is {@code
+ * hive.imetastoreclient.factory.class} — not the commonly documented {@code
+ * hive.metastore.client.factory.class}.
+ */
+ public static final String HIVE_METASTORE_CLIENT_FACTORY_CLASS =
+ "hive.imetastoreclient.factory.class";
+
+ /** AWS Glue Data Catalog Hive client factory class name. */
+ public static final String AWS_GLUE_HIVE_CLIENT_FACTORY_CLASS =
+
"com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory";
+
+ /** Maps Gravitino property keys to Spark HiveTableCatalog property keys. */
+ private static final Map<String, String> GRAVITINO_TO_SPARK_KEYS =
+ ImmutableMap.of(
+ AWS_REGION, "aws.region",
+ AWS_GLUE_CATALOG_ID, "aws.glue.catalog.id",
+ AWS_GLUE_ENDPOINT, "aws.glue.endpoint");
+
+ /** Maps Gravitino property keys to Iceberg GlueCatalog property keys. */
+ private static final Map<String, String> GRAVITINO_TO_ICEBERG_KEYS =
+ ImmutableMap.of(
+ AWS_REGION, CLIENT_REGION,
+ AWS_GLUE_CATALOG_ID, GLUE_ID,
+ AWS_GLUE_ENDPOINT, GLUE_ENDPOINT);
+
+ private static class GluePropertiesConverterHolder {
+ private static final GluePropertiesConverter INSTANCE = new
GluePropertiesConverter();
+ }
+
+ private GluePropertiesConverter() {}
+
+ /**
+ * Returns the singleton instance of {@link GluePropertiesConverter}.
+ *
+ * @return the singleton instance
+ */
+ public static GluePropertiesConverter getInstance() {
+ return GluePropertiesConverterHolder.INSTANCE;
+ }
+
+ /**
+ * Transform Gravitino Glue catalog properties to Spark catalog properties
for HiveTableCatalog.
+ *
+ * <p>Sets {@code hive.imetastoreclient.factory.class} to {@link
+ * #AWS_GLUE_HIVE_CLIENT_FACTORY_CLASS}, replacing the embedded Derby
metastore with a direct Glue
+ * API connection. This eliminates Derby and makes the catalog work
correctly across multiple
+ * Spark applications.
+ *
+ * <p>AWS credentials are read via the default AWS credential chain
(environment variables,
+ * instance profile, etc.) inside the isolated Hive classloader. The static
credentials in
+ * Gravitino catalog properties are used by the Iceberg GlueCatalog path
instead.
+ */
+ @Override
+ public Map<String, String> toSparkCatalogProperties(Map<String, String>
properties) {
+ Preconditions.checkArgument(properties != null, "Glue Catalog properties
should not be null");
+ HashMap<String, String> all = new HashMap<>();
+ all.put(HIVE_METASTORE_CLIENT_FACTORY_CLASS,
AWS_GLUE_HIVE_CLIENT_FACTORY_CLASS);
+ GRAVITINO_TO_SPARK_KEYS.forEach(
+ (gravitinoKey, sparkKey) -> {
+ String value = properties.get(gravitinoKey);
+ if (StringUtils.isNotBlank(value)) {
+ all.put(sparkKey, value);
+ }
+ });
+ return all;
+ }
+
+ /**
+ * Transform Gravitino Glue catalog properties to Iceberg SparkCatalog
properties for Iceberg
+ * tables stored in Glue.
+ *
+ * <p>This maps Gravitino's AWS Glue properties to Iceberg's GlueCatalog
configuration:
+ *
+ * <ul>
+ * <li>{@code aws-region} → {@code client.region} (optional; falls back to
SDK default chain)
+ * <li>{@code aws-glue-catalog-id} → {@code glue.id}
+ * <li>{@code aws-glue-endpoint} → {@code glue.endpoint} (optional)
+ * <li>{@code aws-access-key-id} + {@code aws-secret-access-key} → {@code
+ * client.credentials-provider=GravitinoGlueCredentialsProvider} with
credentials passed via
+ * {@code client.credentials-provider.*} properties
+ * </ul>
+ */
+ @VisibleForTesting
+ Map<String, String> toIcebergCatalogProperties(Map<String, String>
properties) {
+ Preconditions.checkArgument(properties != null, "Glue Catalog properties
should not be null");
+ HashMap<String, String> all = new HashMap<>();
+ all.put(CatalogProperties.CATALOG_IMPL, GLUE_CATALOG_IMPL);
+ GRAVITINO_TO_ICEBERG_KEYS.forEach(
+ (gravitinoKey, icebergKey) -> {
+ String value = properties.get(gravitinoKey);
+ if (StringUtils.isNotBlank(value)) {
+ all.put(icebergKey, value);
+ }
+ });
+ String accessKey = properties.get(AWS_ACCESS_KEY_ID);
+ String secretKey = properties.get(AWS_SECRET_ACCESS_KEY);
+ if (StringUtils.isNotBlank(accessKey) &&
StringUtils.isNotBlank(secretKey)) {
+ // Iceberg 1.10+ reads credentials via client.credentials-provider (a
class with create(Map)).
+ // The client.credentials-provider.* properties are stripped of their
prefix and passed to
+ // GravitinoGlueCredentialsProvider.create(Map) as {"access-key-id":
..., "secret-access-key":
+ // ...}.
+ all.put(CLIENT_CREDENTIALS_PROVIDER,
GRAVITINO_GLUE_CREDENTIALS_PROVIDER);
+ all.put("client.credentials-provider.access-key-id", accessKey);
+ all.put("client.credentials-provider.secret-access-key", secretKey);
+ } else {
+ LOG.debug(
+ "aws-access-key-id or aws-secret-access-key not configured; "
+ + "falling back to the default AWS credential chain (instance
profile, env vars, etc.).");
+ }
+ return all;
+ }
+
+ @Override
+ public Map<String, String> toGravitinoTableProperties(Map<String, String>
properties) {
+ HashMap<String, String> all = new HashMap<>(properties);
+ String provider = all.remove(TableCatalog.PROP_PROVIDER);
+ if (SPARK_PROVIDER_ICEBERG.equalsIgnoreCase(provider)) {
+ all.put(GlueConstants.TABLE_FORMAT, GlueConstants.TABLE_FORMAT_ICEBERG);
+ }
+ return all;
+ }
+
+ @Override
+ public Map<String, String> toSparkTableProperties(Map<String, String>
properties) {
+ return new HashMap<>(properties);
+ }
+}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalog.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalog.java
new file mode 100644
index 0000000000..fc7dbce568
--- /dev/null
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalog.java
@@ -0,0 +1,234 @@
+/*
+ * 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.glue;
+
+import com.google.common.base.Preconditions;
+import java.util.Map;
+import org.apache.gravitino.catalog.glue.GlueConstants;
+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.hive.SparkHiveTable;
+import org.apache.gravitino.spark.connector.hive.SparkHiveTypeConverter;
+import org.apache.gravitino.spark.connector.iceberg.SparkIcebergTable;
+import org.apache.iceberg.spark.SparkCatalog;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.kyuubi.spark.connector.hive.HiveTable;
+import org.apache.kyuubi.spark.connector.hive.HiveTableCatalog;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+/**
+ * Gravitino Glue catalog implementation for Apache Spark.
+ *
+ * <p>This catalog handles mixed table types stored in AWS Glue Data Catalog:
+ *
+ * <ul>
+ * <li>Non-Iceberg tables (Hive, Delta, Parquet): routed to {@link
HiveTableCatalog} backed by the
+ * AWS Glue Data Catalog Hive client ({@code
AWSGlueDataCatalogHiveClientFactory})
+ * <li>Iceberg tables: routed to Iceberg's {@link SparkCatalog}
(GlueCatalog) for I/O
+ * </ul>
+ *
+ * <p>Table routing is based on the {@code table-format} property in Glue
table parameters. Tables
+ * with {@code table-format=ICEBERG} are delegated to the Iceberg backend.
+ *
+ * <p>{@link HiveTableCatalog} is configured with {@code
hive.metastore.client.factory.class} set to
+ * {@code AWSGlueDataCatalogHiveClientFactory}, which replaces the embedded
Derby metastore with a
+ * direct AWS Glue API connection. This means no local Derby sync is required
and the catalog works
+ * correctly across multiple concurrent Spark applications sharing the same
Glue catalog.
+ */
+public class GravitinoGlueCatalog extends BaseCatalog {
+
+ // Lazily initialized Iceberg GlueCatalog for Iceberg tables
+ private volatile SparkCatalog icebergGlueCatalog;
+
+ // Store original config for Iceberg catalog initialization
+ private String catalogName;
+ private Map<String, String> catalogProperties;
+
+ /** Creates a new GravitinoGlueCatalog. */
+ public GravitinoGlueCatalog() {}
+
+ /**
+ * Creates a new HiveTableCatalog instance. Override in tests to inject mock
instances.
+ *
+ * @return a new HiveTableCatalog
+ */
+ protected HiveTableCatalog createHiveTableCatalog() {
+ return new HiveTableCatalog();
+ }
+
+ @Override
+ protected TableCatalog createAndInitSparkCatalog(
+ String name, CaseInsensitiveStringMap options, Map<String, String>
properties) {
+ this.catalogName = name;
+ this.catalogProperties = properties;
+
+ TableCatalog hiveCatalog = createHiveTableCatalog();
+ Map<String, String> all =
+ getPropertiesConverter().toSparkCatalogProperties(options, properties);
+ hiveCatalog.initialize(name, new CaseInsensitiveStringMap(all));
+ return hiveCatalog;
+ }
+
+ /**
+ * Routes Spark table loading to the correct backend.
+ *
+ * <p>Iceberg tables are loaded from the Iceberg GlueCatalog. Non-Iceberg
tables are loaded
+ * directly from {@link HiveTableCatalog}, which reads from Glue via the
factory client.
+ */
+ @Override
+ protected Table loadSparkTable(Identifier ident) {
+ try {
+ org.apache.gravitino.rel.Table gravitinoTable =
loadGravitinoTable(ident);
+ if (isIcebergTable(gravitinoTable)) {
+ return getOrCreateIcebergGlueCatalog().loadTable(ident);
+ }
+ return sparkCatalog.loadTable(ident);
+ } catch (NoSuchTableException e) {
+ throw new RuntimeException(
+ String.format("Failed to load spark table: %s.%s",
getDatabase(ident), ident.name()), e);
+ }
+ }
+
+ /**
+ * Routes table creation to the appropriate Spark wrapper based on the
Gravitino table type.
+ * Iceberg tables are wrapped in {@link SparkIcebergTable}; all others in
{@link SparkHiveTable}.
+ */
+ @Override
+ protected Table createSparkTable(
+ Identifier identifier,
+ org.apache.gravitino.rel.Table gravitinoTable,
+ Table sparkTable,
+ TableCatalog sparkHiveCatalog,
+ PropertiesConverter propertiesConverter,
+ SparkTransformConverter sparkTransformConverter,
+ SparkTypeConverter sparkTypeConverter) {
+
+ if (isIcebergTable(gravitinoTable)) {
+ Preconditions.checkArgument(
+ sparkTable instanceof SparkTable,
+ "Iceberg table %s expected SparkTable from Iceberg backend, got %s",
+ identifier,
+ sparkTable.getClass().getName());
+ return new SparkIcebergTable(
+ identifier,
+ gravitinoTable,
+ (SparkTable) sparkTable,
+ getOrCreateIcebergGlueCatalog(),
+ propertiesConverter,
+ sparkTransformConverter,
+ sparkTypeConverter);
+ }
+
+ return new SparkHiveTable(
+ identifier,
+ gravitinoTable,
+ (HiveTable) sparkTable,
+ (HiveTableCatalog) sparkHiveCatalog,
+ propertiesConverter,
+ sparkTransformConverter,
+ sparkTypeConverter);
+ }
+
+ /** {@inheritDoc} Returns the Glue-specific properties converter singleton.
*/
+ @Override
+ protected PropertiesConverter getPropertiesConverter() {
+ return GluePropertiesConverter.getInstance();
+ }
+
+ /** {@inheritDoc} Returns a transform converter with identity partition
support disabled. */
+ @Override
+ protected SparkTransformConverter getSparkTransformConverter() {
+ return new SparkTransformConverter(false);
+ }
+
+ /** {@inheritDoc} Returns the Hive-compatible type converter used for Glue
tables. */
+ @Override
+ protected SparkTypeConverter getSparkTypeConverter() {
+ return new SparkHiveTypeConverter();
+ }
+
+ /**
+ * Returns true if the Gravitino table is an Iceberg-format table based on
its properties.
+ *
+ * @param gravitinoTable the Gravitino table to inspect
+ * @return true for Iceberg tables, false otherwise
+ */
+ static boolean isIcebergTable(org.apache.gravitino.rel.Table gravitinoTable)
{
+ Map<String, String> properties = gravitinoTable.properties();
+ if (properties == null) {
+ return false;
+ }
+ // Gravitino convention: table-format=ICEBERG
+ String tableFormat = properties.get(GlueConstants.TABLE_FORMAT);
+ if (GlueConstants.TABLE_FORMAT_ICEBERG.equalsIgnoreCase(tableFormat)) {
+ return true;
+ }
+ // Iceberg Glue catalog convention: table_type=ICEBERG stored in Glue
table parameters
+ return GlueConstants.ICEBERG_TABLE_TYPE_VALUE.equalsIgnoreCase(
+ properties.get(GlueConstants.TABLE_TYPE_PARAM));
+ }
+
+ /**
+ * Gets or creates the Iceberg GlueCatalog using double-checked locking.
+ *
+ * @return the Iceberg SparkCatalog
+ */
+ private SparkCatalog getOrCreateIcebergGlueCatalog() {
+ if (icebergGlueCatalog == null) {
+ synchronized (this) {
+ if (icebergGlueCatalog == null) {
+ Preconditions.checkArgument(
+ catalogName != null && catalogProperties != null,
+ "Catalog name and properties must be set before accessing
Iceberg catalog");
+ try {
+ icebergGlueCatalog = createIcebergGlueCatalog();
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to initialize Iceberg GlueCatalog for catalog
'%s'. "
+ + "Check aws-region, aws-access-key-id, and
aws-secret-access-key properties.",
+ catalogName),
+ e);
+ }
+ }
+ }
+ }
+ return icebergGlueCatalog;
+ }
+
+ /**
+ * Creates a new Iceberg GlueCatalog with appropriate configuration.
+ *
+ * @return the configured Iceberg SparkCatalog
+ */
+ private SparkCatalog createIcebergGlueCatalog() {
+ GluePropertiesConverter converter = GluePropertiesConverter.getInstance();
+ Map<String, String> icebergProperties =
converter.toIcebergCatalogProperties(catalogProperties);
+ SparkCatalog catalog = new SparkCatalog();
+ catalog.initialize(catalogName + "_iceberg", new
CaseInsensitiveStringMap(icebergProperties));
+ return catalog;
+ }
+}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCredentialsProvider.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCredentialsProvider.java
new file mode 100644
index 0000000000..b805d40d54
--- /dev/null
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCredentialsProvider.java
@@ -0,0 +1,74 @@
+/*
+ * 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.glue;
+
+import com.google.common.base.Preconditions;
+import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+
+/**
+ * AWS credentials provider for Iceberg {@code GlueCatalog} that reads static
credentials from a
+ * properties map.
+ *
+ * <p>Iceberg 1.10+ loads credentials via {@code client.credentials-provider}.
This class is
+ * instantiated dynamically by Iceberg's {@code AwsClientProperties} using the
{@code create(Map)}
+ * static factory. The properties map contains the {@code
client.credentials-provider.*} entries
+ * with their prefix stripped, i.e. {@code access-key-id} and {@code
secret-access-key}.
+ */
+public class GravitinoGlueCredentialsProvider implements
AwsCredentialsProvider {
+
+ private static final String ACCESS_KEY_ID = "access-key-id";
+ private static final String SECRET_ACCESS_KEY = "secret-access-key";
+
+ private final String accessKeyId;
+ private final String secretAccessKey;
+
+ /**
+ * Creates a credentials provider from the given properties map.
+ *
+ * @param properties map containing {@code access-key-id} and {@code
secret-access-key}
+ * @return a new {@link AwsCredentialsProvider} instance
+ */
+ public static AwsCredentialsProvider create(Map<String, String> properties) {
+ Preconditions.checkArgument(properties != null, "Credentials properties
must not be null");
+ String accessKeyId = properties.get(ACCESS_KEY_ID);
+ String secretAccessKey = properties.get(SECRET_ACCESS_KEY);
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(accessKeyId),
+ "Glue credentials provider requires 'access-key-id' in
client.credentials-provider.* properties");
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(secretAccessKey),
+ "Glue credentials provider requires 'secret-access-key' in
client.credentials-provider.* properties");
+ return new GravitinoGlueCredentialsProvider(accessKeyId, secretAccessKey);
+ }
+
+ GravitinoGlueCredentialsProvider(String accessKeyId, String secretAccessKey)
{
+ this.accessKeyId = accessKeyId;
+ this.secretAccessKey = secretAccessKey;
+ }
+
+ @Override
+ public AwsCredentials resolveCredentials() {
+ return AwsBasicCredentials.create(accessKeyId, secretAccessKey);
+ }
+}
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/HivePropertiesConverter.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/HivePropertiesConverter.java
index 77fb6f3f05..bf8d0bbc3a 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/HivePropertiesConverter.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/hive/HivePropertiesConverter.java
@@ -47,16 +47,17 @@ public class HivePropertiesConverter implements
PropertiesConverter {
// Transform Spark Hive file format to Gravitino hive file format
static final Map<String, String> fileFormatMap =
- ImmutableMap.of(
- "sequencefile",
HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_SEQUENCEFILE,
- "rcfile", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_RCFILE,
- "orc", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_ORC,
- "parquet", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_PARQUET,
- "textfile", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_TEXTFILE,
- "hive", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_TEXTFILE,
- "json", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_JSON,
- "csv", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_CSV,
- "avro", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_AVRO);
+ ImmutableMap.<String, String>builder()
+ .put("sequencefile",
HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_SEQUENCEFILE)
+ .put("rcfile", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_RCFILE)
+ .put("orc", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_ORC)
+ .put("parquet",
HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_PARQUET)
+ .put("textfile",
HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_TEXTFILE)
+ .put("hive", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_TEXTFILE)
+ .put("json", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_JSON)
+ .put("csv", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_CSV)
+ .put("avro", HivePropertiesConstants.GRAVITINO_HIVE_FORMAT_AVRO)
+ .build();
static final Map<String, String> sparkToGravitinoPropertyMap =
ImmutableMap.of(
diff --git
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/version/CatalogNameAdaptor.java
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/version/CatalogNameAdaptor.java
index 8724941529..c5c4bbcc78 100644
---
a/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/version/CatalogNameAdaptor.java
+++
b/spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/version/CatalogNameAdaptor.java
@@ -25,26 +25,44 @@ import org.apache.spark.package$;
import org.apache.spark.util.VersionUtils$;
public class CatalogNameAdaptor {
- private static final Map<String, String> catalogNames =
- ImmutableMap.of(
- "hive-3.3",
-
"org.apache.gravitino.spark.connector.hive.GravitinoHiveCatalogSpark33",
- "hive-3.4",
-
"org.apache.gravitino.spark.connector.hive.GravitinoHiveCatalogSpark34",
- "hive-3.5",
-
"org.apache.gravitino.spark.connector.hive.GravitinoHiveCatalogSpark35",
- "lakehouse-iceberg-3.3",
-
"org.apache.gravitino.spark.connector.iceberg.GravitinoIcebergCatalogSpark33",
- "lakehouse-iceberg-3.4",
-
"org.apache.gravitino.spark.connector.iceberg.GravitinoIcebergCatalogSpark34",
- "lakehouse-iceberg-3.5",
-
"org.apache.gravitino.spark.connector.iceberg.GravitinoIcebergCatalogSpark35",
- "lakehouse-paimon-3.3",
-
"org.apache.gravitino.spark.connector.paimon.GravitinoPaimonCatalogSpark33",
- "lakehouse-paimon-3.4",
-
"org.apache.gravitino.spark.connector.paimon.GravitinoPaimonCatalogSpark34",
- "lakehouse-paimon-3.5",
-
"org.apache.gravitino.spark.connector.paimon.GravitinoPaimonCatalogSpark35");
+ private static final Map<String, String> catalogNames;
+
+ static {
+ ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
+ catalogNames =
+ builder
+ .put(
+ "hive-3.3",
"org.apache.gravitino.spark.connector.hive.GravitinoHiveCatalogSpark33")
+ .put(
+ "hive-3.4",
"org.apache.gravitino.spark.connector.hive.GravitinoHiveCatalogSpark34")
+ .put(
+ "hive-3.5",
"org.apache.gravitino.spark.connector.hive.GravitinoHiveCatalogSpark35")
+ .put(
+ "lakehouse-iceberg-3.3",
+
"org.apache.gravitino.spark.connector.iceberg.GravitinoIcebergCatalogSpark33")
+ .put(
+ "lakehouse-iceberg-3.4",
+
"org.apache.gravitino.spark.connector.iceberg.GravitinoIcebergCatalogSpark34")
+ .put(
+ "lakehouse-iceberg-3.5",
+
"org.apache.gravitino.spark.connector.iceberg.GravitinoIcebergCatalogSpark35")
+ .put(
+ "lakehouse-paimon-3.3",
+
"org.apache.gravitino.spark.connector.paimon.GravitinoPaimonCatalogSpark33")
+ .put(
+ "lakehouse-paimon-3.4",
+
"org.apache.gravitino.spark.connector.paimon.GravitinoPaimonCatalogSpark34")
+ .put(
+ "lakehouse-paimon-3.5",
+
"org.apache.gravitino.spark.connector.paimon.GravitinoPaimonCatalogSpark35")
+ .put(
+ "glue-3.3",
"org.apache.gravitino.spark.connector.glue.GravitinoGlueCatalogSpark33")
+ .put(
+ "glue-3.4",
"org.apache.gravitino.spark.connector.glue.GravitinoGlueCatalogSpark34")
+ .put(
+ "glue-3.5",
"org.apache.gravitino.spark.connector.glue.GravitinoGlueCatalogSpark35")
+ .build();
+ }
private static final Map<String, String> jdbcCatalogNames =
ImmutableMap.of(
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGluePropertiesConverter.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGluePropertiesConverter.java
new file mode 100644
index 0000000000..1340a85935
--- /dev/null
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGluePropertiesConverter.java
@@ -0,0 +1,274 @@
+/*
+ * 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.glue;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.TestInstance.Lifecycle;
+
+@TestInstance(Lifecycle.PER_CLASS)
+public class TestGluePropertiesConverter {
+ private final GluePropertiesConverter converter =
GluePropertiesConverter.getInstance();
+
+ // --- toIcebergCatalogProperties tests ---
+
+ @Test
+ void testToIcebergCatalogPropertiesWithRegion() {
+ Map<String, String> icebergProps =
+ converter.toIcebergCatalogProperties(
+ ImmutableMap.of(GluePropertiesConverter.AWS_REGION, "us-east-1"));
+ Assertions.assertEquals(
+ "org.apache.iceberg.aws.glue.GlueCatalog",
icebergProps.get("catalog-impl"));
+ Assertions.assertEquals("us-east-1",
icebergProps.get(GluePropertiesConverter.CLIENT_REGION));
+ Assertions.assertEquals(2, icebergProps.size());
+ }
+
+ @Test
+ void testToIcebergCatalogPropertiesWithAllProperties() {
+ Map<String, String> icebergProps =
+ converter.toIcebergCatalogProperties(
+ ImmutableMap.of(
+ GluePropertiesConverter.AWS_REGION, "us-west-2",
+ GluePropertiesConverter.AWS_GLUE_CATALOG_ID, "123456789012",
+ GluePropertiesConverter.AWS_GLUE_ENDPOINT,
"http://localhost:4566";,
+ GluePropertiesConverter.AWS_ACCESS_KEY_ID,
"AKIAIOSFODNN7EXAMPLE",
+ GluePropertiesConverter.AWS_SECRET_ACCESS_KEY,
+ "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"));
+ Assertions.assertEquals(
+ "org.apache.iceberg.aws.glue.GlueCatalog",
icebergProps.get("catalog-impl"));
+ Assertions.assertEquals("us-west-2",
icebergProps.get(GluePropertiesConverter.CLIENT_REGION));
+ Assertions.assertEquals("123456789012",
icebergProps.get(GluePropertiesConverter.GLUE_ID));
+ Assertions.assertEquals(
+ "http://localhost:4566";,
icebergProps.get(GluePropertiesConverter.GLUE_ENDPOINT));
+ Assertions.assertEquals(
+
"org.apache.gravitino.spark.connector.glue.GravitinoGlueCredentialsProvider",
+ icebergProps.get(GluePropertiesConverter.CLIENT_CREDENTIALS_PROVIDER));
+ Assertions.assertEquals(
+ "AKIAIOSFODNN7EXAMPLE",
icebergProps.get("client.credentials-provider.access-key-id"));
+ Assertions.assertEquals(
+ "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+ icebergProps.get("client.credentials-provider.secret-access-key"));
+ Assertions.assertEquals(7, icebergProps.size());
+ }
+
+ @Test
+ void testToIcebergCatalogPropertiesWithOptionalCatalogId() {
+ Map<String, String> icebergProps =
+ converter.toIcebergCatalogProperties(
+ ImmutableMap.of(GluePropertiesConverter.AWS_REGION,
"eu-central-1"));
+ Assertions.assertEquals(
+ "eu-central-1",
icebergProps.get(GluePropertiesConverter.CLIENT_REGION));
+ Assertions.assertNull(icebergProps.get(GluePropertiesConverter.GLUE_ID));
+ }
+
+ @Test
+ void testToIcebergCatalogPropertiesWithOptionalEndpoint() {
+ Map<String, String> icebergProps =
+ converter.toIcebergCatalogProperties(
+ ImmutableMap.of(
+ GluePropertiesConverter.AWS_REGION,
+ "us-east-1",
+ GluePropertiesConverter.AWS_GLUE_ENDPOINT,
+ "http://localhost:4566";));
+ Assertions.assertEquals(
+ "http://localhost:4566";,
icebergProps.get(GluePropertiesConverter.GLUE_ENDPOINT));
+ }
+
+ @Test
+ void testToIcebergCatalogPropertiesWithStaticCredentials() {
+ Map<String, String> icebergProps =
+ converter.toIcebergCatalogProperties(
+ ImmutableMap.of(
+ GluePropertiesConverter.AWS_REGION, "us-east-1",
+ GluePropertiesConverter.AWS_ACCESS_KEY_ID, "access-key",
+ GluePropertiesConverter.AWS_SECRET_ACCESS_KEY, "secret-key"));
+ Assertions.assertEquals(
+
"org.apache.gravitino.spark.connector.glue.GravitinoGlueCredentialsProvider",
+ icebergProps.get(GluePropertiesConverter.CLIENT_CREDENTIALS_PROVIDER));
+ Assertions.assertEquals(
+ "access-key",
icebergProps.get("client.credentials-provider.access-key-id"));
+ Assertions.assertEquals(
+ "secret-key",
icebergProps.get("client.credentials-provider.secret-access-key"));
+ }
+
+ @Test
+ void testToIcebergCatalogPropertiesMissingRegionOmitsClientRegion() {
+ // Region is optional; when absent the AWS SDK default region chain is
used.
+ Map<String, String> icebergProps =
converter.toIcebergCatalogProperties(ImmutableMap.of());
+
Assertions.assertNull(icebergProps.get(GluePropertiesConverter.CLIENT_REGION));
+
+ Map<String, String> blankRegion =
+ converter.toIcebergCatalogProperties(
+ ImmutableMap.of(GluePropertiesConverter.AWS_REGION, ""));
+
Assertions.assertNull(blankRegion.get(GluePropertiesConverter.CLIENT_REGION));
+ }
+
+ @Test
+ void testToIcebergCatalogPropertiesNullInputThrows() {
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
converter.toIcebergCatalogProperties(null));
+ }
+
+ @Test
+ void testToIcebergCatalogPropertiesOnlyAccessKeyOmitsCredentialsProvider() {
+ Map<String, String> props =
+ converter.toIcebergCatalogProperties(
+ ImmutableMap.of(GluePropertiesConverter.AWS_ACCESS_KEY_ID,
"AKID"));
+
Assertions.assertNull(props.get(GluePropertiesConverter.CLIENT_CREDENTIALS_PROVIDER));
+ }
+
+ @Test
+ void testToIcebergCatalogPropertiesOnlySecretKeyOmitsCredentialsProvider() {
+ Map<String, String> props =
+ converter.toIcebergCatalogProperties(
+ ImmutableMap.of(GluePropertiesConverter.AWS_SECRET_ACCESS_KEY,
"SECRET"));
+
Assertions.assertNull(props.get(GluePropertiesConverter.CLIENT_CREDENTIALS_PROVIDER));
+ }
+
+ // --- toSparkCatalogProperties tests (single-arg version) ---
+
+ @Test
+ void testToSparkCatalogPropertiesAlwaysContainsFactoryClass() {
+ // Factory class must always be set so HiveTableCatalog uses Glue instead
of Derby.
+ // The patched Hive 2.3.10 reads IMETASTORE_CLIENT_FACTORY_CLASS whose key
is
+ // "hive.imetastoreclient.factory.class" (not
"hive.metastore.client.factory.class").
+ Map<String, String> sparkProps =
converter.toSparkCatalogProperties(ImmutableMap.of());
+ Assertions.assertEquals(
+ GluePropertiesConverter.AWS_GLUE_HIVE_CLIENT_FACTORY_CLASS,
+
sparkProps.get(GluePropertiesConverter.HIVE_METASTORE_CLIENT_FACTORY_CLASS));
+ Assertions.assertEquals(
+ "hive.imetastoreclient.factory.class",
+ GluePropertiesConverter.HIVE_METASTORE_CLIENT_FACTORY_CLASS);
+ Assertions.assertEquals(1, sparkProps.size());
+ }
+
+ @Test
+ void testToSparkCatalogPropertiesWithRegion() {
+ Map<String, String> sparkProps =
+ converter.toSparkCatalogProperties(
+ ImmutableMap.of(GluePropertiesConverter.AWS_REGION, "us-east-1"));
+ Assertions.assertEquals("us-east-1", sparkProps.get("aws.region"));
+ Assertions.assertEquals(
+ GluePropertiesConverter.AWS_GLUE_HIVE_CLIENT_FACTORY_CLASS,
+
sparkProps.get(GluePropertiesConverter.HIVE_METASTORE_CLIENT_FACTORY_CLASS));
+ }
+
+ @Test
+ void testToSparkCatalogPropertiesWithCatalogId() {
+ Map<String, String> sparkProps =
+ converter.toSparkCatalogProperties(
+ ImmutableMap.of(
+ GluePropertiesConverter.AWS_REGION, "us-east-1",
+ GluePropertiesConverter.AWS_GLUE_CATALOG_ID, "123456789012"));
+ Assertions.assertEquals("123456789012",
sparkProps.get("aws.glue.catalog.id"));
+ Assertions.assertEquals("us-east-1", sparkProps.get("aws.region"));
+ }
+
+ @Test
+ void testToSparkCatalogPropertiesWithEndpoint() {
+ Map<String, String> sparkProps =
+ converter.toSparkCatalogProperties(
+ ImmutableMap.of(
+ GluePropertiesConverter.AWS_REGION, "us-east-1",
+ GluePropertiesConverter.AWS_GLUE_ENDPOINT,
"http://localhost:4566";));
+ Assertions.assertEquals("http://localhost:4566";,
sparkProps.get("aws.glue.endpoint"));
+ }
+
+ @Test
+ void testToSparkCatalogPropertiesCredentialsNotForwarded() {
+ // Static credentials are NOT forwarded to HiveTableCatalog — the
+ // AWSGlueDataCatalogHiveClientFactory
+ // uses DefaultAWSCredentialsProviderChain (env vars, instance profile,
etc.) instead.
+ Map<String, String> sparkProps =
+ converter.toSparkCatalogProperties(
+ ImmutableMap.of(
+ GluePropertiesConverter.AWS_ACCESS_KEY_ID, "my-access-key",
+ GluePropertiesConverter.AWS_SECRET_ACCESS_KEY,
"my-secret-key"));
+
Assertions.assertNull(sparkProps.get(GluePropertiesConverter.AWS_ACCESS_KEY_ID));
+
Assertions.assertNull(sparkProps.get(GluePropertiesConverter.AWS_SECRET_ACCESS_KEY));
+ }
+
+ @Test
+ void testToSparkCatalogPropertiesNullInputThrows() {
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
converter.toSparkCatalogProperties(null));
+ }
+
+ // --- toGravitinoTableProperties tests ---
+
+ @Test
+ void testToGravitinoTableProperties() {
+ Map<String, String> tableProps =
+ converter.toGravitinoTableProperties(ImmutableMap.of("key1", "value1",
"key2", "value2"));
+ Assertions.assertEquals(ImmutableMap.of("key1", "value1", "key2",
"value2"), tableProps);
+ Assertions.assertNotSame(tableProps, ImmutableMap.of("key1", "value1",
"key2", "value2"));
+ }
+
+ @Test
+ void testToGravitinoTablePropertiesIcebergProviderMapped() {
+ Map<String, String> tableProps =
+ converter.toGravitinoTableProperties(
+ ImmutableMap.of("provider", "iceberg", "key1", "value1"));
+ Assertions.assertEquals("ICEBERG", tableProps.get("table-format"));
+ Assertions.assertFalse(tableProps.containsKey("provider"), "provider
should be removed");
+ Assertions.assertEquals("value1", tableProps.get("key1"));
+ }
+
+ @Test
+ void testToGravitinoTablePropertiesIcebergProviderCaseInsensitive() {
+ Map<String, String> tableProps =
+ converter.toGravitinoTableProperties(ImmutableMap.of("provider",
"ICEBERG"));
+ Assertions.assertEquals("ICEBERG", tableProps.get("table-format"));
+ Assertions.assertFalse(tableProps.containsKey("provider"));
+ }
+
+ @Test
+ void testToGravitinoTablePropertiesNonIcebergProviderPassthrough() {
+ Map<String, String> tableProps =
+ converter.toGravitinoTableProperties(ImmutableMap.of("provider",
"parquet"));
+ Assertions.assertFalse(tableProps.containsKey("table-format"));
+ Assertions.assertFalse(tableProps.containsKey("provider"), "provider
should be removed");
+ }
+
+ @Test
+ void testToGravitinoTablePropertiesEmpty() {
+ Map<String, String> tableProps =
converter.toGravitinoTableProperties(ImmutableMap.of());
+ Assertions.assertTrue(tableProps.isEmpty());
+ }
+
+ // --- toSparkTableProperties tests ---
+
+ @Test
+ void testToSparkTableProperties() {
+ Map<String, String> sparkTableProps =
+ converter.toSparkTableProperties(ImmutableMap.of("key1", "value1",
"key2", "value2"));
+ Assertions.assertEquals(ImmutableMap.of("key1", "value1", "key2",
"value2"), sparkTableProps);
+ Assertions.assertNotSame(sparkTableProps, ImmutableMap.of("key1",
"value1", "key2", "value2"));
+ }
+
+ @Test
+ void testToSparkTablePropertiesEmpty() {
+ Map<String, String> sparkTableProps =
converter.toSparkTableProperties(ImmutableMap.of());
+ Assertions.assertTrue(sparkTableProps.isEmpty());
+ }
+}
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
new file mode 100644
index 0000000000..130ddee519
--- /dev/null
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGravitinoGlueCatalog.java
@@ -0,0 +1,217 @@
+/*
+ * 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.glue;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.HashMap;
+import org.apache.gravitino.catalog.glue.GlueConstants;
+import org.apache.gravitino.client.GravitinoClient;
+import org.apache.gravitino.rel.Column;
+import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.rel.types.Types;
+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.GravitinoCatalogManager;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.TestInstance.Lifecycle;
+
+@TestInstance(Lifecycle.PER_CLASS)
+public class TestGravitinoGlueCatalog {
+
+ private GravitinoGlueCatalog gravitinoGlueCatalog;
+
+ @BeforeAll
+ void initCatalogManager() {
+ // 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);
+ }
+
+ @AfterAll
+ void cleanupCatalogManager() {
+ GravitinoCatalogManager.get().close();
+ }
+
+ @BeforeEach
+ void setUp() {
+ gravitinoGlueCatalog = new GravitinoGlueCatalog();
+ }
+
+ // -------------------------------------------------------------------------
+ // Test isIcebergTable (static package-private method)
+ // -------------------------------------------------------------------------
+
+ @Test
+ void testIsIcebergTableWithIcebergFormat() {
+ Table mockTable =
+ createMockGravitinoTable(
+ ImmutableMap.of(GlueConstants.TABLE_FORMAT,
GlueConstants.TABLE_FORMAT_ICEBERG));
+ Assertions.assertTrue(GravitinoGlueCatalog.isIcebergTable(mockTable));
+ }
+
+ @Test
+ void testIsIcebergTableWithIcebergFormatLowercase() {
+ Table mockTable =
+ createMockGravitinoTable(ImmutableMap.of(GlueConstants.TABLE_FORMAT,
"iceberg"));
+ Assertions.assertTrue(GravitinoGlueCatalog.isIcebergTable(mockTable));
+ }
+
+ @Test
+ void testIsIcebergTableWithIcebergFormatMixedCase() {
+ Table mockTable =
+ createMockGravitinoTable(ImmutableMap.of(GlueConstants.TABLE_FORMAT,
"IcEbErG"));
+ Assertions.assertTrue(GravitinoGlueCatalog.isIcebergTable(mockTable));
+ }
+
+ @Test
+ void testIsIcebergTableWithHiveFormat() {
+ Table mockTable =
createMockGravitinoTable(ImmutableMap.of(GlueConstants.TABLE_FORMAT, "HIVE"));
+ Assertions.assertFalse(GravitinoGlueCatalog.isIcebergTable(mockTable));
+ }
+
+ @Test
+ void testIsIcebergTableWithDeltaFormat() {
+ Table mockTable =
+ createMockGravitinoTable(ImmutableMap.of(GlueConstants.TABLE_FORMAT,
"DELTA"));
+ Assertions.assertFalse(GravitinoGlueCatalog.isIcebergTable(mockTable));
+ }
+
+ @Test
+ void testIsIcebergTableWithEmptyProperties() {
+ Table mockTable = createMockGravitinoTable(ImmutableMap.of());
+ Assertions.assertFalse(GravitinoGlueCatalog.isIcebergTable(mockTable));
+ }
+
+ @Test
+ void testIsIcebergTableWithNullProperties() {
+ Table mockTable = mock(Table.class);
+ when(mockTable.properties()).thenReturn(null);
+ Assertions.assertFalse(GravitinoGlueCatalog.isIcebergTable(mockTable));
+ }
+
+ @Test
+ void testIsIcebergTableWithNoTableFormatProperty() {
+ Table mockTable =
+ createMockGravitinoTable(ImmutableMap.of("aws-location",
"s3://bucket/table"));
+ Assertions.assertFalse(GravitinoGlueCatalog.isIcebergTable(mockTable));
+ }
+
+ @Test
+ void testIsIcebergTableWithEmptyTableFormat() {
+ Table mockTable =
createMockGravitinoTable(ImmutableMap.of(GlueConstants.TABLE_FORMAT, ""));
+ Assertions.assertFalse(GravitinoGlueCatalog.isIcebergTable(mockTable));
+ }
+
+ @Test
+ void testIsIcebergTableWithNativeIcebergTableType() {
+ // Tables created directly by Iceberg GlueCatalog use table_type=ICEBERG
(no table-format key).
+ Table mockTable = createMockGravitinoTable(ImmutableMap.of("table_type",
"ICEBERG"));
+ Assertions.assertTrue(GravitinoGlueCatalog.isIcebergTable(mockTable));
+ }
+
+ @Test
+ void testIsIcebergTableWithNativeIcebergTableTypeLowercase() {
+ Table mockTable = createMockGravitinoTable(ImmutableMap.of("table_type",
"iceberg"));
+ Assertions.assertTrue(GravitinoGlueCatalog.isIcebergTable(mockTable));
+ }
+
+ // -------------------------------------------------------------------------
+ // Test loadSparkTable routes non-Iceberg tables to sparkCatalog directly
+ // -------------------------------------------------------------------------
+
+ @Test
+ void testLoadSparkTableRoutesNonIcebergToSparkCatalog() throws Exception {
+ org.apache.spark.sql.connector.catalog.TableCatalog mockCatalog =
+ mock(org.apache.spark.sql.connector.catalog.TableCatalog.class);
+ org.apache.spark.sql.connector.catalog.Table mockSparkTable =
+ mock(org.apache.spark.sql.connector.catalog.Table.class);
+ Table mockGravitinoTable = createMockGravitinoTable(ImmutableMap.of());
+
+ Identifier ident = Identifier.of(new String[] {"db"}, "tbl");
+ when(mockCatalog.loadTable(any())).thenReturn(mockSparkTable);
+
+ GravitinoGlueCatalog catalog =
+ new GravitinoGlueCatalog() {
+ {
+ sparkCatalog = mockCatalog;
+ }
+
+ @Override
+ protected Table loadGravitinoTable(Identifier ident) throws
NoSuchTableException {
+ return mockGravitinoTable;
+ }
+ };
+
+ org.apache.spark.sql.connector.catalog.Table result =
catalog.loadSparkTable(ident);
+ Assertions.assertSame(mockSparkTable, result);
+ }
+
+ // -------------------------------------------------------------------------
+ // Test converter methods
+ // -------------------------------------------------------------------------
+
+ @Test
+ void testGetSparkTypeConverter() {
+ SparkTypeConverter typeConverter =
gravitinoGlueCatalog.getSparkTypeConverter();
+ Assertions.assertNotNull(typeConverter);
+ Assertions.assertInstanceOf(
+
org.apache.gravitino.spark.connector.hive.SparkHiveTypeConverter.class,
typeConverter);
+ }
+
+ @Test
+ void testGetPropertiesConverter() {
+ PropertiesConverter converter =
gravitinoGlueCatalog.getPropertiesConverter();
+ Assertions.assertNotNull(converter);
+ Assertions.assertInstanceOf(GluePropertiesConverter.class, converter);
+ }
+
+ @Test
+ void testGetSparkTransformConverter() {
+ SparkTransformConverter transformer =
gravitinoGlueCatalog.getSparkTransformConverter();
+ Assertions.assertNotNull(transformer);
+ }
+
+ // -------------------------------------------------------------------------
+ // Helper methods
+ // -------------------------------------------------------------------------
+
+ /** Creates a mock Gravitino Table with the given properties. */
+ private Table createMockGravitinoTable(java.util.Map<String, String>
properties) {
+ Table mockTable = mock(Table.class);
+ when(mockTable.properties()).thenReturn(new HashMap<>(properties));
+ when(mockTable.name()).thenReturn("test_db.test_table");
+ when(mockTable.columns())
+ .thenReturn(new Column[] {Column.of("id", Types.IntegerType.get(), "id
column")});
+ return mockTable;
+ }
+}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGravitinoGlueCredentialsProvider.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGravitinoGlueCredentialsProvider.java
new file mode 100644
index 0000000000..5a5113f07b
--- /dev/null
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/glue/TestGravitinoGlueCredentialsProvider.java
@@ -0,0 +1,51 @@
+/*
+ * 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.glue;
+
+import java.util.Map;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+
+public class TestGravitinoGlueCredentialsProvider {
+
+ @Test
+ void testCreateAndResolveValidCredentials() {
+ AwsCredentialsProvider provider =
+ GravitinoGlueCredentialsProvider.create(
+ Map.of("access-key-id", "AKID", "secret-access-key", "SECRET"));
+ AwsCredentials creds = provider.resolveCredentials();
+ Assertions.assertEquals("AKID", creds.accessKeyId());
+ Assertions.assertEquals("SECRET", creds.secretAccessKey());
+ }
+
+ @Test
+ void testCreateWithMissingKeysThrows() {
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
GravitinoGlueCredentialsProvider.create(Map.of()));
+ }
+
+ @Test
+ void testCreateWithNullPropertiesThrows() {
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
GravitinoGlueCredentialsProvider.create(null));
+ }
+}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java
index 8a96f3ab44..7f637ff78b 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/SparkCommonIT.java
@@ -222,7 +222,7 @@ public abstract class SparkCommonIT extends SparkEnvIT {
}
@Test
- void testLoadCatalogs() {
+ protected void testLoadCatalogs() {
Set<String> catalogs = getCatalogs();
Assertions.assertTrue(catalogs.contains(getCatalogName()));
}
@@ -281,7 +281,7 @@ public abstract class SparkCommonIT extends SparkEnvIT {
}
@Test
- void testDropSchema() {
+ protected void testDropSchema() {
String testDatabaseName = "t_drop";
dropDatabaseIfExists(testDatabaseName);
Set<String> databases = getDatabases();
@@ -381,7 +381,7 @@ public abstract class SparkCommonIT extends SparkEnvIT {
}
@Test
- void testRenameTable() {
+ protected void testRenameTable() {
String tableName = "rename1";
String newTableName = "rename2";
dropTableIfExists(tableName);
@@ -408,7 +408,7 @@ public abstract class SparkCommonIT extends SparkEnvIT {
}
@Test
- void testListTable() {
+ protected void testListTable() {
String table1 = "list1";
String table2 = "list2";
dropTableIfExists(table1);
@@ -472,6 +472,7 @@ public abstract class SparkCommonIT extends SparkEnvIT {
}
@Test
+ @EnabledIf("supportsSchemaEvolution")
void testAlterTableAddAndDeleteColumn() {
String tableName = "test_column";
dropTableIfExists(tableName);
@@ -491,6 +492,7 @@ public abstract class SparkCommonIT extends SparkEnvIT {
}
@Test
+ @EnabledIf("supportsSchemaEvolution")
void testAlterTableUpdateColumnType() {
String tableName = "test_column_type";
dropTableIfExists(tableName);
@@ -508,6 +510,7 @@ public abstract class SparkCommonIT extends SparkEnvIT {
}
@Test
+ @EnabledIf("supportsSchemaEvolution")
void testAlterTableRenameColumn() {
String tableName = "test_rename_column";
dropTableIfExists(tableName);
@@ -858,7 +861,7 @@ public abstract class SparkCommonIT extends SparkEnvIT {
}
@Test
- void testDropAndWriteTable() {
+ protected void testDropAndWriteTable() {
String tableName = "drop_then_create_write_table";
createSimpleTable(tableName);
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 a9da4aaa8f..fcd2c2d819 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
@@ -96,7 +96,7 @@ public abstract class SparkEnvIT extends SparkUtilIT {
}
@BeforeAll
- void startUp() throws Exception {
+ protected void startUp() throws Exception {
initHiveEnv();
// initialize the hiveMetastoreUri and warehouse at first to inject
properties to
// IcebergRestService
@@ -120,7 +120,7 @@ public abstract class SparkEnvIT extends SparkUtilIT {
}
@AfterAll
- void stop() throws IOException, InterruptedException {
+ protected void stop() throws IOException, InterruptedException {
if (hdfs != null) {
try {
hdfs.close();
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkGlueCatalogIT.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkGlueCatalogIT.java
new file mode 100644
index 0000000000..79b9ca4b33
--- /dev/null
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkGlueCatalogIT.java
@@ -0,0 +1,573 @@
+/*
+ * 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.glue;
+
+import com.google.common.base.Preconditions;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.gravitino.catalog.glue.GlueConstants;
+import
org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfo;
+import
org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfo.SparkColumnInfo;
+import
org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfoChecker;
+import org.apache.spark.sql.types.DataTypes;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Integration test for GravitinoGlueCatalog in Spark connector.
+ *
+ * <p>Tests mixed table type support (Hive format + Iceberg format) in a
single Glue database. Uses
+ * Moto server to mock AWS Glue API, similar to MotoGlueCatalogIT in the
server module.
+ */
+public abstract class SparkGlueCatalogIT extends SparkGlueEnvIT {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(SparkGlueCatalogIT.class);
+
+ private String glueEndpoint;
+ private String awsRegion;
+ private String awsAccessKeyId;
+ private String awsSecretAccessKey;
+
+ @Override
+ protected String getCatalogName() {
+ return "glue";
+ }
+
+ @Override
+ protected String getProvider() {
+ return "glue";
+ }
+
+ @Override
+ protected Map<String, String> getCatalogConfigs() {
+ Preconditions.checkArgument(
+ awsRegion != null, "awsRegion must be set before getCatalogConfigs()");
+ Preconditions.checkArgument(
+ awsAccessKeyId != null, "awsAccessKeyId must be set before
getCatalogConfigs()");
+ Preconditions.checkArgument(
+ awsSecretAccessKey != null, "awsSecretAccessKey must be set before
getCatalogConfigs()");
+ Map<String, String> catalogProperties = new java.util.HashMap<>();
+ catalogProperties.put(GlueConstants.AWS_REGION, awsRegion);
+ catalogProperties.put(GlueConstants.AWS_ACCESS_KEY_ID, awsAccessKeyId);
+ catalogProperties.put(GlueConstants.AWS_SECRET_ACCESS_KEY,
awsSecretAccessKey);
+ catalogProperties.put(GlueConstants.WAREHOUSE, warehouse);
+ if (glueEndpoint != null) {
+ catalogProperties.put(GlueConstants.AWS_GLUE_ENDPOINT, glueEndpoint);
+ }
+ return catalogProperties;
+ }
+
+ @Override
+ protected boolean supportsSparkSQLClusteredBy() {
+ return true;
+ }
+
+ @Override
+ protected boolean supportsPartition() {
+ return true;
+ }
+
+ @Override
+ protected boolean supportsDelete() {
+ return false;
+ }
+
+ @Override
+ protected boolean supportsSchemaEvolution() {
+ return false;
+ }
+
+ @Override
+ protected boolean supportsReplaceColumns() {
+ return false;
+ }
+
+ @Override
+ protected boolean supportsSchemaAndTableProperties() {
+ return true;
+ }
+
+ @Override
+ protected boolean supportsComplexType() {
+ // Glue does not support Gravitino complex types (LIST, MAP, STRUCT) in
table columns.
+ return false;
+ }
+
+ @Override
+ protected boolean supportsUpdateColumnPosition() {
+ return false;
+ }
+
+ @Override
+ protected boolean supportsFunction() {
+ return false;
+ }
+
+ @BeforeAll
+ @Override
+ protected void startUp() throws Exception {
+ String accessKeyId = System.getenv("AWS_ACCESS_KEY_ID");
+ String secretAccessKey = System.getenv("AWS_SECRET_ACCESS_KEY");
+ setGlueEndpoint(null);
+ setAwsCredentials(accessKeyId, secretAccessKey);
+ setAwsRegion(System.getenv("AWS_DEFAULT_REGION"));
+ setS3Credentials(null, accessKeyId, secretAccessKey);
+ setS3BucketName(System.getenv("AWS_S3_TEST_BUCKET"));
+ super.startUp();
+ }
+
+ /**
+ * Sets the Glue endpoint for testing. Called by subclasses after Moto
container is started.
+ *
+ * @param endpoint the Glue API endpoint URL
+ */
+ protected void setGlueEndpoint(String endpoint) {
+ this.glueEndpoint = endpoint;
+ }
+
+ protected void setAwsRegion(String region) {
+ this.awsRegion = region;
+ }
+
+ /**
+ * Sets AWS credentials for testing.
+ *
+ * @param accessKeyId AWS access key ID
+ * @param secretAccessKey AWS secret access key
+ */
+ protected void setAwsCredentials(String accessKeyId, String secretAccessKey)
{
+ this.awsAccessKeyId = accessKeyId;
+ this.awsSecretAccessKey = secretAccessKey;
+ }
+
+ @Override
+ protected String getDefaultAwsRegion() {
+ return awsRegion;
+ }
+
+ @Override
+ protected String getGlueEndpoint() {
+ return glueEndpoint;
+ }
+
+ protected String getAwsRegion() {
+ return awsRegion;
+ }
+
+ /**
+ * Overrides to use CASCADE so that databases with stale tables (e.g., from
prior test runs) can
+ * be cleaned up. Glue tables persist across JVM restarts and may be left
behind after a crash.
+ */
+ @Override
+ protected void dropDatabaseIfExists(String database) {
+ sql("DROP DATABASE IF EXISTS " + database + " CASCADE");
+ }
+
+ /**
+ * Overrides to always use PARQUET format for Glue. Without an explicit
USING clause, Spark may
+ * route CREATE TABLE through the V1 Hive path, bypassing the
location-derivation logic in
+ * GravitinoGlueCatalog.createTable and leaving tables without a stored S3
location.
+ */
+ @Override
+ protected void createSimpleTable(String identifier) {
+ sql(getCreateSimpleTableString(identifier) + " USING PARQUET");
+ }
+
+ /**
+ * Overrides to recreate the database with the correct S3 location. The base
implementation uses
+ * CREATE DATABASE IF NOT EXISTS with a local HDFS path (/user/hive/db),
causing tables to inherit
+ * a local path as their default location. We drop and recreate to ensure
the S3 location is
+ * always set correctly, so stale data cleanup via dropTableIfExists works
reliably.
+ */
+ @Override
+ protected void createDatabaseIfNotExists(String database, String provider) {
+ String dbLocation = warehouse + "/" + database;
+ dropDatabaseIfExists(database);
+ // Delete S3 data directory so stale data files from prior runs don't
cause duplicate rows.
+ deleteDirIfExists(dbLocation);
+ sql(String.format("CREATE DATABASE %s LOCATION '%s'", database,
dbLocation));
+ }
+
+ /**
+ * Overrides to also delete the S3 data directory after dropping the table.
Unlike HDFS managed
+ * tables, Glue external tables do not delete S3 data on DROP TABLE. Stale
data files cause
+ * duplicate rows on the next create+insert cycle.
+ */
+ @Override
+ protected void dropTableIfExists(String tableName) {
+ String location = null;
+ try {
+ location = getTableInfo(tableName).getTableLocation();
+ } catch (Exception e) {
+ // Table may not exist yet — location stays null, nothing to delete.
+ LOG.debug("Could not get location for table {}: {}", tableName,
e.getMessage());
+ }
+ super.dropTableIfExists(tableName);
+ if (location != null) {
+ deleteDirIfExists(location);
+ }
+ }
+
+ /**
+ * Overrides base class: use USING PARQUET to ensure the table goes through
the Gravitino Glue
+ * catalog (V2 path).
+ */
+ @Test
+ @Override
+ protected void testDropAndWriteTable() {
+ String tableName = "drop_then_create_write_table";
+ dropTableIfExists(tableName);
+ sql(getCreateSimpleTableString(tableName) + " USING PARQUET");
+ SparkTableInfo info = getTableInfo(tableName);
+ checkTableReadWrite(info);
+
+ // External tables on S3 do not delete data on DROP TABLE; clean up
explicitly.
+ String location = info.getTableLocation();
+ dropTableIfExists(tableName);
+ if (location != null) {
+ deleteDirIfExists(location);
+ }
+
+ sql(getCreateSimpleTableString(tableName) + " USING PARQUET");
+ checkTableReadWrite(getTableInfo(tableName));
+ }
+
+ @Disabled("Glue does not support table rename for non-Iceberg tables")
+ @Test
+ @Override
+ protected void testRenameTable() {}
+
+ // -------------------------------------------------------------------------
+ // Test mixed table types (Hive format + Iceberg format)
+ // -------------------------------------------------------------------------
+
+ @Test
+ void testCreateHiveFormatTable() {
+ String tableName = "test_hive_format_table";
+ dropTableIfExists(tableName);
+ String createTableSql = getCreateSimpleTableString(tableName);
+ createTableSql += " USING PARQUET";
+ sql(createTableSql);
+
+ SparkTableInfo tableInfo = getTableInfo(tableName);
+ SparkTableInfoChecker checker =
+
SparkTableInfoChecker.create().withName(tableName).withColumns(getSimpleTableColumn());
+ checker.check(tableInfo);
+ checkTableReadWrite(tableInfo);
+ }
+
+ @Test
+ void testCreateIcebergFormatTable() {
+ String tableName = "test_iceberg_format_table";
+ dropTableIfExists(tableName);
+ String createTableSql = getCreateSimpleTableString(tableName);
+ createTableSql += " USING iceberg";
+ sql(createTableSql);
+
+ SparkTableInfo tableInfo = getTableInfo(tableName);
+ SparkTableInfoChecker checker =
+
SparkTableInfoChecker.create().withName(tableName).withColumns(getSimpleTableColumn());
+ checker.check(tableInfo);
+ checkTableReadWrite(tableInfo);
+ }
+
+ @Test
+ void testMixedTableTypesInSameDatabase() {
+ String hiveTable = "mixed_hive_table";
+ String icebergTable = "mixed_iceberg_table";
+
+ // Create non-partitioned Hive format table
+ dropTableIfExists(hiveTable);
+ sql(getCreateSimpleTableString(hiveTable) + " USING PARQUET");
+
+ // Create non-partitioned Iceberg format table
+ dropTableIfExists(icebergTable);
+ sql(getCreateSimpleTableString(icebergTable) + " USING iceberg");
+
+ // Both tables should be accessible
+ SparkTableInfo hiveTableInfo = getTableInfo(hiveTable);
+ SparkTableInfoChecker hiveChecker =
+
SparkTableInfoChecker.create().withName(hiveTable).withColumns(getSimpleTableColumn());
+ hiveChecker.check(hiveTableInfo);
+ checkTableReadWrite(hiveTableInfo);
+
+ SparkTableInfo icebergTableInfo = getTableInfo(icebergTable);
+ SparkTableInfoChecker icebergChecker =
+
SparkTableInfoChecker.create().withName(icebergTable).withColumns(getSimpleTableColumn());
+ icebergChecker.check(icebergTableInfo);
+ checkTableReadWrite(icebergTableInfo);
+ }
+
+ @Test
+ void testHivePartitionedTable() {
+ String tableName = "test_hive_partitioned_table";
+ dropTableIfExists(tableName);
+ // Use existing columns as partition keys (datasource-style) so partition
columns stay in
+ // schema.
+ // Spark places partition columns last; columns = [id, age, name] with
name as partition.
+ sql(
+ "CREATE TABLE "
+ + tableName
+ + " (id INT COMMENT 'id comment', age INT, name STRING COMMENT '')
USING PARQUET"
+ + " PARTITIONED BY (name)");
+
+ SparkTableInfo tableInfo = getTableInfo(tableName);
+ SparkTableInfoChecker checker =
+ SparkTableInfoChecker.create()
+ .withName(tableName)
+ .withColumns(
+ Arrays.asList(
+ SparkColumnInfo.of("id", DataTypes.IntegerType, "id
comment"),
+ SparkColumnInfo.of("age", DataTypes.IntegerType, null),
+ SparkColumnInfo.of("name", DataTypes.StringType, "")))
+ .withIdentifyPartition(Arrays.asList("name"));
+ checker.check(tableInfo);
+ checkTableReadWrite(tableInfo);
+ }
+
+ @Test
+ void testIcebergPartitionedTable() {
+ String tableName = "test_iceberg_partitioned_table";
+ dropTableIfExists(tableName);
+ // Partition by an existing column (id). Iceberg stores partition columns
separately from
+ // schema.
+ sql(getCreateSimpleTableString(tableName) + " USING iceberg PARTITIONED BY
(id)");
+
+ SparkTableInfo tableInfo = getTableInfo(tableName);
+ SparkTableInfoChecker checker =
+
SparkTableInfoChecker.create().withName(tableName).withColumns(getSimpleTableColumn());
+ checker.check(tableInfo);
+ checkTableReadWrite(tableInfo);
+ }
+
+ @Test
+ void testInsertHiveTable() {
+ String tableName = "test_insert_hive_table";
+ dropTableIfExists(tableName);
+ String createTableSql = getCreateSimpleTableString(tableName);
+ createTableSql += " USING PARQUET";
+ sql(createTableSql);
+
+ sql(String.format("INSERT INTO %s VALUES (1, 'name1', 25)", tableName));
+ List<String> tableData = getTableData(tableName);
+ Assertions.assertFalse(tableData.isEmpty());
+ Assertions.assertEquals("1,name1,25", tableData.get(0));
+ }
+
+ @Test
+ void testInsertIcebergTable() {
+ String tableName = "test_insert_iceberg_table";
+ dropTableIfExists(tableName);
+ String createTableSql = getCreateSimpleTableString(tableName);
+ createTableSql += " USING iceberg";
+ sql(createTableSql);
+
+ sql(String.format("INSERT INTO %s VALUES (1, 'name1', 25)", tableName));
+ List<String> tableData = getTableData(tableName);
+ Assertions.assertFalse(tableData.isEmpty());
+ Assertions.assertEquals("1,name1,25", tableData.get(0));
+ }
+
+ @Test
+ void testCreateTableWithComment() {
+ String tableName = "test_table_with_comment";
+ dropTableIfExists(tableName);
+ String createTableSql = getCreateSimpleTableString(tableName);
+ createTableSql += " USING PARQUET COMMENT 'Test table comment'";
+ sql(createTableSql);
+
+ SparkTableInfo tableInfo = getTableInfo(tableName);
+ Assertions.assertEquals("Test table comment", tableInfo.getComment());
+ checkTableReadWrite(tableInfo);
+ }
+
+ @Test
+ void testExternalTableLocation() {
+ String tableName = "test_external_table";
+ dropTableIfExists(tableName);
+ String externalLocation = warehouse + "/external_glue_db/external_table";
+ deleteDirIfExists(externalLocation);
+
+ String createTableSql = getCreateSimpleTableString(tableName);
+ createTableSql += String.format(" USING PARQUET LOCATION '%s'",
externalLocation);
+ sql(createTableSql);
+
+ SparkTableInfo tableInfo = getTableInfo(tableName);
+ Assertions.assertEquals(externalLocation, tableInfo.getTableLocation());
+ checkTableReadWrite(tableInfo);
+ }
+
+ @Test
+ @Override
+ protected void testLoadCatalogs() {
+ // Glue catalog is not shown in SHOW CATALOGS output (Gravitino registers
it lazily via Spark
+ // plugin). Verify accessibility by listing databases instead.
+ Assertions.assertDoesNotThrow(() -> sql("SHOW DATABASES IN " +
getCatalogName()));
+ }
+
+ /**
+ * Overrides base: skips S3 directory verification when no explicit LOCATION
was given. Glue does
+ * not store the auto-assigned warehouse location in table properties, so we
cannot reconstruct
+ * the exact S3 path. Data read/write correctness is already validated by
{@link
+ * #checkTableReadWrite}.
+ */
+ @Override
+ protected void checkPartitionDirExists(SparkTableInfo table) {
+ if (table.getTableLocation() == null) {
+ return;
+ }
+ super.checkPartitionDirExists(table);
+ }
+
+ // -------------------------------------------------------------------------
+ // Override exception-assertion tests (Glue throws different exception types)
+ // -------------------------------------------------------------------------
+
+ /**
+ * Override: Glue does not support local filesystem paths for database
locations; use S3 path.
+ * Also, Glue does not return Owner as "anonymous".
+ */
+ @Test
+ @Override
+ protected void testCreateAndLoadSchema() {
+ String testDatabaseName = "t_create1";
+ dropDatabaseIfExists(testDatabaseName);
+ sql("CREATE DATABASE " + testDatabaseName + " WITH DBPROPERTIES
(ID=001);");
+ Map<String, String> databaseMeta = getDatabaseMetadata(testDatabaseName);
+ // Glue does not auto-assign a location when none is specified, so no
"Location" row appears
+ String properties = databaseMeta.get("Properties");
+ Assertions.assertTrue(properties.contains("(ID,001)"));
+
+ testDatabaseName = "t_create2";
+ dropDatabaseIfExists(testDatabaseName);
+ String testDatabaseLocation = warehouse + "/" + testDatabaseName;
+ sql(
+ String.format(
+ "CREATE DATABASE %s COMMENT 'comment' LOCATION '%s' WITH
DBPROPERTIES (ID=002);",
+ testDatabaseName, testDatabaseLocation));
+ databaseMeta = getDatabaseMetadata(testDatabaseName);
+ String comment = databaseMeta.get("Comment");
+ Assertions.assertEquals("comment", comment);
+
Assertions.assertTrue(databaseMeta.get("Location").contains(testDatabaseName));
+ properties = databaseMeta.get("Properties");
+ Assertions.assertTrue(properties.contains("(ID,002)"));
+ }
+
+ /**
+ * Override: Glue may throw AnalysisException instead of
NoSuchNamespaceException when listing
+ * tables in a nonexistent schema.
+ */
+ @Test
+ void testListTables() {
+ String tableName = "t_list";
+ Set<String> tableNames = listTableNames();
+ Assertions.assertFalse(tableNames.contains(tableName));
+ createSimpleTable(tableName);
+ tableNames = listTableNames();
+ Assertions.assertTrue(tableNames.contains(tableName));
+ // Glue throws AnalysisException or other runtime exception through
Spark's Hive catalog adapter
+ Assertions.assertThrows(Exception.class, () -> sql("SHOW TABLES IN
nonexistent_schema"));
+ }
+
+ /**
+ * Override: Glue may throw AnalysisException instead of
NoSuchNamespaceException when altering a
+ * nonexistent schema.
+ */
+ @Test
+ @Override
+ protected void testAlterSchema() {
+ String testDatabaseName = "t_alter";
+ dropDatabaseIfExists(testDatabaseName);
+ sql("CREATE DATABASE " + testDatabaseName + " WITH DBPROPERTIES
(ID=001);");
+ Assertions.assertTrue(
+
getDatabaseMetadata(testDatabaseName).get("Properties").contains("(ID,001)"));
+
+ sql(String.format("ALTER DATABASE %s SET DBPROPERTIES ('ID'='002')",
testDatabaseName));
+ Assertions.assertFalse(
+
getDatabaseMetadata(testDatabaseName).get("Properties").contains("(ID,001)"));
+ Assertions.assertTrue(
+
getDatabaseMetadata(testDatabaseName).get("Properties").contains("(ID,002)"));
+
+ // Glue may throw AnalysisException instead of NoSuchNamespaceException
+ Assertions.assertThrows(
+ Exception.class, () -> sql("ALTER DATABASE notExists SET DBPROPERTIES
('ID'='001')"));
+ }
+
+ /**
+ * Override: Glue may throw AnalysisException instead of
NoSuchNamespaceException when dropping a
+ * nonexistent schema.
+ */
+ @Test
+ @Override
+ protected void testDropSchema() {
+ String testDatabaseName = "t_drop";
+ dropDatabaseIfExists(testDatabaseName);
+ Set<String> databases = getDatabases();
+ Assertions.assertFalse(databases.contains(testDatabaseName));
+
+ sql("CREATE DATABASE " + testDatabaseName);
+ databases = getDatabases();
+ Assertions.assertTrue(databases.contains(testDatabaseName));
+
+ sql("DROP DATABASE " + testDatabaseName);
+ databases = getDatabases();
+ Assertions.assertFalse(databases.contains(testDatabaseName));
+
+ // Glue may throw AnalysisException instead of NoSuchNamespaceException
+ Assertions.assertThrows(Exception.class, () -> sql("DROP DATABASE
notExists"));
+ }
+
+ /**
+ * Override: Glue may throw AnalysisException instead of
NoSuchNamespaceException when listing
+ * tables from a nonexistent database.
+ */
+ @Test
+ @Override
+ protected void testListTable() {
+ String table1 = "list1";
+ String table2 = "list2";
+ createSimpleTable(table1);
+ createSimpleTable(table2);
+ Set<String> tables = listTableNames();
+ Assertions.assertTrue(tables.contains(table1));
+ Assertions.assertTrue(tables.contains(table2));
+
+ String database = "db_list";
+ String table3 = "list3";
+ String table4 = "list4";
+ createDatabaseIfNotExists(database, getProvider());
+ dropTableIfExists(String.join(".", database, table3));
+ dropTableIfExists(String.join(".", database, table4));
+ createSimpleTable(String.join(".", database, table3));
+ createSimpleTable(String.join(".", database, table4));
+ tables = listTableNames(database);
+
+ Assertions.assertTrue(tables.contains(table3));
+ Assertions.assertTrue(tables.contains(table4));
+
+ // Glue may throw AnalysisException instead of NoSuchNamespaceException
+ Assertions.assertThrows(Exception.class, () ->
listTableNames("not_exists_db"));
+ }
+}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkGlueEnvIT.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkGlueEnvIT.java
new file mode 100644
index 0000000000..a502826776
--- /dev/null
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkGlueEnvIT.java
@@ -0,0 +1,363 @@
+/*
+ * 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.glue;
+
+import com.google.common.base.Preconditions;
+import java.io.IOException;
+import java.util.Collections;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.client.GravitinoMetalake;
+import org.apache.gravitino.spark.connector.GravitinoSparkConfig;
+import org.apache.gravitino.spark.connector.integration.test.SparkCommonIT;
+import org.apache.gravitino.spark.connector.plugin.GravitinoSparkPlugin;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.SparkConf;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base environment setup for Spark Glue connector integration tests.
+ *
+ * <p>Replaces the Hive/HDFS initialization chain from SparkEnvIT with
Glue-appropriate setup. Glue
+ * does not require Hive metastore or HDFS — it uses AWS Glue API (mocked by
LocalStack) and S3
+ * storage.
+ *
+ * <p>Subclasses must:
+ *
+ * <ul>
+ * <li>Start LocalStack container in their {@link #startUp()} and call {@link
+ * SparkGlueCatalogIT#setGlueEndpoint(String)}
+ * <li>Set AWS credentials via {@link
SparkGlueCatalogIT#setAwsCredentials(String, String)}
+ * <li>Set S3 credentials via {@link #setS3Credentials(String, String,
String)}
+ * <li>Call {@code super.startUp()} after configuring Glue
+ * </ul>
+ */
+public abstract class SparkGlueEnvIT extends SparkCommonIT {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(SparkGlueEnvIT.class);
+
+ private SparkSession sparkSession;
+
+ private String s3AccessKey;
+ private String s3SecretKey;
+ private String s3Endpoint;
+ private String s3BucketName;
+
+ @Override
+ protected SparkSession getSparkSession() {
+ return sparkSession;
+ }
+
+ /**
+ * Starts the test environment. Subclasses must call super.startUp() after
configuring the Glue
+ * endpoint and credentials.
+ */
+ @BeforeAll
+ protected void startUp() throws Exception {
+ Preconditions.checkArgument(s3BucketName != null, "s3BucketName must be
set before startUp()");
+ Preconditions.checkArgument(s3AccessKey != null, "s3AccessKey must be set
before startUp()");
+ Preconditions.checkArgument(s3SecretKey != null, "s3SecretKey must be set
before startUp()");
+ warehouse = "s3a://" + s3BucketName + "/warehouse";
+ hiveMetastoreUri = null;
+ hdfs = null;
+
+ // Start the embedded Gravitino server. SparkEnvIT.startIntegrationTest()
is an empty
+ // @BeforeAll override that prevents JUnit from auto-invoking
BaseIT.startIntegrationTest().
+ // We call startServer() directly (a non-@BeforeAll method) to avoid the
virtual-dispatch
+ // problem that would occur with reflection + Method.invoke().
+ startServer();
+ int gravitinoPort = getGravitinoServerPort();
+ String gravitinoUri = String.format("http://127.0.0.1:%d";, gravitinoPort);
+ initMetalakeAndCatalogs();
+ initSparkEnv(gravitinoUri);
+ // Create the default database in the Glue catalog so that tests can USE
it.
+ // Note: SparkCommonIT.initDefaultDatabase() is package-private and cannot
be @Override'd
+ // from this package, so we call this class's own implementation directly.
+ initDefaultDatabase();
+
+ LOG.info(
+ "Startup Spark env for Glue successfully, Gravitino uri: {},
Warehouse: {}",
+ gravitinoUri,
+ warehouse);
+ }
+
+ @BeforeEach
+ protected void init() {
+ // Spark's Hive-style catalog integration treats USE <name> as switching
databases,
+ // not catalogs. We must use SET to set the default catalog instead.
+ sql("SET spark.sql.defaultCatalog=" + getCatalogName());
+ sql("USE " + getDefaultDatabase());
+ }
+
+ protected String getDefaultAwsRegion() {
+ return "us-east-1";
+ }
+
+ /**
+ * Skips the Hive container initialization. Glue uses AWS Glue API (catalog)
and S3 (storage), not
+ * a Hive metastore or HDFS. This method shadows the private {@link
SparkEnvIT#initHiveEnv()} from
+ * the grandparent class — no @Override since the parent method is private
(not inheritable).
+ */
+ protected void initHiveEnv() {
+ // Skip Hive container startup. Glue uses AWS Glue API + S3, not HMS.
+ // warehouse and hiveMetastoreUri were already set in startUp():
+ // - warehouse = "s3a://bucket/warehouse" (already set)
+ // - hiveMetastoreUri = null (already set)
+ }
+
+ /**
+ * Overrides HDFS filesystem initialization. Glue uses S3, not HDFS, so this
is a no-op. Note:
+ * no @Override because the parent method is private (not inheritable).
+ */
+ protected void initHdfsFileSystem() {
+ // Glue uses S3, not HDFS. hdfs is already set to null in startUp().
+ }
+
+ /**
+ * Shadows parent's HDFS-based database initialization. The parent {@link
+ * org.apache.gravitino.spark.connector.integration.test.SparkCommonIT}
creates a database with an
+ * HDFS location ('/user/hive/{db}') which is wrong for Glue. This
implementation drops and
+ * recreates the database with the correct S3 location to remove any stale
tables left by prior
+ * test runs (e.g., when multiple Spark versions share the same Glue
catalog).
+ *
+ * <p>Called explicitly from {@link #startUp()}, not via JUnit lifecycle
inheritance. JUnit 5
+ * treats this protected method as hiding the package-private {@code
@BeforeAll
+ * initDefaultDatabase()} from {@code SparkCommonIT}, so the parent's
drop-and-recreate logic
+ * would not run — this override must perform the same cleanup itself.
+ */
+ protected void initDefaultDatabase() {
+ String defaultDbName = getDefaultDatabase();
+ String dbLocation = warehouse + "/" + defaultDbName;
+ sql("SET spark.sql.defaultCatalog=" + getCatalogName());
+ sql("DROP DATABASE IF EXISTS " + defaultDbName + " CASCADE");
+ deleteDirIfExists(dbLocation);
+ sql(String.format("CREATE DATABASE %s LOCATION '%s'", defaultDbName,
dbLocation));
+ sql("USE " + defaultDbName);
+ }
+
+ /**
+ * Overrides HDFS-based directory check. Glue uses S3 storage, not HDFS.
Uses S3A filesystem to
+ * verify directory existence on S3.
+ */
+ @Override
+ protected void checkDirExists(Path dir) {
+ try {
+ Configuration conf = newS3Config();
+ FileSystem fs = FileSystem.get(dir.toUri(), conf);
+ boolean exists = fs.exists(dir);
+ fs.close();
+ org.junit.jupiter.api.Assertions.assertTrue(exists, "S3 directory not
exists: " + dir);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Overrides HDFS-based file existence check. Glue uses S3 storage, not
HDFS. Uses S3A filesystem
+ * to verify that at least one data file exists in the partition directory.
+ */
+ @Override
+ protected void checkDataFileExists(Path dir) {
+ try {
+ Configuration conf = newS3Config();
+ FileSystem fs = FileSystem.get(dir.toUri(), conf);
+ FileStatus[] files = fs.listStatus(dir);
+ boolean hasFile = false;
+ for (FileStatus file : files) {
+ if (file.isFile()) {
+ hasFile = true;
+ break;
+ }
+ }
+ fs.close();
+ org.junit.jupiter.api.Assertions.assertTrue(hasFile, "No data file found
in: " + dir);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @AfterAll
+ protected void stop() throws IOException, InterruptedException {
+ if (sparkSession != null) {
+ sparkSession.close();
+ }
+ super.stopIntegrationTest();
+ }
+
+ /**
+ * Deletes a directory (file or directory) from S3 if it exists. Overrides
the HDFS-based
+ * implementation in SparkCommonIT since Glue uses S3 storage instead of
HDFS.
+ */
+ @Override
+ protected void deleteDirIfExists(String path) {
+ try {
+ Configuration conf = newS3Config();
+ Path dir = new Path(path);
+ FileSystem fs = FileSystem.get(dir.toUri(), conf);
+ if (fs.exists(dir)) {
+ fs.delete(dir, true);
+ }
+ fs.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /** Creates a Hadoop Configuration configured for S3A access with the test
credentials. */
+ private Configuration newS3Config() {
+ Configuration conf = new Configuration();
+ conf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem");
+ conf.set("fs.s3a.access.key", s3AccessKey);
+ conf.set("fs.s3a.secret.key", s3SecretKey);
+ conf.set(
+ "fs.s3a.aws.credentials.provider",
"org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider");
+ if (s3Endpoint != null) {
+ conf.set("fs.s3a.endpoint", s3Endpoint);
+ conf.set("fs.s3a.path.style.access", "true");
+ conf.set("fs.s3a.connection.ssl.enabled", "false");
+ }
+ return conf;
+ }
+
+ private void initMetalakeAndCatalogs() {
+ client.createMetalake(metalakeName, "", Collections.emptyMap());
+ GravitinoMetalake metalake = client.loadMetalake(metalakeName);
+ metalake.createCatalog(
+ getCatalogName(), Catalog.Type.RELATIONAL, getProvider(), "",
getCatalogConfigs());
+ }
+
+ /**
+ * Returns the Glue API endpoint override, or null to use the default AWS
endpoint. Subclasses
+ * that use a mock Glue service (e.g., Moto/LocalStack) should override this
to return the mock
+ * URL.
+ */
+ protected String getGlueEndpoint() {
+ return null;
+ }
+
+ private void initSparkEnv(String gravitinoUri) {
+ String awsRegion = getDefaultAwsRegion();
+ // Set AWS region as system properties so the AWS SDK's
DefaultAwsRegionProviderChain
+ // can find it. Without this, S3A operations fail with "Unable to find a
region".
+ // aws.region is checked first by DefaultAwsRegionProviderChain;
AWS_DEFAULT_REGION is
+ // fallback for env-var-based provider.
+ if (awsRegion != null && !awsRegion.isEmpty()) {
+ System.setProperty("AWS_DEFAULT_REGION", awsRegion);
+ System.setProperty("aws.region", awsRegion);
+ }
+
+ String glueEndpoint = getGlueEndpoint();
+ SparkConf sparkConf =
+ new SparkConf()
+ .set("spark.plugins", GravitinoSparkPlugin.class.getName())
+ .set(GravitinoSparkConfig.GRAVITINO_URI, gravitinoUri)
+ .set(GravitinoSparkConfig.GRAVITINO_METALAKE, metalakeName)
+ .set(GravitinoSparkConfig.GRAVITINO_ENABLE_ICEBERG_SUPPORT, "true")
+ .set(GravitinoSparkConfig.GRAVITINO_ENABLE_PAIMON_SUPPORT, "true")
+ .set("hive.exec.dynamic.partition.mode", "nonstrict")
+ .set("spark.sql.warehouse.dir", warehouse)
+ .set("spark.sql.session.timeZone", TIME_ZONE_UTC)
+ // S3A filesystem configuration
+ .set("spark.hadoop.fs.s3a.impl",
"org.apache.hadoop.fs.s3a.S3AFileSystem")
+ .set("spark.hadoop.fs.s3a.access.key", s3AccessKey)
+ .set("spark.hadoop.fs.s3a.secret.key", s3SecretKey)
+ .set(
+ "spark.hadoop.fs.s3a.aws.credentials.provider",
+ "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider");
+
+ if (awsRegion != null && !awsRegion.isEmpty()) {
+ sparkConf.set("spark.hadoop.aws.region", awsRegion);
+ }
+ if (glueEndpoint != null) {
+ // For mock Glue services (Moto/LocalStack), override the Glue API
endpoint.
+ sparkConf.set("spark.hadoop.aws.glue.endpoint", glueEndpoint);
+ }
+ // Only set endpoint for custom S3 services (e.g., LocalStack).
+ // For real AWS, omit this to use the default AWS endpoint.
+ if (s3Endpoint != null) {
+ sparkConf.set("spark.hadoop.fs.s3a.endpoint", s3Endpoint);
+ // LocalStack uses HTTP with path-style access
+ sparkConf.set("spark.hadoop.fs.s3a.path.style.access", "true");
+ sparkConf.set("spark.hadoop.fs.s3a.connection.ssl.enabled", "false");
+ } else if (awsRegion != null && !awsRegion.isEmpty()) {
+ // For real AWS S3 (no custom endpoint), configure the region so S3A can
+ // determine the correct AWS partition and endpoint without an explicit
URL.
+ sparkConf.set("spark.hadoop.fs.s3a.endpoint.region", awsRegion);
+ }
+ String glueHiveJarsDir = System.getProperty("glue.hive-jars-dir");
+ Preconditions.checkNotNull(
+ glueHiveJarsDir,
+ "System property glue.hive-jars-dir not set. "
+ + "Download jars from
https://github.com/datastrato/spark-hive-glue-libs "
+ + "and set -Dglue.hive-jars-dir=<path> or run via Gradle test
task.");
+ String sharedPrefixes =
+ String.join(
+ ",",
+ "com.mysql.jdbc",
+ "org.postgresql",
+ "com.microsoft.sqlserver.jdbc",
+ "org.apache.thrift",
+ "org.slf4j",
+ "org.apache.log4j",
+ "com.google.protobuf",
+ "com.google.common",
+ "javax.jdo",
+ "org.apache.derby",
+ "org.antlr");
+ sparkConf
+ .set("spark.sql.hive.metastore.version", "2.3.10")
+ .set("spark.sql.hive.metastore.jars", "path")
+ .set("spark.sql.hive.metastore.jars.path", glueHiveJarsDir + "/*")
+ .set("spark.sql.hive.metastore.sharedPrefixes", sharedPrefixes);
+
+ sparkSession =
+ SparkSession.builder()
+ .master("local[1]")
+ .appName("Spark connector Glue integration test")
+ .config(sparkConf)
+ .enableHiveSupport()
+ .getOrCreate();
+ }
+
+ /**
+ * Sets S3 credentials for Spark S3A filesystem access.
+ *
+ * @param endpoint S3 endpoint URL (e.g., http://localhost:4566)
+ * @param accessKey S3 access key
+ * @param secretKey S3 secret key
+ */
+ public void setS3Credentials(String endpoint, String accessKey, String
secretKey) {
+ this.s3Endpoint = endpoint;
+ this.s3AccessKey = accessKey;
+ this.s3SecretKey = secretKey;
+ }
+
+ public void setS3BucketName(String bucketName) {
+ this.s3BucketName = bucketName;
+ }
+
+ private final String metalakeName = "test";
+}
diff --git
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkUtilIT.java
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkUtilIT.java
index 0ea383e406..8bf9548768 100644
---
a/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkUtilIT.java
+++
b/spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkUtilIT.java
@@ -30,12 +30,13 @@ import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.gravitino.integration.test.util.BaseIT;
import org.apache.spark.sql.AnalysisException;
-import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.catalyst.analysis.ResolvedTable;
-import org.apache.spark.sql.catalyst.plans.logical.CommandResult;
-import org.apache.spark.sql.catalyst.plans.logical.DescribeRelation;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.connector.catalog.CatalogManager;
+import org.apache.spark.sql.connector.catalog.CatalogPlugin;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
import org.junit.jupiter.api.Assertions;
/**
@@ -134,12 +135,45 @@ public abstract class SparkUtilIT extends BaseIT {
}
// Create SparkTableInfo from SparkBaseTable retrieved from LogicalPlan.
+ // In Spark 3.3/3.5: DESC TABLE EXTENDED returns DescribeRelation.
+ // In Spark 3.4: DESC TABLE EXTENDED returns DescribeTableCommand (different
class hierarchy).
+ // Use the v2 Catalog API (CatalogManager + TableCatalog.loadTable) for
cross-version
+ // compatibility.
protected SparkTableInfo getTableInfo(String tableName) {
- Dataset ds = getSparkSession().sql("DESC TABLE EXTENDED " + tableName);
- CommandResult result = (CommandResult) ds.logicalPlan();
- DescribeRelation relation = (DescribeRelation) result.commandLogicalPlan();
- ResolvedTable table = (ResolvedTable) relation.child();
- return SparkTableInfo.create(table.table());
+ CatalogManager catalogManager =
getSparkSession().sessionState().catalogManager();
+
+ // Parse tableName: could be short (tbl), partially-qualified (db.tbl),
+ // or fully-qualified (cat.db.tbl).
+ String[] parts = tableName.split("\\.");
+ Identifier identifier;
+ TableCatalog tableCatalog;
+ if (parts.length == 1) {
+ // Short table name: use current catalog + current V2 namespace.
+ // catalog().currentDatabase() returns the V1 Hive session catalog
database and is NOT
+ // updated when USE <db> is issued against a V2 catalog (e.g. Glue) in
Spark 3.3.
+ // catalogManager.currentNamespace() reflects the V2 namespace correctly.
+ CatalogPlugin currentCatalog = catalogManager.currentCatalog();
+ String[] currentNamespace = catalogManager.currentNamespace();
+ identifier = Identifier.of(currentNamespace, parts[0]);
+ tableCatalog = (TableCatalog) currentCatalog;
+ } else if (parts.length == 2) {
+ // Partially qualified: db.table
+ identifier = Identifier.of(new String[] {parts[0]}, parts[1]);
+ CatalogPlugin currentCatalog = catalogManager.currentCatalog();
+ tableCatalog = (TableCatalog) currentCatalog;
+ } else if (parts.length == 3) {
+ // Fully qualified: cat.db.table — namespace is only the db part, not
the catalog
+ identifier = Identifier.of(new String[] {parts[1]}, parts[2]);
+ CatalogPlugin catalog = catalogManager.catalog(parts[0]);
+ tableCatalog = (TableCatalog) catalog;
+ } else {
+ throw new IllegalArgumentException("Invalid table name format: " +
tableName);
+ }
+ try {
+ return SparkTableInfo.create(tableCatalog.loadTable(identifier));
+ } catch (NoSuchTableException e) {
+ throw new RuntimeException(e);
+ }
}
protected List<Object[]> getTablePartitions(String tableName) {
@@ -159,6 +193,10 @@ public abstract class SparkUtilIT extends BaseIT {
if (e instanceof AnalysisException) {
return false;
}
+ // getTableInfo() wraps NoSuchTableException (extends AnalysisException)
in RuntimeException
+ if (e.getCause() instanceof AnalysisException) {
+ return false;
+ }
throw e;
}
}
diff --git a/spark-connector/v3.3/spark/build.gradle.kts
b/spark-connector/v3.3/spark/build.gradle.kts
index 2fbd6218c2..3581801337 100644
--- a/spark-connector/v3.3/spark/build.gradle.kts
+++ b/spark-connector/v3.3/spark/build.gradle.kts
@@ -16,6 +16,8 @@
* specific language governing permissions and limitations
* under the License.
*/
+evaluationDependsOn(":spark-connector:spark-common")
+
plugins {
`maven-publish`
id("java")
@@ -143,6 +145,16 @@ dependencies {
testImplementation(libs.mysql.driver)
testImplementation(libs.postgresql.driver)
testImplementation(libs.testcontainers)
+ testImplementation(libs.hadoop3.aws)
+ // hadoop-aws declares hadoop-client-api as provided; add it explicitly so
S3AFileSystem can load
+ // org.apache.hadoop.fs.impl.prefetch.PrefetchingStatistics (added in 3.3.5)
at runtime.
+ testImplementation(libs.hadoop3.client.api)
+ // Iceberg's GlueCatalog references several AWS SDK modules at runtime; must
be on test classpath
+ testImplementation(libs.aws.dynamodb)
+ testImplementation(libs.aws.glue)
+ testImplementation(libs.aws.sts)
+ testImplementation(libs.aws.s3)
+ testImplementation(libs.aws.kms)
// org.apache.iceberg.rest.RESTSerializers#registerAll(ObjectMapper) has
different method signature for iceberg-core and iceberg-spark-runtime package,
we must make sure iceberg-core is in front to start up MiniGravitino server.
testImplementation("org.apache.iceberg:iceberg-core:$icebergVersion")
@@ -172,6 +184,9 @@ dependencies {
testRuntimeOnly(libs.junit.jupiter.engine)
}
+val glueHiveJarsDir: String? =
+ project(":spark-connector:spark-common").extra["glueHiveJarsDir"] as String?
+
tasks.test {
val skipITs = project.hasProperty("skipITs")
val enableSparkSQLITs = project.hasProperty("enableSparkSQLITs")
@@ -182,6 +197,10 @@ tasks.test {
// Exclude integration tests
exclude("**/integration/test/**")
} else {
+ if (glueHiveJarsDir != null) {
+ dependsOn(":spark-connector:spark-common:downloadGlueHiveJars")
+ jvmArgs("-Dglue.hive-jars-dir=$glueHiveJarsDir")
+ }
dependsOn(tasks.jar)
dependsOn(":catalogs:catalog-lakehouse-iceberg:jar")
dependsOn(":catalogs:catalog-hive:jar")
diff --git
a/spark-connector/v3.3/spark/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalogSpark33.java
b/spark-connector/v3.3/spark/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalogSpark33.java
new file mode 100644
index 0000000000..4a8075682d
--- /dev/null
+++
b/spark-connector/v3.3/spark/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalogSpark33.java
@@ -0,0 +1,22 @@
+/*
+ * 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.glue;
+
+/** Spark 3.3 specific Gravitino Glue catalog implementation. */
+public class GravitinoGlueCatalogSpark33 extends GravitinoGlueCatalog {}
diff --git
a/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkAwsGlueCatalogIT33.java
b/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkAwsGlueCatalogIT33.java
new file mode 100644
index 0000000000..17b8a78ea7
--- /dev/null
+++
b/spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkAwsGlueCatalogIT33.java
@@ -0,0 +1,37 @@
+/*
+ * 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.glue;
+
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+
+/**
+ * Integration test for Spark 3.3 Glue catalog connector against a real AWS
Glue endpoint.
+ *
+ * <p>This test is <b>skipped by default</b> and only runs when {@code
AWS_ACCESS_KEY_ID} is set.
+ * Required environment variables:
+ *
+ * <ul>
+ * <li>{@code AWS_ACCESS_KEY_ID}
+ * <li>{@code AWS_SECRET_ACCESS_KEY}
+ * <li>{@code AWS_DEFAULT_REGION}
+ * <li>{@code AWS_S3_TEST_BUCKET}
+ * </ul>
+ */
+@EnabledIfEnvironmentVariable(named = "AWS_ACCESS_KEY_ID", matches = ".+")
+public class SparkAwsGlueCatalogIT33 extends SparkGlueCatalogIT {}
diff --git a/spark-connector/v3.4/spark/build.gradle.kts
b/spark-connector/v3.4/spark/build.gradle.kts
index d271ba3f94..d4da6062fc 100644
--- a/spark-connector/v3.4/spark/build.gradle.kts
+++ b/spark-connector/v3.4/spark/build.gradle.kts
@@ -16,6 +16,8 @@
* specific language governing permissions and limitations
* under the License.
*/
+evaluationDependsOn(":spark-connector:spark-common")
+
plugins {
`maven-publish`
id("java")
@@ -96,6 +98,16 @@ dependencies {
}
testImplementation(libs.awaitility)
+ // Iceberg's GlueCatalog references several AWS SDK modules at runtime; must
be on test classpath
+ testImplementation(libs.aws.dynamodb)
+ testImplementation(libs.aws.glue)
+ testImplementation(libs.aws.kms)
+ testImplementation(libs.aws.s3)
+ testImplementation(libs.aws.sts)
+ testImplementation(libs.hadoop3.aws)
+ // hadoop-aws declares hadoop-client-api as provided; add it explicitly so
S3AFileSystem can load
+ // org.apache.hadoop.fs.impl.prefetch.PrefetchingStatistics (added in 3.3.5)
at runtime.
+ testImplementation(libs.hadoop3.client.api)
testImplementation(libs.hive2.common) {
exclude("com.sun.jersey")
exclude("org.apache.curator")
@@ -163,6 +175,9 @@ dependencies {
testRuntimeOnly(libs.junit.jupiter.engine)
}
+val glueHiveJarsDir: String? =
+ project(":spark-connector:spark-common").extra["glueHiveJarsDir"] as String?
+
tasks.test {
val skipITs = project.hasProperty("skipITs")
val enableSparkSQLITs = project.hasProperty("enableSparkSQLITs")
@@ -173,6 +188,10 @@ tasks.test {
// Exclude integration tests
exclude("**/integration/test/**")
} else {
+ if (glueHiveJarsDir != null) {
+ dependsOn(":spark-connector:spark-common:downloadGlueHiveJars")
+ jvmArgs("-Dglue.hive-jars-dir=$glueHiveJarsDir")
+ }
dependsOn(tasks.jar)
dependsOn(":catalogs:catalog-lakehouse-iceberg:jar")
dependsOn(":catalogs:catalog-hive:jar")
diff --git
a/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalogSpark34.java
b/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalogSpark34.java
new file mode 100644
index 0000000000..3ca04141be
--- /dev/null
+++
b/spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalogSpark34.java
@@ -0,0 +1,38 @@
+/*
+ * 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.glue;
+
+import org.apache.gravitino.spark.connector.SparkTableChangeConverter;
+import org.apache.gravitino.spark.connector.SparkTableChangeConverter34;
+import org.apache.gravitino.spark.connector.SparkTypeConverter;
+import org.apache.gravitino.spark.connector.hive.SparkHiveTypeConverter34;
+
+/** Spark 3.4 specific Gravitino Glue catalog implementation. */
+public class GravitinoGlueCatalogSpark34 extends GravitinoGlueCatalog {
+ @Override
+ protected SparkTypeConverter getSparkTypeConverter() {
+ return new SparkHiveTypeConverter34();
+ }
+
+ @Override
+ protected SparkTableChangeConverter getSparkTableChangeConverter(
+ SparkTypeConverter sparkTypeConverter) {
+ return new SparkTableChangeConverter34(sparkTypeConverter);
+ }
+}
diff --git
a/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkAwsGlueCatalogIT34.java
b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkAwsGlueCatalogIT34.java
new file mode 100644
index 0000000000..c6a18f4827
--- /dev/null
+++
b/spark-connector/v3.4/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkAwsGlueCatalogIT34.java
@@ -0,0 +1,37 @@
+/*
+ * 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.glue;
+
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+
+/**
+ * Integration test for Spark 3.4 Glue catalog connector against a real AWS
Glue endpoint.
+ *
+ * <p>This test is <b>skipped by default</b> and only runs when {@code
AWS_ACCESS_KEY_ID} is set.
+ * Required environment variables:
+ *
+ * <ul>
+ * <li>{@code AWS_ACCESS_KEY_ID}
+ * <li>{@code AWS_SECRET_ACCESS_KEY}
+ * <li>{@code AWS_DEFAULT_REGION}
+ * <li>{@code AWS_S3_TEST_BUCKET}
+ * </ul>
+ */
+@EnabledIfEnvironmentVariable(named = "AWS_ACCESS_KEY_ID", matches = ".+")
+public class SparkAwsGlueCatalogIT34 extends SparkGlueCatalogIT {}
diff --git a/spark-connector/v3.5/spark/build.gradle.kts
b/spark-connector/v3.5/spark/build.gradle.kts
index c17a21d394..4b2a7b7402 100644
--- a/spark-connector/v3.5/spark/build.gradle.kts
+++ b/spark-connector/v3.5/spark/build.gradle.kts
@@ -16,6 +16,8 @@
* specific language governing permissions and limitations
* under the License.
*/
+evaluationDependsOn(":spark-connector:spark-common")
+
plugins {
`maven-publish`
id("java")
@@ -27,6 +29,9 @@ repositories {
mavenCentral()
}
+val glueHiveJarsDir: String? =
+ project(":spark-connector:spark-common").extra["glueHiveJarsDir"] as String?
+
val scalaVersion: String = project.properties["scalaVersion"] as? String ?:
extra["defaultScalaVersion"].toString()
val sparkVersion: String = libs.versions.spark35.get()
val sparkMajorVersion: String = sparkVersion.substringBeforeLast(".")
@@ -98,6 +103,16 @@ dependencies {
}
testImplementation(libs.awaitility)
+ // Iceberg's GlueCatalog references several AWS SDK modules at runtime; must
be on test classpath
+ testImplementation(libs.aws.dynamodb)
+ testImplementation(libs.aws.glue)
+ testImplementation(libs.aws.kms)
+ testImplementation(libs.aws.s3)
+ testImplementation(libs.aws.sts)
+ testImplementation(libs.hadoop3.aws)
+ // hadoop-aws declares hadoop-client-api as provided; add it explicitly so
S3AFileSystem can load
+ // org.apache.hadoop.fs.impl.prefetch.PrefetchingStatistics (added in 3.3.5)
at runtime.
+ testImplementation(libs.hadoop3.client.api)
testImplementation(libs.hive2.common) {
exclude("com.sun.jersey")
exclude("org.apache.curator")
@@ -175,6 +190,10 @@ tasks.test {
// Exclude integration tests
exclude("**/integration/test/**")
} else {
+ if (glueHiveJarsDir != null) {
+ dependsOn(":spark-connector:spark-common:downloadGlueHiveJars")
+ jvmArgs("-Dglue.hive-jars-dir=$glueHiveJarsDir")
+ }
dependsOn(tasks.jar)
dependsOn(":catalogs:catalog-lakehouse-iceberg:jar")
dependsOn(":catalogs:catalog-hive:jar")
diff --git
a/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalogSpark35.java
b/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalogSpark35.java
new file mode 100644
index 0000000000..0203ef9793
--- /dev/null
+++
b/spark-connector/v3.5/spark/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalogSpark35.java
@@ -0,0 +1,34 @@
+/*
+ * 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.glue;
+
+import java.util.Set;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.catalog.TableWritePrivilege;
+
+/** Spark 3.5 specific Gravitino Glue catalog implementation. */
+public class GravitinoGlueCatalogSpark35 extends GravitinoGlueCatalogSpark34 {
+ @Override
+ public Table loadTable(Identifier ident, Set<TableWritePrivilege>
writePrivileges)
+ throws NoSuchTableException {
+ return loadTableForWriting(ident);
+ }
+}
diff --git
a/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkAwsGlueCatalogIT35.java
b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkAwsGlueCatalogIT35.java
new file mode 100644
index 0000000000..449e371765
--- /dev/null
+++
b/spark-connector/v3.5/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkAwsGlueCatalogIT35.java
@@ -0,0 +1,37 @@
+/*
+ * 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.glue;
+
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+
+/**
+ * Integration test for Spark 3.5 Glue catalog connector against a real AWS
Glue endpoint.
+ *
+ * <p>This test is <b>skipped by default</b> and only runs when {@code
AWS_ACCESS_KEY_ID} is set.
+ * Required environment variables:
+ *
+ * <ul>
+ * <li>{@code AWS_ACCESS_KEY_ID}
+ * <li>{@code AWS_SECRET_ACCESS_KEY}
+ * <li>{@code AWS_DEFAULT_REGION}
+ * <li>{@code AWS_S3_TEST_BUCKET}
+ * </ul>
+ */
+@EnabledIfEnvironmentVariable(named = "AWS_ACCESS_KEY_ID", matches = ".+")
+public class SparkAwsGlueCatalogIT35 extends SparkGlueCatalogIT {}
