[Feat](opt) Optimize Python UDF/UDAF/UDTF for Doris

Author: yangshijie

be/src/common/config.cpp

@@ -1094,14 +1094,9 @@ DEFINE_String(python_conda_root_path, "");
 DEFINE_String(python_venv_root_path, "${DORIS_HOME}/lib/udf/python");
 // python interpreter paths used by venv, e.g. /usr/bin/python3.7:/usr/bin/python3.6
 DEFINE_String(python_venv_interpreter_paths, "");
-// python deps index url
-DEFINE_String(python_deps_index_url, "https://pypi.org/simple/");
-// min number of python process
-DEFINE_Int32(min_python_process_nums, "16");
-// max number of python process
-DEFINE_Int32(max_python_process_nums, "256");
-// timeout in milliseconds when waiting for available python process
-DEFINE_Int32(python_process_pool_wait_timeout_ms, "30000");
+// max python processes in global shared pool, each version can have up to this many processes
+// 0 means use CPU core count as default, otherwise use the specified value
+DEFINE_mInt32(max_python_process_num, "0");
 
 // Set config randomly to check more issues in github workflow
 DEFINE_Bool(enable_fuzzy_mode, "false");

be/src/common/config.h

@@ -1141,14 +1141,8 @@ DECLARE_String(python_conda_root_path);
 DECLARE_String(python_venv_root_path);
 // python interpreter paths used by venv, e.g. /usr/bin/python3.7:/usr/bin/python3.6
 DECLARE_String(python_venv_interpreter_paths);
-// python deps index url
-DECLARE_String(python_deps_index_url);
-// min number of python process
-DECLARE_Int32(min_python_process_nums);
-// max number of python process
-DECLARE_Int32(max_python_process_nums);
-// timeout in milliseconds when waiting for available python process
-DECLARE_Int32(python_process_pool_wait_timeout_ms);
+// max python processes in global shared pool, each version can have up to this many processes
+DECLARE_mInt32(max_python_process_num);
 
 // Set config randomly to check more issues in github workflow
 DECLARE_Bool(enable_fuzzy_mode);

be/src/udf/python/python_client.cpp

@@ -20,6 +20,7 @@
 #include "arrow/flight/client.h"
 #include "arrow/flight/server.h"
 #include "common/compiler_util.h"
+#include "common/config.h"
 #include "common/status.h"
 #include "udf/python/python_udf_meta.h"
 #include "udf/python/python_udf_runtime.h"
@@ -86,11 +87,7 @@ Status PythonClient::close() {
     _arrow_client.reset();
     _writer.reset();
     _reader.reset();
-
-    // Return process to pool if available
-    if (auto* pool = _process->pool(); pool) {
-        pool->return_process(std::move(_process));
-    }
+    _process.reset();
 
     return Status::OK();
 }
@@ -101,7 +98,6 @@ Status PythonClient::handle_error(arrow::Status status) {
     // Clean up resources
     _writer.reset();
     _reader.reset();
-    _process->shutdown();
 
     // Extract and clean error message
     std::string msg = status.message();
@@ -116,13 +112,6 @@ Status PythonClient::handle_error(arrow::Status status) {
     return Status::RuntimeError(trim(msg));
 }
 
-Status PythonClient::check_process_alive() const {
-    if (UNLIKELY(!_process->is_alive())) {
-        return Status::RuntimeError("{} process is not alive", _operation_name);
-    }
-    return Status::OK();
-}
-
 Status PythonClient::begin_stream(const std::shared_ptr<arrow::Schema>& schema) {
     if (UNLIKELY(!_begin)) {
         auto begin_res = _writer->Begin(schema);
@@ -150,7 +139,6 @@ Status PythonClient::read_batch(std::shared_ptr<arrow::RecordBatch>* output) {
 
     arrow::flight::FlightStreamChunk chunk = std::move(*read_res);
     if (!chunk.data) {
-        _process->shutdown();
         return Status::InternalError("Received null RecordBatch from {} server", _operation_name);
     }
 

be/src/udf/python/python_client.h

@@ -71,15 +71,15 @@ class PythonClient {
      * Get process information for debugging
      * @return Process string representation
      */
-    std::string print_process() const { return _process->to_string(); }
+    std::string print_process() const { return _process ? _process->to_string() : "null"; }
 
-protected:
     /**
-     * Check if process is alive
-     * @return Status
+     * Get the underlying Python process
+     * @return Process pointer
      */
-    Status check_process_alive() const;
+    ProcessPtr get_process() const { return _process; }
 
+protected:
     /**
      * Begin Flight stream with schema (called only once per stream)
      * @param schema Input schema

be/src/udf/python/python_env.cpp

@@ -283,7 +283,6 @@ Status PythonVersionManager::init(PythonEnvType env_type, const fs::path& python
     std::vector<PythonVersion> versions;
     RETURN_IF_ERROR(_env_scanner->scan());
     RETURN_IF_ERROR(_env_scanner->get_versions(&versions));
-    RETURN_IF_ERROR(PythonServerManager::instance().init(versions));
     return Status::OK();
 }