rahil-c commented on code in PR #752: URL: https://github.com/apache/incubator-xtable/pull/752#discussion_r2467158116
########## xtable-spark-plugin/src/main/java/org/apache/xtable/spark/XTableSparkCatalog.java: ########## @@ -0,0 +1,506 @@ +/* + * 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.xtable.spark; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +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.NonEmptyNamespaceException; +import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException; +import org.apache.spark.sql.connector.catalog.CatalogPlugin; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.NamespaceChange; +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.catalog.TableChange; +import org.apache.spark.sql.connector.expressions.Transform; +import org.apache.spark.sql.hudi.catalog.HoodieCatalog; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.iceberg.aws.glue.GlueCatalog; +import org.apache.iceberg.spark.SparkCatalog; + +import com.google.common.annotations.VisibleForTesting; + +import org.apache.xtable.catalog.TableFormatUtils; +import org.apache.xtable.model.storage.TableFormat; + +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.glue.model.GetTableRequest; +import software.amazon.awssdk.services.glue.model.GetTableResponse; +import software.amazon.awssdk.services.glue.model.GlueException; +import software.amazon.awssdk.services.glue.model.StorageDescriptor; + +/** + * XTable Spark Catalog Plugin - Unified catalog implementation for both Hudi and Iceberg tables. + * This plugin leverages XTable's existing format detection capabilities and provides a seamless + * experience for querying mixed table formats through a single Spark catalog. + */ +public class XTableSparkCatalog implements CatalogPlugin, TableCatalog, SupportsNamespaces { + + private static final Logger LOG = LoggerFactory.getLogger(XTableSparkCatalog.class); + + private String catalogName; + private SparkCatalog icebergCatalog; + private HoodieCatalog hudiCatalog; + private GlueClient glueClient; + private CaseInsensitiveStringMap options; + + @Override + public void initialize(String name, CaseInsensitiveStringMap options) { + this.catalogName = name; + this.options = options; + + LOG.info("Initializing XTable Spark Catalog Plugin with name: {}", name); + + this.glueClient = buildGlueClient(options); + + SparkSession spark = SparkSession.active(); + spark.conf().set("hoodie.schema.on.read.enable", "true"); + + initializeIcebergCatalog(name, options); + initializeHudiCatalog(options); + } + + private GlueClient buildGlueClient(CaseInsensitiveStringMap options) { + String region = + options.getOrDefault("glue.region", options.getOrDefault("aws.region", "us-west-2")); + + LOG.info("Configuring Glue client with region: {}", region); + + GlueClient client = + GlueClient.builder().region(software.amazon.awssdk.regions.Region.of(region)).build(); + + // Log catalog ID if specified (for cross-account access) + if (options.containsKey("catalog-id")) { + String catalogId = options.get("catalog-id"); + LOG.info("Using Glue catalog ID: {} (for cross-account access)", catalogId); + } + + return client; + } + + private void initializeIcebergCatalog(String name, CaseInsensitiveStringMap options) { + LOG.info("Initializing Iceberg sub-catalog"); + + icebergCatalog = new SparkCatalog(); + + Map<String, String> icebergOptions = new HashMap<>(options); + icebergOptions.put("catalog-impl", GlueCatalog.class.getName()); + + if (options.containsKey("warehouse")) { + icebergOptions.put("warehouse", options.get("warehouse")); + } + + // Pass catalog-id for cross-account Glue access if specified + if (options.containsKey("catalog-id")) { + icebergOptions.put("catalog-id", options.get("catalog-id")); + LOG.info("Configured Iceberg catalog with catalog-id: {}", options.get("catalog-id")); + } + + icebergCatalog.initialize(name, new CaseInsensitiveStringMap(icebergOptions)); + } + + private void initializeHudiCatalog(CaseInsensitiveStringMap options) { + LOG.info("Initializing Hudi sub-catalog"); + + hudiCatalog = new HoodieCatalog(); + + // Set up delegation to spark_catalog for metastore operations + // This allows Hudi to access the underlying Glue/Hive metastore through spark_catalog + try { + SparkSession spark = SparkSession.active(); + CatalogPlugin sparkCatalog = spark.sessionState().catalogManager().catalog("spark_catalog"); + hudiCatalog.setDelegateCatalog(sparkCatalog); + LOG.info("Successfully set spark_catalog as delegate for HoodiCatalog"); + } catch (Exception e) { + LOG.error( + "Failed to set delegate catalog for HoodieCatalog. This may cause table operations to fail.", + e); + throw new RuntimeException( + "Failed to initialize HoodieCatalog with spark_catalog delegate", e); + } + + Map<String, String> hudiOptions = new HashMap<>(options); + + hudiOptions.put("provider", "hudi"); Review Comment: Can you also try removing these properties I think it should be handled by default. -- 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]
