be/src/format_v2/table/hudi_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/table/hudi_reader.h" |
19 | | |
20 | | #include <utility> |
21 | | |
22 | | #include "exprs/vexpr_context.h" |
23 | | #include "format_v2/column_mapper.h" |
24 | | #include "format_v2/jni/hudi_jni_reader.h" |
25 | | #include "format_v2/table/schema_history_util.h" |
26 | | #include "gen_cpp/PlanNodes_types.h" |
27 | | |
28 | | namespace doris::format::hudi { |
29 | | |
30 | 7 | Status HudiReader::prepare_split(const format::SplitReadOptions& options) { |
31 | 7 | _split_schema_id = -1; |
32 | 7 | if (options.current_range.__isset.table_format_params && |
33 | 7 | options.current_range.table_format_params.__isset.hudi_params && |
34 | 7 | options.current_range.table_format_params.hudi_params.__isset.schema_id) { |
35 | 5 | _split_schema_id = options.current_range.table_format_params.hudi_params.schema_id; |
36 | 5 | } |
37 | 7 | RETURN_IF_ERROR(format::TableReader::prepare_split(options)); |
38 | 7 | if (current_split_pruned()) { |
39 | 0 | return Status::OK(); |
40 | 0 | } |
41 | | // This native reader only receives Hudi base-file splits that do not require merging delta |
42 | | // logs. Hudi creates a new versioned base file for updates and compaction instead of modifying |
43 | | // an existing Parquet/ORC base file in place, so missing mtime does not make its page-cache key |
44 | | // ambiguous. Merge-on-read log files remain on the JNI path because they may be appended and |
45 | | // must never be marked immutable here. |
46 | 7 | mark_current_data_file_immutable(); |
47 | 7 | return Status::OK(); |
48 | 7 | } |
49 | | |
50 | 8 | format::TableColumnMappingMode HudiReader::mapping_mode() const { |
51 | 8 | return format::can_map_by_history_schema(_scan_params, _split_schema_id) |
52 | 8 | ? format::TableColumnMappingMode::BY_FIELD_ID |
53 | 8 | : format::TableColumnMappingMode::BY_NAME; |
54 | 8 | } |
55 | | |
56 | 3 | Status HudiReader::annotate_file_schema(std::vector<format::ColumnDefinition>* file_schema) { |
57 | 3 | DORIS_CHECK(file_schema != nullptr); |
58 | 3 | if (mapping_mode() != format::TableColumnMappingMode::BY_FIELD_ID) { |
59 | 2 | return Status::OK(); |
60 | 2 | } |
61 | 1 | return format::annotate_file_schema_from_history(_scan_params, _split_schema_id, file_schema); |
62 | 3 | } |
63 | | |
64 | 1 | Status HudiHybridReader::init(format::TableReadOptions&& options) { |
65 | 1 | return format::TableReader::init(std::move(options)); |
66 | 1 | } |
67 | | |
68 | 1 | Status HudiHybridReader::prepare_split(const format::SplitReadOptions& options) { |
69 | 1 | RETURN_IF_ERROR(_ensure_current_split_reader(options)); |
70 | 1 | DORIS_CHECK(_current_split_reader != nullptr); |
71 | 1 | return _current_split_reader->prepare_split(options); |
72 | 1 | } |
73 | | |
74 | 1 | Status HudiHybridReader::get_block(Block* block, bool* eos) { |
75 | 1 | DORIS_CHECK(_current_split_reader != nullptr); |
76 | 1 | return _current_split_reader->get_block(block, eos); |
77 | 1 | } |
78 | | |
79 | 0 | bool HudiHybridReader::current_split_pruned() const { |
80 | 0 | DORIS_CHECK(_current_split_reader != nullptr); |
81 | 0 | return _current_split_reader->current_split_pruned(); |
82 | 0 | } |
83 | | |
84 | 1 | bool HudiHybridReader::current_split_uses_metadata_count() const { |
85 | 1 | DORIS_CHECK(_current_split_reader != nullptr); |
86 | 1 | return _current_split_reader->current_split_uses_metadata_count(); |
87 | 1 | } |
88 | | |
89 | 0 | Status HudiHybridReader::abort_split() { |
90 | 0 | DORIS_CHECK(_current_split_reader != nullptr); |
91 | 0 | return _current_split_reader->abort_split(); |
92 | 0 | } |
93 | | |
94 | 1 | Status HudiHybridReader::close() { |
95 | 1 | Status close_status = Status::OK(); |
96 | 1 | if (_native_reader != nullptr) { |
97 | 1 | close_status = _native_reader->close(); |
98 | 1 | } |
99 | 1 | if (_jni_reader != nullptr) { |
100 | 0 | auto status = _jni_reader->close(); |
101 | 0 | if (!status.ok() && close_status.ok()) { |
102 | 0 | close_status = std::move(status); |
103 | 0 | } |
104 | 0 | } |
105 | 1 | _current_split_reader = nullptr; |
106 | 1 | return close_status; |
107 | 1 | } |
108 | | |
109 | 1 | void HudiHybridReader::set_batch_size(size_t batch_size) { |
110 | 1 | format::TableReader::set_batch_size(batch_size); |
111 | 1 | if (_native_reader != nullptr) { |
112 | 1 | _native_reader->set_batch_size(_batch_size); |
113 | 1 | } |
114 | 1 | if (_jni_reader != nullptr) { |
115 | 1 | _jni_reader->set_batch_size(_batch_size); |
116 | 1 | } |
117 | 1 | } |
118 | | |
119 | 1 | Status HudiHybridReader::_ensure_current_split_reader(const format::SplitReadOptions& options) { |
120 | 1 | DORIS_CHECK(_scan_params != nullptr); |
121 | 1 | if (_is_jni_split(*_scan_params, options.current_range)) { |
122 | 0 | if (_jni_reader == nullptr) { |
123 | 0 | _jni_reader = std::make_unique<format::hudi::HudiJniReader>(); |
124 | 0 | RETURN_IF_ERROR(_init_child_reader(_jni_reader.get(), format::FileFormat::JNI)); |
125 | 0 | } |
126 | 0 | _current_split_reader = _jni_reader.get(); |
127 | 1 | } else { |
128 | 1 | format::FileFormat file_format; |
129 | 1 | RETURN_IF_ERROR(_to_file_format(*_scan_params, options.current_range, &file_format)); |
130 | 1 | if (_native_reader == nullptr) { |
131 | 1 | _native_reader = format::hudi::HudiReader::create_unique(); |
132 | 1 | RETURN_IF_ERROR(_init_child_reader(_native_reader.get(), file_format)); |
133 | 1 | } |
134 | 1 | _current_split_reader = _native_reader.get(); |
135 | 1 | } |
136 | 1 | return Status::OK(); |
137 | 1 | } |
138 | | |
139 | | Status HudiHybridReader::_init_child_reader(format::TableReader* reader, |
140 | 1 | format::FileFormat file_format) { |
141 | 1 | DORIS_CHECK(reader != nullptr); |
142 | 1 | VExprContextSPtrs conjuncts; |
143 | 1 | RETURN_IF_ERROR(_clone_conjuncts(&conjuncts)); |
144 | 1 | RETURN_IF_ERROR(reader->init({ |
145 | 1 | .projected_columns = _projected_columns, |
146 | 1 | .conjuncts = std::move(conjuncts), |
147 | 1 | .format = file_format, |
148 | 1 | .scan_params = _scan_params, |
149 | 1 | .io_ctx = _io_ctx, |
150 | 1 | .runtime_state = _runtime_state, |
151 | 1 | .scanner_profile = _scanner_profile, |
152 | 1 | .push_down_agg_type = _push_down_agg_type, |
153 | 1 | .push_down_count_columns = _push_down_count_columns, |
154 | 1 | .condition_cache_digest = _condition_cache_digest, |
155 | 1 | })); |
156 | | // Zero means no adaptive prediction has been produced yet. Preserve the child's normal |
157 | | // runtime default until FileScannerV2 supplies the first positive prediction. |
158 | 1 | if (_batch_size > 0) { |
159 | 0 | reader->set_batch_size(_batch_size); |
160 | 0 | } |
161 | 1 | return Status::OK(); |
162 | 1 | } |
163 | | |
164 | 1 | Status HudiHybridReader::_clone_conjuncts(VExprContextSPtrs* conjuncts) const { |
165 | 1 | DORIS_CHECK(conjuncts != nullptr); |
166 | 1 | conjuncts->clear(); |
167 | 1 | conjuncts->reserve(_conjuncts.size()); |
168 | 1 | for (const auto& conjunct : _conjuncts) { |
169 | 0 | VExprSPtr root; |
170 | 0 | RETURN_IF_ERROR(format::clone_table_expr_tree(conjunct->root(), &root)); |
171 | 0 | conjuncts->push_back(VExprContext::create_shared(std::move(root))); |
172 | 0 | } |
173 | 1 | return Status::OK(); |
174 | 1 | } |
175 | | |
176 | | TFileFormatType::type HudiHybridReader::_range_format_type(const TFileScanRangeParams& params, |
177 | 2 | const TFileRangeDesc& range) { |
178 | 2 | return range.__isset.format_type ? range.format_type : params.format_type; |
179 | 2 | } |
180 | | |
181 | | bool HudiHybridReader::_is_jni_split(const TFileScanRangeParams& params, |
182 | 1 | const TFileRangeDesc& range) { |
183 | 1 | return _range_format_type(params, range) == TFileFormatType::FORMAT_JNI; |
184 | 1 | } |
185 | | |
186 | | Status HudiHybridReader::_to_file_format(const TFileScanRangeParams& params, |
187 | | const TFileRangeDesc& range, |
188 | 1 | format::FileFormat* file_format) { |
189 | 1 | DORIS_CHECK(file_format != nullptr); |
190 | 1 | const auto format_type = _range_format_type(params, range); |
191 | 1 | switch (format_type) { |
192 | 1 | case TFileFormatType::FORMAT_PARQUET: |
193 | 1 | *file_format = format::FileFormat::PARQUET; |
194 | 1 | return Status::OK(); |
195 | 0 | case TFileFormatType::FORMAT_ORC: |
196 | 0 | *file_format = format::FileFormat::ORC; |
197 | 0 | return Status::OK(); |
198 | 0 | default: |
199 | 0 | return Status::NotSupported("Unsupported native Hudi file format {}", |
200 | 0 | to_string(format_type)); |
201 | 1 | } |
202 | 1 | } |
203 | | |
204 | | } // namespace doris::format::hudi |