This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 827e388c0a4 branch-4.1: [Fix](udf) Key UDF class cache by function ID
and enable cleanup in cloud mode (#67046) (#67314)
827e388c0a4 is described below
commit 827e388c0a4a128bae4b1aad5049ba11667f0291
Author: linrrarity <[email protected]>
AuthorDate: Mon Aug 31 15:28:48 2026 +0800
branch-4.1: [Fix](udf) Key UDF class cache by function ID and enable
cleanup in cloud mode (#67046) (#67314)
pick: #67046
---
be/src/agent/agent_server.cpp | 3 +
be/src/agent/task_worker_pool.cpp | 23 ++++--
be/src/util/jni-util.cpp | 6 +-
be/src/util/jni-util.h | 3 +-
be/test/agent/task_worker_pool_test.cpp | 18 +++++
.../doris/common/classloader/ScannerLoader.java | 90 +++++++++++++++-------
.../common/classloader/ScannerLoaderTest.java | 77 ++++++++++++++++++
.../java/org/apache/doris/udf/BaseExecutor.java | 13 ++--
.../java/org/apache/doris/catalog/Database.java | 16 +++-
.../apache/doris/catalog/GlobalFunctionMgr.java | 10 ++-
.../trees/plans/commands/DropFunctionCommand.java | 52 ++++---------
.../apache/doris/catalog/CreateFunctionTest.java | 45 +++++++++++
.../org/apache/doris/catalog/DropFunctionTest.java | 14 ++++
13 files changed, 290 insertions(+), 80 deletions(-)
diff --git a/be/src/agent/agent_server.cpp b/be/src/agent/agent_server.cpp
index c1fb065ac6c..13a707a8adb 100644
--- a/be/src/agent/agent_server.cpp
+++ b/be/src/agent/agent_server.cpp
@@ -250,6 +250,9 @@ void AgentServer::cloud_start_workers(CloudStorageEngine&
engine, ExecEnv* exec_
return make_cloud_committed_rs_visible_callback(engine, task);
});
+ _workers[TTaskType::CLEAN_UDF_CACHE] = std::make_unique<TaskWorkerPool>(
+ "CLEAN_UDF_CACHE", 1, [](auto&& task) { return
clean_udf_cache_callback(task); });
+
_report_workers.push_back(std::make_unique<ReportWorker>(
"REPORT_TASK", _cluster_info, config::report_task_interval_seconds,
[&cluster_info = _cluster_info] {
report_task_callback(cluster_info); }));
diff --git a/be/src/agent/task_worker_pool.cpp
b/be/src/agent/task_worker_pool.cpp
index 2605112673c..6ee1554a874 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -2577,17 +2577,30 @@ void clean_trash_callback(StorageEngine& engine, const
TAgentTaskRequest& req) {
void clean_udf_cache_callback(const TAgentTaskRequest& req) {
const auto& clean_req = req.clean_udf_cache_req;
-
- if (doris::config::enable_java_support) {
-
static_cast<void>(Jni::Util::clean_udf_class_load_cache(clean_req.function_signature));
+ if (clean_req.__isset.function_id && clean_req.function_id <= 0) {
+ LOG(WARNING) << "skip clean udf cache request with invalid
function_id="
+ << clean_req.function_id
+ << ", function_signature=" <<
clean_req.function_signature;
+ return;
}
+ // Requests from old FEs do not set function_id and must keep
signature-based cleanup.
+ const bool drop_by_function_id = clean_req.__isset.function_id;
- if (clean_req.__isset.function_id && clean_req.function_id > 0) {
+ if (doris::config::enable_java_support) {
+ WARN_IF_ERROR(
+ Jni::Util::clean_udf_class_load_cache(
+ clean_req.function_signature,
+ drop_by_function_id ? clean_req.function_id : 0),
+ fmt::format("failed to clean Java UDF cache,
function_signature={}, function_id={}",
+ clean_req.function_signature,
clean_req.function_id));
+ }
+ if (drop_by_function_id) {
UserFunctionCache::instance()->drop_function_cache(clean_req.function_id);
PythonServerManager::instance().clear_udaf_state_cache(clean_req.function_id);
}
- LOG(INFO) << "clean udf cache finish: function_signature=" <<
clean_req.function_signature;
+ LOG(INFO) << "clean udf cache callback finish: function_signature="
+ << clean_req.function_signature << ", function_id=" <<
clean_req.function_id;
}
void report_index_policy_callback(const ClusterInfo* cluster_info) {
diff --git a/be/src/util/jni-util.cpp b/be/src/util/jni-util.cpp
index 9f5f3999c17..9854d3a49ee 100644
--- a/be/src/util/jni-util.cpp
+++ b/be/src/util/jni-util.cpp
@@ -316,7 +316,7 @@ Status Util::_init_jni_scanner_loader() {
jni_scanner_loader_cls.get_method(env, "loadAllScannerJars",
"()V", &load_jni_scanner));
RETURN_IF_ERROR(jni_scanner_loader_cls.get_method(
- env, "cleanUdfClassLoader", "(Ljava/lang/String;)V",
&_clean_udf_cache_method_id));
+ env, "cleanUdfClassLoader", "(Ljava/lang/String;J)V",
&_clean_udf_cache_method_id));
RETURN_IF_ERROR(jni_scanner_loader_cls.new_object(env,
jni_scanner_loader_constructor)
.call(&jni_scanner_loader_obj_));
@@ -325,7 +325,8 @@ Status Util::_init_jni_scanner_loader() {
return Status::OK();
}
-Status Util::clean_udf_class_load_cache(const std::string& function_signature)
{
+Status Util::clean_udf_class_load_cache(const std::string& function_signature,
+ int64_t function_id) {
JNIEnv* env = nullptr;
RETURN_IF_ERROR(Jni::Env::Get(&env));
@@ -335,6 +336,7 @@ Status Util::clean_udf_class_load_cache(const std::string&
function_signature) {
RETURN_IF_ERROR(jni_scanner_loader_obj_.call_void_method(env,
_clean_udf_cache_method_id)
.with_arg(function_signature_jstr)
+ .with_arg((jlong)function_id)
.call());
return Status::OK();
diff --git a/be/src/util/jni-util.h b/be/src/util/jni-util.h
index 956e4c4df7d..03be54d9395 100644
--- a/be/src/util/jni-util.h
+++ b/be/src/util/jni-util.h
@@ -1156,7 +1156,8 @@ public:
return Status::OK();
}
- static Status clean_udf_class_load_cache(const std::string&
function_signature);
+ static Status clean_udf_class_load_cache(const std::string&
function_signature,
+ int64_t function_id);
static Status Init();
diff --git a/be/test/agent/task_worker_pool_test.cpp
b/be/test/agent/task_worker_pool_test.cpp
index 9cd7ddd640d..b8b2de0fc4b 100644
--- a/be/test/agent/task_worker_pool_test.cpp
+++ b/be/test/agent/task_worker_pool_test.cpp
@@ -25,7 +25,10 @@
#include <chrono>
#include <thread>
+#include "agent/agent_server.h"
+#include "cloud/cloud_storage_engine.h"
#include "runtime/cluster_info.h"
+#include "runtime/exec_env.h"
#include "storage/options.h"
#include "storage/storage_engine.h"
@@ -181,4 +184,19 @@ TEST(TaskWorkerPoolTest, ReportWorkerPool) {
EXPECT_EQ(count.load(), 3);
}
+TEST(AgentServerTest, CloudRegistersCleanUdfCacheWorker) {
+ auto* exec_env = ExecEnv::GetInstance();
+ auto engine = std::make_unique<CloudStorageEngine>(EngineOptions {});
+ auto* cloud_engine = engine.get();
+ exec_env->set_storage_engine(std::move(engine));
+ Defer defer {[exec_env] { exec_env->set_storage_engine(nullptr); }};
+
+ ClusterInfo cluster_info;
+ AgentServer agent_server(exec_env, &cluster_info);
+
+ agent_server.cloud_start_workers(*cloud_engine, exec_env);
+
+ EXPECT_TRUE(agent_server._workers.contains(TTaskType::CLEAN_UDF_CACHE));
+}
+
} // namespace doris
diff --git
a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/classloader/ScannerLoader.java
b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/classloader/ScannerLoader.java
index f8a119efaa9..35f2eb856c3 100644
---
a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/classloader/ScannerLoader.java
+++
b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/classloader/ScannerLoader.java
@@ -97,12 +97,14 @@ public class ScannerLoader {
// 2) rebuilding a fresh URLClassLoader on every eviction produced
multiple coexisting
// ClassLoaders for the same UDF, which broke lazy class resolution
and reflective
// lookups inside user UDF code.
+ // Cache by function id so a recreated function with the same signature
does not reuse
+ // the previous function's class loader.
// NOTE: a cache miss in BaseExecutor.getClassCache() is NOT only
reachable after
- // cleanUdfClassLoader() — concurrent first-time loads of the same
signature can also
+ // cleanUdfClassLoader() — concurrent first-time loads of the same
function can also
// both observe a miss. cacheClassLoader() must therefore insert
atomically via
// putIfAbsent and must never close a cache that was already published to
the map,
// because another executor may already be holding it.
- private static final Map<String, UdfClassCache> udfLoadedClasses = new
ConcurrentHashMap<>();
+ private static final Map<Long, UdfClassCacheEntry> udfLoadedClasses = new
ConcurrentHashMap<>();
private static final String CLASS_SUFFIX = ".class";
private static final String LOAD_PACKAGE = "org.apache.doris";
@@ -126,15 +128,26 @@ public class ScannerLoader {
LOG.info("Finished loading scanner JARs");
}
- public static UdfClassCache getUdfClassLoader(String functionSignature) {
- return udfLoadedClasses.get(functionSignature);
+ private static class UdfClassCacheEntry {
+ private final String functionSignature;
+ private final UdfClassCache classCache;
+
+ UdfClassCacheEntry(String functionSignature, UdfClassCache classCache)
{
+ this.functionSignature = functionSignature;
+ this.classCache = classCache;
+ }
+ }
+
+ public static UdfClassCache getUdfClassLoader(long functionId) {
+ UdfClassCacheEntry entry = udfLoadedClasses.get(functionId);
+ return entry == null ? null : entry.classCache;
}
/**
- * Cache the UDF class metadata for the given function signature.
+ * Cache the UDF class metadata for the given catalog function id.
*
- * <p>Insertion is atomic via {@link Map#putIfAbsent}: if another executor
thread has
- * already published a cache entry for {@code functionSignature}, the
{@code classCache}
+ * <p>Insertion is atomic via {@link Map#putIfAbsent}: if another executor
+ * thread has already published a cache entry for {@code functionId}, the
{@code classCache}
* argument is treated as a redundant build and closed here (it has not
yet been handed
* to any executor, so closing its URLClassLoader is safe). The
already-published entry
* is returned to the caller so the current executor can switch to it.</p>
@@ -142,16 +155,17 @@ public class ScannerLoader {
* <p>The {@code expirationTime} parameter is kept for backward
compatibility with the
* existing call sites and DDL property {@code expiration_time}, but is no
longer used:
* cached entries are not evicted by time. Removal happens only via
- * {@link #cleanUdfClassLoader(String)} on DROP FUNCTION.</p>
+ * {@link #cleanUdfClassLoader(String, long)} on DROP FUNCTION.</p>
*
* @return the {@link UdfClassCache} actually held in the map after this
call —
* either {@code classCache} (we won the race) or the pre-existing
entry
* (another thread won; {@code classCache} has been closed and
must not be used).
*/
- public static UdfClassCache cacheClassLoader(String functionSignature,
UdfClassCache classCache,
- long expirationTime) {
- LOG.info("Cache UDF for: " + functionSignature);
- UdfClassCache existing =
udfLoadedClasses.putIfAbsent(functionSignature, classCache);
+ public static UdfClassCache cacheClassLoader(String functionSignature,
long functionId,
+ UdfClassCache classCache, long expirationTime) {
+ LOG.info("Cache UDF for function signature: {}, function id: {}",
functionSignature, functionId);
+ UdfClassCacheEntry newEntry = new
UdfClassCacheEntry(functionSignature, classCache);
+ UdfClassCacheEntry existing = udfLoadedClasses.putIfAbsent(functionId,
newEntry);
if (existing == null) {
return classCache;
}
@@ -159,28 +173,48 @@ public class ScannerLoader {
// never been exposed to any executor, so closing its URLClassLoader
here cannot
// affect anyone. Do NOT touch `existing` — another executor may
already be using it.
try {
- classCache.close();
+ newEntry.classCache.close();
} catch (Exception e) {
- LOG.warn("Failed to close redundant UdfClassCache for " +
functionSignature, e);
+ LOG.warn("Failed to close UdfClassCache for function signature:
{}, function id: {}",
+ newEntry.functionSignature, functionId, e);
}
- return existing;
+ return existing.classCache;
}
- public void cleanUdfClassLoader(String functionSignature) {
- LOG.info("cleanUdfClassLoader for: " + functionSignature);
- UdfClassCache removed = udfLoadedClasses.remove(functionSignature);
- if (removed != null) {
- // Immediately close the URLClassLoader. NOTE: any in-flight query
still holding a
- // reference to this cache (e.g. via JNIContext.executor) will
fail with
- // NoClassDefFoundError on lazy class resolution after this point.
This is the
- // accepted semantic of DROP FUNCTION: the function is gone,
queries against it
- // are expected to fail.
- try {
- removed.close();
- } catch (Exception e) {
- LOG.warn("Failed to close UdfClassCache for " +
functionSignature, e);
+ public void cleanUdfClassLoader(String functionSignature, long functionId)
{
+ LOG.info("cleanUdfClassLoader for function signature: {}, function id:
{}",
+ functionSignature, functionId);
+ if (functionId > 0) {
+ UdfClassCacheEntry removed = udfLoadedClasses.remove(functionId);
+ if (removed != null) {
+ // Immediately close the URLClassLoader. NOTE: any in-flight
query still holding a
+ // reference to this cache (e.g. via JNIContext.executor) will
fail with
+ // NoClassDefFoundError on lazy class resolution after this
point. This is the
+ // accepted semantic of DROP FUNCTION: the function is gone,
queries against it
+ // are expected to fail.
+ try {
+ removed.classCache.close();
+ } catch (Exception e) {
+ LOG.warn("Failed to close UdfClassCache for function
signature: {}, function id: {}",
+ removed.functionSignature, functionId, e);
+ }
}
+ return;
}
+
+ // Old FEs do not set function_id in cleanup requests, so remove every
cache with
+ // the requested signature.
+ udfLoadedClasses.forEach((cachedFunctionId, entry) -> {
+ if (entry.functionSignature.equals(functionSignature)
+ && udfLoadedClasses.remove(cachedFunctionId, entry)) {
+ try {
+ entry.classCache.close();
+ } catch (Exception e) {
+ LOG.warn("Failed to close UdfClassCache for function
signature: {}, function id: {}",
+ entry.functionSignature, cachedFunctionId, e);
+ }
+ }
+ });
}
/**
diff --git
a/fe/be-java-extensions/java-common/src/test/java/org/apache/doris/common/classloader/ScannerLoaderTest.java
b/fe/be-java-extensions/java-common/src/test/java/org/apache/doris/common/classloader/ScannerLoaderTest.java
new file mode 100644
index 00000000000..6d630d7ce42
--- /dev/null
+++
b/fe/be-java-extensions/java-common/src/test/java/org/apache/doris/common/classloader/ScannerLoaderTest.java
@@ -0,0 +1,77 @@
+// 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.doris.common.classloader;
+
+import org.apache.doris.common.jni.utils.UdfClassCache;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ScannerLoaderTest {
+ @Test
+ public void testCleanCacheByFunctionId() {
+ long oldFunctionId = 10001;
+ long newFunctionId = 10002;
+ String functionSignature = "recreated_function(INT)";
+ UdfClassCache oldCache = new UdfClassCache();
+ UdfClassCache newCache = new UdfClassCache();
+ ScannerLoader loader = new ScannerLoader();
+
+ try {
+ ScannerLoader.cacheClassLoader(functionSignature, oldFunctionId,
oldCache, 0);
+ ScannerLoader.cacheClassLoader(functionSignature, newFunctionId,
newCache, 0);
+
+ loader.cleanUdfClassLoader(functionSignature, oldFunctionId);
+
+ Assert.assertNull(ScannerLoader.getUdfClassLoader(oldFunctionId));
+ Assert.assertSame(newCache,
ScannerLoader.getUdfClassLoader(newFunctionId));
+ } finally {
+ loader.cleanUdfClassLoader(functionSignature, oldFunctionId);
+ loader.cleanUdfClassLoader(functionSignature, newFunctionId);
+ }
+ }
+
+ @Test
+ public void testCleanAllCachesByFunctionSignatureWithoutFunctionId() {
+ long firstFunctionId = 10003;
+ long secondFunctionId = 10004;
+ long otherFunctionId = 10005;
+ String functionSignature = "legacy_function(INT)";
+ String otherFunctionSignature = "other_function(INT)";
+ UdfClassCache firstCache = new UdfClassCache();
+ UdfClassCache secondCache = new UdfClassCache();
+ UdfClassCache otherCache = new UdfClassCache();
+ ScannerLoader loader = new ScannerLoader();
+
+ try {
+ ScannerLoader.cacheClassLoader(functionSignature, firstFunctionId,
firstCache, 0);
+ ScannerLoader.cacheClassLoader(functionSignature,
secondFunctionId, secondCache, 0);
+ ScannerLoader.cacheClassLoader(otherFunctionSignature,
otherFunctionId, otherCache, 0);
+
+ loader.cleanUdfClassLoader(functionSignature, 0);
+
+
Assert.assertNull(ScannerLoader.getUdfClassLoader(firstFunctionId));
+
Assert.assertNull(ScannerLoader.getUdfClassLoader(secondFunctionId));
+ Assert.assertSame(otherCache,
ScannerLoader.getUdfClassLoader(otherFunctionId));
+ } finally {
+ loader.cleanUdfClassLoader(functionSignature, firstFunctionId);
+ loader.cleanUdfClassLoader(functionSignature, secondFunctionId);
+ loader.cleanUdfClassLoader(otherFunctionSignature,
otherFunctionId);
+ }
+ }
+}
diff --git
a/fe/be-java-extensions/java-udf/src/main/java/org/apache/doris/udf/BaseExecutor.java
b/fe/be-java-extensions/java-udf/src/main/java/org/apache/doris/udf/BaseExecutor.java
index 6356576baaf..87756adc289 100644
---
a/fe/be-java-extensions/java-udf/src/main/java/org/apache/doris/udf/BaseExecutor.java
+++
b/fe/be-java-extensions/java-udf/src/main/java/org/apache/doris/udf/BaseExecutor.java
@@ -109,8 +109,8 @@ public abstract class BaseExecutor {
if (request.getFn().isSetExpirationTime()) {
expirationTime = request.getFn().getExpirationTime();
}
- objCache = getClassCache(jarPath, request.getFn().getSignature(),
expirationTime,
- funcRetType, parameterTypes);
+ objCache = getClassCache(jarPath, request.getFn().getSignature(),
request.getFn().getId(),
+ expirationTime, funcRetType, parameterTypes);
Constructor<?> ctor = objCache.udfClass.getConstructor();
udf = ctor.newInstance();
} catch (MalformedURLException e) {
@@ -131,13 +131,13 @@ public abstract class BaseExecutor {
}
- public UdfClassCache getClassCache(String jarPath, String signature, long
expirationTime,
- Type funcRetType, Type... parameterTypes)
+ public UdfClassCache getClassCache(String jarPath, String
functionSignature, long functionId,
+ long expirationTime, Type funcRetType, Type... parameterTypes)
throws MalformedURLException, FileNotFoundException,
ClassNotFoundException, InternalException,
UdfRuntimeException {
UdfClassCache cache = null;
if (isStaticLoad) {
- cache = ScannerLoader.getUdfClassLoader(signature);
+ cache = ScannerLoader.getUdfClassLoader(functionId);
if (cache != null) {
// Reuse the cached classLoader to ensure dependent classes
can be loaded.
// NOTE: cache.classLoader may be null when the UDF was
originally loaded via
@@ -166,7 +166,8 @@ public abstract class BaseExecutor {
cache.classLoader = classLoader;
checkAndCacheUdfClass(cache, funcRetType, parameterTypes);
if (isStaticLoad) {
- UdfClassCache effective =
ScannerLoader.cacheClassLoader(signature, cache, expirationTime);
+ UdfClassCache effective = ScannerLoader.cacheClassLoader(
+ functionSignature, functionId, cache, expirationTime);
if (effective != cache) {
// Another thread won the publish race. Our locally-built
cache (and its
// URLClassLoader) was already closed inside
cacheClassLoader(); switch to
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Database.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/Database.java
index 3583e0dc807..5d3a12ed84e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Database.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Database.java
@@ -837,7 +837,7 @@ public class Database extends MetaObject implements
Writable, DatabaseIf<Table>,
}
}
- public synchronized void dropFunction(FunctionSearchDesc function, boolean
ifExists) throws UserException {
+ public synchronized List<Long> dropFunction(FunctionSearchDesc function,
boolean ifExists) throws UserException {
Function udfFunction = null;
try {
// here we must first getFunction, as dropFunctionImpl will remove
it
@@ -847,11 +847,13 @@ public class Database extends MetaObject implements
Writable, DatabaseIf<Table>,
throw new UserException(e);
} else {
// ignore it, as drop it if exist, so can't sure it must exist
- return;
+ return ImmutableList.of();
}
}
+ List<Long> droppedFunctionIds = Lists.newArrayList();
dropFunctionImpl(function, ifExists);
+ droppedFunctionIds.add(udfFunction.getId());
if (udfFunction != null && udfFunction.isUDTFunction()) {
// all of the table function in doris will have two function
// one is the normal, and another is outer, the different of them
is deal with
@@ -860,8 +862,18 @@ public class Database extends MetaObject implements
Writable, DatabaseIf<Table>,
function.getName().getFunction() + "_outer");
FunctionSearchDesc functionOuter = new FunctionSearchDesc(name,
function.getArgTypes(),
function.isVariadic());
+ Function udfOuterFunction = null;
+ try {
+ udfOuterFunction = getFunction(functionOuter);
+ } catch (AnalysisException e) {
+ // Let dropFunctionImpl preserve the existing IF EXISTS and
error behavior.
+ }
dropFunctionImpl(functionOuter, ifExists);
+ if (udfOuterFunction != null) {
+ droppedFunctionIds.add(udfOuterFunction.getId());
+ }
}
+ return droppedFunctionIds;
}
public synchronized void dropFunctionImpl(FunctionSearchDesc function,
boolean ifExists) throws UserException {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/GlobalFunctionMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/GlobalFunctionMgr.java
index c2f4fb3c1c0..f9e3ef66d0e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/GlobalFunctionMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/GlobalFunctionMgr.java
@@ -96,11 +96,19 @@ public class GlobalFunctionMgr extends MetaObject
implements GsonPostProcessable
}
}
- public synchronized void dropFunction(FunctionSearchDesc function, boolean
ifExists) throws UserException {
+ public synchronized List<Long> dropFunction(FunctionSearchDesc function,
boolean ifExists) throws UserException {
+ Function droppedFunction = null;
+ try {
+ droppedFunction = FunctionUtil.getFunction(function,
name2Function);
+ } catch (AnalysisException e) {
+ // Let dropFunctionImpl preserve the existing IF EXISTS and error
behavior.
+ }
if (FunctionUtil.dropFunctionImpl(function, ifExists, name2Function)) {
Env.getCurrentEnv().getEditLog().logDropGlobalFunction(function);
FunctionUtil.dropFromNereids(null, function);
+ return ImmutableList.of(droppedFunction.getId());
}
+ return ImmutableList.of();
}
public synchronized void replayDropFunction(FunctionSearchDesc
functionSearchDesc) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropFunctionCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropFunctionCommand.java
index 6199b406746..fcd9871cf84 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropFunctionCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropFunctionCommand.java
@@ -22,9 +22,7 @@ import org.apache.doris.analysis.SetType;
import org.apache.doris.analysis.StmtType;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
-import org.apache.doris.catalog.Function;
import org.apache.doris.catalog.FunctionSearchDesc;
-import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
import org.apache.doris.mysql.privilege.PrivPredicate;
@@ -43,6 +41,8 @@ import com.google.common.collect.ImmutableMap;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import java.util.List;
+
/**
* drop a alias or user defined function
*/
@@ -74,35 +74,9 @@ public class DropFunctionCommand extends Command implements
ForwardWithSync {
argsDef.analyze();
FunctionSearchDesc function = new FunctionSearchDesc(functionName,
argsDef.getArgTypes(), argsDef.isVariadic());
- // Get function id before dropping, for cleaning cached library files
in BE
- long functionId = -1;
- try {
- Function fn = null;
- if (SetType.GLOBAL.equals(setType)) {
- fn =
Env.getCurrentEnv().getGlobalFunctionMgr().getFunction(function);
- } else {
- String dbName = functionName.getDb();
- if (dbName == null) {
- dbName = ctx.getDatabase();
- functionName.setDb(dbName);
- }
- Database db =
Env.getCurrentInternalCatalog().getDbNullable(dbName);
- if (db != null) {
- fn = db.getFunction(function);
- }
- }
- if (fn != null) {
- functionId = fn.getId();
- } else {
- LOG.warn("Function not found: {}, setType: {}",
function.getName(), setType);
- }
- } catch (AnalysisException e) {
- LOG.warn("Function not found when getting function id: {}, error:
{}",
- function.getName(), e.getMessage());
- }
-
+ List<Long> functionIds;
if (SetType.GLOBAL.equals(setType)) {
- Env.getCurrentEnv().getGlobalFunctionMgr().dropFunction(function,
ifExists);
+ functionIds =
Env.getCurrentEnv().getGlobalFunctionMgr().dropFunction(function, ifExists);
} else {
String dbName = functionName.getDb();
if (dbName == null) {
@@ -113,17 +87,25 @@ public class DropFunctionCommand extends Command
implements ForwardWithSync {
if (db == null) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_NO_DB_ERROR);
}
- db.dropFunction(function, ifExists);
+ functionIds = db.dropFunction(function, ifExists);
+ }
+ if (functionIds.isEmpty()) {
+ // No function generation was removed. A signature-based cleanup
task could arrive
+ // after a same-signature function is created and delete the new
generation's cache.
+ return;
}
// BE will cache classload, when drop function, BE need clear cache
ImmutableMap<Long, Backend> backendsInfo =
Env.getCurrentSystemInfo().getAllBackendsByAllCluster();
String functionSignature = getSignatureString();
AgentBatchTask batchTask = new AgentBatchTask();
for (Backend backend : backendsInfo.values()) {
- CleanUDFCacheTask cleanUDFCacheTask = new
CleanUDFCacheTask(backend.getId(), functionSignature, functionId);
- batchTask.addTask(cleanUDFCacheTask);
- LOG.info("clean udf cache in be {}, beId {}, functionId {}",
- backend.getHost(), backend.getId(), functionId);
+ for (long functionId : functionIds) {
+ CleanUDFCacheTask cleanUDFCacheTask = new CleanUDFCacheTask(
+ backend.getId(), functionSignature, functionId);
+ batchTask.addTask(cleanUDFCacheTask);
+ LOG.info("clean udf cache in be {}, beId {}, functionId {}",
+ backend.getHost(), backend.getId(), functionId);
+ }
}
AgentTaskExecutor.submit(batchTask);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateFunctionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateFunctionTest.java
index 5b8829473cc..c73951d8ffd 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateFunctionTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateFunctionTest.java
@@ -19,9 +19,11 @@ package org.apache.doris.catalog;
import org.apache.doris.analysis.Expr;
import org.apache.doris.analysis.FunctionCallExpr;
+import org.apache.doris.analysis.FunctionName;
import org.apache.doris.analysis.StringLiteral;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.common.util.URI;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.trees.plans.commands.CreateDatabaseCommand;
@@ -34,9 +36,11 @@ import org.apache.doris.planner.UnionNode;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.QueryState;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.thrift.TFunctionBinaryType;
import org.apache.doris.utframe.DorisAssert;
import org.apache.doris.utframe.UtFrameUtils;
+import com.google.common.collect.ImmutableList;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
@@ -172,6 +176,37 @@ public class CreateFunctionTest {
"ARRAY unsupported sub-type: bitmap");
}
+ @Test
+ public void testDropFunctionReturnsCurrentGenerationId() throws Exception {
+ ConnectContext ctx = UtFrameUtils.createDefaultCtx();
+ createDatabase(ctx, "create database drop_function_id_db;");
+ Database db =
Env.getCurrentInternalCatalog().getDbNullable("drop_function_id_db");
+ Assert.assertNotNull(db);
+
+ Function firstGeneration = createJavaUdf("drop_function_id_db",
"generation_fn", Type.INT);
+ db.addFunction(firstGeneration, false);
+ Assert.assertEquals(ImmutableList.of(firstGeneration.getId()),
+ db.dropFunction(searchDesc(firstGeneration), false));
+
+ Function secondGeneration = createJavaUdf("drop_function_id_db",
"generation_fn", Type.INT);
+ db.addFunction(secondGeneration, false);
+ Assert.assertNotEquals(firstGeneration.getId(),
secondGeneration.getId());
+ Assert.assertEquals(ImmutableList.of(secondGeneration.getId()),
+ db.dropFunction(searchDesc(secondGeneration), false));
+ }
+
+ @Test
+ public void testDropGlobalFunctionReturnsCurrentGenerationId() throws
Exception {
+ GlobalFunctionMgr globalFunctionMgr =
Env.getCurrentEnv().getGlobalFunctionMgr();
+ Function function = createJavaUdf(null, "drop_global_function_id_fn",
Type.INT);
+ FunctionSearchDesc functionDesc = searchDesc(function);
+ globalFunctionMgr.dropFunction(functionDesc, true);
+
+ globalFunctionMgr.addFunction(function, false);
+ Assert.assertEquals(ImmutableList.of(function.getId()),
+ globalFunctionMgr.dropFunction(functionDesc, false));
+ }
+
@Test
public void testCreateGlobalFunction() throws Exception {
ConnectContext ctx = UtFrameUtils.createDefaultCtx();
@@ -275,4 +310,14 @@ public class CreateFunctionTest {
}
throw new AssertionError("function not found: " + functionName);
}
+
+ private Function createJavaUdf(String dbName, String functionName, Type...
argTypes) throws Exception {
+ return ScalarFunction.createUdf(TFunctionBinaryType.JAVA_UDF,
+ new FunctionName(dbName, functionName), argTypes, Type.INT,
false,
+ URI.create("file:///tmp/" + functionName + ".jar"),
"evaluate", null, null);
+ }
+
+ private FunctionSearchDesc searchDesc(Function function) {
+ return new FunctionSearchDesc(function.getFunctionName(),
function.getArgs(), function.hasVarArgs());
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/DropFunctionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/DropFunctionTest.java
index bef0bb92d20..935bf5cbb68 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/DropFunctionTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/DropFunctionTest.java
@@ -26,6 +26,7 @@ import
org.apache.doris.nereids.trees.plans.commands.DropFunctionCommand;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.task.AgentTaskExecutor;
import org.apache.doris.utframe.DorisAssert;
import org.apache.doris.utframe.UtFrameUtils;
@@ -33,6 +34,8 @@ import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
import java.io.File;
import java.util.List;
@@ -86,6 +89,17 @@ public class DropFunctionTest {
Assert.assertEquals(0, functions.size());
}
+ @Test
+ public void testDropIfExistsMissingFunctionDoesNotSubmitCacheCleanup()
throws Exception {
+ ConnectContext ctx = UtFrameUtils.createDefaultCtx();
+ try (MockedStatic<AgentTaskExecutor> mockedAgentTaskExecutor =
+ Mockito.mockStatic(AgentTaskExecutor.class)) {
+ dropFunction("drop global function if exists
missing_function(bigint)", ctx);
+
+ mockedAgentTaskExecutor.verifyNoInteractions();
+ }
+ }
+
private void createFunction(String sql, ConnectContext connectContext)
throws Exception {
NereidsParser nereidsParser = new NereidsParser();
LogicalPlan parsed = nereidsParser.parseSingle(sql);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
