yuqi1129 commented on code in PR #11186: URL: https://github.com/apache/gravitino/pull/11186#discussion_r3318294602
########## spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalog.java: ########## @@ -0,0 +1,417 @@ +/* + * 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.Arrays; +import java.util.Collections; +import java.util.HashMap; +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.NamespaceAlreadyExistsException; +import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException; +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; +import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.SupportsNamespaces; +import org.apache.spark.sql.connector.catalog.Table; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.connector.expressions.Transform; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 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 HiveTableCatalog for I/O + * <li>Iceberg tables: routed to Iceberg's 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>Derby sync: Gravitino creates/modifies tables in AWS Glue. HiveTableCatalog (used as Review Comment: This should not be described as Derby-specific. The catalog is writing a shadow entry into whatever Hive metastore `HiveTableCatalog` is configured to use; embedded Derby is only one possible backend. Please rename the methods/comments to Spark/Hive metastore shadow metadata and document that Glue/Gravitino remains the source of truth. ########## spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalog.java: ########## @@ -0,0 +1,417 @@ +/* + * 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.Arrays; +import java.util.Collections; +import java.util.HashMap; +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.NamespaceAlreadyExistsException; +import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException; +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; +import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.SupportsNamespaces; +import org.apache.spark.sql.connector.catalog.Table; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.connector.expressions.Transform; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 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 HiveTableCatalog for I/O + * <li>Iceberg tables: routed to Iceberg's 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>Derby sync: Gravitino creates/modifies tables in AWS Glue. HiveTableCatalog (used as + * sparkCatalog) uses an embedded Derby metastore for metadata validation. We must keep Derby in + * sync with Glue for loadSparkTable() to succeed. Derby is populated lazily on createTable() and + * loadTable(), and cleaned up on dropTable()/purgeTable()/renameTable(). + */ +public class GravitinoGlueCatalog extends BaseCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(GravitinoGlueCatalog.class); + + // 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 after Gravitino creates the table. + * + * <p>Iceberg tables are loaded from the Iceberg GlueCatalog; they are never registered in Derby. + * Hive tables are loaded from Derby, syncing from Glue first if the entry is missing. + */ + @Override + protected Table loadSparkTable(Identifier ident) { + try { + org.apache.gravitino.rel.Table gravitinoTable = loadGravitinoTable(ident); + if (isIcebergTable(gravitinoTable)) { + return loadIcebergSparkTable(ident, getOrCreateIcebergGlueCatalog()); + } + syncNamespaceToDerby(ident.namespace()); + try { + return sparkCatalog.loadTable(ident); Review Comment: This only syncs the Hive metastore shadow entry when `loadTable` misses. If the shadow entry already exists but Glue/Gravitino metadata changed (for example after alterTable, schema changes, location/serde changes, or a previous stale run), Spark will keep using the old `HiveTable.catalogTable()` metadata. Please refresh/recreate the shadow entry after metadata-changing operations, or validate that the existing entry matches the Gravitino table before returning it. ########## spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalog.java: ########## @@ -0,0 +1,417 @@ +/* + * 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.Arrays; +import java.util.Collections; +import java.util.HashMap; +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.NamespaceAlreadyExistsException; +import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException; +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; +import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.SupportsNamespaces; +import org.apache.spark.sql.connector.catalog.Table; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.connector.expressions.Transform; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 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 HiveTableCatalog for I/O + * <li>Iceberg tables: routed to Iceberg's 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>Derby sync: Gravitino creates/modifies tables in AWS Glue. HiveTableCatalog (used as + * sparkCatalog) uses an embedded Derby metastore for metadata validation. We must keep Derby in + * sync with Glue for loadSparkTable() to succeed. Derby is populated lazily on createTable() and + * loadTable(), and cleaned up on dropTable()/purgeTable()/renameTable(). + */ +public class GravitinoGlueCatalog extends BaseCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(GravitinoGlueCatalog.class); + + // 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 after Gravitino creates the table. + * + * <p>Iceberg tables are loaded from the Iceberg GlueCatalog; they are never registered in Derby. + * Hive tables are loaded from Derby, syncing from Glue first if the entry is missing. + */ + @Override + protected Table loadSparkTable(Identifier ident) { + try { + org.apache.gravitino.rel.Table gravitinoTable = loadGravitinoTable(ident); + if (isIcebergTable(gravitinoTable)) { + return loadIcebergSparkTable(ident, getOrCreateIcebergGlueCatalog()); + } + syncNamespaceToDerby(ident.namespace()); + try { + return sparkCatalog.loadTable(ident); + } catch (NoSuchTableException e) { + syncTableToDerby(ident, gravitinoTable); + return sparkCatalog.loadTable(ident); + } + } catch (NoSuchTableException e) { + throw new RuntimeException( + String.format("Failed to load spark table: %s.%s", getDatabase(ident), ident.name()), e); + } + } + + /** + * Overrides dropTable to also remove the table from Derby. Without this, Derby accumulates stale + * entries that cause TableAlreadyExistsException on the next createTable call for the same name. + * Iceberg tables are never registered in Derby, so Derby cleanup is skipped for them. + */ + @Override + public boolean dropTable(Identifier ident) { + try { + if (!isIcebergTable(loadGravitinoTable(ident))) { + dropFromDerby(ident); + } + } catch (NoSuchTableException e) { + return false; + } + return super.dropTable(ident); + } + + /** + * Overrides purgeTable to also remove the table from Derby. Iceberg tables are never registered + * in Derby, so Derby cleanup is skipped for them. + */ + @Override + public boolean purgeTable(Identifier ident) { + try { + if (!isIcebergTable(loadGravitinoTable(ident))) { + dropFromDerby(ident); Review Comment: Same ordering issue as `dropTable`, and it is more visible for Glue because the Glue catalog operations do not override `purgeTable` today. `super.purgeTable(ident)` can throw `UnsupportedOperationException` after this code has already removed the local shadow entry. Please call the authoritative operation first and clean the shadow entry only after success, or do not override purge until Glue supports it. ########## spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkGlueCatalogIT.java: ########## @@ -0,0 +1,582 @@ +/* + * 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 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.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 = "us-east-1"; + private String awsAccessKeyId = "test"; + private String awsSecretAccessKey = "test"; + + @Override + protected String getCatalogName() { + return "glue"; + } + + @Override + protected String getProvider() { + return "glue"; + } + + @Override + protected Map<String, String> 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; + } + + /** + * 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)); + } + + /** + * Overrides base class: skip this test due to a known issue where ALTER TABLE RENAME fails for + * non-Iceberg (PARQUET) tables via the Glue catalog path. The rename operation triggers + * tableExists() which calls tableCatalog.loadTable() → GravitinoGlueCatalog.loadTable() → + * loadSparkTable() → HiveTableCatalog.loadTable(ident), but the table lookup may fail because + * Derby and Glue are out of sync for renamed tables. This requires further investigation into the + * HiveTableCatalog.loadTable() behavior and the renameTable dispatch chain in + * RenameTableExec.apply(). + */ + @Test + @Override + protected void testRenameTable() { Review Comment: This turns a known failing/unsupported rename path into a passing empty test. Please use `@Disabled` with the tracking issue/reason, or introduce a capability switch such as `supportsRenameTable()` so the report clearly shows the case is skipped instead of passed. ########## spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalog.java: ########## @@ -0,0 +1,417 @@ +/* + * 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.Arrays; +import java.util.Collections; +import java.util.HashMap; +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.NamespaceAlreadyExistsException; +import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException; +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; +import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.SupportsNamespaces; +import org.apache.spark.sql.connector.catalog.Table; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.connector.expressions.Transform; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 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 HiveTableCatalog for I/O + * <li>Iceberg tables: routed to Iceberg's 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>Derby sync: Gravitino creates/modifies tables in AWS Glue. HiveTableCatalog (used as + * sparkCatalog) uses an embedded Derby metastore for metadata validation. We must keep Derby in + * sync with Glue for loadSparkTable() to succeed. Derby is populated lazily on createTable() and + * loadTable(), and cleaned up on dropTable()/purgeTable()/renameTable(). + */ +public class GravitinoGlueCatalog extends BaseCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(GravitinoGlueCatalog.class); + + // 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 after Gravitino creates the table. + * + * <p>Iceberg tables are loaded from the Iceberg GlueCatalog; they are never registered in Derby. + * Hive tables are loaded from Derby, syncing from Glue first if the entry is missing. + */ + @Override + protected Table loadSparkTable(Identifier ident) { + try { + org.apache.gravitino.rel.Table gravitinoTable = loadGravitinoTable(ident); + if (isIcebergTable(gravitinoTable)) { + return loadIcebergSparkTable(ident, getOrCreateIcebergGlueCatalog()); + } + syncNamespaceToDerby(ident.namespace()); + try { + return sparkCatalog.loadTable(ident); + } catch (NoSuchTableException e) { + syncTableToDerby(ident, gravitinoTable); + return sparkCatalog.loadTable(ident); + } + } catch (NoSuchTableException e) { + throw new RuntimeException( + String.format("Failed to load spark table: %s.%s", getDatabase(ident), ident.name()), e); + } + } + + /** + * Overrides dropTable to also remove the table from Derby. Without this, Derby accumulates stale + * entries that cause TableAlreadyExistsException on the next createTable call for the same name. + * Iceberg tables are never registered in Derby, so Derby cleanup is skipped for them. + */ + @Override + public boolean dropTable(Identifier ident) { + try { + if (!isIcebergTable(loadGravitinoTable(ident))) { + dropFromDerby(ident); + } + } catch (NoSuchTableException e) { + return false; + } + return super.dropTable(ident); + } + + /** + * Overrides purgeTable to also remove the table from Derby. Iceberg tables are never registered + * in Derby, so Derby cleanup is skipped for them. + */ + @Override + public boolean purgeTable(Identifier ident) { + try { + if (!isIcebergTable(loadGravitinoTable(ident))) { + dropFromDerby(ident); + } + } catch (NoSuchTableException e) { + return false; + } + return super.purgeTable(ident); + } + + /** + * Overrides renameTable to keep Derby in sync after the Gravitino rename. The old Derby entry is + * dropped eagerly; the new entry is synced lazily on the next loadTable call. + */ + @Override + public void renameTable(Identifier oldIdent, Identifier newIdent) + throws NoSuchTableException, TableAlreadyExistsException { + super.renameTable(oldIdent, newIdent); + dropFromDerby(oldIdent); + } + + /** + * 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); + } + + Preconditions.checkArgument( + sparkTable instanceof HiveTable, + "Glue table %s expected HiveTable from HiveTableCatalog, got %s", + identifier, + sparkTable.getClass().getName()); + Preconditions.checkArgument( + sparkHiveCatalog instanceof HiveTableCatalog, + "Glue catalog expected HiveTableCatalog, got %s", + sparkHiveCatalog.getClass().getName()); + 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; + } + + /** + * Loads the raw Spark table from the Iceberg GlueCatalog. + * + * @param identifier the table identifier + * @param icebergCatalog the Iceberg SparkCatalog + * @return the Spark table + */ + private Table loadIcebergSparkTable(Identifier identifier, SparkCatalog icebergCatalog) + throws NoSuchTableException { + return icebergCatalog.loadTable(identifier); + } + + /** Creates the namespace in Derby if it does not already exist. */ + private void syncNamespaceToDerby(String[] namespace) { + if (!(sparkCatalog instanceof SupportsNamespaces)) { + LOG.warn( + "sparkCatalog {} does not implement SupportsNamespaces; " + + "skipping Derby namespace sync for {}. Derby createTable may fail.", + sparkCatalog.getClass().getName(), + Arrays.toString(namespace)); + return; + } + SupportsNamespaces ns = (SupportsNamespaces) sparkCatalog; + try { + ns.loadNamespaceMetadata(namespace); + } catch (NoSuchNamespaceException e) { + try { + ns.createNamespace(namespace, Collections.emptyMap()); + } catch (NamespaceAlreadyExistsException ex) { + // Created concurrently — OK. + } + } + } + + /** + * Creates a Derby entry from an already-loaded Gravitino table. Called when loadTable() detects + * the table is missing from Derby (e.g., after a JVM restart or against a pre-populated catalog). + * The caller supplies the Gravitino table to avoid a redundant load and eliminate the TOCTOU + * window that would exist if we re-fetched it here. + * + * @param ident the table identifier + * @param gravitinoTable the Gravitino table metadata + */ + private void syncTableToDerby(Identifier ident, org.apache.gravitino.rel.Table gravitinoTable) { + SparkTypeConverter typeConverter = getSparkTypeConverter(); + SparkTransformConverter transformConverter = getSparkTransformConverter(); + + StructField[] fields = + Arrays.stream(gravitinoTable.columns()) + .map( + col -> + DataTypes.createStructField( + col.name(), typeConverter.toSparkType(col.dataType()), col.nullable())) + .toArray(StructField[]::new); + StructType schema = new StructType(fields); + + Transform[] sparkPartitions = + transformConverter.toSparkTransform(gravitinoTable.partitioning(), null, null); + + Map<String, String> props = new HashMap<>(); + Map<String, String> gravitinoProps = gravitinoTable.properties(); + String location = gravitinoProps.get(GlueConstants.LOCATION); + if (location != null) { + props.put(TableCatalog.PROP_LOCATION, location); + } + String inputFormat = gravitinoProps.get(GlueConstants.INPUT_FORMAT_CLASS); + if (inputFormat != null) { + props.put("hive.input-format", inputFormat); + } + String outputFormat = gravitinoProps.get(GlueConstants.OUTPUT_FORMAT); + if (outputFormat != null) { + props.put("hive.output-format", outputFormat); + } + String serdeLib = gravitinoProps.get(GlueConstants.SERDE_LIB); + if (serdeLib != null) { + props.put("hive.serde", serdeLib); + } + + syncNamespaceToDerby(ident.namespace()); + try { + sparkCatalog.createTable(ident, schema, sparkPartitions, props); Review Comment: The shadow table is created with only schema, partition transforms, location, input/output format, and serde. It drops table comments, column comments, serde parameters, and other table properties that may be needed by Spark/Hive readers or writers. Please either sync the full Hive-compatible metadata from Glue/Gravitino, or document and test the intentionally supported subset. ########## spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkGlueEnvIT.java: ########## @@ -0,0 +1,358 @@ +/* + * 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 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 = S3_BUCKET_NAME; + + protected static final String S3_BUCKET_NAME = "ice-glue-test-01"; + protected static final int DEFAULT_GRAVITINO_PORT = 8090; + + @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 { + warehouse = "s3a://" + s3BucketName + "/warehouse"; + hiveMetastoreUri = null; + hdfs = null; + + int gravitinoPort; + boolean serverWasStartedByThisClass = false; + if (serverConfig != null) { + // Gravitino server already started by an external process (e.g., real AWS integration + // test environment). Use existing config. + gravitinoPort = getGravitinoServerPort(); + } else { + // 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(). + try { Review Comment: Catching all embedded-server startup failures and falling back to port 8090 hides the real failure and makes the test fail later with a less useful connection error. Since the external-server path is already represented by `serverConfig != null`, please let embedded startup failures propagate here. ########## spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalog.java: ########## @@ -0,0 +1,417 @@ +/* + * 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.Arrays; +import java.util.Collections; +import java.util.HashMap; +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.NamespaceAlreadyExistsException; +import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException; +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; +import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.SupportsNamespaces; +import org.apache.spark.sql.connector.catalog.Table; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.connector.expressions.Transform; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 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 HiveTableCatalog for I/O + * <li>Iceberg tables: routed to Iceberg's 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>Derby sync: Gravitino creates/modifies tables in AWS Glue. HiveTableCatalog (used as + * sparkCatalog) uses an embedded Derby metastore for metadata validation. We must keep Derby in + * sync with Glue for loadSparkTable() to succeed. Derby is populated lazily on createTable() and + * loadTable(), and cleaned up on dropTable()/purgeTable()/renameTable(). + */ +public class GravitinoGlueCatalog extends BaseCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(GravitinoGlueCatalog.class); + + // 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 after Gravitino creates the table. + * + * <p>Iceberg tables are loaded from the Iceberg GlueCatalog; they are never registered in Derby. + * Hive tables are loaded from Derby, syncing from Glue first if the entry is missing. + */ + @Override + protected Table loadSparkTable(Identifier ident) { + try { + org.apache.gravitino.rel.Table gravitinoTable = loadGravitinoTable(ident); + if (isIcebergTable(gravitinoTable)) { + return loadIcebergSparkTable(ident, getOrCreateIcebergGlueCatalog()); + } + syncNamespaceToDerby(ident.namespace()); + try { + return sparkCatalog.loadTable(ident); + } catch (NoSuchTableException e) { + syncTableToDerby(ident, gravitinoTable); + return sparkCatalog.loadTable(ident); + } + } catch (NoSuchTableException e) { + throw new RuntimeException( + String.format("Failed to load spark table: %s.%s", getDatabase(ident), ident.name()), e); + } + } + + /** + * Overrides dropTable to also remove the table from Derby. Without this, Derby accumulates stale + * entries that cause TableAlreadyExistsException on the next createTable call for the same name. + * Iceberg tables are never registered in Derby, so Derby cleanup is skipped for them. + */ + @Override + public boolean dropTable(Identifier ident) { + try { + if (!isIcebergTable(loadGravitinoTable(ident))) { + dropFromDerby(ident); Review Comment: The local shadow entry is dropped before the authoritative Glue/Gravitino drop succeeds. If `super.dropTable` fails, the source table still exists but Spark has already lost its local Hive metastore entry. Please perform the shadow cleanup only after `super.dropTable(ident)` returns true. ########## spark-connector/v3.3/spark/src/test/java/org/apache/gravitino/spark/connector/integration/test/glue/SparkAwsGlueCatalogIT33.java: ########## @@ -0,0 +1,71 @@ +/* + * 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 java.io.IOException; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +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} (optional, defaults to us-east-1) + * <li>{@code S3_BUCKET_NAME} (optional, uses default bucket if not set) + * </ul> + */ +@EnabledIfEnvironmentVariable(named = "AWS_ACCESS_KEY_ID", matches = ".+") Review Comment: This enables the real AWS test when only `AWS_ACCESS_KEY_ID` is set, but the setup also requires `AWS_SECRET_ACCESS_KEY` and uses a real S3 bucket. Please require the secret key too, and strongly consider requiring `S3_BUCKET_NAME` instead of falling back to the fixed default bucket. The same applies to the Spark 3.4 and 3.5 variants. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
