[Enhancement](pyudf) Support empty arg pyudf && udtf

linrrzqqq · linrrzqqq · commit 25ea1db020ea · 2026-04-13T17:51:10.000+08:00
diff --git a/be/src/exprs/function/function_python_udf.cpp b/be/src/exprs/function/function_python_udf.cpp
@@ -113,7 +113,7 @@ Status PythonFunctionCall::execute_impl(FunctionContext* context, Block& block,
         return Status::InternalError("Python UDF client is null");
     }
 
-    int64_t input_rows = block.rows();
+    int64_t input_rows = num_rows;
     uint32_t input_columns = block.columns();
     DCHECK(input_columns > 0 && result < input_columns &&
            _argument_types.size() == arguments.size());
@@ -142,8 +142,13 @@ Status PythonFunctionCall::execute_impl(FunctionContext* context, Block& block,
     std::shared_ptr<arrow::RecordBatch> input_batch;
     std::shared_ptr<arrow::RecordBatch> output_batch;
     cctz::time_zone _timezone_obj; // default UTC
-    RETURN_IF_ERROR(convert_to_arrow_batch(input_block, schema, arrow::default_memory_pool(),
-                                           &input_batch, _timezone_obj));
+    if (arguments.empty()) {
+        input_batch = arrow::RecordBatch::Make(schema, input_rows,
+                                               std::vector<std::shared_ptr<arrow::Array>> {});
+    } else {
+        RETURN_IF_ERROR(convert_to_arrow_batch(input_block, schema, arrow::default_memory_pool(),
+                                               &input_batch, _timezone_obj));
+    }
     RETURN_IF_ERROR(client->evaluate(*input_batch, &output_batch));
     int64_t output_rows = output_batch->num_rows();
 
diff --git a/be/src/exprs/table_function/python_udtf_function.cpp b/be/src/exprs/table_function/python_udtf_function.cpp
@@ -132,12 +132,19 @@ Status PythonUDTFFunction::process_init(Block* block, RuntimeState* state) {
     for (uint32_t i = 0; i < child_column_idxs.size(); ++i) {
         input_block.insert(block->get_by_position(child_column_idxs[i]));
     }
+    int64_t input_rows = block->rows();
     std::shared_ptr<arrow::Schema> input_schema;
     std::shared_ptr<arrow::RecordBatch> input_batch;
     RETURN_IF_ERROR(get_arrow_schema_from_block(input_block, &input_schema,
                                                 TimezoneUtils::default_time_zone));
-    RETURN_IF_ERROR(convert_to_arrow_batch(input_block, input_schema, arrow::default_memory_pool(),
-                                           &input_batch, _timezone_obj));
+    if (child_column_idxs.empty()) {
+        input_batch = arrow::RecordBatch::Make(input_schema, input_rows,
+                                               std::vector<std::shared_ptr<arrow::Array>> {});
+    } else {
+        RETURN_IF_ERROR(convert_to_arrow_batch(input_block, input_schema,
+                                               arrow::default_memory_pool(), &input_batch,
+                                               _timezone_obj));
+    }
 
     // Step 3: Call Python UDTF to evaluate all rows at once (similar to Java UDTF's JNI call)
     // Python returns a ListArray where each element contains outputs for one input row
diff --git a/be/src/udf/python/python_udf_meta.cpp b/be/src/udf/python/python_udf_meta.cpp
@@ -32,7 +32,6 @@ namespace doris {
 
 Status PythonUDFMeta::convert_types_to_schema(const DataTypes& types, const std::string& timezone,
                                               std::shared_ptr<arrow::Schema>* schema) {
-    assert(!types.empty());
     arrow::SchemaBuilder builder;
     for (size_t i = 0; i < types.size(); ++i) {
         std::shared_ptr<arrow::DataType> arrow_type;
@@ -152,8 +151,9 @@ Status PythonUDFMeta::check() const {
         return Status::InvalidArgument("Python UDF runtime version is empty");
     }
 
-    if (input_types.empty()) {
-        return Status::InvalidArgument("Python UDF input types is empty");
+    if (input_types.empty() &&
+        (client_type == PythonClientType::UDAF || type == PythonUDFLoadType::UNKNOWN)) {
+        return Status::InvalidArgument("Python UDAF input types is empty");
     }
 
     if (!return_type) {
diff --git a/be/test/udf/python/python_udf_meta_test.cpp b/be/test/udf/python/python_udf_meta_test.cpp
@@ -109,14 +109,43 @@ TEST_F(PythonUDFMetaTest, CheckEmptyRuntimeVersion) {
     EXPECT_TRUE(status.to_string().find("runtime version is empty") != std::string::npos);
 }
 
-TEST_F(PythonUDFMetaTest, CheckEmptyInputTypes) {
+TEST_F(PythonUDFMetaTest, CheckEmptyInputTypesAllowedForUdf) {
     PythonUDFMeta meta;
     meta.name = "test_udf";
     meta.symbol = "test_func";
     meta.runtime_version = "3.9.16";
     meta.input_types = {};
     meta.return_type = nullable_int32_;
     meta.type = PythonUDFLoadType::INLINE;
+    meta.client_type = PythonClientType::UDF;
+
+    Status status = meta.check();
+    EXPECT_TRUE(status.ok()) << status.to_string();
+}
+
+TEST_F(PythonUDFMetaTest, CheckEmptyInputTypesAllowedForUdtf) {
+    PythonUDFMeta meta;
+    meta.name = "test_udtf";
+    meta.symbol = "test_func";
+    meta.runtime_version = "3.9.16";
+    meta.input_types = {};
+    meta.return_type = nullable_string_;
+    meta.type = PythonUDFLoadType::INLINE;
+    meta.client_type = PythonClientType::UDTF;
+
+    Status status = meta.check();
+    EXPECT_TRUE(status.ok()) << status.to_string();
+}
+
+TEST_F(PythonUDFMetaTest, CheckEmptyInputTypesRejectedForUdaf) {
+    PythonUDFMeta meta;
+    meta.name = "test_udaf";
+    meta.symbol = "test_func";
+    meta.runtime_version = "3.9.16";
+    meta.input_types = {};
+    meta.return_type = nullable_int32_;
+    meta.type = PythonUDFLoadType::INLINE;
+    meta.client_type = PythonClientType::UDAF;
 
     Status status = meta.check();
     EXPECT_FALSE(status.ok());
@@ -401,6 +430,27 @@ TEST_F(PythonUDFMetaTest, SerializeToJsonMultipleInputTypes) {
     EXPECT_TRUE(doc.HasMember("input_types"));
 }
 
+TEST_F(PythonUDFMetaTest, SerializeToJsonEmptyInputTypesForUdf) {
+    PythonUDFMeta meta;
+    meta.name = "zero_arg_udf";
+    meta.symbol = "func";
+    meta.runtime_version = "3.9.16";
+    meta.input_types = {};
+    meta.return_type = nullable_int32_;
+    meta.type = PythonUDFLoadType::INLINE;
+    meta.client_type = PythonClientType::UDF;
+
+    std::string json_str;
+    Status status = meta.serialize_to_json(&json_str);
+    EXPECT_TRUE(status.ok()) << status.to_string();
+
+    rapidjson::Document doc;
+    doc.Parse(json_str.c_str());
+    EXPECT_FALSE(doc.HasParseError());
+    EXPECT_TRUE(doc.HasMember("input_types"));
+    EXPECT_FALSE(std::string(doc["input_types"].GetString()).empty());
+}
+
 // ============================================================================
 // PythonUDFMeta convert_types_to_schema() tests
 // ============================================================================
@@ -429,6 +479,17 @@ TEST_F(PythonUDFMetaTest, ConvertTypesToSchemaSingleType) {
     EXPECT_EQ(schema->num_fields(), 1);
 }
 
+TEST_F(PythonUDFMetaTest, ConvertTypesToSchemaEmpty) {
+    DataTypes types = {};
+    std::shared_ptr<arrow::Schema> schema;
+
+    Status status = PythonUDFMeta::convert_types_to_schema(types, TimezoneUtils::default_time_zone,
+                                                           &schema);
+    EXPECT_TRUE(status.ok()) << status.to_string();
+    EXPECT_NE(schema, nullptr);
+    EXPECT_EQ(schema->num_fields(), 0);
+}
+
 // ============================================================================
 // PythonUDFMeta serialize_arrow_schema() tests
 // ============================================================================
diff --git a/regression-test/suites/pythonudf_p0/test_pythonudf_no_input.groovy b/regression-test/suites/pythonudf_p0/test_pythonudf_no_input.groovy
@@ -0,0 +1,66 @@
+// 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.
+
+suite("test_pythonudf_no_input") {
+    def runtime_version = getPythonUdfRuntimeVersion()
+    def table_name = "test_pythonudf_no_input_tbl"
+
+    try {
+        sql """ DROP FUNCTION IF EXISTS py_const_no_input(); """
+        sql """ DROP TABLE IF EXISTS ${table_name}; """
+
+        sql """
+        CREATE FUNCTION py_const_no_input()
+        RETURNS INT
+        PROPERTIES (
+            "type" = "PYTHON_UDF",
+            "symbol" = "evaluate",
+            "runtime_version" = "${runtime_version}"
+        )
+        AS \$\$
+def evaluate():
+    return 7
+\$\$;
+        """
+
+        assert sql(""" SELECT py_const_no_input(); """)[0][0] == 7
+
+        sql """
+        CREATE TABLE ${table_name} (
+            id INT
+        ) ENGINE=OLAP
+        DUPLICATE KEY(id)
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES("replication_num" = "1");
+        """
+
+        sql """ INSERT INTO ${table_name} VALUES (1), (2), (3); """
+
+        def rows = sql("""
+            SELECT id, py_const_no_input() AS v
+            FROM ${table_name}
+            ORDER BY id
+        """)
+
+        assert rows.size() == 3 : "Expected 3 rows, got ${rows.size()}"
+        assert rows.collect { it[0] as int } == [1, 2, 3]
+        assert rows.every { (it[1] as int) == 7 }
+    } finally {
+        try_sql(""" DROP FUNCTION IF EXISTS py_const_no_input(); """)
+        try_sql(""" DROP TABLE IF EXISTS ${table_name}; """)
+    }
+}
diff --git a/regression-test/suites/pythonudtf_p0/test_pythonudtf_no_input.groovy b/regression-test/suites/pythonudtf_p0/test_pythonudtf_no_input.groovy
@@ -0,0 +1,70 @@
+// 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.
+
+suite("test_pythonudtf_no_input") {
+    def runtime_version = getPythonUdfRuntimeVersion()
+    def table_name = "test_pythonudtf_no_input_tbl"
+
+    try {
+        sql """ DROP FUNCTION IF EXISTS py_emit_no_input(); """
+        sql """ DROP TABLE IF EXISTS ${table_name}; """
+
+        sql """
+        CREATE TABLES FUNCTION py_emit_no_input()
+        RETURNS ARRAY<STRING>
+        PROPERTIES (
+            "type" = "PYTHON_UDF",
+            "symbol" = "emit_values",
+            "runtime_version" = "${runtime_version}"
+        )
+        AS \$\$
+def emit_values():
+    yield ('left',)
+    yield ('right',)
+\$\$;
+        """
+
+        sql """
+        CREATE TABLE ${table_name} (
+            id INT
+        ) ENGINE=OLAP
+        DUPLICATE KEY(id)
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES("replication_num" = "1");
+        """
+
+        sql """ INSERT INTO ${table_name} VALUES (1), (2); """
+
+        def rows = sql("""
+            SELECT id, value
+            FROM ${table_name}
+            LATERAL VIEW py_emit_no_input() tmp AS value
+            ORDER BY id, value
+        """)
+
+        assert rows.size() == 4 : "Expected 4 rows, got ${rows.size()}"
+        assert rows.collect { [(it[0] as int), it[1].toString()] } == [
+            [1, "left"],
+            [1, "right"],
+            [2, "left"],
+            [2, "right"]
+        ]
+    } finally {
+        try_sql(""" DROP FUNCTION IF EXISTS py_emit_no_input(); """)
+        try_sql(""" DROP TABLE IF EXISTS ${table_name}; """)
+    }
+}
