This is an automated email from the ASF dual-hosted git repository. wenchen pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push: new f62ffde0457 [SPARK-42928][SQL] Make resolvePersistentFunction synchronized f62ffde0457 is described below commit f62ffde045771b3275acf3dfb24573804e7daf93 Author: allisonwang-db <allison.w...@databricks.com> AuthorDate: Tue Mar 28 16:43:02 2023 +0800 [SPARK-42928][SQL] Make resolvePersistentFunction synchronized ### What changes were proposed in this pull request? This PR makes the function `resolvePersistentFunctionInternal` synchronized. ### Why are the changes needed? To make function resolution thread-safe. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing UTs. Closes #40557 from allisonwang-db/SPARK-42928-sync-func. Authored-by: allisonwang-db <allison.w...@databricks.com> Signed-off-by: Wenchen Fan <wenc...@databricks.com> --- .../sql/catalyst/catalog/SessionCatalog.scala | 56 ++++++++++++---------- 1 file changed, 31 insertions(+), 25 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala index 4f7196e2dc9..cd4b4cfaf6b 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala @@ -1720,31 +1720,37 @@ class SessionCatalog( arguments: Seq[Expression], registry: FunctionRegistryBase[T], createFunctionBuilder: CatalogFunction => FunctionRegistryBase[T]#FunctionBuilder): T = { - val qualifiedIdent = qualifyIdentifier(name) - val db = qualifiedIdent.database.get - val funcName = qualifiedIdent.funcName - if (registry.functionExists(qualifiedIdent)) { - // This function has been already loaded into the function registry. - registry.lookupFunction(qualifiedIdent, arguments) - } else { - // The function has not been loaded to the function registry, which means - // that the function is a persistent function (if it actually has been registered - // in the metastore). We need to first put the function in the function registry. - val catalogFunction = externalCatalog.getFunction(db, funcName) - loadFunctionResources(catalogFunction.resources) - // Please note that qualifiedName is provided by the user. However, - // catalogFunction.identifier.unquotedString is returned by the underlying - // catalog. So, it is possible that qualifiedName is not exactly the same as - // catalogFunction.identifier.unquotedString (difference is on case-sensitivity). - // At here, we preserve the input from the user. - val funcMetadata = catalogFunction.copy(identifier = qualifiedIdent) - registerFunction( - funcMetadata, - overrideIfExists = false, - registry = registry, - functionBuilder = createFunctionBuilder(funcMetadata)) - // Now, we need to create the Expression. - registry.lookupFunction(qualifiedIdent, arguments) + // `synchronized` is used to prevent multiple threads from concurrently resolving the + // same function that has not yet been loaded into the function registry. This is needed + // because calling `registerFunction` twice with `overrideIfExists = false` can lead to + // a FunctionAlreadyExistsException. + synchronized { + val qualifiedIdent = qualifyIdentifier(name) + val db = qualifiedIdent.database.get + val funcName = qualifiedIdent.funcName + if (registry.functionExists(qualifiedIdent)) { + // This function has been already loaded into the function registry. + registry.lookupFunction(qualifiedIdent, arguments) + } else { + // The function has not been loaded to the function registry, which means + // that the function is a persistent function (if it actually has been registered + // in the metastore). We need to first put the function in the function registry. + val catalogFunction = externalCatalog.getFunction(db, funcName) + loadFunctionResources(catalogFunction.resources) + // Please note that qualifiedName is provided by the user. However, + // catalogFunction.identifier.unquotedString is returned by the underlying + // catalog. So, it is possible that qualifiedName is not exactly the same as + // catalogFunction.identifier.unquotedString (difference is on case-sensitivity). + // At here, we preserve the input from the user. + val funcMetadata = catalogFunction.copy(identifier = qualifiedIdent) + registerFunction( + funcMetadata, + overrideIfExists = false, + registry = registry, + functionBuilder = createFunctionBuilder(funcMetadata)) + // Now, we need to create the Expression. + registry.lookupFunction(qualifiedIdent, arguments) + } } } --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org