be/src/format/table/iceberg_default_value.h
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 | | #pragma once |
19 | | |
20 | | #include <gen_cpp/ExternalTableSchema_types.h> |
21 | | #include <rapidjson/document.h> |
22 | | #include <rapidjson/stringbuffer.h> |
23 | | #include <rapidjson/writer.h> |
24 | | |
25 | | #include <cstddef> |
26 | | #include <deque> |
27 | | #include <limits> |
28 | | #include <string> |
29 | | #include <string_view> |
30 | | #include <unordered_map> |
31 | | #include <utility> |
32 | | |
33 | | #include "common/status.h" |
34 | | #include "core/assert_cast.h" |
35 | | #include "core/column/column.h" |
36 | | #include "core/data_type/data_type.h" |
37 | | #include "core/data_type/data_type_array.h" |
38 | | #include "core/data_type/data_type_map.h" |
39 | | #include "core/data_type/data_type_nullable.h" |
40 | | #include "core/data_type/data_type_struct.h" |
41 | | #include "core/data_type/primitive_type.h" |
42 | | #include "core/field.h" |
43 | | #include "util/string_util.h" |
44 | | #include "util/url_coding.h" |
45 | | |
46 | | namespace doris::iceberg { |
47 | | |
48 | | namespace detail { |
49 | | |
50 | | inline bool parse_non_finite_default(doris::PrimitiveType type, std::string_view value, |
51 | 59 | Field* result) { |
52 | 59 | DORIS_CHECK(result != nullptr); |
53 | 59 | if (type != TYPE_FLOAT && type != TYPE_DOUBLE) { |
54 | 52 | return false; |
55 | 52 | } |
56 | 7 | double parsed; |
57 | 7 | if (value == "NaN") { |
58 | 3 | parsed = std::numeric_limits<double>::quiet_NaN(); |
59 | 4 | } else if (value == "Infinity") { |
60 | 2 | parsed = std::numeric_limits<double>::infinity(); |
61 | 2 | } else if (value == "-Infinity") { |
62 | 2 | parsed = -std::numeric_limits<double>::infinity(); |
63 | 2 | } else { |
64 | 0 | return false; |
65 | 0 | } |
66 | | // Iceberg serializes non-finite defaults as strings, which generic Doris numeric parsers reject. |
67 | 7 | *result = type == TYPE_FLOAT ? Field::create_field<TYPE_FLOAT>(static_cast<float>(parsed)) |
68 | 7 | : Field::create_field<TYPE_DOUBLE>(parsed); |
69 | 7 | return true; |
70 | 7 | } |
71 | | |
72 | 18 | inline const schema::external::TField* get_field_ptr(const schema::external::TFieldPtr& field_ptr) { |
73 | 18 | if (!field_ptr.__isset.field_ptr || field_ptr.field_ptr == nullptr) { |
74 | 0 | return nullptr; |
75 | 0 | } |
76 | 18 | return field_ptr.field_ptr.get(); |
77 | 18 | } |
78 | | |
79 | | inline const schema::external::TField* find_struct_child( |
80 | 8 | const schema::external::TStructField& struct_field, const std::string& name) { |
81 | 8 | if (!struct_field.__isset.fields) { |
82 | 0 | return nullptr; |
83 | 0 | } |
84 | 13 | for (const auto& child_ptr : struct_field.fields) { |
85 | 13 | const auto* child = get_field_ptr(child_ptr); |
86 | 13 | if (child != nullptr && child->__isset.name && iequal(child->name, name)) { |
87 | 6 | return child; |
88 | 6 | } |
89 | 13 | } |
90 | 2 | for (const auto& child_ptr : struct_field.fields) { |
91 | 2 | const auto* child = get_field_ptr(child_ptr); |
92 | 2 | if (child == nullptr || !child->__isset.name_mapping) { |
93 | 0 | continue; |
94 | 0 | } |
95 | 2 | for (const auto& alias : child->name_mapping) { |
96 | 2 | if (iequal(alias, name)) { |
97 | 2 | return child; |
98 | 2 | } |
99 | 2 | } |
100 | 2 | } |
101 | 0 | return nullptr; |
102 | 2 | } |
103 | | |
104 | 76 | inline int hex_value(char c) { |
105 | 76 | if (c >= '0' && c <= '9') { |
106 | 52 | return c - '0'; |
107 | 52 | } |
108 | 24 | if (c >= 'a' && c <= 'f') { |
109 | 17 | return c - 'a' + 10; |
110 | 17 | } |
111 | 7 | if (c >= 'A' && c <= 'F') { |
112 | 6 | return c - 'A' + 10; |
113 | 6 | } |
114 | 1 | return -1; |
115 | 7 | } |
116 | | |
117 | 5 | inline Status decode_hex(std::string_view encoded, std::string* decoded) { |
118 | 5 | DORIS_CHECK(decoded != nullptr); |
119 | 5 | if ((encoded.size() & 1U) != 0) { |
120 | 1 | return Status::InvalidArgument("Invalid odd-length Iceberg binary default"); |
121 | 1 | } |
122 | 4 | decoded->resize(encoded.size() / 2); |
123 | 41 | for (size_t index = 0; index < encoded.size(); index += 2) { |
124 | 38 | const int high = hex_value(encoded[index]); |
125 | 38 | const int low = hex_value(encoded[index + 1]); |
126 | 38 | if (high < 0 || low < 0) { |
127 | 1 | return Status::InvalidArgument("Invalid hexadecimal Iceberg binary default"); |
128 | 1 | } |
129 | 37 | (*decoded)[index / 2] = static_cast<char>((high << 4) | low); |
130 | 37 | } |
131 | 3 | return Status::OK(); |
132 | 4 | } |
133 | | |
134 | 2 | inline Status decode_json_binary(std::string_view encoded, std::string* decoded) { |
135 | 2 | DORIS_CHECK(decoded != nullptr); |
136 | 2 | const bool is_uuid = encoded.size() == 36 && encoded[8] == '-' && encoded[13] == '-' && |
137 | 2 | encoded[18] == '-' && encoded[23] == '-'; |
138 | 2 | if (is_uuid) { |
139 | 1 | std::string uuid_hex; |
140 | 1 | uuid_hex.reserve(32); |
141 | 37 | for (size_t index = 0; index < encoded.size(); ++index) { |
142 | 36 | if (index != 8 && index != 13 && index != 18 && index != 23) { |
143 | 32 | uuid_hex.push_back(encoded[index]); |
144 | 32 | } |
145 | 36 | } |
146 | 1 | return decode_hex(uuid_hex, decoded); |
147 | 1 | } |
148 | 1 | return decode_hex(encoded, decoded); |
149 | 2 | } |
150 | | |
151 | 5 | inline std::string json_scalar_text(const rapidjson::Value& value) { |
152 | 5 | if (value.IsString()) { |
153 | 3 | return {value.GetString(), value.GetStringLength()}; |
154 | 3 | } |
155 | 2 | rapidjson::StringBuffer buffer; |
156 | 2 | rapidjson::Writer<rapidjson::StringBuffer> writer(buffer); |
157 | 2 | value.Accept(writer); |
158 | 2 | return {buffer.GetString(), buffer.GetSize()}; |
159 | 5 | } |
160 | | |
161 | 7 | inline void normalize_timestamp_for_doris(doris::PrimitiveType primitive_type, std::string* value) { |
162 | 7 | if (primitive_type != TYPE_DATETIME && primitive_type != TYPE_DATETIMEV2 && |
163 | 7 | primitive_type != TYPE_TIMESTAMPTZ) { |
164 | 3 | return; |
165 | 3 | } |
166 | 4 | if (const size_t separator = value->find('T'); separator != std::string::npos) { |
167 | 3 | (*value)[separator] = ' '; |
168 | 3 | } |
169 | 4 | if (primitive_type == TYPE_TIMESTAMPTZ) { |
170 | 1 | return; |
171 | 1 | } |
172 | 3 | if (value->ends_with('Z')) { |
173 | 1 | value->pop_back(); |
174 | 1 | return; |
175 | 1 | } |
176 | 2 | const size_t time_start = value->find(' '); |
177 | 2 | if (time_start == std::string::npos) { |
178 | 1 | return; |
179 | 1 | } |
180 | 1 | const size_t offset = value->find_first_of("+-", time_start + 1); |
181 | 1 | if (offset != std::string::npos) { |
182 | 1 | value->erase(offset); |
183 | 1 | } |
184 | 1 | } |
185 | | |
186 | | inline Status make_null_field(const schema::external::TField& field, const DataTypePtr& data_type, |
187 | 8 | Field* result) { |
188 | 8 | DORIS_CHECK(data_type != nullptr); |
189 | 8 | DORIS_CHECK(result != nullptr); |
190 | 8 | if (field.__isset.is_optional && !field.is_optional) { |
191 | 1 | return Status::InvalidArgument("Required Iceberg field '{}' has a null default", |
192 | 1 | field.name); |
193 | 1 | } |
194 | 7 | if (!data_type->is_nullable()) { |
195 | 1 | return Status::InternalError( |
196 | 1 | "Optional Iceberg field '{}' has a null default, but its Doris type '{}' is not " |
197 | 1 | "nullable", |
198 | 1 | field.name, data_type->get_name()); |
199 | 1 | } |
200 | 6 | *result = Field(); |
201 | 6 | return Status::OK(); |
202 | 7 | } |
203 | | |
204 | | inline Status build_initial_default_field(const schema::external::TField& field, |
205 | | const DataTypePtr& data_type, |
206 | | std::deque<std::string>* binary_storage, Field* result); |
207 | | |
208 | | inline Status build_json_default_field(const schema::external::TField& field, |
209 | | const DataTypePtr& data_type, |
210 | | const rapidjson::Value& json_value, |
211 | | std::deque<std::string>* binary_storage, Field* result); |
212 | | |
213 | | inline Status build_json_struct_default(const schema::external::TField& field, |
214 | | const DataTypePtr& value_type, |
215 | | const rapidjson::Value& json_value, |
216 | 5 | std::deque<std::string>* binary_storage, Field* result) { |
217 | 5 | if (!json_value.IsObject() || !field.__isset.nestedField || |
218 | 5 | !field.nestedField.__isset.struct_field || !field.nestedField.struct_field.__isset.fields) { |
219 | 0 | return Status::InvalidArgument("Invalid Iceberg struct default for field '{}'", field.name); |
220 | 0 | } |
221 | | |
222 | 5 | const auto& struct_type = assert_cast<const DataTypeStruct&>(*value_type); |
223 | 5 | Struct struct_value; |
224 | 5 | struct_value.reserve(struct_type.get_elements().size()); |
225 | 13 | for (size_t index = 0; index < struct_type.get_elements().size(); ++index) { |
226 | 8 | const auto& child_name = struct_type.get_element_name(index); |
227 | 8 | const auto* child = find_struct_child(field.nestedField.struct_field, child_name); |
228 | 8 | if (child == nullptr || !child->__isset.id) { |
229 | 0 | return Status::InvalidArgument( |
230 | 0 | "Iceberg struct default for field '{}' is missing metadata for projected " |
231 | 0 | "child '{}'", |
232 | 0 | field.name, child_name); |
233 | 0 | } |
234 | | |
235 | 8 | const std::string child_id = std::to_string(child->id); |
236 | 8 | const auto member = json_value.FindMember(child_id.c_str()); |
237 | 8 | Field child_value; |
238 | 8 | if (member == json_value.MemberEnd()) { |
239 | 6 | RETURN_IF_ERROR(build_initial_default_field(*child, struct_type.get_element(index), |
240 | 6 | binary_storage, &child_value)); |
241 | 6 | } else { |
242 | 2 | RETURN_IF_ERROR(build_json_default_field(*child, struct_type.get_element(index), |
243 | 2 | member->value, binary_storage, &child_value)); |
244 | 2 | } |
245 | 8 | struct_value.push_back(std::move(child_value)); |
246 | 8 | } |
247 | 5 | *result = Field::create_field<TYPE_STRUCT>(std::move(struct_value)); |
248 | 5 | return Status::OK(); |
249 | 5 | } |
250 | | |
251 | | // The recursive item TField describes the element schema and its field-level default metadata. It |
252 | | // cannot represent a particular list literal's length or per-position values, so the parent |
253 | | // initial-default keeps those values in Iceberg's single-value JSON array. |
254 | | inline Status build_json_array_default(const schema::external::TField& field, |
255 | | const DataTypePtr& value_type, |
256 | | const rapidjson::Value& json_value, |
257 | 1 | std::deque<std::string>* binary_storage, Field* result) { |
258 | 1 | if (!json_value.IsArray() || !field.__isset.nestedField || |
259 | 1 | !field.nestedField.__isset.array_field || |
260 | 1 | !field.nestedField.array_field.__isset.item_field) { |
261 | 0 | return Status::InvalidArgument("Invalid Iceberg list default for field '{}'", field.name); |
262 | 0 | } |
263 | 1 | const auto* element = get_field_ptr(field.nestedField.array_field.item_field); |
264 | 1 | if (element == nullptr) { |
265 | 0 | return Status::InvalidArgument( |
266 | 0 | "Iceberg list default for field '{}' has incomplete element metadata", field.name); |
267 | 0 | } |
268 | | |
269 | 1 | const auto& array_type = assert_cast<const DataTypeArray&>(*value_type); |
270 | 1 | Array array_value; |
271 | 1 | array_value.reserve(json_value.Size()); |
272 | 2 | for (const auto& json_element : json_value.GetArray()) { |
273 | 2 | Field element_value; |
274 | 2 | RETURN_IF_ERROR(build_json_default_field(*element, array_type.get_nested_type(), |
275 | 2 | json_element, binary_storage, &element_value)); |
276 | 2 | array_value.push_back(std::move(element_value)); |
277 | 2 | } |
278 | 1 | *result = Field::create_field<TYPE_ARRAY>(std::move(array_value)); |
279 | 1 | return Status::OK(); |
280 | 1 | } |
281 | | |
282 | | // The recursive key/value TFields describe entry schemas and field-level default metadata. They |
283 | | // cannot represent the number, order, or concrete values of map entries, so the parent |
284 | | // initial-default keeps the entries in Iceberg's single-value JSON key/value arrays. |
285 | | inline Status build_json_map_default(const schema::external::TField& field, |
286 | | const DataTypePtr& value_type, |
287 | | const rapidjson::Value& json_value, |
288 | 1 | std::deque<std::string>* binary_storage, Field* result) { |
289 | 1 | if (!json_value.IsObject() || !json_value.HasMember("keys") || !json_value["keys"].IsArray() || |
290 | 1 | !json_value.HasMember("values") || !json_value["values"].IsArray() || |
291 | 1 | !field.__isset.nestedField || !field.nestedField.__isset.map_field || |
292 | 1 | !field.nestedField.map_field.__isset.key_field || |
293 | 1 | !field.nestedField.map_field.__isset.value_field) { |
294 | 0 | return Status::InvalidArgument("Invalid Iceberg map default for field '{}'", field.name); |
295 | 0 | } |
296 | 1 | const auto& keys = json_value["keys"]; |
297 | 1 | const auto& values = json_value["values"]; |
298 | 1 | if (keys.Size() != values.Size()) { |
299 | 0 | return Status::InvalidArgument( |
300 | 0 | "Iceberg map default for field '{}' has {} keys but {} values", field.name, |
301 | 0 | keys.Size(), values.Size()); |
302 | 0 | } |
303 | | |
304 | 1 | const auto* key = get_field_ptr(field.nestedField.map_field.key_field); |
305 | 1 | const auto* value = get_field_ptr(field.nestedField.map_field.value_field); |
306 | 1 | if (key == nullptr || value == nullptr) { |
307 | 0 | return Status::InvalidArgument( |
308 | 0 | "Iceberg map default for field '{}' has incomplete key/value metadata", field.name); |
309 | 0 | } |
310 | | |
311 | 1 | const auto& map_type = assert_cast<const DataTypeMap&>(*value_type); |
312 | 1 | Array key_fields; |
313 | 1 | Array value_fields; |
314 | 1 | key_fields.reserve(keys.Size()); |
315 | 1 | value_fields.reserve(values.Size()); |
316 | 2 | for (rapidjson::SizeType index = 0; index < keys.Size(); ++index) { |
317 | 1 | Field key_value; |
318 | 1 | Field mapped_value; |
319 | 1 | RETURN_IF_ERROR(build_json_default_field(*key, map_type.get_key_type(), keys[index], |
320 | 1 | binary_storage, &key_value)); |
321 | 1 | RETURN_IF_ERROR(build_json_default_field(*value, map_type.get_value_type(), values[index], |
322 | 1 | binary_storage, &mapped_value)); |
323 | 1 | key_fields.push_back(std::move(key_value)); |
324 | 1 | value_fields.push_back(std::move(mapped_value)); |
325 | 1 | } |
326 | 1 | Map map_value; |
327 | 1 | map_value.push_back(Field::create_field<TYPE_ARRAY>(std::move(key_fields))); |
328 | 1 | map_value.push_back(Field::create_field<TYPE_ARRAY>(std::move(value_fields))); |
329 | 1 | *result = Field::create_field<TYPE_MAP>(std::move(map_value)); |
330 | 1 | return Status::OK(); |
331 | 1 | } |
332 | | |
333 | | inline Status build_json_scalar_default(const schema::external::TField& field, |
334 | | const DataTypePtr& value_type, |
335 | | const rapidjson::Value& json_value, |
336 | 5 | std::deque<std::string>* binary_storage, Field* result) { |
337 | 5 | const auto primitive_type = value_type->get_primitive_type(); |
338 | 5 | std::string serialized_value = json_scalar_text(json_value); |
339 | 5 | const bool binary_like = (field.__isset.initial_default_value_is_base64 && |
340 | 5 | field.initial_default_value_is_base64) || |
341 | 5 | primitive_type == TYPE_VARBINARY; |
342 | 5 | if (binary_like) { |
343 | 1 | if (!json_value.IsString()) { |
344 | 0 | return Status::InvalidArgument( |
345 | 0 | "Iceberg binary default for field '{}' is not a JSON string", field.name); |
346 | 0 | } |
347 | 1 | binary_storage->emplace_back(); |
348 | 1 | RETURN_IF_ERROR(decode_json_binary(serialized_value, &binary_storage->back())); |
349 | 1 | if (primitive_type == TYPE_VARBINARY) { |
350 | 0 | *result = Field::create_field<TYPE_VARBINARY>(StringView(binary_storage->back())); |
351 | 1 | } else if (is_string_type(primitive_type)) { |
352 | 1 | *result = Field::create_field<TYPE_STRING>(binary_storage->back()); |
353 | 1 | } else { |
354 | 0 | return Status::InvalidArgument( |
355 | 0 | "Iceberg binary default for field '{}' has incompatible Doris type '{}'", |
356 | 0 | field.name, value_type->get_name()); |
357 | 0 | } |
358 | 1 | return Status::OK(); |
359 | 1 | } |
360 | | |
361 | 4 | if (is_string_type(primitive_type)) { |
362 | 2 | if (!json_value.IsString()) { |
363 | 0 | return Status::InvalidArgument("Iceberg string default for field '{}' is not a string", |
364 | 0 | field.name); |
365 | 0 | } |
366 | 2 | *result = Field::create_field<TYPE_STRING>(std::move(serialized_value)); |
367 | 2 | return Status::OK(); |
368 | 2 | } |
369 | 2 | normalize_timestamp_for_doris(primitive_type, &serialized_value); |
370 | 2 | if (parse_non_finite_default(primitive_type, serialized_value, result)) { |
371 | 0 | return Status::OK(); |
372 | 0 | } |
373 | 2 | RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(serialized_value, *result)); |
374 | 2 | return Status::OK(); |
375 | 2 | } |
376 | | |
377 | | inline Status build_json_default_field(const schema::external::TField& field, |
378 | | const DataTypePtr& data_type, |
379 | | const rapidjson::Value& json_value, |
380 | 13 | std::deque<std::string>* binary_storage, Field* result) { |
381 | 13 | DORIS_CHECK(data_type != nullptr); |
382 | 13 | DORIS_CHECK(binary_storage != nullptr); |
383 | 13 | DORIS_CHECK(result != nullptr); |
384 | 13 | if (json_value.IsNull()) { |
385 | 1 | return make_null_field(field, data_type, result); |
386 | 1 | } |
387 | | |
388 | 12 | const auto value_type = remove_nullable(data_type); |
389 | 12 | switch (value_type->get_primitive_type()) { |
390 | 5 | case TYPE_STRUCT: |
391 | 5 | return build_json_struct_default(field, value_type, json_value, binary_storage, result); |
392 | 1 | case TYPE_ARRAY: |
393 | 1 | return build_json_array_default(field, value_type, json_value, binary_storage, result); |
394 | 1 | case TYPE_MAP: |
395 | 1 | return build_json_map_default(field, value_type, json_value, binary_storage, result); |
396 | 5 | default: |
397 | 5 | return build_json_scalar_default(field, value_type, json_value, binary_storage, result); |
398 | 12 | } |
399 | 12 | } |
400 | | |
401 | | inline Status build_initial_default_field(const schema::external::TField& field, |
402 | | const DataTypePtr& data_type, |
403 | 41 | std::deque<std::string>* binary_storage, Field* result) { |
404 | 41 | DORIS_CHECK(data_type != nullptr); |
405 | 41 | DORIS_CHECK(binary_storage != nullptr); |
406 | 41 | DORIS_CHECK(result != nullptr); |
407 | 41 | if (!field.__isset.initial_default_value) { |
408 | 5 | if (field.__isset.is_optional && !field.is_optional) { |
409 | 1 | return Status::InvalidArgument( |
410 | 1 | "Required Iceberg field '{}' is missing from the data file and has no initial " |
411 | 1 | "default", |
412 | 1 | field.name); |
413 | 1 | } |
414 | 4 | return make_null_field(field, data_type, result); |
415 | 5 | } |
416 | | |
417 | 36 | const auto value_type = remove_nullable(data_type); |
418 | 36 | const auto primitive_type = value_type->get_primitive_type(); |
419 | 36 | if (is_complex_type(primitive_type)) { |
420 | 7 | rapidjson::Document document; |
421 | 7 | document.Parse(field.initial_default_value.data(), field.initial_default_value.size()); |
422 | 7 | if (document.HasParseError()) { |
423 | 0 | return Status::InvalidArgument("Invalid Iceberg JSON initial default for field '{}'", |
424 | 0 | field.name); |
425 | 0 | } |
426 | 7 | return build_json_default_field(field, data_type, document, binary_storage, result); |
427 | 7 | } |
428 | | |
429 | 29 | const bool default_is_base64 = (field.__isset.initial_default_value_is_base64 && |
430 | 29 | field.initial_default_value_is_base64) || |
431 | 29 | primitive_type == TYPE_VARBINARY; |
432 | 29 | if (default_is_base64) { |
433 | 3 | binary_storage->emplace_back(); |
434 | 3 | if (!base64_decode(field.initial_default_value, &binary_storage->back())) { |
435 | 0 | return Status::InvalidArgument("Invalid Base64 Iceberg initial default for field '{}'", |
436 | 0 | field.name); |
437 | 0 | } |
438 | 3 | if (primitive_type == TYPE_VARBINARY) { |
439 | 2 | *result = Field::create_field<TYPE_VARBINARY>(StringView(binary_storage->back())); |
440 | 2 | } else if (is_string_type(primitive_type)) { |
441 | 1 | *result = Field::create_field<TYPE_STRING>(binary_storage->back()); |
442 | 1 | } else { |
443 | 0 | return Status::InvalidArgument( |
444 | 0 | "Iceberg field '{}' marks its initial default as Base64, but Doris type '{}' " |
445 | 0 | "cannot contain binary data", |
446 | 0 | field.name, value_type->get_name()); |
447 | 0 | } |
448 | 3 | return Status::OK(); |
449 | 3 | } |
450 | | |
451 | 26 | if (parse_non_finite_default(primitive_type, field.initial_default_value, result)) { |
452 | 1 | return Status::OK(); |
453 | 1 | } |
454 | 25 | RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(field.initial_default_value, *result)); |
455 | 25 | return Status::OK(); |
456 | 25 | } |
457 | | |
458 | | } // namespace detail |
459 | | |
460 | | // Builds an owned one-row column for an Iceberg field that is absent from an old data file. |
461 | | // Complex values follow Iceberg's JSON single-value encoding. Struct members omitted from the |
462 | | // encoded value are recursively populated from the child field's own initial default. |
463 | | inline Status create_initial_default_column(const schema::external::TField& field, |
464 | 35 | const DataTypePtr& data_type, ColumnPtr* result) { |
465 | 35 | DORIS_CHECK(data_type != nullptr); |
466 | 35 | DORIS_CHECK(result != nullptr); |
467 | | |
468 | 35 | auto column = data_type->create_column(); |
469 | 35 | std::deque<std::string> binary_storage; |
470 | 35 | Field value; |
471 | 35 | RETURN_IF_ERROR(detail::build_initial_default_field(field, data_type, &binary_storage, &value)); |
472 | | // The column copies every String/StringView leaf before binary_storage is destroyed. |
473 | 34 | column->insert(value); |
474 | | |
475 | 34 | *result = std::move(column); |
476 | 34 | return Status::OK(); |
477 | 35 | } |
478 | | |
479 | 16 | inline ColumnPtr repeat_initial_default_column(const ColumnPtr& default_column, size_t rows) { |
480 | 16 | DORIS_CHECK(default_column); |
481 | 16 | DORIS_CHECK_EQ(default_column->size(), 1); |
482 | | |
483 | 16 | auto repeated_column = default_column->clone_empty(); |
484 | 16 | repeated_column->insert_many_from(*default_column, 0, rows); |
485 | 16 | return repeated_column; |
486 | 16 | } |
487 | | |
488 | | inline Status append_initial_default( |
489 | | const schema::external::TField& field, const DataTypePtr& data_type, size_t rows, |
490 | | std::unordered_map<int32_t, std::pair<DataTypePtr, ColumnPtr>>* prepared_values, |
491 | 9 | ColumnPtr* destination) { |
492 | 9 | DORIS_CHECK(data_type != nullptr); |
493 | 9 | DORIS_CHECK(prepared_values != nullptr); |
494 | 9 | DORIS_CHECK(destination != nullptr); |
495 | 9 | DORIS_CHECK(field.__isset.id); |
496 | | |
497 | 9 | auto prepared_value = prepared_values->find(field.id); |
498 | 9 | if (prepared_value == prepared_values->end()) { |
499 | 8 | ColumnPtr default_column; |
500 | 8 | RETURN_IF_ERROR(create_initial_default_column(field, data_type, &default_column)); |
501 | 8 | prepared_value = |
502 | 8 | prepared_values |
503 | 8 | ->emplace(field.id, std::make_pair(data_type, std::move(default_column))) |
504 | 8 | .first; |
505 | 8 | } else { |
506 | | // One Iceberg field ID resolves to one query type. Hold the first DataTypePtr so equivalent |
507 | | // complex types reconstructed for later Blocks reuse the same prepared value. |
508 | 1 | DORIS_CHECK(prepared_value->second.first->equals(*data_type)); |
509 | 1 | } |
510 | | |
511 | 9 | auto mutable_destination = IColumn::mutate(std::move(*destination)); |
512 | 9 | mutable_destination->insert_many_from(*prepared_value->second.second, 0, rows); |
513 | 9 | *destination = std::move(mutable_destination); |
514 | 9 | return Status::OK(); |
515 | 9 | } |
516 | | |
517 | | } // namespace doris::iceberg |