jerryshao commented on code in PR #13209: URL: https://github.com/apache/gravitino/pull/13209#discussion_r4025087032
########## spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/SparkCatalogExtensions.java: ########## @@ -0,0 +1,116 @@ +/* + * 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.plugin; + +import java.util.HashMap; +import java.util.Iterator; +import java.util.Locale; +import java.util.Map; +import java.util.ServiceConfigurationError; +import java.util.ServiceLoader; +import javax.annotation.Nullable; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Resolves Spark catalog classes contributed through {@link SparkCatalogExtension}. */ +public class SparkCatalogExtensions { + + private static final Logger LOG = LoggerFactory.getLogger(SparkCatalogExtensions.class); + + private static volatile Map<String, String> catalogClassNamesByProvider; + + private SparkCatalogExtensions() {} + + /** + * Looks up the Spark catalog class an extension registered for a provider. + * + * @param provider the Gravitino catalog provider + * @return the catalog class name, or null when no extension serves the provider + */ + @Nullable + public static String catalogClassName(String provider) { + return extensions().get(provider.toLowerCase(Locale.ROOT)); + } + + private static Map<String, String> extensions() { + Map<String, String> loaded = catalogClassNamesByProvider; + if (loaded == null) { + synchronized (SparkCatalogExtensions.class) { + loaded = catalogClassNamesByProvider; + if (loaded == null) { + loaded = load(); + catalogClassNamesByProvider = loaded; + } + } + } + return loaded; + } + + private static Map<String, String> load() { + Map<String, String> byProvider = new HashMap<>(); + Iterator<SparkCatalogExtension> iterator = + ServiceLoader.load(SparkCatalogExtension.class, extensionClassLoader()).iterator(); + while (true) { + try { + if (!iterator.hasNext()) { + break; + } + SparkCatalogExtension extension = iterator.next(); + String provider = extension.provider(); + String catalogClassName = extension.catalogClassName(); + if (StringUtils.isBlank(provider) || StringUtils.isBlank(catalogClassName)) { + LOG.error( + "Skip Spark catalog extension {}: provider and catalog class name must not be blank," + + " got provider={}, catalogClassName={}.", + extension.getClass().getName(), + provider, + catalogClassName); + continue; + } + String normalized = provider.toLowerCase(Locale.ROOT); + String previous = byProvider.putIfAbsent(normalized, catalogClassName); + if (previous != null) { + LOG.warn( + "Ignore Spark catalog extension {} for provider {}: already served by {}.", + catalogClassName, + normalized, + previous); + } + } catch (ServiceConfigurationError | LinkageError e) { Review Comment: Only `ServiceConfigurationError` and `LinkageError` are caught here. If an extension's `provider()` or `catalogClassName()` throws a plain `RuntimeException` (e.g. an `IllegalStateException` because some required setting is missing), it escapes `load()`. In that case: - `catalogClassNamesByProvider` stays `null`, so the whole `ServiceLoader` scan is repeated on every lookup and throws again. - `CatalogNameAdaptor.getCatalogName` now throws for **every** provider, including the built-in `hive` / `lakehouse-iceberg` / `jdbc-*` ones. - `GravitinoDriverPlugin.registerGravitinoCatalogs` catches it per catalog and only logs "Register catalog ... failed", so none of the metalake's catalogs get registered in Spark. One broken third-party jar would take down all catalogs, which is the opposite of "skip extensions that fail and keep the rest". Could we also catch `RuntimeException` around the `provider()` / `catalogClassName()` calls (and add a test fixture whose `provider()` throws one)? ########## spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/plugin/SparkCatalogExtension.java: ########## @@ -0,0 +1,44 @@ +/* + * 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.plugin; + +/** + * Service provider interface that lets a jar outside the connector supply the Spark catalog + * implementation for a Gravitino catalog provider. Implementations are discovered through {@link + * java.util.ServiceLoader} and take precedence over the catalogs the connector ships itself. When Review Comment: This says extensions take precedence over the built-in catalogs, but that isn't true for `lakehouse-iceberg` and `lakehouse-paimon`. `GravitinoDriverPlugin.registerGravitinoCatalogs` returns early for these providers when `enableIcebergSupport` / `enablePaimonSupport` is false, before `CatalogNameAdaptor.getCatalogName` (and therefore the extension lookup) is ever called. So an extension registered for `lakehouse-iceberg` is silently ignored with the default config. The only workaround is to turn on Iceberg support, which also forces `IcebergSparkSessionExtensions` / `GravitinoIcebergSparkSessionExtensions` into `spark.sql.extensions`. That fails session creation if the Iceberg Spark runtime isn't on the class path. Should the extension lookup happen before the support-flag check, or should the Javadoc call out that these two providers can't be overridden through this SPI? -- 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]
