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 | 6 | Status HudiReader::prepare_split(const format::SplitReadOptions& options) { |
31 | 6 | _split_schema_id = -1; |
32 | 6 | if (options.current_range.__isset.table_format_params && |
33 | 6 | options.current_range.table_format_params.__isset.hudi_params && |
34 | 6 | 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 | 6 | RETURN_IF_ERROR(format::TableReader::prepare_split(options)); |
38 | 6 | 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 | 6 | mark_current_data_file_immutable(); |
47 | 6 | return Status::OK(); |
48 | 6 | } |
49 | | |
50 | 6 | format::TableColumnMappingMode HudiReader::mapping_mode() const { |
51 | 6 | return format::can_map_by_history_schema(_scan_params, _split_schema_id) |
52 | 6 | ? format::TableColumnMappingMode::BY_FIELD_ID |
53 | 6 | : format::TableColumnMappingMode::BY_NAME; |
54 | 6 | } |
55 | | |
56 | 2 | Status HudiReader::annotate_file_schema(std::vector<format::ColumnDefinition>* file_schema) { |
57 | 2 | DORIS_CHECK(file_schema != nullptr); |
58 | 2 | if (mapping_mode() != format::TableColumnMappingMode::BY_FIELD_ID) { |
59 | 1 | return Status::OK(); |
60 | 1 | } |
61 | 1 | return format::annotate_file_schema_from_history(_scan_params, _split_schema_id, file_schema); |
62 | 2 | } |
63 | | |
64 | 0 | Status HudiHybridReader::init(format::TableReadOptions&& options) { |
65 | 0 | return format::TableReader::init(std::move(options)); |
66 | 0 | } |
67 | | |
68 | 0 | Status HudiHybridReader::prepare_split(const format::SplitReadOptions& options) { |
69 | 0 | RETURN_IF_ERROR(_ensure_current_split_reader(options)); |
70 | 0 | DORIS_CHECK(_current_split_reader != nullptr); |
71 | 0 | return _current_split_reader->prepare_split(options); |
72 | 0 | } |
73 | | |
74 | 0 | Status HudiHybridReader::get_block(Block* block, bool* eos) { |
75 | 0 | DORIS_CHECK(_current_split_reader != nullptr); |
76 | 0 | return _current_split_reader->get_block(block, eos); |
77 | 0 | } |
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 | 0 | Status HudiHybridReader::abort_split() { |
85 | 0 | DORIS_CHECK(_current_split_reader != nullptr); |
86 | 0 | return _current_split_reader->abort_split(); |
87 | 0 | } |
88 | | |
89 | 0 | Status HudiHybridReader::close() { |
90 | 0 | Status close_status = Status::OK(); |
91 | 0 | if (_native_reader != nullptr) { |
92 | 0 | close_status = _native_reader->close(); |
93 | 0 | } |
94 | 0 | if (_jni_reader != nullptr) { |
95 | 0 | auto status = _jni_reader->close(); |
96 | 0 | if (!status.ok() && close_status.ok()) { |
97 | 0 | close_status = std::move(status); |
98 | 0 | } |
99 | 0 | } |
100 | 0 | _current_split_reader = nullptr; |
101 | 0 | return close_status; |
102 | 0 | } |
103 | | |
104 | 1 | void HudiHybridReader::set_batch_size(size_t batch_size) { |
105 | 1 | format::TableReader::set_batch_size(batch_size); |
106 | 1 | if (_native_reader != nullptr) { |
107 | 1 | _native_reader->set_batch_size(_batch_size); |
108 | 1 | } |
109 | 1 | if (_jni_reader != nullptr) { |
110 | 1 | _jni_reader->set_batch_size(_batch_size); |
111 | 1 | } |
112 | 1 | } |
113 | | |
114 | 0 | Status HudiHybridReader::_ensure_current_split_reader(const format::SplitReadOptions& options) { |
115 | 0 | DORIS_CHECK(_scan_params != nullptr); |
116 | 0 | if (_is_jni_split(*_scan_params, options.current_range)) { |
117 | 0 | if (_jni_reader == nullptr) { |
118 | 0 | _jni_reader = std::make_unique<format::hudi::HudiJniReader>(); |
119 | 0 | RETURN_IF_ERROR(_init_child_reader(_jni_reader.get(), format::FileFormat::JNI)); |
120 | 0 | } |
121 | 0 | _current_split_reader = _jni_reader.get(); |
122 | 0 | } else { |
123 | 0 | format::FileFormat file_format; |
124 | 0 | RETURN_IF_ERROR(_to_file_format(*_scan_params, options.current_range, &file_format)); |
125 | 0 | if (_native_reader == nullptr) { |
126 | 0 | _native_reader = format::hudi::HudiReader::create_unique(); |
127 | 0 | RETURN_IF_ERROR(_init_child_reader(_native_reader.get(), file_format)); |
128 | 0 | } |
129 | 0 | _current_split_reader = _native_reader.get(); |
130 | 0 | } |
131 | 0 | return Status::OK(); |
132 | 0 | } |
133 | | |
134 | | Status HudiHybridReader::_init_child_reader(format::TableReader* reader, |
135 | 0 | format::FileFormat file_format) { |
136 | 0 | DORIS_CHECK(reader != nullptr); |
137 | 0 | VExprContextSPtrs conjuncts; |
138 | 0 | RETURN_IF_ERROR(_clone_conjuncts(&conjuncts)); |
139 | 0 | RETURN_IF_ERROR(reader->init({ |
140 | 0 | .projected_columns = _projected_columns, |
141 | 0 | .conjuncts = std::move(conjuncts), |
142 | 0 | .format = file_format, |
143 | 0 | .scan_params = _scan_params, |
144 | 0 | .io_ctx = _io_ctx, |
145 | 0 | .runtime_state = _runtime_state, |
146 | 0 | .scanner_profile = _scanner_profile, |
147 | 0 | .push_down_agg_type = _push_down_agg_type, |
148 | 0 | .condition_cache_digest = _condition_cache_digest, |
149 | 0 | })); |
150 | | // Zero means no adaptive prediction has been produced yet. Preserve the child's normal |
151 | | // runtime default until FileScannerV2 supplies the first positive prediction. |
152 | 0 | if (_batch_size > 0) { |
153 | 0 | reader->set_batch_size(_batch_size); |
154 | 0 | } |
155 | 0 | return Status::OK(); |
156 | 0 | } |
157 | | |
158 | 0 | Status HudiHybridReader::_clone_conjuncts(VExprContextSPtrs* conjuncts) const { |
159 | 0 | DORIS_CHECK(conjuncts != nullptr); |
160 | 0 | conjuncts->clear(); |
161 | 0 | conjuncts->reserve(_conjuncts.size()); |
162 | 0 | for (const auto& conjunct : _conjuncts) { |
163 | 0 | VExprSPtr root; |
164 | 0 | RETURN_IF_ERROR(format::clone_table_expr_tree(conjunct->root(), &root)); |
165 | 0 | conjuncts->push_back(VExprContext::create_shared(std::move(root))); |
166 | 0 | } |
167 | 0 | return Status::OK(); |
168 | 0 | } |
169 | | |
170 | | TFileFormatType::type HudiHybridReader::_range_format_type(const TFileScanRangeParams& params, |
171 | 0 | const TFileRangeDesc& range) { |
172 | 0 | return range.__isset.format_type ? range.format_type : params.format_type; |
173 | 0 | } |
174 | | |
175 | | bool HudiHybridReader::_is_jni_split(const TFileScanRangeParams& params, |
176 | 0 | const TFileRangeDesc& range) { |
177 | 0 | return _range_format_type(params, range) == TFileFormatType::FORMAT_JNI; |
178 | 0 | } |
179 | | |
180 | | Status HudiHybridReader::_to_file_format(const TFileScanRangeParams& params, |
181 | | const TFileRangeDesc& range, |
182 | 0 | format::FileFormat* file_format) { |
183 | 0 | DORIS_CHECK(file_format != nullptr); |
184 | 0 | const auto format_type = _range_format_type(params, range); |
185 | 0 | switch (format_type) { |
186 | 0 | case TFileFormatType::FORMAT_PARQUET: |
187 | 0 | *file_format = format::FileFormat::PARQUET; |
188 | 0 | return Status::OK(); |
189 | 0 | case TFileFormatType::FORMAT_ORC: |
190 | 0 | *file_format = format::FileFormat::ORC; |
191 | 0 | return Status::OK(); |
192 | 0 | default: |
193 | 0 | return Status::NotSupported("Unsupported native Hudi file format {}", |
194 | 0 | to_string(format_type)); |
195 | 0 | } |
196 | 0 | } |
197 | | |
198 | | } // namespace doris::format::hudi |