github-actions[bot] commented on code in PR #67046:
URL: https://github.com/apache/doris/pull/67046#discussion_r3860373764
##########
fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/classloader/ScannerLoader.java:
##########
@@ -126,59 +128,82 @@ public void loadAllScannerJars() {
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>
*
* <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;
}
// Lost the race against a concurrent first-time load. The cache we
just built has
// 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();
- } catch (Exception e) {
- LOG.warn("Failed to close redundant UdfClassCache for " +
functionSignature, e);
+ closeUdfClassLoader(functionId, newEntry);
+ return existing.classCache;
+ }
+
+ public void cleanUdfClassLoader(String functionSignature, long functionId)
{
+ boolean dropByFunctionId = functionId > 0;
+ LOG.info("cleanUdfClassLoader for function signature: {}, function id:
{}, drop by function id: {}",
+ functionSignature, functionId, dropByFunctionId);
+ if (dropByFunctionId) {
+ closeUdfClassLoader(functionId,
udfLoadedClasses.remove(functionId));
+ return;
}
- return existing;
+ udfLoadedClasses.forEach((cachedFunctionId, entry) -> {
+ if (entry.functionSignature.equals(functionSignature)
Review Comment:
[P1] Do not map a legacy signature to unrelated ID entries
This fallback key is not a safe generation identity. Older FEs send only an
unqualified signature, so on a new BE a DROP of `db1.f(INT)` matches and closes
both db1/ID10 and db2/ID20; the comment below notes that this can fail an
in-flight db2 query. It also cannot clean a variadic static UDF: execution
stores `f(INT...)` via `Function.signatureString()`, while
`DropFunctionCommand.getSignatureString()` sends `f(INT)`, so exact equality
leaks it indefinitely. These paths predate `function_id` and are the
compatibility case this branch claims to support. Please use a
qualified/canonical compatibility identity or avoid destructive cleanup when
the legacy request is ambiguous, and test both same-signature namespaces and
varargs.
##########
fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/classloader/ScannerLoader.java:
##########
@@ -126,59 +128,82 @@ public void loadAllScannerJars() {
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>
*
* <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);
Review Comment:
[P2] Prevent publishing this generation after DROP cleanup
An already-planned query can miss ID N here and spend time
loading/reflection before this insertion. If the exact-ID DROP task runs in
that window, `cleanUdfClassLoader` removes nothing and returns; this later
`putIfAbsent` then republishes N. Static executors do not close the cache and
time eviction is disabled, so the removed generation keeps its URLClassLoader
for the BE lifetime. This is distinct from the existing FE lookup/drop race
because the task carries the correct ID. Please add a per-ID
in-progress/tombstone protocol (or equivalent) so cleanup can make a later
builder close rather than publish, with a latch-based test for this ordering.
##########
be/src/agent/task_worker_pool.cpp:
##########
@@ -2582,17 +2582,23 @@ 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;
+ const bool drop_by_function_id = clean_req.__isset.function_id &&
clean_req.function_id > 0;
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) {
+ 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);
Review Comment:
[P2] Fence native and Python publication after this cleanup
This exact-ID erase is still racy with an already-planned fragment in cloud
mode. The callback can find no entry and return, after which
`UserFunctionCache::_get_cache_entry()` inserts N or the Python server creates
N's UDAF manager; neither path checks a dropped-ID tombstone, and no later DROP
remains to remove that state. This is the native/Python parallel of the Java
publication race and still occurs with the correct ID. Please coordinate
cleanup with later cache creation for every ID-keyed subsystem and add a
deterministic pause-clean-resume test; the current test only checks worker
registration.
##########
be/src/agent/task_worker_pool.cpp:
##########
@@ -2582,17 +2582,23 @@ 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;
+ const bool drop_by_function_id = clean_req.__isset.function_id &&
clean_req.function_id > 0;
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) {
+ 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);
Review Comment:
[P2] Fix the Python module-cache key used by this cleanup
Even when this exact-ID path finds the `UserFunctionCache` entry, the Python
action cannot remove the loaded module. `ModuleUDFLoader` stores
`_module_cache` under the full location string, but
`_clear_modules_from_location` filters with `key[0] == location` and then
unpacks `key` as `(location, module_name)`. For a path string `key[0]` is only
its first character, so the list is empty and the action reports success while
retaining the module in every worker. Please make the cache/cleanup key
representation consistent and add a live Python-process test that loads then
clears a location.
##########
be/src/agent/agent_server.cpp:
##########
@@ -251,6 +251,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); });
+
Review Comment:
[P2] Preserve legacy cleanup for the non-Java caches
Registering this cloud worker exposes the callback to materially old FEs
that legitimately omit `function_id`. In that case the changed callback falls
back by signature only for `ScannerLoader`, while its positive-ID gate skips
`UserFunctionCache` and both Python cleanup actions even though execution
cached their positive `TFunction.id`. Repeated create/execute/drop cycles
during rolling operation therefore retain native/JAR/ZIP artifacts and Python
state. Please define a safe no-ID compatibility identity or bounded reclamation
path for every cache, and test a signature-only request after populating the
native and Python caches.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
