github-actions[bot] commented on code in PR #69255:
URL: https://github.com/apache/doris/pull/69255#discussion_r4310257763
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionUtil.java:
##########
@@ -48,6 +51,37 @@
public class FunctionUtil {
private static final Logger LOG = LogManager.getLogger(FunctionUtil.class);
+ public static void cleanUdfCache(Map<Long, String> functionSignatures) {
+ if (functionSignatures.isEmpty()) {
+ // No function generation was removed. Never fall back to
signature-based cleanup,
+ // which could delete a newly created function's cache when an old
task arrives late.
+ return;
+ }
+ AgentBatchTask batchTask = new AgentBatchTask();
+ for (long backendId :
Env.getCurrentSystemInfo().getAllBackendIds(false)) {
+ for (Map.Entry<Long, String> function :
functionSignatures.entrySet()) {
+ CleanUDFCacheTask task = new CleanUDFCacheTask(backendId,
function.getValue(), function.getKey());
+ batchTask.addTask(task);
+ LOG.info("clean udf cache in be {}, functionId {}", backendId,
function.getKey());
+ }
+ }
+ if (batchTask.getTaskNum() > 0) {
+ AgentTaskExecutor.submit(batchTask);
Review Comment:
[P2] Retry database cleanup when a backend becomes available
If a BE temporarily loses its heartbeat while its Python processes remain
alive when a recycled database is finally erased, this cleanup is lost
permanently: `AgentBatchTask.run()` skips `!backend.isAlive()` and only marks
the tasks failed. These tasks are never registered in `AgentTaskQueue`, so
heartbeat recovery cannot resend them. The database metadata has already been
removed, leaving no later recycler pass to retry. With the default
`auto_suspend=-1`, every missed function generation retains its dedicated
Python pool and package handles indefinitely; repeated drops can accumulate
processes, unlike the former reusable runtime-wide pool. Retain and retry these
ID-scoped cleanup notifications after backend recovery.
##########
be/src/udf/python/python_server.cpp:
##########
@@ -38,104 +38,167 @@
#include "common/status.h"
#include "cpp/sync_point.h"
#include "runtime/thread_context.h"
+#include "runtime/user_function_cache.h"
#include "udf/python/python_udaf_client.h"
#include "udf/python/python_udf_client.h"
#include "udf/python/python_udtf_client.h"
-#include "util/cpu_info.h"
+#include "util/defer_op.h"
namespace doris {
-Result<std::shared_ptr<PythonServerManager::VersionedProcessPool>>
-PythonServerManager::_get_or_create_process_pool(const PythonVersion& version)
{
- std::lock_guard<std::mutex> lock(_pools_mutex);
+// A client holds this from pool acquisition through stream/state cleanup. The
pool may outlive
+// its manager after DROP; releasing a lease must therefore only access the
pool itself.
+class PythonUDFPoolLease {
+public:
+ explicit
PythonUDFPoolLease(std::shared_ptr<PythonServerManager::FunctionProcessPool>
pool)
+ : _pool(std::move(pool)) {}
+
+ ~PythonUDFPoolLease() {
+ std::scoped_lock lock(_pool->mutex);
+ DCHECK_GT(_pool->active_clients, 0);
+ if (--_pool->active_clients == 0) {
+ _pool->idle_since = std::chrono::steady_clock::now();
+ }
+ }
+
+private:
+ std::shared_ptr<PythonServerManager::FunctionProcessPool> _pool;
+};
+
+void PythonServerManager::ProcessRegistry::add(const ProcessPtr& process) {
+ std::scoped_lock lock(mutex);
+ processes.emplace_back(process);
+}
+
+std::vector<ProcessPtr> PythonServerManager::ProcessRegistry::snapshot() {
+ std::vector<ProcessPtr> snapshot;
+ std::scoped_lock lock(mutex);
+ std::erase_if(processes, [&snapshot](const auto& weak_process) {
+ if (auto process = weak_process.lock()) {
+ snapshot.emplace_back(std::move(process));
+ return false;
+ }
+ return true;
+ });
+ return snapshot;
+}
+
+Result<std::shared_ptr<PythonServerManager::FunctionProcessPool>>
+PythonServerManager::_get_or_create_process_pool(
+ const PythonUDFMeta& meta, const PythonVersion& version,
+ std::shared_ptr<UserFunctionCacheEntry> package_cache_entry,
+ std::shared_ptr<PythonUDFPoolLease>* lease) {
+ std::scoped_lock lock(_pools_mutex);
// shutdown() owns the manager lifecycle. Once it starts, creating a new
pool would let detached
// init workers publish Python processes that the manager no longer tracks.
if (_shutdown_flag.load(std::memory_order_acquire)) {
return ResultError(Status::Error<ErrorCode::SERVICE_UNAVAILABLE>(
"Python server manager is shutting down"));
}
- auto& pool = _process_pools[version];
+ if (meta.pool_size <= 0) {
+ return ResultError(
+ Status::InvalidArgument("Python UDF pool_size must be a
positive integer"));
+ }
+ auto& pool = _process_pools[meta.id];
Review Comment:
[P2] Prevent late queries from recreating dropped generations
`drop_function()` erases the pool without remembering the retired ID, so
work already dispatched with the old `TFunction` can enter this branch after
cleanup. For example, a Python UDAF can finish `open()` (pinning its package)
before DROP and call `create()` afterward; this recreates the removed
generation's pool, and the default `auto_suspend=-1` keeps its processes and
package alive permanently. A late scalar/UDTF open can also recreate a cache
entry for the same ID/path while the earlier entry is retained by an old
process; releasing that old entry then deletes files used by the new entry.
Preserve DROP retirement across cache and pool acquisition, while allowing
idle-suspended pools to reopen, so late query work cannot create durable new
owners for a dropped generation.
##########
be/src/udf/python/python_server.py:
##########
@@ -1113,39 +1024,7 @@ def _get_or_import_module(self, location: str,
full_module_name: str) -> Any:
f"server-critical modules."
)
- cache_key = location
-
- # Use a per-module lock to prevent race conditions during import
- import_lock = ModuleUDFLoader._get_import_lock(full_module_name)
-
- with import_lock:
- # Fast path: check cache first
- if cache_key in ModuleUDFLoader._module_cache:
- cached_module = ModuleUDFLoader._module_cache[cache_key]
- if cached_module is not None and (
- hasattr(cached_module, "__file__")
- or hasattr(cached_module, "__path__")
- ):
- return cached_module
- else:
- del ModuleUDFLoader._module_cache[cache_key]
-
- self._clear_modules_from_sys(full_module_name)
-
- with temporary_sys_path(location):
- try:
- module = importlib.import_module(full_module_name)
- ModuleUDFLoader._module_cache[cache_key] = module
- # Evict from sys.modules so future imports from a
- # different location are not poisoned by this one.
- self._clear_modules_from_sys(full_module_name)
- return module
- except Exception:
- # Clean up any partially-imported modules
- self._clear_modules_from_sys(full_module_name)
- if cache_key in ModuleUDFLoader._module_cache:
- del ModuleUDFLoader._module_cache[cache_key]
- raise
+ return importlib.import_module(full_module_name)
Review Comment:
[P2] Resolve UDF entry points against their supplied package
A dedicated function process still has modules preloaded by server startup.
For example, `inspect` loads stdlib `types`, but `types` is absent from
`_FORBIDDEN_MODULE_NAMES`. With a valid package containing `types.py` and
symbol `types.evaluate`, `_load_single_file_udf()` verifies that the supplied
file exists, then this call returns the cached stdlib module and
`_extract_function()` fails because it has no `evaluate`. The previous loader
removed that module key before importing; an isolated probe of the actual
base/head loaders succeeds on the base and fails on this head. Ensure
entry-point resolution uses the requested package even when its name is already
in `sys.modules`, and cover this compatibility case.
--
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]
