Copilot commented on code in PR #11186:
URL: https://github.com/apache/gravitino/pull/11186#discussion_r3308125781
##########
spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkUtilIT.java:
##########
@@ -134,12 +135,45 @@ protected List<String> getTableMetadata(String
getTableMetadataSql) {
}
// 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
+ identifier = Identifier.of(new String[] {parts[0], parts[1]}, parts[2]);
+ CatalogPlugin catalog = catalogManager.catalog(parts[0]);
+ tableCatalog = (TableCatalog) catalog;
Review Comment:
In the fully-qualified case (cat.db.table), the Identifier namespace should
be the table namespace *within* the selected catalog (db), not include the
catalog name. Including the catalog in the namespace will cause
TableCatalog.loadTable() lookups to fail.
##########
spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkTableInfo.java:
##########
@@ -118,27 +119,31 @@ static SparkTableInfo create(Table baseTable) {
items.length == 2, "Table name format should be $db.$table, but is: "
+ identifier);
sparkTableInfo.tableName = items[1];
sparkTableInfo.database = items[0];
- sparkTableInfo.columns =
- // using `baseTable.schema()` directly will failed because the method
named `schema` is
- // Deprecated in Spark Table interface
- Arrays.stream(getSchema(baseTable).fields())
- .map(
- sparkField ->
- new SparkColumnInfo(
- sparkField.name(),
- sparkField.dataType(),
- sparkField.getComment().isDefined() ?
sparkField.getComment().get() : null,
- sparkField.nullable()))
- .collect(Collectors.toList());
sparkTableInfo.comment =
baseTable.properties().remove(ConnectorConstants.COMMENT);
sparkTableInfo.tableProperties = baseTable.properties();
+ // V1Table.schema() includes partition columns in the data schema. We must
filter them
+ // out so that columns contains only the data columns (non-partition
columns).
+ // Collect partition column names first, then filter schema fields.
+ java.util.Set<String> partitionColNames = new java.util.HashSet<>();
Review Comment:
Avoid using fully-qualified java.util types here; Set/HashSet are already
imported in this file and using FQNs violates the repo import convention
(AGENTS.md:26-30).
##########
spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkTableInfo.java:
##########
@@ -35,6 +35,7 @@
import org.apache.spark.sql.connector.catalog.SupportsMetadataColumns;
import org.apache.spark.sql.connector.catalog.Table;
import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.catalog.V1Table;
import org.apache.spark.sql.connector.expressions.ApplyTransform;
Review Comment:
Avoid fully qualified Spark expression types in code; add a normal import
for NamedReference (AGENTS.md:26-30) so the loop below can use the simple name.
##########
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);
+ }
+
+ return new SparkHiveTable(
+ identifier,
+ gravitinoTable,
+ (HiveTable) sparkTable,
+ (HiveTableCatalog) sparkHiveCatalog,
+ propertiesConverter,
+ sparkTransformConverter,
+ sparkTypeConverter);
+ }
+
+ @Override
+ protected PropertiesConverter getPropertiesConverter() {
+ return GluePropertiesConverter.getInstance();
+ }
+
+ @Override
+ protected SparkTransformConverter getSparkTransformConverter() {
+ return new SparkTransformConverter(false);
+ }
+
+ @Override
+ protected SparkTypeConverter getSparkTypeConverter() {
+ return new SparkHiveTypeConverter();
+ }
+
+ private static boolean isIcebergProperties(Map<String, String> properties) {
+ if (properties == null) {
+ return false;
+ }
+ String provider = properties.get("provider");
+ if ("iceberg".equalsIgnoreCase(provider)) {
+ return true;
+ }
+ String tableFormat = properties.get(GlueConstants.TABLE_FORMAT);
+ return GlueConstants.TABLE_FORMAT_ICEBERG.equalsIgnoreCase(tableFormat);
+ }
+
+ /**
+ * 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 "ICEBERG".equalsIgnoreCase(properties.get("table_type"));
+ }
+
+ /**
+ * 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");
+ icebergGlueCatalog = createIcebergGlueCatalog();
+ }
+ }
+ }
+ 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);
+
+ // Iceberg GlueCatalog requires a warehouse path to derive default table
locations.
+ // Read from the active SparkSession if not explicitly set in catalog
properties.
+ if (!icebergProperties.containsKey("warehouse")) {
+ try {
+ SparkSession spark = SparkSession.active();
+ String warehouseDir = spark.conf().get("spark.sql.warehouse.dir",
null);
+ if (warehouseDir != null) {
+ icebergProperties.put("warehouse", warehouseDir);
+ }
+ } catch (Exception e) {
+ LOG.warn("Could not read spark.sql.warehouse.dir for Iceberg Glue
catalog", e);
+ }
+ }
+
+ 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) {
+ try {
+ return icebergCatalog.loadTable(identifier);
+ } catch (NoSuchTableException e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to load Iceberg table: %s",
+ String.join(".", getDatabase(identifier), identifier.name())),
+ e);
+ }
Review Comment:
loadIcebergSparkTable() wraps NoSuchTableException in a RuntimeException,
which means GravitinoGlueCatalog.loadTable() can throw an unexpected unchecked
exception instead of NoSuchTableException. Propagate NoSuchTableException so
Spark callers can handle "table not found" correctly.
##########
spark-connector/v3.4/spark/src/main/java/org/apache/gravitino/spark/connector/glue/GravitinoGlueCatalogSpark34.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.glue;
+
+import org.apache.gravitino.spark.connector.SparkTableChangeConverter;
+import org.apache.gravitino.spark.connector.SparkTableChangeConverter34;
+import org.apache.gravitino.spark.connector.SparkTypeConverter;
+
+/** Spark 3.4 specific Gravitino Glue catalog implementation. */
+public class GravitinoGlueCatalogSpark34 extends GravitinoGlueCatalog {
+ @Override
+ protected SparkTypeConverter getSparkTypeConverter() {
+ return new
org.apache.gravitino.spark.connector.hive.SparkHiveTypeConverter34();
+ }
Review Comment:
Repo convention is to avoid fully qualified class names in code when an
import works (AGENTS.md:26-30). Use a normal import for
SparkHiveTypeConverter34 instead of the inline FQN.
##########
spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/util/SparkTableInfo.java:
##########
@@ -118,27 +119,31 @@ static SparkTableInfo create(Table baseTable) {
items.length == 2, "Table name format should be $db.$table, but is: "
+ identifier);
sparkTableInfo.tableName = items[1];
sparkTableInfo.database = items[0];
- sparkTableInfo.columns =
- // using `baseTable.schema()` directly will failed because the method
named `schema` is
- // Deprecated in Spark Table interface
- Arrays.stream(getSchema(baseTable).fields())
- .map(
- sparkField ->
- new SparkColumnInfo(
- sparkField.name(),
- sparkField.dataType(),
- sparkField.getComment().isDefined() ?
sparkField.getComment().get() : null,
- sparkField.nullable()))
- .collect(Collectors.toList());
sparkTableInfo.comment =
baseTable.properties().remove(ConnectorConstants.COMMENT);
sparkTableInfo.tableProperties = baseTable.properties();
+ // V1Table.schema() includes partition columns in the data schema. We must
filter them
+ // out so that columns contains only the data columns (non-partition
columns).
+ // Collect partition column names first, then filter schema fields.
+ java.util.Set<String> partitionColNames = new java.util.HashSet<>();
Arrays.stream(baseTable.partitioning())
.forEach(
transform -> {
+ if (transform instanceof IdentityTransform) {
+ partitionColNames.add(((IdentityTransform)
transform).reference().fieldNames()[0]);
+ }
if (transform instanceof BucketTransform
|| transform instanceof SortedBucketTransform) {
if (isBucketPartition(baseTable, transform)) {
sparkTableInfo.addPartition(transform);
+ // Collect bucket partition column names
+ if (transform instanceof BucketTransform) {
+ for
(org.apache.spark.sql.connector.expressions.NamedReference ref :
+ ((BucketTransform) transform).references()) {
Review Comment:
Now that NamedReference is imported, use it directly instead of an inline
fully qualified name (AGENTS.md:26-30).
##########
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:
createSparkTable() currently falls back to calling loadIcebergSparkTable(),
which wraps NoSuchTableException into a RuntimeException (breaking Spark’s
TableCatalog contract). Since all current call sites already pass an Iceberg
SparkTable for Iceberg tables, prefer to require SparkTable here and fail fast
if it’s missing.
--
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]
