diqiu50 commented on code in PR #11186: URL: https://github.com/apache/gravitino/pull/11186#discussion_r3309242085
########## spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalog.java: ########## @@ -0,0 +1,484 @@ +/* + * 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.exceptions.ForbiddenException; +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.SparkSession; +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; + } + + /** + * Overrides createTable to handle Iceberg and non-Iceberg tables differently. + * + * <p>For Iceberg tables: delegates directly to the Iceberg GlueCatalog, which creates both the + * Iceberg metadata in S3 and the Glue table entry with {@code table_type=ICEBERG}. BaseCatalog's + * {@code loadSparkTable()} would fail because Derby never has Iceberg entries. + * + * <p>For non-Iceberg tables: pre-creates a placeholder in Derby before calling Gravitino. + * BaseCatalog.createTable() calls loadSparkTable() after creating in Gravitino, which routes to + * HiveTableCatalog.loadTable() → Derby. Without the placeholder, Derby returns + * NoSuchTableException. + */ + @Override + public Table createTable( + Identifier ident, StructType schema, Transform[] partitions, Map<String, String> properties) + throws TableAlreadyExistsException, NoSuchNamespaceException { + syncNamespaceToDerby(ident.namespace()); + if (isIcebergProperties(properties)) { + SparkCatalog icebergCatalog = getOrCreateIcebergGlueCatalog(); + icebergCatalog.createTable(ident, schema, partitions, properties); + try { + org.apache.gravitino.rel.Table gravitinoTable = loadGravitinoTable(ident); + Table icebergSparkTable = loadIcebergSparkTable(ident, icebergCatalog); + return createSparkTable( + ident, + gravitinoTable, + icebergSparkTable, + sparkCatalog, + getPropertiesConverter(), + getSparkTransformConverter(), + getSparkTypeConverter()); + } catch (NoSuchTableException e) { + try { + icebergCatalog.dropTable(ident); + } catch (Exception rollbackEx) { + LOG.warn("Failed to rollback Iceberg table creation for {}", ident, rollbackEx); + } + throw new RuntimeException("Failed to load Iceberg table after creation: " + ident, e); + } + } + // Hive tables require a non-null location in the StorageDescriptor for write operations. + // If not explicitly set, derive the default from spark.sql.warehouse.dir. + if (!properties.containsKey("location")) { + try { + SparkSession spark = SparkSession.active(); + String warehouseDir = spark.conf().get("spark.sql.warehouse.dir", null); + if (warehouseDir != null) { + String db = ident.namespace()[ident.namespace().length - 1]; + Map<String, String> withLocation = new HashMap<>(properties); + withLocation.put( + TableCatalog.PROP_LOCATION, warehouseDir + "/" + db + "/" + ident.name()); + properties = withLocation; + } + } catch (Exception e) { + LOG.warn("Could not derive default table location", e); + } + } + try { + sparkCatalog.createTable(ident, schema, partitions, properties); + } catch (TableAlreadyExistsException e) { + // Already in Derby — OK. + } catch (Exception e) { + LOG.warn("Pre-create in Derby failed for {}", ident, e); + } + return super.createTable(ident, schema, partitions, properties); + } + + /** + * Overrides loadTable to handle Iceberg and non-Iceberg tables differently. + * + * <p>For Iceberg tables: loads the Gravitino metadata from Glue and delegates table loading to + * the Iceberg GlueCatalog, bypassing Derby entirely. Iceberg tables are never registered in Derby + * because HiveTableCatalog cannot represent them. + * + * <p>For non-Iceberg tables: ensures the table exists in Derby before calling super.loadTable(). + * This handles tables that exist in Gravitino/Glue but were not created through this catalog + * instance (e.g., tables from a previous test run, or tables loaded after a JVM restart). + */ + @Override + public Table loadTable(Identifier ident) throws NoSuchTableException { + org.apache.gravitino.rel.Table gravitinoTable = loadGravitinoTable(ident); + if (isIcebergTable(gravitinoTable)) { + SparkCatalog icebergCatalog = getOrCreateIcebergGlueCatalog(); + Table icebergSparkTable = loadIcebergSparkTable(ident, icebergCatalog); + return createSparkTable( + ident, + gravitinoTable, + icebergSparkTable, + sparkCatalog, + getPropertiesConverter(), + getSparkTransformConverter(), + getSparkTypeConverter()); + } + // Non-Iceberg: ensure Derby is in sync, then delegate to BaseCatalog.loadTable(). + try { + sparkCatalog.loadTable(ident); + } catch (NoSuchTableException e) { + syncTableToDerby(ident, gravitinoTable); + } + return super.loadTable(ident); + } + + /** + * Overrides loadTableForWriting (called by Spark's write path) to route Iceberg tables to the + * Iceberg GlueCatalog. The base implementation calls loadSparkTable (HiveTableCatalog) which does + * not hold Iceberg entries. + */ + @Override + protected Table loadTableForWriting(Identifier ident) + throws NoSuchTableException, ForbiddenException { + org.apache.gravitino.rel.Table gravitinoTable = loadGravitinoTableForWriting(ident); + if (isIcebergTable(gravitinoTable)) { + SparkCatalog icebergCatalog = getOrCreateIcebergGlueCatalog(); + Table icebergSparkTable = loadIcebergSparkTable(ident, icebergCatalog); + return createSparkTable( + ident, + gravitinoTable, + icebergSparkTable, + sparkCatalog, + getPropertiesConverter(), + getSparkTransformConverter(), + getSparkTypeConverter()); + } + return super.loadTableForWriting(ident); + } + + /** + * 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. + */ + @Override + public boolean dropTable(Identifier ident) { + dropFromDerby(ident); + return super.dropTable(ident); + } + + /** Overrides purgeTable to also remove the table from Derby. */ + @Override + public boolean purgeTable(Identifier ident) { + dropFromDerby(ident); + 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); + } + + @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)) { + SparkCatalog icebergCatalog = getOrCreateIcebergGlueCatalog(); + // Reuse the already-loaded sparkTable when the caller has it; load only when missing. + Table icebergSparkTable = + (sparkTable instanceof SparkTable) + ? sparkTable + : loadIcebergSparkTable(identifier, icebergCatalog); + return new SparkIcebergTable( + identifier, + gravitinoTable, + (SparkTable) icebergSparkTable, + icebergCatalog, + propertiesConverter, + sparkTransformConverter, + sparkTypeConverter); Review Comment: fallback to iceberg is required -- 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]
