be/src/udf/python/python_udf_meta.cpp
Line | Count | Source |
1 | | // Licensed to the Apache Software Foundation (ASF) under one |
2 | | // or more contributor license agreements. See the NOTICE file |
3 | | // distributed with this work for additional information |
4 | | // regarding copyright ownership. The ASF licenses this file |
5 | | // to you under the Apache License, Version 2.0 (the |
6 | | // "License"); you may not use this file except in compliance |
7 | | // with the License. You may obtain a copy of the License at |
8 | | // |
9 | | // http://www.apache.org/licenses/LICENSE-2.0 |
10 | | // |
11 | | // Unless required by applicable law or agreed to in writing, |
12 | | // software distributed under the License is distributed on an |
13 | | // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
14 | | // KIND, either express or implied. See the License for the |
15 | | // specific language governing permissions and limitations |
16 | | // under the License. |
17 | | |
18 | | #include "udf/python/python_udf_meta.h" |
19 | | |
20 | | #include <arrow/util/base64.h> |
21 | | #include <fmt/core.h> |
22 | | #include <rapidjson/document.h> |
23 | | #include <rapidjson/stringbuffer.h> |
24 | | #include <rapidjson/writer.h> |
25 | | |
26 | | #include <sstream> |
27 | | |
28 | | #include "common/status.h" |
29 | | #include "format/arrow/arrow_utils.h" |
30 | | #include "util/string_util.h" |
31 | | |
32 | | namespace doris { |
33 | | |
34 | | Status PythonUDFMeta::convert_types_to_schema(const DataTypes& types, const std::string& timezone, |
35 | 15 | std::shared_ptr<arrow::Schema>* schema) { |
36 | 15 | arrow::SchemaBuilder builder; |
37 | 31 | for (size_t i = 0; i < types.size(); ++i) { |
38 | 16 | std::shared_ptr<arrow::DataType> arrow_type; |
39 | 16 | RETURN_IF_ERROR(convert_to_arrow_type(types[i], &arrow_type, timezone)); |
40 | 16 | std::shared_ptr<arrow::Field> field = std::make_shared<arrow::Field>( |
41 | 16 | "arg" + std::to_string(i), arrow_type, types[i]->is_nullable()); |
42 | 16 | RETURN_DORIS_STATUS_IF_ERROR(builder.AddField(field)); |
43 | 16 | } |
44 | 15 | RETURN_DORIS_STATUS_IF_RESULT_ERROR(schema, builder.Finish()); |
45 | 15 | return Status::OK(); |
46 | 15 | } |
47 | | |
48 | | Status PythonUDFMeta::serialize_arrow_schema(const std::shared_ptr<arrow::Schema>& schema, |
49 | 13 | std::shared_ptr<arrow::Buffer>* out) { |
50 | 13 | RETURN_DORIS_STATUS_IF_RESULT_ERROR( |
51 | 13 | out, arrow::ipc::SerializeSchema(*schema, arrow::default_memory_pool())); |
52 | 13 | return Status::OK(); |
53 | 13 | } |
54 | | |
55 | | /* |
56 | | json format: |
57 | | { |
58 | | "name": "xxx", |
59 | | "id": 123, |
60 | | "symbol": "xxx", |
61 | | "location": "xxx", |
62 | | "udf_load_type": 0 or 1, |
63 | | "client_type": 0 (UDF) or 1 (UDAF) or 2 (UDTF), |
64 | | "runtime_version": "x.xx.xx", |
65 | | "always_nullable": true, |
66 | | "inline_code": "base64_inline_code", |
67 | | "input_types": "base64_input_types", |
68 | | "return_type": "base64_return_type" |
69 | | } |
70 | | */ |
71 | 6 | Status PythonUDFMeta::serialize_to_json(std::string* json_str) const { |
72 | 6 | rapidjson::Document doc; |
73 | 6 | doc.SetObject(); |
74 | 6 | auto& allocator = doc.GetAllocator(); |
75 | 6 | doc.AddMember("name", rapidjson::Value().SetString(name.c_str(), allocator), allocator); |
76 | 6 | doc.AddMember("id", rapidjson::Value().SetInt64(id), allocator); |
77 | 6 | doc.AddMember("symbol", rapidjson::Value().SetString(symbol.c_str(), allocator), allocator); |
78 | 6 | doc.AddMember("location", rapidjson::Value().SetString(location.c_str(), allocator), allocator); |
79 | 6 | doc.AddMember("udf_load_type", rapidjson::Value().SetInt(static_cast<int>(type)), allocator); |
80 | 6 | doc.AddMember("client_type", rapidjson::Value().SetInt(static_cast<int>(client_type)), |
81 | 6 | allocator); |
82 | 6 | doc.AddMember("runtime_version", |
83 | 6 | rapidjson::Value().SetString(runtime_version.c_str(), allocator), allocator); |
84 | 6 | doc.AddMember("always_nullable", rapidjson::Value().SetBool(always_nullable), allocator); |
85 | | |
86 | 6 | { |
87 | | // Serialize base64 inline code to json |
88 | 6 | std::string base64_str = arrow::util::base64_encode(inline_code); |
89 | 6 | doc.AddMember("inline_code", rapidjson::Value().SetString(base64_str.c_str(), allocator), |
90 | 6 | allocator); |
91 | 6 | } |
92 | 6 | { |
93 | | // Serialize base64 input types to json |
94 | 6 | std::shared_ptr<arrow::Schema> input_schema; |
95 | 6 | RETURN_IF_ERROR(convert_types_to_schema(input_types, TimezoneUtils::default_time_zone, |
96 | 6 | &input_schema)); |
97 | 6 | std::shared_ptr<arrow::Buffer> input_schema_buffer; |
98 | 6 | RETURN_IF_ERROR(serialize_arrow_schema(input_schema, &input_schema_buffer)); |
99 | 6 | std::string base64_str = |
100 | 6 | arrow::util::base64_encode({input_schema_buffer->data_as<char>(), |
101 | 6 | static_cast<size_t>(input_schema_buffer->size())}); |
102 | 6 | doc.AddMember("input_types", rapidjson::Value().SetString(base64_str.c_str(), allocator), |
103 | 6 | allocator); |
104 | 6 | } |
105 | 0 | { |
106 | | // Serialize base64 return type to json |
107 | 6 | std::shared_ptr<arrow::Schema> return_schema; |
108 | 6 | RETURN_IF_ERROR(convert_types_to_schema({return_type}, TimezoneUtils::default_time_zone, |
109 | 6 | &return_schema)); |
110 | 6 | std::shared_ptr<arrow::Buffer> return_schema_buffer; |
111 | 6 | RETURN_IF_ERROR(serialize_arrow_schema(return_schema, &return_schema_buffer)); |
112 | 6 | std::string base64_str = |
113 | 6 | arrow::util::base64_encode({return_schema_buffer->data_as<char>(), |
114 | 6 | static_cast<size_t>(return_schema_buffer->size())}); |
115 | 6 | doc.AddMember("return_type", rapidjson::Value().SetString(base64_str.c_str(), allocator), |
116 | 6 | allocator); |
117 | 6 | } |
118 | | |
119 | | // Convert document to json string |
120 | 0 | rapidjson::StringBuffer buffer; |
121 | 6 | rapidjson::Writer<rapidjson::StringBuffer> writer(buffer); |
122 | 6 | doc.Accept(writer); |
123 | 6 | *json_str = std::string(buffer.GetString(), buffer.GetSize()); |
124 | 6 | return Status::OK(); |
125 | 6 | } |
126 | | |
127 | 2 | std::string PythonUDFMeta::to_string() const { |
128 | 2 | std::stringstream input_types_ss; |
129 | 2 | input_types_ss << "<"; |
130 | 7 | for (size_t i = 0; i < input_types.size(); ++i) { |
131 | 5 | input_types_ss << input_types[i]->get_name(); |
132 | 5 | if (i != input_types.size() - 1) { |
133 | 3 | input_types_ss << ", "; |
134 | 3 | } |
135 | 5 | } |
136 | 2 | input_types_ss << ">"; |
137 | 2 | return fmt::format( |
138 | 2 | "[name: {}, symbol: {}, location: {}, runtime_version: {}, always_nullable: {}, " |
139 | 2 | "inline_code: {}][input_types: {}][return_type: {}]", |
140 | 2 | name, symbol, location, runtime_version, always_nullable, inline_code, |
141 | 2 | input_types_ss.str(), return_type->get_name()); |
142 | 2 | } |
143 | | |
144 | 13 | Status PythonUDFMeta::check() const { |
145 | 13 | if (trim(name).empty()) { |
146 | 2 | return Status::InvalidArgument("Python UDF name is empty"); |
147 | 2 | } |
148 | | |
149 | 11 | if (trim(symbol).empty()) { |
150 | 1 | return Status::InvalidArgument("Python UDF symbol is empty"); |
151 | 1 | } |
152 | | |
153 | 10 | if (trim(runtime_version).empty()) { |
154 | 1 | return Status::InvalidArgument("Python UDF runtime version is empty"); |
155 | 1 | } |
156 | | |
157 | 9 | if (input_types.empty() && |
158 | 9 | (client_type == PythonClientType::UDAF || type == PythonUDFLoadType::UNKNOWN)) { |
159 | 1 | return Status::InvalidArgument("Python UDAF input types is empty"); |
160 | 1 | } |
161 | | |
162 | 8 | if (!return_type) { |
163 | 1 | return Status::InvalidArgument("Python UDF return type is empty"); |
164 | 1 | } |
165 | | |
166 | 7 | if (type == PythonUDFLoadType::UNKNOWN) { |
167 | 1 | return Status::InvalidArgument( |
168 | 1 | "Python UDF load type is invalid, please check inline code or file path"); |
169 | 1 | } |
170 | | |
171 | 6 | if (type == PythonUDFLoadType::MODULE) { |
172 | 3 | if (trim(location).empty()) { |
173 | 1 | return Status::InvalidArgument("Non-inline Python UDF location is empty"); |
174 | 1 | } |
175 | 2 | if (trim(checksum).empty()) { |
176 | 1 | return Status::InvalidArgument("Non-inline Python UDF checksum is empty"); |
177 | 1 | } |
178 | 2 | } |
179 | | |
180 | 4 | return Status::OK(); |
181 | 6 | } |
182 | | |
183 | | } // namespace doris |