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 |