be/src/format/table/iceberg_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/table/iceberg_reader.h" |
19 | | |
20 | | #include <gen_cpp/Descriptors_types.h> |
21 | | #include <gen_cpp/Metrics_types.h> |
22 | | #include <gen_cpp/PlanNodes_types.h> |
23 | | #include <gen_cpp/parquet_types.h> |
24 | | #include <glog/logging.h> |
25 | | #include <parallel_hashmap/phmap.h> |
26 | | #include <rapidjson/document.h> |
27 | | |
28 | | #include <algorithm> |
29 | | #include <cstring> |
30 | | #include <functional> |
31 | | #include <memory> |
32 | | #include <optional> |
33 | | |
34 | | #include "common/compiler_util.h" // IWYU pragma: keep |
35 | | #include "common/consts.h" |
36 | | #include "common/status.h" |
37 | | #include "core/assert_cast.h" |
38 | | #include "core/block/block.h" |
39 | | #include "core/block/column_with_type_and_name.h" |
40 | | #include "core/column/column.h" |
41 | | #include "core/column/column_nullable.h" |
42 | | #include "core/column/column_string.h" |
43 | | #include "core/column/column_vector.h" |
44 | | #include "core/data_type/data_type_factory.hpp" |
45 | | #include "core/data_type/data_type_nullable.h" |
46 | | #include "core/data_type/data_type_struct.h" |
47 | | #include "core/data_type/define_primitive_type.h" |
48 | | #include "core/data_type/primitive_type.h" |
49 | | #include "core/string_ref.h" |
50 | | #include "core/typeid_cast.h" |
51 | | #include "exprs/aggregate/aggregate_function.h" |
52 | | #include "format/format_common.h" |
53 | | #include "format/generic_reader.h" |
54 | | #include "format/orc/vorc_reader.h" |
55 | | #include "format/parquet/schema_desc.h" |
56 | | #include "format/parquet/vparquet_column_chunk_reader.h" |
57 | | #include "format/table/deletion_vector_reader.h" |
58 | | #include "format/table/iceberg/iceberg_orc_nested_column_utils.h" |
59 | | #include "format/table/iceberg/iceberg_parquet_nested_column_utils.h" |
60 | | #include "format/table/iceberg_scan_semantics.h" |
61 | | #include "format/table/nested_column_access_helper.h" |
62 | | #include "format/table/table_schema_change_helper.h" |
63 | | #include "runtime/runtime_state.h" |
64 | | #include "util/coding.h" |
65 | | #include "util/string_util.h" |
66 | | |
67 | | namespace cctz { |
68 | | class time_zone; |
69 | | } // namespace cctz |
70 | | namespace doris { |
71 | | class RowDescriptor; |
72 | | class SlotDescriptor; |
73 | | class TupleDescriptor; |
74 | | |
75 | | namespace io { |
76 | | struct IOContext; |
77 | | } // namespace io |
78 | | class VExprContext; |
79 | | } // namespace doris |
80 | | |
81 | | namespace doris { |
82 | | namespace { |
83 | | |
84 | | constexpr auto kIcebergOrcAttribute = "iceberg.id"; |
85 | | |
86 | 50 | bool orc_subtree_has_iceberg_id(const orc::Type* type, const std::string& attribute) { |
87 | 50 | if (type->hasAttributeKey(attribute)) { |
88 | 13 | return true; |
89 | 13 | } |
90 | 55 | for (uint64_t idx = 0; idx < type->getSubtypeCount(); ++idx) { |
91 | 31 | if (orc_subtree_has_iceberg_id(type->getSubtype(idx), attribute)) { |
92 | 13 | return true; |
93 | 13 | } |
94 | 31 | } |
95 | 24 | return false; |
96 | 37 | } |
97 | | |
98 | 80 | bool parquet_subtree_has_iceberg_id(const FieldSchema& field) { |
99 | 80 | if (field.field_id >= 0) { |
100 | 57 | return true; |
101 | 57 | } |
102 | 23 | return std::ranges::any_of(field.children, parquet_subtree_has_iceberg_id); |
103 | 80 | } |
104 | | |
105 | | struct ParquetEqualityFieldPath { |
106 | | std::vector<const FieldSchema*> fields; |
107 | | std::vector<size_t> child_indexes; |
108 | | }; |
109 | | |
110 | | bool find_parquet_equality_field_path_by_id(const FieldDescriptor* descriptor, int32_t field_id, |
111 | 25 | ParquetEqualityFieldPath* result) { |
112 | 25 | DORIS_CHECK(descriptor != nullptr); |
113 | 25 | DORIS_CHECK(result != nullptr); |
114 | 25 | const auto find = [field_id](const auto& self, const FieldSchema* field, |
115 | 82 | ParquetEqualityFieldPath* path) -> bool { |
116 | 82 | DORIS_CHECK(field != nullptr); |
117 | 82 | path->fields.push_back(field); |
118 | 82 | if (field->field_id == field_id) { |
119 | 19 | return true; |
120 | 19 | } |
121 | 71 | for (size_t index = 0; index < field->children.size(); ++index) { |
122 | 26 | path->child_indexes.push_back(index); |
123 | 26 | if (self(self, &field->children[index], path)) { |
124 | 18 | return true; |
125 | 18 | } |
126 | 8 | path->child_indexes.pop_back(); |
127 | 8 | } |
128 | 45 | path->fields.pop_back(); |
129 | 45 | return false; |
130 | 63 | }; |
131 | 62 | for (int index = 0; index < descriptor->size(); ++index) { |
132 | 56 | if (find(find, descriptor->get_column(index), result)) { |
133 | 19 | return true; |
134 | 19 | } |
135 | 56 | } |
136 | 6 | return false; |
137 | 25 | } |
138 | | |
139 | | bool find_parquet_equality_field_prefix_by_id_path( |
140 | | const FieldDescriptor* descriptor, |
141 | | const std::vector<const schema::external::TField*>& table_path, |
142 | 6 | ParquetEqualityFieldPath* result) { |
143 | 6 | DORIS_CHECK(descriptor != nullptr); |
144 | 6 | DORIS_CHECK(result != nullptr); |
145 | 6 | DORIS_CHECK(!table_path.empty()); |
146 | 6 | const std::vector<FieldSchema>* candidates = nullptr; |
147 | 7 | for (size_t path_index = 0; path_index < table_path.size(); ++path_index) { |
148 | 7 | const auto* table_field = table_path[path_index]; |
149 | 7 | DORIS_CHECK(table_field != nullptr); |
150 | 7 | DORIS_CHECK(table_field->__isset.id); |
151 | 7 | const FieldSchema* match = nullptr; |
152 | 7 | size_t match_index = 0; |
153 | 7 | const size_t candidate_count = |
154 | 7 | candidates == nullptr ? cast_set<size_t>(descriptor->size()) : candidates->size(); |
155 | 25 | for (size_t candidate_index = 0; candidate_index < candidate_count; ++candidate_index) { |
156 | 18 | const auto* candidate = candidates == nullptr |
157 | 18 | ? descriptor->get_column(cast_set<int>(candidate_index)) |
158 | 18 | : &(*candidates)[candidate_index]; |
159 | 18 | if (candidate != nullptr && candidate->field_id == table_field->id) { |
160 | 0 | match = candidate; |
161 | 0 | match_index = candidate_index; |
162 | 0 | break; |
163 | 0 | } |
164 | 18 | } |
165 | 7 | if (match == nullptr) { |
166 | 7 | const auto wrapper = |
167 | 7 | candidates == nullptr |
168 | 7 | ? TableSchemaChangeHelper::BuildTableInfoUtil:: |
169 | 6 | find_unique_idless_parquet_wrapper_index( |
170 | 6 | *table_field, descriptor->get_fields_schema()) |
171 | 7 | : TableSchemaChangeHelper::BuildTableInfoUtil:: |
172 | 1 | find_unique_idless_parquet_wrapper_index(*table_field, |
173 | 1 | *candidates); |
174 | 7 | if (wrapper.has_value()) { |
175 | 1 | match_index = *wrapper; |
176 | 1 | match = candidates == nullptr ? descriptor->get_column(cast_set<int>(match_index)) |
177 | 1 | : &(*candidates)[match_index]; |
178 | 1 | } |
179 | 7 | } |
180 | 7 | if (match == nullptr) { |
181 | 6 | return false; |
182 | 6 | } |
183 | 1 | if (!result->fields.empty()) { |
184 | 0 | result->child_indexes.push_back(match_index); |
185 | 0 | } |
186 | 1 | result->fields.push_back(match); |
187 | 1 | candidates = &match->children; |
188 | 1 | } |
189 | 0 | return true; |
190 | 6 | } |
191 | | |
192 | | std::vector<std::string> equality_field_name_candidates(const schema::external::TField& table_field, |
193 | 21 | const std::string* leaf_fallback) { |
194 | 21 | std::vector<std::string> candidates; |
195 | 21 | if (table_field.__isset.name_mapping) { |
196 | 11 | candidates.insert(candidates.end(), table_field.name_mapping.begin(), |
197 | 11 | table_field.name_mapping.end()); |
198 | 11 | if (table_field.__isset.name_mapping_is_authoritative && |
199 | 11 | table_field.name_mapping_is_authoritative) { |
200 | 10 | return candidates; |
201 | 10 | } |
202 | 11 | } |
203 | 11 | if (table_field.__isset.name) { |
204 | 11 | candidates.push_back(table_field.name); |
205 | 11 | } |
206 | 11 | if (leaf_fallback != nullptr) { |
207 | 11 | candidates.push_back(*leaf_fallback); |
208 | 11 | } |
209 | 11 | return candidates; |
210 | 21 | } |
211 | | |
212 | | bool find_parquet_equality_field_prefix_by_name_path( |
213 | | const FieldDescriptor* descriptor, |
214 | | const std::vector<const schema::external::TField*>& table_path, |
215 | 7 | const std::string& leaf_fallback, ParquetEqualityFieldPath* result) { |
216 | 7 | DORIS_CHECK(descriptor != nullptr); |
217 | 7 | DORIS_CHECK(result != nullptr); |
218 | 7 | DORIS_CHECK(!table_path.empty()); |
219 | 7 | const std::vector<FieldSchema>* children = nullptr; |
220 | 17 | for (size_t path_index = 0; path_index < table_path.size(); ++path_index) { |
221 | 11 | const auto* table_field = table_path[path_index]; |
222 | 11 | DORIS_CHECK(table_field != nullptr); |
223 | 11 | const auto names = equality_field_name_candidates( |
224 | 11 | *table_field, path_index + 1 == table_path.size() ? &leaf_fallback : nullptr); |
225 | 11 | const FieldSchema* match = nullptr; |
226 | 11 | size_t match_index = 0; |
227 | 11 | const size_t child_count = |
228 | 11 | children == nullptr ? cast_set<size_t>(descriptor->size()) : children->size(); |
229 | 15 | for (const auto& name : names) { |
230 | 29 | for (size_t child_index = 0; child_index < child_count; ++child_index) { |
231 | 24 | const auto* child = children == nullptr |
232 | 24 | ? descriptor->get_column(cast_set<int>(child_index)) |
233 | 24 | : &(*children)[child_index]; |
234 | 24 | if (child != nullptr && iequal(child->name, name)) { |
235 | 10 | match = child; |
236 | 10 | match_index = child_index; |
237 | 10 | break; |
238 | 10 | } |
239 | 24 | } |
240 | 15 | if (match != nullptr) { |
241 | 10 | break; |
242 | 10 | } |
243 | 15 | } |
244 | 11 | if (match == nullptr) { |
245 | 1 | return false; |
246 | 1 | } |
247 | 10 | if (!result->fields.empty()) { |
248 | 4 | result->child_indexes.push_back(match_index); |
249 | 4 | } |
250 | 10 | result->fields.push_back(match); |
251 | 10 | children = &match->children; |
252 | 10 | } |
253 | 6 | return true; |
254 | 7 | } |
255 | | |
256 | | struct OrcEqualityFieldPath { |
257 | | std::vector<const orc::Type*> fields; |
258 | | std::vector<std::string> names; |
259 | | std::vector<size_t> child_indexes; |
260 | | }; |
261 | | |
262 | | bool find_orc_equality_field_path_by_id(const orc::Type* root, int32_t field_id, |
263 | 12 | OrcEqualityFieldPath* result) { |
264 | 12 | DORIS_CHECK(root != nullptr); |
265 | 12 | DORIS_CHECK(result != nullptr); |
266 | 12 | const auto find = [field_id](const auto& self, const orc::Type* field, |
267 | 12 | const std::string& field_name, |
268 | 34 | OrcEqualityFieldPath* path) -> bool { |
269 | 34 | DORIS_CHECK(field != nullptr); |
270 | 34 | path->fields.push_back(field); |
271 | 34 | path->names.push_back(field_name); |
272 | 34 | if (field->hasAttributeKey(kIcebergOrcAttribute) && |
273 | 34 | std::stoi(field->getAttributeValue(kIcebergOrcAttribute)) == field_id) { |
274 | 5 | return true; |
275 | 5 | } |
276 | 30 | for (size_t index = 0; index < field->getSubtypeCount(); ++index) { |
277 | 5 | path->child_indexes.push_back(index); |
278 | 5 | if (self(self, field->getSubtype(index), field->getFieldName(index), path)) { |
279 | 4 | return true; |
280 | 4 | } |
281 | 1 | path->child_indexes.pop_back(); |
282 | 1 | } |
283 | 25 | path->fields.pop_back(); |
284 | 25 | path->names.pop_back(); |
285 | 25 | return false; |
286 | 29 | }; |
287 | 36 | for (size_t index = 0; index < root->getSubtypeCount(); ++index) { |
288 | 29 | if (find(find, root->getSubtype(index), root->getFieldName(index), result)) { |
289 | 5 | return true; |
290 | 5 | } |
291 | 29 | } |
292 | 7 | return false; |
293 | 12 | } |
294 | | |
295 | | bool find_orc_equality_field_prefix_by_id_path( |
296 | | const orc::Type* root, const std::vector<const schema::external::TField*>& table_path, |
297 | 6 | OrcEqualityFieldPath* result) { |
298 | 6 | DORIS_CHECK(root != nullptr); |
299 | 6 | DORIS_CHECK(result != nullptr); |
300 | 6 | DORIS_CHECK(!table_path.empty()); |
301 | 6 | const orc::Type* parent = root; |
302 | 7 | for (const auto* table_field : table_path) { |
303 | 7 | DORIS_CHECK(table_field != nullptr); |
304 | 7 | DORIS_CHECK(table_field->__isset.id); |
305 | 7 | const orc::Type* match = nullptr; |
306 | 7 | size_t match_index = 0; |
307 | 25 | for (size_t candidate_index = 0; candidate_index < parent->getSubtypeCount(); |
308 | 18 | ++candidate_index) { |
309 | 18 | const auto* candidate = parent->getSubtype(candidate_index); |
310 | 18 | if (candidate->hasAttributeKey(kIcebergOrcAttribute) && |
311 | 18 | std::stoi(candidate->getAttributeValue(kIcebergOrcAttribute)) == table_field->id) { |
312 | 0 | match = candidate; |
313 | 0 | match_index = candidate_index; |
314 | 0 | break; |
315 | 0 | } |
316 | 18 | } |
317 | 7 | if (match == nullptr) { |
318 | 7 | const auto wrapper = TableSchemaChangeHelper::BuildTableInfoUtil:: |
319 | 7 | find_unique_idless_orc_wrapper_index(*table_field, parent, |
320 | 7 | kIcebergOrcAttribute); |
321 | 7 | if (wrapper.has_value()) { |
322 | 1 | match_index = *wrapper; |
323 | 1 | match = parent->getSubtype(match_index); |
324 | 1 | } |
325 | 7 | } |
326 | 7 | if (match == nullptr) { |
327 | 6 | return false; |
328 | 6 | } |
329 | 1 | if (!result->fields.empty()) { |
330 | 0 | result->child_indexes.push_back(match_index); |
331 | 0 | } |
332 | 1 | result->fields.push_back(match); |
333 | 1 | result->names.push_back(parent->getFieldName(match_index)); |
334 | 1 | parent = match; |
335 | 1 | } |
336 | 0 | return true; |
337 | 6 | } |
338 | | |
339 | | bool find_orc_equality_field_prefix_by_name_path( |
340 | | const orc::Type* root, const std::vector<const schema::external::TField*>& table_path, |
341 | 6 | const std::string& leaf_fallback, OrcEqualityFieldPath* result) { |
342 | 6 | DORIS_CHECK(root != nullptr); |
343 | 6 | DORIS_CHECK(result != nullptr); |
344 | 6 | DORIS_CHECK(!table_path.empty()); |
345 | 6 | const orc::Type* parent = root; |
346 | 15 | for (size_t path_index = 0; path_index < table_path.size(); ++path_index) { |
347 | 10 | const auto* table_field = table_path[path_index]; |
348 | 10 | DORIS_CHECK(table_field != nullptr); |
349 | 10 | const auto names = equality_field_name_candidates( |
350 | 10 | *table_field, path_index + 1 == table_path.size() ? &leaf_fallback : nullptr); |
351 | 10 | const orc::Type* match = nullptr; |
352 | 10 | size_t match_index = 0; |
353 | 14 | for (const auto& name : names) { |
354 | 26 | for (size_t child_index = 0; child_index < parent->getSubtypeCount(); ++child_index) { |
355 | 21 | if (iequal(parent->getFieldName(child_index), name)) { |
356 | 9 | match = parent->getSubtype(child_index); |
357 | 9 | match_index = child_index; |
358 | 9 | break; |
359 | 9 | } |
360 | 21 | } |
361 | 14 | if (match != nullptr) { |
362 | 9 | break; |
363 | 9 | } |
364 | 14 | } |
365 | 10 | if (match == nullptr) { |
366 | 1 | return false; |
367 | 1 | } |
368 | 9 | if (!result->fields.empty()) { |
369 | 4 | result->child_indexes.push_back(match_index); |
370 | 4 | } |
371 | 9 | result->fields.push_back(match); |
372 | 9 | result->names.push_back(parent->getFieldName(match_index)); |
373 | 9 | parent = match; |
374 | 9 | } |
375 | 5 | return true; |
376 | 6 | } |
377 | | |
378 | | struct ProjectedIcebergStructChild { |
379 | | size_t index; |
380 | | const schema::external::TField* field; |
381 | | }; |
382 | | |
383 | | std::optional<ProjectedIcebergStructChild> find_projected_iceberg_struct_child( |
384 | | const schema::external::TField& parent, const schema::external::TField& target, |
385 | 11 | const DataTypeStruct& projected_type) { |
386 | 11 | DORIS_CHECK(parent.__isset.nestedField); |
387 | 11 | DORIS_CHECK(parent.nestedField.__isset.struct_field); |
388 | 11 | DORIS_CHECK(parent.nestedField.struct_field.__isset.fields); |
389 | 11 | DORIS_CHECK(target.__isset.name); |
390 | 11 | DORIS_CHECK(target.__isset.id); |
391 | | |
392 | | // FE keeps current children first and appends historical equality fields to the schema carrier |
393 | | // by field ID. A dropped and re-added field may therefore appear twice with the same name, |
394 | | // while the query's DataTypeStruct contains only the current occurrence. |
395 | 11 | size_t schema_name_ordinal = 0; |
396 | 11 | const schema::external::TField* schema_child = nullptr; |
397 | 16 | for (const auto& child_ptr : parent.nestedField.struct_field.fields) { |
398 | 16 | DORIS_CHECK(child_ptr.__isset.field_ptr && child_ptr.field_ptr != nullptr); |
399 | 16 | const auto& child = child_ptr.field_ptr; |
400 | 16 | DORIS_CHECK(child->__isset.name); |
401 | 16 | DORIS_CHECK(child->__isset.id); |
402 | 16 | if (child->name != target.name) { |
403 | 5 | continue; |
404 | 5 | } |
405 | 11 | if (child->id == target.id) { |
406 | 11 | schema_child = child.get(); |
407 | 11 | break; |
408 | 11 | } |
409 | 0 | ++schema_name_ordinal; |
410 | 0 | } |
411 | 11 | if (schema_child == nullptr) { |
412 | 0 | return std::nullopt; |
413 | 0 | } |
414 | | |
415 | 11 | size_t projected_name_ordinal = 0; |
416 | 16 | for (size_t index = 0; index < projected_type.get_elements().size(); ++index) { |
417 | 15 | if (projected_type.get_element_name(index) != target.name) { |
418 | 5 | continue; |
419 | 5 | } |
420 | 10 | if (projected_name_ordinal == schema_name_ordinal) { |
421 | 10 | return ProjectedIcebergStructChild {.index = index, .field = schema_child}; |
422 | 10 | } |
423 | 0 | ++projected_name_ordinal; |
424 | 0 | } |
425 | 1 | return std::nullopt; |
426 | 11 | } |
427 | | |
428 | | // Resolves a nested equality key inside the projected root column. That column has the table's |
429 | | // projected struct type, not the file's physical struct, so the child indexes come from the |
430 | | // projected type. `found` stays false when the projected type does not hold the key, e.g. when the |
431 | | // key was dropped from the table or the struct was pruned. |
432 | | Status find_projected_equality_key_path( |
433 | | const schema::external::TField& projected_root, |
434 | | const std::vector<const schema::external::TField*>& key_path, const DataTypePtr& root_type, |
435 | 11 | std::vector<size_t>* child_indexes, DataTypePtr* leaf_type, bool* found) { |
436 | 11 | DORIS_CHECK(!key_path.empty()); |
437 | 11 | DORIS_CHECK(child_indexes != nullptr); |
438 | 11 | DORIS_CHECK(leaf_type != nullptr); |
439 | 11 | DORIS_CHECK(found != nullptr); |
440 | 11 | child_indexes->clear(); |
441 | 11 | *found = false; |
442 | 11 | const auto* projected_parent = &projected_root; |
443 | 11 | DataTypePtr current_type = root_type; |
444 | 21 | for (size_t path_index = 1; path_index < key_path.size(); ++path_index) { |
445 | 11 | const auto* child = key_path[path_index]; |
446 | 11 | DORIS_CHECK(child != nullptr); |
447 | 11 | if (!projected_parent->__isset.nestedField || |
448 | 11 | !projected_parent->nestedField.__isset.struct_field || |
449 | 11 | !projected_parent->nestedField.struct_field.__isset.fields) { |
450 | 0 | return Status::NotSupported( |
451 | 0 | "Iceberg equality-delete field {} has a non-struct projected-schema parent", |
452 | 0 | child->name); |
453 | 0 | } |
454 | 11 | const auto* struct_type = |
455 | 11 | typeid_cast<const DataTypeStruct*>(remove_nullable(current_type).get()); |
456 | 11 | if (struct_type == nullptr) { |
457 | 0 | return Status::OK(); |
458 | 0 | } |
459 | 11 | const auto projected_child = |
460 | 11 | find_projected_iceberg_struct_child(*projected_parent, *child, *struct_type); |
461 | 11 | if (!projected_child.has_value()) { |
462 | 1 | return Status::OK(); |
463 | 1 | } |
464 | 10 | child_indexes->push_back(projected_child->index); |
465 | 10 | current_type = struct_type->get_element(projected_child->index); |
466 | 10 | projected_parent = projected_child->field; |
467 | 10 | } |
468 | 10 | *leaf_type = current_type; |
469 | 10 | *found = true; |
470 | 10 | return Status::OK(); |
471 | 11 | } |
472 | | |
473 | | // A projected column is read through its own schema mapping, as the struct reader resolves each |
474 | | // child (`children_file_column_name`), while a hidden key is located in the file by field ID or by |
475 | | // its name path. The two can bind different file fields: a plan without the Iceberg scan semantics |
476 | | // marker maps a struct level by name when not all of its file children carry field IDs. The key |
477 | | // can therefore come from the projected column only when that mapping reads the same file field at |
478 | | // every level of the key's physical path. |
479 | | bool projected_root_reads_physical_path(const TableSchemaChangeHelper::Node& root_node, |
480 | | const std::string& root_name, const DataTypePtr& root_type, |
481 | | const std::vector<size_t>& child_indexes, |
482 | 10 | const ParquetEqualityFieldPath& file_path) { |
483 | 10 | if (child_indexes.size() + 1 != file_path.fields.size() || |
484 | 10 | !root_node.children_column_exists(root_name) || |
485 | 10 | root_node.children_file_column_name(root_name) != file_path.fields.front()->name) { |
486 | 0 | return false; |
487 | 0 | } |
488 | 10 | auto node = root_node.get_children_node(root_name); |
489 | 10 | DataTypePtr type = root_type; |
490 | 19 | for (size_t level = 0; level < child_indexes.size(); ++level) { |
491 | 10 | const auto& struct_type = assert_cast<const DataTypeStruct&>(*remove_nullable(type)); |
492 | 10 | const auto& child_name = struct_type.get_element_name(child_indexes[level]); |
493 | 10 | if (!node->children_column_exists(child_name) || |
494 | 10 | node->children_file_column_name(child_name) != file_path.fields[level + 1]->name) { |
495 | 1 | return false; |
496 | 1 | } |
497 | 9 | node = node->get_children_node(child_name); |
498 | 9 | type = struct_type.get_element(child_indexes[level]); |
499 | 9 | } |
500 | 9 | return true; |
501 | 10 | } |
502 | | |
503 | | } // namespace |
504 | | |
505 | | const std::string IcebergOrcReader::ICEBERG_ORC_ATTRIBUTE = kIcebergOrcAttribute; |
506 | | |
507 | | bool IcebergTableReader::_is_fully_dictionary_encoded( |
508 | 8 | const tparquet::ColumnMetaData& column_metadata) { |
509 | 14 | const auto is_dictionary_encoding = [](tparquet::Encoding::type encoding) { |
510 | 14 | return encoding == tparquet::Encoding::PLAIN_DICTIONARY || |
511 | 14 | encoding == tparquet::Encoding::RLE_DICTIONARY; |
512 | 14 | }; |
513 | 12 | const auto is_data_page = [](tparquet::PageType::type page_type) { |
514 | 12 | return page_type == tparquet::PageType::DATA_PAGE || |
515 | 12 | page_type == tparquet::PageType::DATA_PAGE_V2; |
516 | 12 | }; |
517 | 8 | const auto is_level_encoding = [](tparquet::Encoding::type encoding) { |
518 | 2 | return encoding == tparquet::Encoding::RLE || encoding == tparquet::Encoding::BIT_PACKED; |
519 | 2 | }; |
520 | | |
521 | | // A column chunk may have a dictionary page but still contain plain-encoded data pages. |
522 | | // Only treat it as dictionary-coded when all data pages are dictionary encoded. |
523 | 8 | if (column_metadata.__isset.encoding_stats) { |
524 | 7 | bool has_data_page_stats = false; |
525 | 12 | for (const tparquet::PageEncodingStats& enc_stat : column_metadata.encoding_stats) { |
526 | 12 | if (is_data_page(enc_stat.page_type) && enc_stat.count > 0) { |
527 | 8 | has_data_page_stats = true; |
528 | 8 | if (!is_dictionary_encoding(enc_stat.encoding)) { |
529 | 2 | return false; |
530 | 2 | } |
531 | 8 | } |
532 | 12 | } |
533 | 5 | if (has_data_page_stats) { |
534 | 4 | return true; |
535 | 4 | } |
536 | 5 | } |
537 | | |
538 | 2 | bool has_dict_encoding = false; |
539 | 2 | bool has_nondict_encoding = false; |
540 | 3 | for (const tparquet::Encoding::type& encoding : column_metadata.encodings) { |
541 | 3 | if (is_dictionary_encoding(encoding)) { |
542 | 1 | has_dict_encoding = true; |
543 | 1 | } |
544 | | |
545 | 3 | if (!is_dictionary_encoding(encoding) && !is_level_encoding(encoding)) { |
546 | 2 | has_nondict_encoding = true; |
547 | 2 | break; |
548 | 2 | } |
549 | 3 | } |
550 | 2 | if (!has_dict_encoding || has_nondict_encoding) { |
551 | 2 | return false; |
552 | 2 | } |
553 | | |
554 | 0 | return true; |
555 | 2 | } |
556 | | |
557 | | // ============================================================================ |
558 | | // IcebergParquetReader: on_before_init_reader (Parquet-specific schema matching) |
559 | | // ============================================================================ |
560 | | // This format-specific setup mirrors the existing reader initialization sequence. |
561 | | // NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size) |
562 | 32 | Status IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) { |
563 | 32 | _column_descs = ctx->column_descs; |
564 | 32 | _fill_col_name_to_block_idx = ctx->col_name_to_block_idx; |
565 | 32 | _file_format = Fileformat::PARQUET; |
566 | 32 | set_preserve_binary_uuid(true); |
567 | | |
568 | | // Get file metadata schema first (available because _open_file() already ran) |
569 | 32 | const FieldDescriptor* field_desc = nullptr; |
570 | 32 | RETURN_IF_ERROR(this->get_file_metadata_schema(&field_desc)); |
571 | 32 | DCHECK(field_desc != nullptr); |
572 | | |
573 | | // Build table_info_node by field_id or name matching. |
574 | | // This must happen BEFORE column classification so we can use children_column_exists |
575 | | // to check if a column exists in the file (by field ID, not name). |
576 | 32 | if (!get_scan_params().__isset.history_schema_info || |
577 | 32 | get_scan_params().history_schema_info.empty()) [[unlikely]] { |
578 | 1 | RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_name(ctx->tuple_descriptor, *field_desc, |
579 | 1 | ctx->table_info_node)); |
580 | 31 | } else { |
581 | 31 | RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_field_id_with_name_mapping( |
582 | 31 | get_scan_params().history_schema_info.front().root_field, *field_desc, |
583 | 31 | ctx->table_info_node, supports_iceberg_scan_semantics_v1(&get_scan_params()))); |
584 | 31 | } |
585 | | |
586 | 32 | std::unordered_set<std::string> partition_col_names; |
587 | 32 | if (ctx->range->__isset.columns_from_path_keys) { |
588 | 0 | partition_col_names.insert(ctx->range->columns_from_path_keys.begin(), |
589 | 0 | ctx->range->columns_from_path_keys.end()); |
590 | 0 | } |
591 | | |
592 | | // Single pass: classify columns, detect $row_id, handle partition fallback. |
593 | 32 | bool has_partition_from_path = false; |
594 | 47 | for (const auto& desc : *ctx->column_descs) { |
595 | 47 | if (desc.category == ColumnCategory::SYNTHESIZED) { |
596 | 0 | if (desc.name == BeConsts::ICEBERG_ROWID_COL) { |
597 | 0 | this->register_synthesized_column_handler( |
598 | 0 | BeConsts::ICEBERG_ROWID_COL, [this](Block* block, size_t rows) -> Status { |
599 | 0 | return _fill_iceberg_row_id(block, rows); |
600 | 0 | }); |
601 | 0 | continue; |
602 | 0 | } else if (desc.name.starts_with(BeConsts::GLOBAL_ROWID_COL)) { |
603 | 0 | auto topn_row_id_column_iter = _create_topn_row_id_column_iterator(); |
604 | 0 | this->register_synthesized_column_handler( |
605 | 0 | desc.name, |
606 | 0 | [iter = std::move(topn_row_id_column_iter), this, &desc]( |
607 | 0 | Block* block, size_t rows) -> Status { |
608 | 0 | return fill_topn_row_id(iter, desc.name, block, rows); |
609 | 0 | }); |
610 | 0 | continue; |
611 | 0 | } |
612 | 47 | } else if (desc.category == ColumnCategory::PARTITION_KEY) { |
613 | 0 | bool has_partition_value = partition_col_names.contains(desc.name); |
614 | 0 | bool exists_in_file = ctx->table_info_node->children_column_exists(desc.name); |
615 | 0 | if (!has_partition_value || exists_in_file) { |
616 | | // Keep PARTITION_KEY category stable for scan planning, but still read |
617 | | // from file when the column exists there. |
618 | 0 | ctx->column_names.push_back(desc.name); |
619 | 0 | continue; |
620 | 0 | } |
621 | 0 | has_partition_from_path = true; |
622 | 47 | } else if (desc.category == ColumnCategory::REGULAR) { |
623 | 47 | ctx->column_names.push_back(desc.name); |
624 | 47 | } else if (desc.category == ColumnCategory::GENERATED) { |
625 | 0 | _init_row_lineage_columns(); |
626 | 0 | if (desc.name == ROW_LINEAGE_ROW_ID) { |
627 | 0 | ctx->column_names.push_back(desc.name); |
628 | 0 | this->register_generated_column_handler( |
629 | 0 | ROW_LINEAGE_ROW_ID, [this](Block* block, size_t rows) -> Status { |
630 | 0 | return _fill_row_lineage_row_id(block, rows); |
631 | 0 | }); |
632 | 0 | continue; |
633 | 0 | } else if (desc.name == ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER) { |
634 | 0 | ctx->column_names.push_back(desc.name); |
635 | 0 | this->register_generated_column_handler( |
636 | 0 | ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER, |
637 | 0 | [this](Block* block, size_t rows) -> Status { |
638 | 0 | return _fill_row_lineage_last_updated_sequence_number(block, rows); |
639 | 0 | }); |
640 | 0 | continue; |
641 | 0 | } |
642 | 0 | } |
643 | 47 | } |
644 | | |
645 | | // Set up partition value extraction if any partition columns need filling from path |
646 | 32 | if (has_partition_from_path) { |
647 | 0 | RETURN_IF_ERROR(_extract_partition_values(*ctx->range, ctx->tuple_descriptor, |
648 | 0 | _fill_partition_values, |
649 | 0 | &_fill_partition_value_is_null)); |
650 | 0 | } |
651 | | |
652 | 32 | _all_required_col_names = ctx->column_names; |
653 | | |
654 | | // Create column IDs from field descriptor |
655 | 32 | auto column_id_result = |
656 | 32 | _create_column_ids(field_desc, ctx->tuple_descriptor, ctx->table_info_node); |
657 | 32 | ctx->column_ids = std::move(column_id_result.column_ids); |
658 | 32 | ctx->filter_column_ids = std::move(column_id_result.filter_column_ids); |
659 | | |
660 | | // Build field_id -> block_column_name mapping for equality delete filtering. |
661 | | // This was previously done in init_reader() column matching (pre-CRTP refactoring). |
662 | 47 | for (const auto* slot : ctx->tuple_descriptor->slots()) { |
663 | 47 | _id_to_block_column_name.emplace(slot->col_unique_id(), slot->col_name()); |
664 | 47 | } |
665 | | |
666 | | // Process delete files (must happen before _do_init_reader so expand col IDs are included) |
667 | 32 | RETURN_IF_ERROR(_init_row_filters()); |
668 | | |
669 | | // Add expand column IDs for equality delete and remap expand column names |
670 | | // to match master's behavior: |
671 | | // - Use field_id to find the actual file column name in Parquet schema |
672 | | // - Prefix with __equality_delete_column__ to avoid name conflicts |
673 | | // - Correctly map table_col_name → file_col_name in table_info_node |
674 | 32 | const static std::string EQ_DELETE_PRE = "__equality_delete_column__"; |
675 | 32 | bool all_file_columns_have_field_ids = true; |
676 | 32 | bool any_file_column_has_field_id = false; |
677 | 105 | for (int i = 0; i < field_desc->size(); ++i) { |
678 | 73 | const auto* field_schema = field_desc->get_column(i); |
679 | 73 | if (field_schema) { |
680 | 73 | if (field_schema->field_id < 0) { |
681 | 19 | all_file_columns_have_field_ids = false; |
682 | 19 | } |
683 | 73 | if (parquet_subtree_has_iceberg_id(*field_schema)) { |
684 | 57 | any_file_column_has_field_id = true; |
685 | 57 | } |
686 | 73 | } |
687 | 73 | } |
688 | 32 | const bool use_field_ids_for_hidden_keys = |
689 | 32 | supports_iceberg_scan_semantics_v1(&get_scan_params()) |
690 | 32 | ? any_file_column_has_field_id |
691 | 32 | : all_file_columns_have_field_ids; |
692 | 32 | const auto find_file_column_by_name = [&](const std::string& name) -> const FieldSchema* { |
693 | 0 | for (int j = 0; j < field_desc->size(); ++j) { |
694 | 0 | const auto* candidate = field_desc->get_column(j); |
695 | 0 | if (candidate != nullptr && iequal(candidate->name, name)) { |
696 | 0 | return candidate; |
697 | 0 | } |
698 | 0 | } |
699 | 0 | return nullptr; |
700 | 0 | }; |
701 | 32 | std::unordered_map<int32_t, const SlotDescriptor*> projected_slots; |
702 | 47 | for (const auto* slot : ctx->tuple_descriptor->slots()) { |
703 | 47 | projected_slots.emplace(slot->col_unique_id(), slot); |
704 | 47 | } |
705 | | // Finds a nested key inside its root struct when the query projects that root, so the key is |
706 | | // taken from the projected column instead of reading the root a second time. |
707 | 32 | const auto find_projected_root = [&](int32_t field_id, const DataTypePtr& key_type, |
708 | 32 | const ParquetEqualityFieldPath& file_path, |
709 | 32 | std::string* root_name, |
710 | 32 | std::vector<size_t>* child_indexes) -> Status { |
711 | 22 | const auto key_path = _find_schema_field_path(field_id); |
712 | 22 | if (key_path.empty()) { |
713 | 0 | return Status::OK(); |
714 | 0 | } |
715 | 22 | DORIS_CHECK(key_path.front()->__isset.id); |
716 | 22 | const auto slot = projected_slots.find(key_path.front()->id); |
717 | 22 | if (slot == projected_slots.end()) { |
718 | 11 | return Status::OK(); |
719 | 11 | } |
720 | 11 | std::vector<const schema::external::TField*> root_path; |
721 | 11 | if (!_find_schema_field_path_in_root(_current_schema_root(), key_path.front()->id, |
722 | 11 | &root_path)) { |
723 | 0 | return Status::OK(); |
724 | 0 | } |
725 | 11 | DORIS_CHECK(root_path.size() == 1); |
726 | 11 | DataTypePtr leaf_type; |
727 | 11 | bool found = false; |
728 | 11 | RETURN_IF_ERROR(find_projected_equality_key_path(*root_path.front(), key_path, |
729 | 11 | slot->second->type(), child_indexes, |
730 | 11 | &leaf_type, &found)); |
731 | | // The projected leaf has the table's current type. A key written before a type promotion |
732 | | // (INT promoted to BIGINT) keeps its own type, so it still reads the physical root. |
733 | 11 | if (found && remove_nullable(leaf_type)->equals(*remove_nullable(key_type)) && |
734 | 11 | projected_root_reads_physical_path(*ctx->table_info_node, slot->second->col_name(), |
735 | 10 | slot->second->type(), *child_indexes, file_path)) { |
736 | 9 | *root_name = slot->second->col_name(); |
737 | 9 | } |
738 | 11 | return Status::OK(); |
739 | 11 | }; |
740 | | // Physical root struct -> the carrier that reads it for every hidden key under that root. |
741 | 32 | std::unordered_map<std::string, std::string> physical_root_carriers; |
742 | | |
743 | | // Rebuild _expand_col_names with proper file-column-based names |
744 | 32 | std::vector<std::string> new_expand_col_names; |
745 | 32 | DORIS_CHECK(_expand_col_names.size() == _expand_col_field_ids.size()); |
746 | 32 | DORIS_CHECK(_expand_col_names.size() == _expand_columns.size()); |
747 | 64 | for (size_t i = 0; i < _expand_col_names.size(); ++i) { |
748 | 32 | const auto& old_name = _expand_col_names[i]; |
749 | 32 | const int32_t field_id = _expand_col_field_ids[i]; |
750 | | |
751 | 32 | const FieldSchema* file_column = nullptr; |
752 | 32 | ParquetEqualityFieldPath file_path; |
753 | 32 | bool complete_file_path = false; |
754 | 32 | if (use_field_ids_for_hidden_keys) { |
755 | 25 | complete_file_path = |
756 | 25 | find_parquet_equality_field_path_by_id(field_desc, field_id, &file_path); |
757 | 25 | if (!complete_file_path && supports_iceberg_scan_semantics_v2(&get_scan_params())) { |
758 | 6 | const auto table_path = _find_schema_field_path(field_id); |
759 | 6 | if (!table_path.empty()) { |
760 | 6 | complete_file_path = find_parquet_equality_field_prefix_by_id_path( |
761 | 6 | field_desc, table_path, &file_path); |
762 | 6 | } |
763 | 6 | } |
764 | 25 | if (!file_path.fields.empty()) { |
765 | 20 | file_column = file_path.fields.front(); |
766 | 20 | } |
767 | 25 | } else { |
768 | 7 | const auto table_path = _find_schema_field_path(field_id); |
769 | 7 | if (!table_path.empty()) { |
770 | 7 | complete_file_path = find_parquet_equality_field_prefix_by_name_path( |
771 | 7 | field_desc, table_path, old_name, &file_path); |
772 | 7 | if (!file_path.fields.empty()) { |
773 | 6 | file_column = file_path.fields.front(); |
774 | 6 | } |
775 | 7 | } else { |
776 | 0 | file_column = find_file_column_by_name(old_name); |
777 | 0 | complete_file_path = file_column != nullptr; |
778 | 0 | } |
779 | 7 | } |
780 | | |
781 | 32 | std::string leaf_name = old_name; |
782 | 32 | if (!file_path.fields.empty()) { |
783 | 26 | leaf_name = file_path.fields.back()->name; |
784 | 26 | } else if (file_column != nullptr) { |
785 | 0 | leaf_name = file_column->name; |
786 | 0 | } |
787 | 32 | const std::string file_col_name = file_column == nullptr ? old_name : file_column->name; |
788 | 32 | std::string table_col_name = EQ_DELETE_PRE + std::to_string(field_id) + "_" + leaf_name; |
789 | | |
790 | | // Update _id_to_block_column_name |
791 | 32 | if (field_id >= 0) { |
792 | 32 | _id_to_block_column_name[field_id] = table_col_name; |
793 | 32 | } |
794 | | |
795 | | // Update _expand_columns name |
796 | 32 | _expand_columns[i].name = table_col_name; |
797 | | |
798 | 32 | if (file_column == nullptr) { |
799 | 6 | RETURN_IF_ERROR(_register_missing_equality_delete_column(field_id, table_col_name, |
800 | 6 | _expand_columns[i].type)); |
801 | | // The old data file predates this equality key. Its logical initial default is appended |
802 | | // after the read; do not advertise it to Parquet as a physical child. |
803 | 6 | new_expand_col_names.push_back(table_col_name); |
804 | 6 | continue; |
805 | 6 | } |
806 | | |
807 | 26 | new_expand_col_names.push_back(table_col_name); |
808 | | |
809 | | // A hidden nested key is read through its containing top-level struct. V1 column IDs are |
810 | | // pre-order ranges, so include the complete subtree before extracting the primitive leaf. |
811 | | // A projected root that serves as the key's source then reads the key's leaf as well. |
812 | 26 | for (uint64_t column_id = file_column->get_column_id(); |
813 | 86 | column_id <= file_column->get_max_column_id(); ++column_id) { |
814 | 60 | ctx->column_ids.insert(column_id); |
815 | 60 | } |
816 | | |
817 | 26 | if (!complete_file_path || !file_path.child_indexes.empty()) { |
818 | | // Parquet reads each table column through its own column reader. Without the two |
819 | | // branches below, a projected root and a carrier, or two carriers of keys under one |
820 | | // root, all named the same file column, and only one of them was read. Example: |
821 | | // `SELECT id, payload` with an equality delete on payload.existing returned an empty |
822 | | // payload, or no rows at all when payload was listed first. |
823 | 23 | std::string projected_root_name; |
824 | 23 | std::vector<size_t> projected_child_indexes; |
825 | 23 | if (complete_file_path) { |
826 | 22 | RETURN_IF_ERROR(find_projected_root(field_id, _expand_columns[i].type, file_path, |
827 | 22 | &projected_root_name, |
828 | 22 | &projected_child_indexes)); |
829 | 22 | } |
830 | 23 | if (!projected_root_name.empty()) { |
831 | | // The query projects payload and the file has payload.existing: extract the key |
832 | | // from the projected payload column. |
833 | 9 | _nested_equality_delete_columns.push_back({ |
834 | 9 | .field_id = field_id, |
835 | 9 | .block_name = table_col_name, |
836 | 9 | .leaf_type = _expand_columns[i].type, |
837 | 9 | .child_indexes = std::move(projected_child_indexes), |
838 | 9 | .missing_value = nullptr, |
839 | 9 | .source_block_name = std::move(projected_root_name), |
840 | 9 | }); |
841 | 9 | continue; |
842 | 9 | } |
843 | 14 | ColumnPtr missing_value; |
844 | 14 | if (!complete_file_path) { |
845 | 1 | RETURN_IF_ERROR(_create_missing_equality_delete_value( |
846 | 1 | field_id, _expand_columns[i].type, file_path.fields.size(), |
847 | 1 | &missing_value)); |
848 | 1 | } |
849 | 14 | const auto root_carrier = physical_root_carriers.find(file_column->name); |
850 | 14 | if (root_carrier != physical_root_carriers.end()) { |
851 | | // Keys payload.a and payload.b: the carrier of payload.a reads payload, and |
852 | | // payload.b is extracted from that carrier's column. |
853 | 2 | _nested_equality_delete_columns.push_back({ |
854 | 2 | .field_id = field_id, |
855 | 2 | .block_name = table_col_name, |
856 | 2 | .leaf_type = _expand_columns[i].type, |
857 | 2 | .child_indexes = file_path.child_indexes, |
858 | 2 | .missing_value = std::move(missing_value), |
859 | 2 | .source_block_name = root_carrier->second, |
860 | 2 | }); |
861 | 2 | continue; |
862 | 2 | } |
863 | 12 | physical_root_carriers.emplace(file_column->name, table_col_name); |
864 | 12 | _nested_equality_delete_columns.push_back({ |
865 | 12 | .field_id = field_id, |
866 | 12 | .block_name = table_col_name, |
867 | 12 | .leaf_type = _expand_columns[i].type, |
868 | 12 | .child_indexes = file_path.child_indexes, |
869 | 12 | .missing_value = std::move(missing_value), |
870 | 12 | }); |
871 | 12 | _expand_columns[i].type = make_nullable(file_column->data_type); |
872 | 12 | _expand_columns[i].column = _expand_columns[i].type->create_column(); |
873 | 12 | } |
874 | | |
875 | | // Register in table_info_node: table_col_name → file_col_name |
876 | 15 | ctx->column_names.push_back(table_col_name); |
877 | 15 | ctx->table_info_node->add_children(table_col_name, file_col_name, |
878 | 15 | TableSchemaChangeHelper::ConstNode::get_instance()); |
879 | 15 | } |
880 | 32 | _expand_col_names = std::move(new_expand_col_names); |
881 | | |
882 | | // Enable group filtering for Iceberg |
883 | 32 | _filter_groups = true; |
884 | | |
885 | 32 | return Status::OK(); |
886 | 32 | } |
887 | | |
888 | | // ============================================================================ |
889 | | // IcebergParquetReader: _create_column_ids |
890 | | // ============================================================================ |
891 | | ColumnIdResult IcebergParquetReader::_create_column_ids( |
892 | | const FieldDescriptor* field_desc, const TupleDescriptor* tuple_descriptor, |
893 | 39 | const std::shared_ptr<TableSchemaChangeHelper::Node>& table_info_node) { |
894 | 39 | auto* mutable_field_desc = const_cast<FieldDescriptor*>(field_desc); |
895 | 39 | mutable_field_desc->assign_ids(); |
896 | | |
897 | 39 | std::unordered_map<int, const FieldSchema*> iceberg_id_to_field_schema_map; |
898 | 162 | for (int i = 0; i < field_desc->size(); ++i) { |
899 | 123 | const auto* field_schema = field_desc->get_column(i); |
900 | 123 | if (!field_schema) { |
901 | 0 | continue; |
902 | 0 | } |
903 | 123 | int iceberg_id = field_schema->field_id; |
904 | 123 | iceberg_id_to_field_schema_map[iceberg_id] = field_schema; |
905 | 123 | } |
906 | | |
907 | 39 | std::set<uint64_t> column_ids; |
908 | 39 | std::set<uint64_t> filter_column_ids; |
909 | | |
910 | 39 | auto process_access_paths = [](const FieldSchema* parquet_field, |
911 | 39 | const std::vector<TColumnAccessPath>& access_paths, |
912 | 39 | std::set<uint64_t>& out_ids) { |
913 | 26 | process_nested_access_paths( |
914 | 26 | parquet_field, access_paths, out_ids, |
915 | 26 | [](const FieldSchema* field) { return field->get_column_id(); }, |
916 | 26 | [](const FieldSchema* field) { return field->get_max_column_id(); }, |
917 | 26 | IcebergParquetNestedColumnUtils::extract_nested_column_ids); |
918 | 26 | }; |
919 | | |
920 | | // The Iceberg schema-mapping root is a StructNode whose registered children are the real |
921 | | // table columns. When present, resolve each column by name through it so the column-id set |
922 | | // stays consistent with the schema-mapping decision (BY_ID or BY_NAME/name-mapping); |
923 | | // otherwise fall back to matching by Iceberg field id. |
924 | 39 | const auto* struct_node = |
925 | 39 | dynamic_cast<const TableSchemaChangeHelper::StructNode*>(table_info_node.get()); |
926 | | |
927 | 62 | for (const auto* slot : tuple_descriptor->slots()) { |
928 | 62 | const FieldSchema* field_schema = nullptr; |
929 | 62 | if (struct_node != nullptr) { |
930 | | // Synthesized/metadata slots (e.g. the TopN global row-id or the $row_id column) are |
931 | | // never registered as children, so check membership before querying: calling |
932 | | // children_column_exists() on an unregistered name DCHECK-aborts in debug builds and |
933 | | // throws std::out_of_range from .at() in release builds. |
934 | 49 | if (struct_node->get_children().contains(slot->col_name()) && |
935 | 49 | struct_node->children_column_exists(slot->col_name())) { |
936 | | // Use the physical child selected by the schema-mapping pass. This keeps partial-id |
937 | | // files in BY_NAME mode from binding a projected column through an unrelated stale |
938 | | // field id. |
939 | 46 | const auto& file_column_name = |
940 | 46 | struct_node->children_file_column_name(slot->col_name()); |
941 | 65 | for (int i = 0; i < field_desc->size(); ++i) { |
942 | 65 | const auto* candidate = field_desc->get_column(i); |
943 | 65 | if (candidate != nullptr && candidate->name == file_column_name) { |
944 | 46 | field_schema = candidate; |
945 | 46 | break; |
946 | 46 | } |
947 | 65 | } |
948 | 46 | DORIS_CHECK(field_schema != nullptr); |
949 | 46 | } |
950 | 49 | } else { |
951 | 13 | auto it = iceberg_id_to_field_schema_map.find(slot->col_unique_id()); |
952 | 13 | if (it != iceberg_id_to_field_schema_map.end()) { |
953 | 13 | field_schema = it->second; |
954 | 13 | } |
955 | 13 | } |
956 | 62 | if (field_schema == nullptr) { |
957 | 3 | continue; |
958 | 3 | } |
959 | | |
960 | 59 | if ((slot->col_type() != TYPE_STRUCT && slot->col_type() != TYPE_ARRAY && |
961 | 59 | slot->col_type() != TYPE_MAP)) { |
962 | 39 | column_ids.insert(field_schema->column_id); |
963 | 39 | if (slot->is_predicate()) { |
964 | 0 | filter_column_ids.insert(field_schema->column_id); |
965 | 0 | } |
966 | 39 | continue; |
967 | 39 | } |
968 | | |
969 | 20 | const auto& all_access_paths = slot->all_access_paths(); |
970 | 20 | process_access_paths(field_schema, all_access_paths, column_ids); |
971 | | |
972 | 20 | const auto& predicate_access_paths = slot->predicate_access_paths(); |
973 | 20 | if (!predicate_access_paths.empty()) { |
974 | 6 | process_access_paths(field_schema, predicate_access_paths, filter_column_ids); |
975 | 6 | } |
976 | 20 | } |
977 | 39 | return {std::move(column_ids), std::move(filter_column_ids)}; |
978 | 39 | } |
979 | | |
980 | | // ============================================================================ |
981 | | // IcebergParquetReader: _read_position_delete_file |
982 | | // ============================================================================ |
983 | | Status IcebergParquetReader::_read_position_delete_file(const TFileRangeDesc* delete_range, |
984 | 2 | DeleteFile* position_delete) { |
985 | 2 | ParquetReader parquet_delete_reader(get_profile(), get_scan_params(), *delete_range, |
986 | 2 | READ_DELETE_FILE_BATCH_SIZE, &get_state()->timezone_obj(), |
987 | 2 | get_io_ctx(), get_state(), _meta_cache); |
988 | | // The delete file range has size=-1 (read whole file). We must disable |
989 | | // row group filtering before init; otherwise _do_init_reader returns EndOfFile |
990 | | // when _filter_groups && _range_size < 0. |
991 | 2 | ParquetInitContext delete_ctx; |
992 | 2 | delete_ctx.filter_groups = false; |
993 | 2 | delete_ctx.column_names = delete_file_col_names; |
994 | 2 | delete_ctx.col_name_to_block_idx = |
995 | 2 | const_cast<std::unordered_map<std::string, uint32_t>*>(&DELETE_COL_NAME_TO_BLOCK_IDX); |
996 | 2 | RETURN_IF_ERROR(parquet_delete_reader.init_reader(&delete_ctx)); |
997 | | |
998 | 0 | const tparquet::FileMetaData* meta_data = parquet_delete_reader.get_meta_data(); |
999 | 0 | bool dictionary_coded = true; |
1000 | 0 | for (const auto& row_group : meta_data->row_groups) { |
1001 | 0 | const auto& column_chunk = row_group.columns[ICEBERG_FILE_PATH_INDEX]; |
1002 | 0 | if (!(column_chunk.__isset.meta_data && has_dict_page(column_chunk.meta_data))) { |
1003 | 0 | dictionary_coded = false; |
1004 | 0 | break; |
1005 | 0 | } |
1006 | 0 | } |
1007 | 0 | DataTypePtr data_type_file_path = make_nullable(std::make_shared<DataTypeString>()); |
1008 | 0 | DataTypePtr data_type_pos = make_nullable(std::make_shared<DataTypeInt64>()); |
1009 | 0 | bool eof = false; |
1010 | 0 | while (!eof) { |
1011 | 0 | Block block = { |
1012 | 0 | dictionary_coded |
1013 | 0 | ? ColumnWithTypeAndName {ColumnNullable::create(ColumnDictI32::create(), |
1014 | 0 | ColumnUInt8::create()), |
1015 | 0 | data_type_file_path, ICEBERG_FILE_PATH} |
1016 | 0 | : ColumnWithTypeAndName {data_type_file_path, ICEBERG_FILE_PATH}, |
1017 | |
|
1018 | 0 | {data_type_pos, ICEBERG_ROW_POS}}; |
1019 | 0 | size_t read_rows = 0; |
1020 | 0 | RETURN_IF_ERROR(parquet_delete_reader.get_next_block(&block, &read_rows, &eof)); |
1021 | | |
1022 | 0 | if (read_rows <= 0) { |
1023 | 0 | break; |
1024 | 0 | } |
1025 | 0 | RETURN_IF_ERROR(_gen_position_delete_file_range(block, position_delete, read_rows, |
1026 | 0 | dictionary_coded)); |
1027 | 0 | } |
1028 | 0 | return Status::OK(); |
1029 | 0 | }; |
1030 | | |
1031 | | // ============================================================================ |
1032 | | // IcebergOrcReader: on_before_init_reader (ORC-specific schema matching) |
1033 | | // ============================================================================ |
1034 | | // This format-specific setup mirrors the existing reader initialization sequence. |
1035 | | // NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size) |
1036 | 20 | Status IcebergOrcReader::on_before_init_reader(ReaderInitContext* ctx) { |
1037 | 20 | _column_descs = ctx->column_descs; |
1038 | 20 | _fill_col_name_to_block_idx = ctx->col_name_to_block_idx; |
1039 | 20 | _file_format = Fileformat::ORC; |
1040 | | |
1041 | | // Get ORC file type first (available because _create_file_reader() already ran) |
1042 | 20 | const orc::Type* orc_type_ptr = nullptr; |
1043 | 20 | RETURN_IF_ERROR(this->get_file_type(&orc_type_ptr)); |
1044 | | |
1045 | | // Build table_info_node by field_id or name matching. |
1046 | | // This must happen BEFORE column classification so we can use children_column_exists |
1047 | | // to check if a column exists in the file (by field ID, not name). |
1048 | 20 | if (!get_scan_params().__isset.history_schema_info || |
1049 | 20 | get_scan_params().history_schema_info.empty()) [[unlikely]] { |
1050 | 1 | RETURN_IF_ERROR(BuildTableInfoUtil::by_orc_name(ctx->tuple_descriptor, orc_type_ptr, |
1051 | 1 | ctx->table_info_node)); |
1052 | 19 | } else { |
1053 | 19 | RETURN_IF_ERROR(BuildTableInfoUtil::by_orc_field_id_with_name_mapping( |
1054 | 19 | get_scan_params().history_schema_info.front().root_field, orc_type_ptr, |
1055 | 19 | ICEBERG_ORC_ATTRIBUTE, ctx->table_info_node, |
1056 | 19 | supports_iceberg_scan_semantics_v1(&get_scan_params()))); |
1057 | 19 | } |
1058 | | |
1059 | 20 | std::unordered_set<std::string> partition_col_names; |
1060 | 20 | if (ctx->range->__isset.columns_from_path_keys) { |
1061 | 0 | partition_col_names.insert(ctx->range->columns_from_path_keys.begin(), |
1062 | 0 | ctx->range->columns_from_path_keys.end()); |
1063 | 0 | } |
1064 | | |
1065 | | // Single pass: classify columns, detect $row_id, handle partition fallback. |
1066 | 20 | bool has_partition_from_path = false; |
1067 | 27 | for (const auto& desc : *ctx->column_descs) { |
1068 | 27 | if (desc.category == ColumnCategory::SYNTHESIZED) { |
1069 | 0 | if (desc.name == BeConsts::ICEBERG_ROWID_COL) { |
1070 | 0 | this->register_synthesized_column_handler( |
1071 | 0 | BeConsts::ICEBERG_ROWID_COL, [this](Block* block, size_t rows) -> Status { |
1072 | 0 | return _fill_iceberg_row_id(block, rows); |
1073 | 0 | }); |
1074 | 0 | continue; |
1075 | 0 | } else if (desc.name.starts_with(BeConsts::GLOBAL_ROWID_COL)) { |
1076 | 0 | auto topn_row_id_column_iter = _create_topn_row_id_column_iterator(); |
1077 | 0 | this->register_synthesized_column_handler( |
1078 | 0 | desc.name, |
1079 | 0 | [iter = std::move(topn_row_id_column_iter), this, &desc]( |
1080 | 0 | Block* block, size_t rows) -> Status { |
1081 | 0 | return fill_topn_row_id(iter, desc.name, block, rows); |
1082 | 0 | }); |
1083 | 0 | continue; |
1084 | 0 | } |
1085 | 27 | } else if (desc.category == ColumnCategory::PARTITION_KEY) { |
1086 | 0 | bool has_partition_value = partition_col_names.contains(desc.name); |
1087 | 0 | bool exists_in_file = ctx->table_info_node->children_column_exists(desc.name); |
1088 | 0 | if (!has_partition_value || exists_in_file) { |
1089 | 0 | ctx->column_names.push_back(desc.name); |
1090 | 0 | continue; |
1091 | 0 | } |
1092 | 0 | has_partition_from_path = true; |
1093 | 27 | } else if (desc.category == ColumnCategory::REGULAR) { |
1094 | 27 | ctx->column_names.push_back(desc.name); |
1095 | 27 | } else if (desc.category == ColumnCategory::GENERATED) { |
1096 | 0 | _init_row_lineage_columns(); |
1097 | 0 | if (desc.name == ROW_LINEAGE_ROW_ID) { |
1098 | 0 | ctx->column_names.push_back(desc.name); |
1099 | 0 | this->register_generated_column_handler( |
1100 | 0 | ROW_LINEAGE_ROW_ID, [this](Block* block, size_t rows) -> Status { |
1101 | 0 | return _fill_row_lineage_row_id(block, rows); |
1102 | 0 | }); |
1103 | 0 | continue; |
1104 | 0 | } else if (desc.name == ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER) { |
1105 | 0 | ctx->column_names.push_back(desc.name); |
1106 | 0 | this->register_generated_column_handler( |
1107 | 0 | ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER, |
1108 | 0 | [this](Block* block, size_t rows) -> Status { |
1109 | 0 | return _fill_row_lineage_last_updated_sequence_number(block, rows); |
1110 | 0 | }); |
1111 | 0 | continue; |
1112 | 0 | } |
1113 | 0 | } |
1114 | 27 | } |
1115 | | |
1116 | 20 | if (has_partition_from_path) { |
1117 | 0 | RETURN_IF_ERROR(_extract_partition_values(*ctx->range, ctx->tuple_descriptor, |
1118 | 0 | _fill_partition_values, |
1119 | 0 | &_fill_partition_value_is_null)); |
1120 | 0 | } |
1121 | | |
1122 | 20 | _all_required_col_names = ctx->column_names; |
1123 | | |
1124 | | // Create column IDs from ORC type |
1125 | 20 | auto column_id_result = |
1126 | 20 | _create_column_ids(orc_type_ptr, ctx->tuple_descriptor, ctx->table_info_node); |
1127 | 20 | ctx->column_ids = std::move(column_id_result.column_ids); |
1128 | 20 | ctx->filter_column_ids = std::move(column_id_result.filter_column_ids); |
1129 | | |
1130 | | // Build field_id -> block_column_name mapping for equality delete filtering. |
1131 | 27 | for (const auto* slot : ctx->tuple_descriptor->slots()) { |
1132 | 27 | _id_to_block_column_name.emplace(slot->col_unique_id(), slot->col_name()); |
1133 | 27 | } |
1134 | | |
1135 | | // Process delete files (must happen before _do_init_reader so expand col IDs are included) |
1136 | 20 | RETURN_IF_ERROR(_init_row_filters()); |
1137 | | |
1138 | | // Add expand column IDs for equality delete and remap expand column names |
1139 | | // (matching master's behavior with __equality_delete_column__ prefix) |
1140 | 20 | const static std::string EQ_DELETE_PRE = "__equality_delete_column__"; |
1141 | 20 | bool all_file_columns_have_field_ids = true; |
1142 | 66 | for (uint64_t i = 0; i < orc_type_ptr->getSubtypeCount(); ++i) { |
1143 | 46 | const orc::Type* sub_type = orc_type_ptr->getSubtype(i); |
1144 | 46 | if (!sub_type->hasAttributeKey(ICEBERG_ORC_ATTRIBUTE)) { |
1145 | 16 | all_file_columns_have_field_ids = false; |
1146 | 16 | } |
1147 | 46 | } |
1148 | 20 | const bool use_field_ids_for_hidden_keys = |
1149 | 20 | supports_iceberg_scan_semantics_v1(&get_scan_params()) |
1150 | 20 | ? orc_subtree_has_iceberg_id(orc_type_ptr, ICEBERG_ORC_ATTRIBUTE) |
1151 | 20 | : all_file_columns_have_field_ids; |
1152 | 20 | const auto find_file_column_by_name = [&](const std::string& name) -> const orc::Type* { |
1153 | 0 | for (uint64_t j = 0; j < orc_type_ptr->getSubtypeCount(); ++j) { |
1154 | 0 | if (iequal(orc_type_ptr->getFieldName(j), name)) { |
1155 | 0 | return orc_type_ptr->getSubtype(j); |
1156 | 0 | } |
1157 | 0 | } |
1158 | 0 | return nullptr; |
1159 | 0 | }; |
1160 | | |
1161 | 20 | std::vector<std::string> new_expand_col_names; |
1162 | 20 | DORIS_CHECK(_expand_col_names.size() == _expand_col_field_ids.size()); |
1163 | 20 | DORIS_CHECK(_expand_col_names.size() == _expand_columns.size()); |
1164 | 38 | for (size_t i = 0; i < _expand_col_names.size(); ++i) { |
1165 | 18 | const auto& old_name = _expand_col_names[i]; |
1166 | 18 | const int32_t field_id = _expand_col_field_ids[i]; |
1167 | | |
1168 | 18 | const orc::Type* file_column = nullptr; |
1169 | 18 | OrcEqualityFieldPath file_path; |
1170 | 18 | bool complete_file_path = false; |
1171 | 18 | if (use_field_ids_for_hidden_keys) { |
1172 | 12 | complete_file_path = |
1173 | 12 | find_orc_equality_field_path_by_id(orc_type_ptr, field_id, &file_path); |
1174 | 12 | if (!complete_file_path && supports_iceberg_scan_semantics_v2(&get_scan_params())) { |
1175 | 6 | const auto table_path = _find_schema_field_path(field_id); |
1176 | 6 | if (!table_path.empty()) { |
1177 | 6 | complete_file_path = find_orc_equality_field_prefix_by_id_path( |
1178 | 6 | orc_type_ptr, table_path, &file_path); |
1179 | 6 | } |
1180 | 6 | } |
1181 | 12 | if (!file_path.fields.empty()) { |
1182 | 6 | file_column = file_path.fields.front(); |
1183 | 6 | } |
1184 | 12 | } else { |
1185 | 6 | const auto table_path = _find_schema_field_path(field_id); |
1186 | 6 | if (!table_path.empty()) { |
1187 | 6 | complete_file_path = find_orc_equality_field_prefix_by_name_path( |
1188 | 6 | orc_type_ptr, table_path, old_name, &file_path); |
1189 | 6 | if (!file_path.fields.empty()) { |
1190 | 5 | file_column = file_path.fields.front(); |
1191 | 5 | } |
1192 | 6 | } else { |
1193 | 0 | file_column = find_file_column_by_name(old_name); |
1194 | 0 | complete_file_path = file_column != nullptr; |
1195 | 0 | } |
1196 | 6 | } |
1197 | | |
1198 | 18 | std::string file_col_name = old_name; |
1199 | 18 | std::string leaf_name = old_name; |
1200 | 18 | if (!file_path.fields.empty()) { |
1201 | 11 | file_col_name = file_path.names.front(); |
1202 | 11 | leaf_name = file_path.names.back(); |
1203 | 11 | } else if (file_column != nullptr) { |
1204 | 0 | for (uint64_t j = 0; j < orc_type_ptr->getSubtypeCount(); ++j) { |
1205 | 0 | if (orc_type_ptr->getSubtype(j) == file_column) { |
1206 | 0 | file_col_name = orc_type_ptr->getFieldName(j); |
1207 | 0 | leaf_name = file_col_name; |
1208 | 0 | break; |
1209 | 0 | } |
1210 | 0 | } |
1211 | 0 | } |
1212 | 18 | std::string table_col_name = EQ_DELETE_PRE + std::to_string(field_id) + "_" + leaf_name; |
1213 | | |
1214 | 18 | if (field_id >= 0) { |
1215 | 18 | _id_to_block_column_name[field_id] = table_col_name; |
1216 | 18 | } |
1217 | 18 | _expand_columns[i].name = table_col_name; |
1218 | 18 | if (file_column == nullptr) { |
1219 | 7 | RETURN_IF_ERROR(_register_missing_equality_delete_column(field_id, table_col_name, |
1220 | 7 | _expand_columns[i].type)); |
1221 | | // The old data file predates this equality key. Its logical initial default is appended |
1222 | | // after the read. Adding it to column_names/table_info_node would mark it as an |
1223 | | // existing ORC child and make OrcReader read a column that is not present in the file. |
1224 | 7 | new_expand_col_names.push_back(table_col_name); |
1225 | 7 | continue; |
1226 | 7 | } |
1227 | 11 | new_expand_col_names.push_back(table_col_name); |
1228 | | |
1229 | 11 | if (!complete_file_path) { |
1230 | 1 | ColumnPtr missing_value; |
1231 | 1 | RETURN_IF_ERROR(_create_missing_equality_delete_value( |
1232 | 1 | field_id, _expand_columns[i].type, file_path.fields.size(), &missing_value)); |
1233 | 1 | _nested_equality_delete_columns.push_back({ |
1234 | 1 | .field_id = field_id, |
1235 | 1 | .block_name = table_col_name, |
1236 | 1 | .leaf_type = _expand_columns[i].type, |
1237 | 1 | .child_indexes = file_path.child_indexes, |
1238 | 1 | .missing_value = std::move(missing_value), |
1239 | 1 | }); |
1240 | 1 | _expand_columns[i].type = make_nullable(convert_to_doris_type(file_column)); |
1241 | 1 | _expand_columns[i].column = _expand_columns[i].type->create_column(); |
1242 | 10 | } else if (!file_path.child_indexes.empty()) { |
1243 | 8 | _nested_equality_delete_columns.push_back({ |
1244 | 8 | .field_id = field_id, |
1245 | 8 | .block_name = table_col_name, |
1246 | 8 | .leaf_type = _expand_columns[i].type, |
1247 | 8 | .child_indexes = file_path.child_indexes, |
1248 | 8 | .missing_value = nullptr, |
1249 | 8 | }); |
1250 | 8 | _expand_columns[i].type = make_nullable(convert_to_doris_type(file_column)); |
1251 | 8 | _expand_columns[i].column = _expand_columns[i].type->create_column(); |
1252 | 8 | } |
1253 | | |
1254 | 11 | for (uint64_t column_id = file_column->getColumnId(); |
1255 | 31 | column_id <= file_column->getMaximumColumnId(); ++column_id) { |
1256 | 20 | ctx->column_ids.insert(column_id); |
1257 | 20 | } |
1258 | | |
1259 | 11 | ctx->column_names.push_back(table_col_name); |
1260 | 11 | ctx->table_info_node->add_children(table_col_name, file_col_name, |
1261 | 11 | TableSchemaChangeHelper::ConstNode::get_instance()); |
1262 | 11 | } |
1263 | 20 | _expand_col_names = std::move(new_expand_col_names); |
1264 | | |
1265 | 20 | return Status::OK(); |
1266 | 20 | } |
1267 | | |
1268 | | // ============================================================================ |
1269 | | // IcebergOrcReader: _create_column_ids |
1270 | | // ============================================================================ |
1271 | | ColumnIdResult IcebergOrcReader::_create_column_ids( |
1272 | | const orc::Type* orc_type, const TupleDescriptor* tuple_descriptor, |
1273 | 27 | const std::shared_ptr<TableSchemaChangeHelper::Node>& table_info_node) { |
1274 | 27 | std::unordered_map<int, const orc::Type*> iceberg_id_to_orc_type_map; |
1275 | 123 | for (uint64_t i = 0; i < orc_type->getSubtypeCount(); ++i) { |
1276 | 96 | const auto* orc_sub_type = orc_type->getSubtype(i); |
1277 | 96 | if (!orc_sub_type) { |
1278 | 0 | continue; |
1279 | 0 | } |
1280 | 96 | if (!orc_sub_type->hasAttributeKey(ICEBERG_ORC_ATTRIBUTE)) { |
1281 | 17 | continue; |
1282 | 17 | } |
1283 | 79 | int iceberg_id = std::stoi(orc_sub_type->getAttributeValue(ICEBERG_ORC_ATTRIBUTE)); |
1284 | 79 | iceberg_id_to_orc_type_map[iceberg_id] = orc_sub_type; |
1285 | 79 | } |
1286 | | |
1287 | 27 | std::set<uint64_t> column_ids; |
1288 | 27 | std::set<uint64_t> filter_column_ids; |
1289 | | |
1290 | 27 | auto process_access_paths = [](const orc::Type* orc_field, |
1291 | 27 | const std::vector<TColumnAccessPath>& access_paths, |
1292 | 27 | std::set<uint64_t>& out_ids) { |
1293 | 19 | process_nested_access_paths( |
1294 | 19 | orc_field, access_paths, out_ids, |
1295 | 19 | [](const orc::Type* type) { return type->getColumnId(); }, |
1296 | 19 | [](const orc::Type* type) { return type->getMaximumColumnId(); }, |
1297 | 19 | IcebergOrcNestedColumnUtils::extract_nested_column_ids); |
1298 | 19 | }; |
1299 | | |
1300 | | // The Iceberg schema-mapping root is a StructNode whose registered children are the real |
1301 | | // table columns. When present, resolve each column by name through it so the column-id set |
1302 | | // stays consistent with the schema-mapping decision (BY_ID or BY_NAME/name-mapping); |
1303 | | // otherwise fall back to matching by Iceberg field id. |
1304 | 27 | const auto* struct_node = |
1305 | 27 | dynamic_cast<const TableSchemaChangeHelper::StructNode*>(table_info_node.get()); |
1306 | | |
1307 | 42 | for (const auto* slot : tuple_descriptor->slots()) { |
1308 | 42 | const orc::Type* orc_field = nullptr; |
1309 | 42 | if (struct_node != nullptr) { |
1310 | | // Synthesized/metadata slots (e.g. the TopN global row-id or the $row_id column) are |
1311 | | // never registered as children, so check membership before querying: calling |
1312 | | // children_column_exists() on an unregistered name DCHECK-aborts in debug builds and |
1313 | | // throws std::out_of_range from .at() in release builds. |
1314 | 29 | if (struct_node->get_children().contains(slot->col_name()) && |
1315 | 29 | struct_node->children_column_exists(slot->col_name())) { |
1316 | | // Select the physical child resolved by the shared schema-mapping pass. Hidden |
1317 | | // equality keys and projected columns must obey the same BY_NAME decision for |
1318 | | // partial-id ORC files. |
1319 | 27 | const auto& file_column_name = |
1320 | 27 | struct_node->children_file_column_name(slot->col_name()); |
1321 | 35 | for (uint64_t i = 0; i < orc_type->getSubtypeCount(); ++i) { |
1322 | 35 | if (orc_type->getFieldName(i) == file_column_name) { |
1323 | 27 | orc_field = orc_type->getSubtype(i); |
1324 | 27 | break; |
1325 | 27 | } |
1326 | 35 | } |
1327 | 27 | DORIS_CHECK(orc_field != nullptr); |
1328 | 27 | } |
1329 | 29 | } else { |
1330 | 13 | auto it = iceberg_id_to_orc_type_map.find(slot->col_unique_id()); |
1331 | 13 | if (it != iceberg_id_to_orc_type_map.end()) { |
1332 | 13 | orc_field = it->second; |
1333 | 13 | } |
1334 | 13 | } |
1335 | 42 | if (orc_field == nullptr) { |
1336 | 2 | continue; |
1337 | 2 | } |
1338 | | |
1339 | 40 | if ((slot->col_type() != TYPE_STRUCT && slot->col_type() != TYPE_ARRAY && |
1340 | 40 | slot->col_type() != TYPE_MAP)) { |
1341 | 27 | column_ids.insert(orc_field->getColumnId()); |
1342 | 27 | if (slot->is_predicate()) { |
1343 | 0 | filter_column_ids.insert(orc_field->getColumnId()); |
1344 | 0 | } |
1345 | 27 | continue; |
1346 | 27 | } |
1347 | | |
1348 | 13 | const auto& all_access_paths = slot->all_access_paths(); |
1349 | 13 | process_access_paths(orc_field, all_access_paths, column_ids); |
1350 | | |
1351 | 13 | const auto& predicate_access_paths = slot->predicate_access_paths(); |
1352 | 13 | if (!predicate_access_paths.empty()) { |
1353 | 6 | process_access_paths(orc_field, predicate_access_paths, filter_column_ids); |
1354 | 6 | } |
1355 | 13 | } |
1356 | | |
1357 | 27 | return {std::move(column_ids), std::move(filter_column_ids)}; |
1358 | 27 | } |
1359 | | |
1360 | | // ============================================================================ |
1361 | | // IcebergOrcReader: _read_position_delete_file |
1362 | | // ============================================================================ |
1363 | | Status IcebergOrcReader::_read_position_delete_file(const TFileRangeDesc* delete_range, |
1364 | 0 | DeleteFile* position_delete) { |
1365 | 0 | OrcReader orc_delete_reader(get_profile(), get_state(), get_scan_params(), *delete_range, |
1366 | 0 | READ_DELETE_FILE_BATCH_SIZE, get_state()->timezone(), get_io_ctx(), |
1367 | 0 | _meta_cache); |
1368 | 0 | OrcInitContext delete_ctx; |
1369 | 0 | delete_ctx.column_names = delete_file_col_names; |
1370 | 0 | delete_ctx.col_name_to_block_idx = |
1371 | 0 | const_cast<std::unordered_map<std::string, uint32_t>*>(&DELETE_COL_NAME_TO_BLOCK_IDX); |
1372 | 0 | RETURN_IF_ERROR(orc_delete_reader.init_reader(&delete_ctx)); |
1373 | | |
1374 | 0 | bool eof = false; |
1375 | 0 | DataTypePtr data_type_file_path {new DataTypeString}; |
1376 | 0 | DataTypePtr data_type_pos {new DataTypeInt64}; |
1377 | 0 | while (!eof) { |
1378 | 0 | Block block = {{data_type_file_path, ICEBERG_FILE_PATH}, {data_type_pos, ICEBERG_ROW_POS}}; |
1379 | |
|
1380 | 0 | size_t read_rows = 0; |
1381 | 0 | RETURN_IF_ERROR(orc_delete_reader.get_next_block(&block, &read_rows, &eof)); |
1382 | | |
1383 | 0 | RETURN_IF_ERROR(_gen_position_delete_file_range(block, position_delete, read_rows, false)); |
1384 | 0 | } |
1385 | 0 | return Status::OK(); |
1386 | 0 | } |
1387 | | |
1388 | | } // namespace doris |