[Feat](udtf) Support Python UDTF for Doris

Author: yangshijie

be/src/udf/python/python_udaf_client.cpp

@@ -440,7 +440,7 @@ Status PythonUDAFClient::close() {
 
     auto writer_res = _writer->Close();
     if (!writer_res.ok()) {
-        return _handle_error(writer_res);
+        return handle_error(writer_res);
     }
 
     _inited = false;
@@ -529,22 +529,22 @@ Status PythonUDAFClient::_send_operation(const arrow::RecordBatch* input,
         // Always use the unified schema for all operations
         auto begin_res = _writer->Begin(kUnifiedUDAFSchema);
         if (!begin_res.ok()) {
-            return _handle_error(begin_res);
+            return handle_error(begin_res);
         }
         _begin = true;
     }
 
     // Step 2: Write the record batch to server
     auto write_res = _writer->WriteRecordBatch(*input);
     if (!write_res.ok()) {
-        return _handle_error(write_res);
+        return handle_error(write_res);
     }
 
     // Step 3: Read response from server (if output is expected)
     if (output != nullptr) {
         auto read_res = _reader->Next();
         if (!read_res.ok()) {
-            return _handle_error(read_res.status());
+            return handle_error(read_res.status());
         }
 
         arrow::flight::FlightStreamChunk chunk = std::move(*read_res);
@@ -581,13 +581,13 @@ Status PythonUDAFClient::_send_operation(const arrow::RecordBatch* input,
     return Status::OK();
 }
 
-Status PythonUDAFClient::_handle_error(arrow::Status status) {
+Status PythonUDAFClient::handle_error(arrow::Status status) {
     DCHECK(!status.ok());
     _writer.reset();
     _reader.reset();
     _process->shutdown();
-
     std::string msg = status.message();
+    LOG(ERROR) << "Python UDAF error: " << msg;
     // Remove Python traceback noise
     size_t pos = msg.find("The above exception was the direct cause");
     if (pos != std::string::npos) {

be/src/udf/python/python_udaf_client.h

@@ -147,15 +147,15 @@ class PythonUDAFClient {
 
     Status close();
 
+    Status handle_error(arrow::Status status);
+
     std::string print_process() const { return _process->to_string(); }
 
     static std::string print_operation(UDAFOperation op);
 
 private:
     DISALLOW_COPY_AND_ASSIGN(PythonUDAFClient);
 
-    Status _handle_error(arrow::Status status);
-
     /**
      * Helper to execute a UDAF operation (CREATE, RESET, DESTROY, etc.)
      * This consolidates the common pattern:

be/src/udf/python/python_udf_client.cpp

@@ -98,6 +98,8 @@ Status PythonUDFClient::handle_error(arrow::Status status) {
     _reader.reset();
     _process->shutdown();
     std::string msg = status.message();
+    LOG(ERROR) << "Python UDF error: " << msg;
+    // Remove Python traceback noise
     size_t pos = msg.find("The above exception was the direct cause");
     if (pos != std::string::npos) {
         msg = msg.substr(0, pos);

be/src/udf/python/python_udf_server.cpp

@@ -29,6 +29,7 @@
 #include "common/config.h"
 #include "udf/python/python_udaf_client.h"
 #include "udf/python/python_udf_client.h"
+#include "udf/python/python_udtf_client.h"
 
 namespace doris {
 
@@ -140,7 +141,7 @@ void PythonUDFServerManager::shutdown() {
     LOG(INFO) << "Python UDF server manager shutdown successfully";
 }
 
-// Explicit template instantiation for UDF and UDAF clients
+// Explicit template instantiation for UDF, UDAF and UDTF clients
 template Status PythonUDFServerManager::get_client<PythonUDFClient>(
         const PythonUDFMeta& func_meta, const PythonVersion& version,
         std::shared_ptr<PythonUDFClient>* client);
@@ -149,4 +150,8 @@ template Status PythonUDFServerManager::get_client<PythonUDAFClient>(
         const PythonUDFMeta& func_meta, const PythonVersion& version,
         std::shared_ptr<PythonUDAFClient>* client);
 
+template Status PythonUDFServerManager::get_client<PythonUDTFClient>(
+        const PythonUDFMeta& func_meta, const PythonVersion& version,
+        std::shared_ptr<PythonUDTFClient>* client);
+
 } // namespace doris