Coverage Report

Created: 2026-03-14 20:54

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
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