be/src/exprs/table_function/udf_table_function.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 "exprs/table_function/udf_table_function.h" |
19 | | |
20 | | #include <glog/logging.h> |
21 | | |
22 | | #include "core/assert_cast.h" |
23 | | #include "core/block/block.h" |
24 | | #include "core/block/column_numbers.h" |
25 | | #include "core/column/column_array.h" |
26 | | #include "core/column/column_nullable.h" |
27 | | #include "core/data_type/data_type_array.h" |
28 | | #include "core/data_type/data_type_factory.hpp" |
29 | | #include "core/types.h" |
30 | | #include "exec/connector/jni_connector.h" |
31 | | #include "exprs/vexpr.h" |
32 | | #include "exprs/vexpr_context.h" |
33 | | #include "runtime/user_function_cache.h" |
34 | | |
35 | | namespace doris { |
36 | | #include "common/compile_check_begin.h" |
37 | | |
38 | | const char* EXECUTOR_CLASS = "org/apache/doris/udf/UdfExecutor"; |
39 | | const char* EXECUTOR_CTOR_SIGNATURE = "([B)V"; |
40 | | const char* EXECUTOR_EVALUATE_SIGNATURE = "(Ljava/util/Map;Ljava/util/Map;)J"; |
41 | | const char* EXECUTOR_CLOSE_SIGNATURE = "()V"; |
42 | 0 | UDFTableFunction::UDFTableFunction(const TFunction& t_fn) : TableFunction(), _t_fn(t_fn) { |
43 | 0 | _fn_name = _t_fn.name.function_name; |
44 | 0 | _return_type = DataTypeFactory::instance().create_data_type(t_fn.ret_type); |
45 | | // as the java-utdf function in java code is eg: ArrayList<String> |
46 | | // so we need a array column to save the execute result, and make_nullable could help deal with nullmap |
47 | 0 | _return_type = make_nullable(std::make_shared<DataTypeArray>(make_nullable(_return_type))); |
48 | 0 | } |
49 | | |
50 | 0 | Status UDFTableFunction::open() { |
51 | 0 | JNIEnv* env = nullptr; |
52 | 0 | RETURN_IF_ERROR(Jni::Env::Get(&env)); |
53 | 0 | _jni_ctx = std::make_shared<JniContext>(); |
54 | 0 | std::string local_location; |
55 | 0 | auto* function_cache = UserFunctionCache::instance(); |
56 | 0 | TJavaUdfExecutorCtorParams ctor_params; |
57 | 0 | ctor_params.__set_fn(_t_fn); |
58 | 0 | if (!_t_fn.hdfs_location.empty() && !_t_fn.checksum.empty()) { |
59 | | // get jar path if both file path location and checksum are null |
60 | 0 | RETURN_IF_ERROR(function_cache->get_jarpath(_t_fn.id, _t_fn.hdfs_location, _t_fn.checksum, |
61 | 0 | &local_location)); |
62 | 0 | ctor_params.__set_location(local_location); |
63 | 0 | } |
64 | 0 | RETURN_IF_ERROR(Jni::Util::find_class(env, EXECUTOR_CLASS, &_jni_ctx->executor_cl)); |
65 | | |
66 | 0 | Jni::LocalArray ctor_params_bytes; |
67 | 0 | RETURN_IF_ERROR(Jni::Util::SerializeThriftMsg(env, &ctor_params, &ctor_params_bytes)); |
68 | | |
69 | 0 | RETURN_IF_ERROR(_jni_ctx->executor_cl.get_method(env, "<init>", EXECUTOR_CTOR_SIGNATURE, |
70 | 0 | &_jni_ctx->executor_ctor_id)); |
71 | | |
72 | 0 | RETURN_IF_ERROR(_jni_ctx->executor_cl.get_method(env, "evaluate", EXECUTOR_EVALUATE_SIGNATURE, |
73 | 0 | &_jni_ctx->executor_evaluate_id)); |
74 | | |
75 | 0 | RETURN_IF_ERROR(_jni_ctx->executor_cl.get_method(env, "close", EXECUTOR_CLOSE_SIGNATURE, |
76 | 0 | &_jni_ctx->executor_close_id)); |
77 | | |
78 | 0 | RETURN_IF_ERROR(_jni_ctx->executor_cl.new_object(env, _jni_ctx->executor_ctor_id) |
79 | 0 | .with_arg(ctor_params_bytes) |
80 | 0 | .call(&_jni_ctx->executor)); |
81 | | |
82 | 0 | _jni_ctx->open_successes = true; |
83 | 0 | return Status::OK(); |
84 | 0 | } |
85 | | |
86 | 0 | Status UDFTableFunction::process_init(Block* block, RuntimeState* state) { |
87 | 0 | auto child_size = _expr_context->root()->children().size(); |
88 | 0 | ColumnNumbers child_column_idxs; |
89 | 0 | child_column_idxs.resize(child_size); |
90 | 0 | for (int i = 0; i < child_size; ++i) { |
91 | 0 | int result_id = -1; |
92 | 0 | RETURN_IF_ERROR(_expr_context->root()->children()[i]->execute(_expr_context.get(), block, |
93 | 0 | &result_id)); |
94 | 0 | DCHECK_NE(result_id, -1); |
95 | 0 | child_column_idxs[i] = result_id; |
96 | 0 | } |
97 | 0 | JNIEnv* env = nullptr; |
98 | 0 | RETURN_IF_ERROR(Jni::Env::Get(&env)); |
99 | 0 | std::unique_ptr<long[]> input_table; |
100 | 0 | RETURN_IF_ERROR( |
101 | 0 | JniConnector::to_java_table(block, block->rows(), child_column_idxs, input_table)); |
102 | 0 | auto input_table_schema = JniConnector::parse_table_schema(block, child_column_idxs, true); |
103 | 0 | std::map<String, String> input_params = { |
104 | 0 | {"meta_address", std::to_string((long)input_table.get())}, |
105 | 0 | {"required_fields", input_table_schema.first}, |
106 | 0 | {"columns_types", input_table_schema.second}}; |
107 | |
|
108 | 0 | Jni::LocalObject input_map; |
109 | 0 | RETURN_IF_ERROR(Jni::Util::convert_to_java_map(env, input_params, &input_map)); |
110 | 0 | _array_result_column = _return_type->create_column(); |
111 | 0 | _result_column_idx = block->columns(); |
112 | 0 | block->insert({_array_result_column, _return_type, "res"}); |
113 | 0 | auto output_table_schema = JniConnector::parse_table_schema(block, {_result_column_idx}, true); |
114 | 0 | std::string output_nullable = _return_type->is_nullable() ? "true" : "false"; |
115 | 0 | std::map<String, String> output_params = {{"is_nullable", output_nullable}, |
116 | 0 | {"required_fields", output_table_schema.first}, |
117 | 0 | {"columns_types", output_table_schema.second}}; |
118 | |
|
119 | 0 | Jni::LocalObject output_map; |
120 | 0 | RETURN_IF_ERROR(Jni::Util::convert_to_java_map(env, output_params, &output_map)); |
121 | 0 | long output_address; |
122 | 0 | RETURN_IF_ERROR(_jni_ctx->executor.call_long_method(env, _jni_ctx->executor_evaluate_id) |
123 | 0 | .with_arg(input_map) |
124 | 0 | .with_arg(output_map) |
125 | 0 | .call(&output_address)); |
126 | 0 | RETURN_IF_ERROR(JniConnector::fill_block(block, {_result_column_idx}, output_address)); |
127 | 0 | block->erase(_result_column_idx); |
128 | 0 | if (!extract_column_array_info(*_array_result_column, _array_column_detail)) { |
129 | 0 | return Status::NotSupported("column type {} not supported now", |
130 | 0 | block->get_by_position(_result_column_idx).column->get_name()); |
131 | 0 | } |
132 | 0 | return Status::OK(); |
133 | 0 | } |
134 | | |
135 | 0 | void UDFTableFunction::process_row(size_t row_idx) { |
136 | 0 | TableFunction::process_row(row_idx); |
137 | 0 | if (!_array_column_detail.array_nullmap_data || |
138 | 0 | !_array_column_detail.array_nullmap_data[row_idx]) { |
139 | 0 | _array_offset = (*_array_column_detail.offsets_ptr)[row_idx - 1]; |
140 | 0 | _cur_size = (*_array_column_detail.offsets_ptr)[row_idx] - _array_offset; |
141 | 0 | } |
142 | | // so when it's NULL of row_idx, will not update _cur_size |
143 | | // it's will be _cur_size == 0, and means current_empty. |
144 | | // if the fn is outer, will be continue insert_default |
145 | | // if the fn is not outer function, will be not insert any value. |
146 | 0 | } |
147 | | |
148 | 0 | void UDFTableFunction::process_close() { |
149 | 0 | _array_result_column = nullptr; |
150 | 0 | _array_column_detail.reset(); |
151 | 0 | _array_offset = 0; |
152 | 0 | } |
153 | | |
154 | 0 | void UDFTableFunction::get_same_many_values(MutableColumnPtr& column, int length) { |
155 | 0 | size_t pos = _array_offset + _cur_offset; |
156 | 0 | if (current_empty() || (_array_column_detail.nested_nullmap_data && |
157 | 0 | _array_column_detail.nested_nullmap_data[pos])) { |
158 | 0 | column->insert_many_defaults(length); |
159 | 0 | } else { |
160 | 0 | if (_is_nullable) { |
161 | 0 | auto* nullable_column = assert_cast<ColumnNullable*>(column.get()); |
162 | 0 | auto nested_column = nullable_column->get_nested_column_ptr(); |
163 | 0 | auto nullmap_column = nullable_column->get_null_map_column_ptr(); |
164 | 0 | nested_column->insert_many_from(*_array_column_detail.nested_col, pos, length); |
165 | 0 | assert_cast<ColumnUInt8*>(nullmap_column.get())->insert_many_defaults(length); |
166 | 0 | } else { |
167 | 0 | column->insert_many_from(*_array_column_detail.nested_col, pos, length); |
168 | 0 | } |
169 | 0 | } |
170 | 0 | } |
171 | | |
172 | 0 | int UDFTableFunction::get_value(MutableColumnPtr& column, int max_step) { |
173 | 0 | max_step = std::min(max_step, (int)(_cur_size - _cur_offset)); |
174 | 0 | size_t pos = _array_offset + _cur_offset; |
175 | 0 | if (current_empty()) { |
176 | 0 | column->insert_default(); |
177 | 0 | max_step = 1; |
178 | 0 | } else { |
179 | 0 | if (_is_nullable) { |
180 | 0 | auto* nullable_column = assert_cast<ColumnNullable*>(column.get()); |
181 | 0 | auto nested_column = nullable_column->get_nested_column_ptr(); |
182 | 0 | auto* nullmap_column = |
183 | 0 | assert_cast<ColumnUInt8*>(nullable_column->get_null_map_column_ptr().get()); |
184 | 0 | nested_column->insert_range_from(*_array_column_detail.nested_col, pos, max_step); |
185 | 0 | size_t old_size = nullmap_column->size(); |
186 | 0 | nullmap_column->resize(old_size + max_step); |
187 | 0 | memcpy(nullmap_column->get_data().data() + old_size, |
188 | 0 | _array_column_detail.nested_nullmap_data + pos * sizeof(UInt8), |
189 | 0 | max_step * sizeof(UInt8)); |
190 | 0 | } else { |
191 | 0 | column->insert_range_from(*_array_column_detail.nested_col, pos, max_step); |
192 | 0 | } |
193 | 0 | } |
194 | 0 | forward(max_step); |
195 | 0 | return max_step; |
196 | 0 | } |
197 | | |
198 | | #include "common/compile_check_end.h" |
199 | | } // namespace doris |