Coverage Report

Created: 2026-07-22 19:19

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/format_v2/jni/jdbc_reader.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 "format_v2/jni/jdbc_reader.h"
19
20
#include <memory>
21
#include <utility>
22
23
#include "common/cast_set.h"
24
#include "core/assert_cast.h"
25
#include "core/block/block.h"
26
#include "core/block/columns_with_type_and_name.h"
27
#include "core/column/column_nullable.h"
28
#include "core/data_type/data_type_nullable.h"
29
#include "core/data_type/data_type_string.h"
30
#include "exprs/function/simple_function_factory.h"
31
#include "exprs/vexpr_context.h"
32
#include "format_v2/table_reader.h"
33
#include "util/jdbc_utils.h"
34
35
namespace doris::format::jdbc {
36
37
0
std::string JdbcJniReader::connector_class() const {
38
0
    return "org/apache/doris/jdbc/JdbcJniScanner";
39
0
}
40
41
0
Status JdbcJniReader::prepare_split(const format::SplitReadOptions& options) {
42
0
    {
43
        // End these scopes before JniTableReader enters the same counters; nested use would count
44
        // this JDBC parameter preparation twice instead of extending the common lifecycle total.
45
0
        SCOPED_TIMER(_profile.total_timer);
46
0
        SCOPED_TIMER(_profile.prepare_split_timer);
47
0
        SCOPED_TIMER(connector_total_timer());
48
0
        _jdbc_params.clear();
49
0
        if (options.current_range.__isset.table_format_params &&
50
0
            options.current_range.table_format_params.table_format_type == "jdbc") {
51
0
            _jdbc_params = std::map<std::string, std::string>(
52
0
                    options.current_range.table_format_params.jdbc_params.begin(),
53
0
                    options.current_range.table_format_params.jdbc_params.end());
54
0
        }
55
0
    }
56
0
    return format::JniTableReader::prepare_split(options);
57
0
}
58
59
// need pass to the java side, so the java scanner can parse the params and construct the JDBC connection
60
0
Status JdbcJniReader::build_scanner_params(std::map<std::string, std::string>* params) const {
61
0
    DORIS_CHECK(params != nullptr);
62
0
    *params = _jdbc_params;
63
0
    if (params->contains("jdbc_driver_url")) {
64
0
        std::string resolved;
65
0
        if (JdbcUtils::resolve_driver_url((*params)["jdbc_driver_url"], &resolved).ok()) {
66
0
            (*params)["jdbc_driver_url"] = resolved;
67
0
        }
68
0
    }
69
0
    return Status::OK();
70
0
}
71
72
Status JdbcJniReader::build_jni_columns(
73
0
        std::vector<format::JniTableReader::JniColumn>* columns) const {
74
0
    DORIS_CHECK(columns != nullptr);
75
0
    columns->clear();
76
0
    columns->reserve(_projected_columns.size());
77
0
    for (size_t i = 0; i < _projected_columns.size(); ++i) {
78
0
        const auto& table_column = _projected_columns[i];
79
0
        const auto primitive_type = remove_nullable(table_column.type)->get_primitive_type();
80
0
        columns->push_back({
81
0
                .java_name = table_column.name,
82
0
                .output_index = i,
83
0
                .output_type = table_column.type,
84
0
                .transfer_type = _transfer_type_for(table_column.type),
85
0
                .replace_type = _replace_type_for(primitive_type),
86
0
        });
87
0
    }
88
0
    return Status::OK();
89
0
}
90
91
0
Status JdbcJniReader::finalize_jni_block(Block* jni_block, Block* output_block, size_t* rows) {
92
0
    DORIS_CHECK(jni_block != nullptr);
93
0
    DORIS_CHECK(output_block != nullptr);
94
0
    DORIS_CHECK(rows != nullptr);
95
0
    const auto original_rows = *rows;
96
0
    const auto& columns = jni_columns();
97
0
    DORIS_CHECK(columns.size() == jni_block->columns());
98
99
0
    for (size_t i = 0; i < columns.size(); ++i) {
100
0
        const auto& column = columns[i];
101
0
        DORIS_CHECK(column.output_type != nullptr);
102
0
        DORIS_CHECK(column.output_index < output_block->columns());
103
0
        if (_is_special_type(remove_nullable(column.output_type)->get_primitive_type())) {
104
0
            RETURN_IF_ERROR(_cast_string_to_special_type(column, jni_block, i, output_block,
105
0
                                                         original_rows));
106
0
            continue;
107
0
        }
108
0
        output_block->get_by_position(column.output_index).type = column.output_type;
109
0
        output_block->replace_by_position(column.output_index,
110
0
                                          jni_block->get_by_position(i).column);
111
0
    }
112
0
    DORIS_CHECK(output_block->rows() == original_rows);
113
0
    if (!_conjuncts.empty()) {
114
0
        RETURN_IF_ERROR(
115
0
                VExprContext::filter_block(_conjuncts, output_block, output_block->columns()));
116
0
    }
117
0
    *rows = output_block->rows();
118
0
    return Status::OK();
119
0
}
120
121
0
std::string JdbcJniReader::_replace_type_for(PrimitiveType type) const {
122
0
    switch (type) {
123
0
    case PrimitiveType::TYPE_BITMAP:
124
0
        return "bitmap";
125
0
    case PrimitiveType::TYPE_HLL:
126
0
        return "hll";
127
0
    case PrimitiveType::TYPE_QUANTILE_STATE:
128
0
        return "quantile_state";
129
0
    case PrimitiveType::TYPE_JSONB:
130
0
        return "jsonb";
131
0
    default:
132
0
        return "not_replace";
133
0
    }
134
0
}
135
136
0
bool JdbcJniReader::_is_special_type(PrimitiveType type) const {
137
0
    return type == PrimitiveType::TYPE_BITMAP || type == PrimitiveType::TYPE_HLL ||
138
0
           type == PrimitiveType::TYPE_QUANTILE_STATE || type == PrimitiveType::TYPE_JSONB;
139
0
}
140
141
0
DataTypePtr JdbcJniReader::_transfer_type_for(const DataTypePtr& output_type) const {
142
0
    DORIS_CHECK(output_type != nullptr);
143
0
    if (!_is_special_type(remove_nullable(output_type)->get_primitive_type())) {
144
0
        return output_type;
145
0
    }
146
0
    DataTypePtr string_type = std::make_shared<DataTypeString>();
147
0
    if (output_type->is_nullable()) {
148
0
        string_type = make_nullable(string_type);
149
0
    }
150
0
    return string_type;
151
0
}
152
153
Status JdbcJniReader::_cast_string_to_special_type(const format::JniTableReader::JniColumn& column,
154
                                                   Block* jni_block, size_t jni_column_index,
155
0
                                                   Block* output_block, size_t rows) {
156
0
    DORIS_CHECK(column.output_type != nullptr);
157
0
    DORIS_CHECK(column.transfer_type != nullptr);
158
0
    const auto target_type = column.output_type;
159
0
    const auto target_type_name = target_type->get_name();
160
161
0
    ColumnPtr input_column = jni_block->get_by_position(jni_column_index).column;
162
0
    ColumnPtr cast_param = target_type->create_column_const_with_default_value(1);
163
164
0
    ColumnsWithTypeAndName argument_template;
165
0
    argument_template.reserve(2);
166
0
    argument_template.emplace_back(std::move(input_column), column.transfer_type,
167
0
                                   "java.sql.String");
168
0
    argument_template.emplace_back(std::move(cast_param), target_type, target_type_name);
169
170
0
    FunctionBasePtr cast_function = SimpleFunctionFactory::instance().get_function(
171
0
            "CAST", argument_template, make_nullable(target_type));
172
0
    if (cast_function == nullptr) {
173
0
        return Status::InternalError("Failed to find CAST function for type {}", target_type_name);
174
0
    }
175
176
0
    Block cast_block(argument_template);
177
0
    const auto result_idx = cast_set<uint32_t>(cast_block.columns());
178
0
    cast_block.insert({nullptr, make_nullable(target_type), "cast_result"});
179
0
    RETURN_IF_ERROR(
180
0
            cast_function->execute(nullptr, cast_block, {0}, result_idx, cast_set<int>(rows)));
181
182
0
    auto result_column = cast_block.get_by_position(result_idx).column;
183
0
    output_block->get_by_position(column.output_index).type = target_type;
184
0
    if (target_type->is_nullable()) {
185
0
        output_block->replace_by_position(column.output_index, result_column);
186
0
    } else {
187
0
        const auto* nullable_column = assert_cast<const ColumnNullable*>(result_column.get());
188
0
        output_block->replace_by_position(column.output_index,
189
0
                                          nullable_column->get_nested_column_ptr());
190
0
    }
191
0
    return Status::OK();
192
0
}
193
194
} // namespace doris::format::jdbc