Gabriel39 commented on code in PR #66825: URL: https://github.com/apache/doris/pull/66825#discussion_r3840355099
########## be/src/exec/operator/iceberg_sorter_reserve_memory.h: ########## @@ -0,0 +1,187 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <algorithm> +#include <functional> +#include <limits> +#include <vector> + +namespace doris { + +class Block; + +struct IcebergSorterReserveMemory { + size_t retained_growth = 0; + size_t retained_growth_trigger_bytes = 0; + size_t retained_sorted_destination = 0; + size_t transient_workspace = 0; +}; + +inline size_t iceberg_saturating_add(size_t lhs, size_t rhs) { + return std::min(std::numeric_limits<size_t>::max() - lhs, rhs) + lhs; +} + +inline size_t iceberg_saturating_multiply(size_t lhs, size_t rhs) { + return lhs != 0 && rhs > std::numeric_limits<size_t>::max() / lhs + ? std::numeric_limits<size_t>::max() + : lhs * rhs; +} + +inline size_t bounded_iceberg_reserve_size( + const std::vector<IcebergSorterReserveMemory>& per_partition_reservations, + size_t incoming_rows = std::numeric_limits<size_t>::max(), + size_t incoming_bytes = std::numeric_limits<size_t>::max()) { + size_t transient_workspace = 0; + for (const auto& reservation : per_partition_reservations) { + transient_workspace = std::max(transient_workspace, reservation.transient_workspace); + } + + std::vector<size_t> sorted_destinations; + sorted_destinations.reserve(per_partition_reservations.size()); + for (const auto& reservation : per_partition_reservations) { + if (reservation.retained_sorted_destination > 0) { + sorted_destinations.push_back(reservation.retained_sorted_destination); + } + } + std::sort(sorted_destinations.begin(), sorted_destinations.end(), std::greater<>()); + // A row can touch only one partition, but each destination survives serial dispatch. At EOS + // every nonempty sorter is closed, so a zero-row final item must retain all destinations. + const size_t destination_count = incoming_rows == 0 Review Comment: Fixed. Non-empty EOS now reserves the destinations touched by incoming rows plus one additional largest destination for serial forced close, with a one-row EOS regression test. ########## be/test/exec/sort/full_sort_test.cpp: ########## @@ -94,6 +102,36 @@ TEST_F(FullSorterTest, test_full_sorter2) { std::cout << sorter->get_reserve_mem_size(&_state, false) << std::endl; } +TEST_F(FullSorterTest, EosReservationIncludesForcedSortBelowAppendThresholds) { + sorter = FullSorter::create_unique(ordering_expr_ctxs, -1, 0, &pool, is_asc_order, nulls_first, + *row_desc, &_state, nullptr); + Block block = ColumnHelper::create_block<DataTypeInt64>({10, 9, 8, 7, 6, 5, 4, 3, 2, 1}); + const size_t buffered_bytes = block.bytes(); + const size_t buffered_rows = block.rows(); + ASSERT_TRUE(sorter->append_block(&block).ok()); + + const auto reservation = sorter->get_reserve_mem_size_components(&_state, true, 0, 0); + + EXPECT_GE(reservation.transient_workspace, Review Comment: Fixed. The test now asserts retained_sorted_destination and transient_workspace separately, and also checks their combined reservation. ########## be/src/exec/operator/iceberg_sorter_reserve_memory.h: ########## @@ -0,0 +1,187 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <algorithm> +#include <functional> +#include <limits> +#include <vector> + +namespace doris { + +class Block; + +struct IcebergSorterReserveMemory { + size_t retained_growth = 0; + size_t retained_growth_trigger_bytes = 0; + size_t retained_sorted_destination = 0; + size_t transient_workspace = 0; +}; + +inline size_t iceberg_saturating_add(size_t lhs, size_t rhs) { + return std::min(std::numeric_limits<size_t>::max() - lhs, rhs) + lhs; +} + +inline size_t iceberg_saturating_multiply(size_t lhs, size_t rhs) { + return lhs != 0 && rhs > std::numeric_limits<size_t>::max() / lhs + ? std::numeric_limits<size_t>::max() + : lhs * rhs; +} + +inline size_t bounded_iceberg_reserve_size( + const std::vector<IcebergSorterReserveMemory>& per_partition_reservations, + size_t incoming_rows = std::numeric_limits<size_t>::max(), + size_t incoming_bytes = std::numeric_limits<size_t>::max()) { + size_t transient_workspace = 0; + for (const auto& reservation : per_partition_reservations) { + transient_workspace = std::max(transient_workspace, reservation.transient_workspace); + } + + std::vector<size_t> sorted_destinations; + sorted_destinations.reserve(per_partition_reservations.size()); + for (const auto& reservation : per_partition_reservations) { + if (reservation.retained_sorted_destination > 0) { + sorted_destinations.push_back(reservation.retained_sorted_destination); + } + } + std::sort(sorted_destinations.begin(), sorted_destinations.end(), std::greater<>()); + // A row can touch only one partition, but each destination survives serial dispatch. At EOS + // every nonempty sorter is closed, so a zero-row final item must retain all destinations. + const size_t destination_count = incoming_rows == 0 + ? sorted_destinations.size() + : std::min(incoming_rows, sorted_destinations.size()); + size_t retained_sorted_destinations = 0; + for (size_t i = 0; i < destination_count; ++i) { + retained_sorted_destinations = + iceberg_saturating_add(retained_sorted_destinations, sorted_destinations[i]); + } + + std::vector<const IcebergSorterReserveMemory*> growth_candidates; + growth_candidates.reserve(per_partition_reservations.size()); + for (const auto& reservation : per_partition_reservations) { + if (reservation.retained_growth > 0) { + growth_candidates.push_back(&reservation); + } + } + + std::sort(growth_candidates.begin(), growth_candidates.end(), + [](const auto* lhs, const auto* rhs) { + return lhs->retained_growth > rhs->retained_growth; + }); + size_t row_bound = 0; + for (size_t i = 0; i < std::min(incoming_rows, growth_candidates.size()); ++i) { + row_bound = iceberg_saturating_add(row_bound, growth_candidates[i]->retained_growth); + } + + size_t byte_bound = 0; + std::vector<const IcebergSorterReserveMemory*> positive_trigger_candidates; + positive_trigger_candidates.reserve(growth_candidates.size()); + for (const auto* reservation : growth_candidates) { + if (reservation->retained_growth_trigger_bytes == 0) { + byte_bound = iceberg_saturating_add(byte_bound, reservation->retained_growth); + } else { + positive_trigger_candidates.push_back(reservation); + } + } + std::sort(positive_trigger_candidates.begin(), positive_trigger_candidates.end(), + [](const auto* lhs, const auto* rhs) { + return static_cast<unsigned __int128>(lhs->retained_growth) * + rhs->retained_growth_trigger_bytes > + static_cast<unsigned __int128>(rhs->retained_growth) * + lhs->retained_growth_trigger_bytes; + }); + size_t remaining_bytes = incoming_bytes; + for (const auto* reservation : positive_trigger_candidates) { + if (reservation->retained_growth_trigger_bytes <= remaining_bytes) { + byte_bound = iceberg_saturating_add(byte_bound, reservation->retained_growth); + remaining_bytes -= reservation->retained_growth_trigger_bytes; + continue; + } + const auto numerator = + static_cast<unsigned __int128>(reservation->retained_growth) * remaining_bytes + + reservation->retained_growth_trigger_bytes - 1; + const auto fractional_growth = + std::min<unsigned __int128>(numerator / reservation->retained_growth_trigger_bytes, + std::numeric_limits<size_t>::max()); + byte_bound = iceberg_saturating_add(byte_bound, static_cast<size_t>(fractional_growth)); + break; + } + + // A block's rows and bytes are divided across partition sorters. The two fractional-relaxation + // bounds avoid charging the complete input block to every active partition while remaining safe. + const size_t retained_growth = std::min(row_bound, byte_bound); + return iceberg_saturating_add( + iceberg_saturating_add(retained_growth, retained_sorted_destinations), + transient_workspace); +} + +inline size_t iceberg_reserve_size( + const std::vector<IcebergSorterReserveMemory>& per_partition_reservations, + size_t incoming_block_reserve, size_t incoming_rows = std::numeric_limits<size_t>::max(), + size_t incoming_bytes = std::numeric_limits<size_t>::max()) { + size_t sorter_reserve = + bounded_iceberg_reserve_size(per_partition_reservations, incoming_rows, incoming_bytes); + // The incoming block creates cold partition writers before they can appear in the published snapshot. + return iceberg_saturating_add(sorter_reserve, incoming_block_reserve); +} + +size_t iceberg_cold_writer_reserve_size(const Block& block, size_t writer_workspace_bytes); + +inline size_t iceberg_spill_merge_fan_in(size_t spill_buffer_bytes, size_t merge_limit_bytes) { + if (spill_buffer_bytes == 0) { + return 0; + } + const size_t reader_bytes = iceberg_saturating_multiply(3, spill_buffer_bytes); + const size_t available_reader_bytes = + merge_limit_bytes > spill_buffer_bytes ? merge_limit_bytes - spill_buffer_bytes : 0; + return std::max<size_t>(2, reader_bytes == 0 ? 0 : available_reader_bytes / reader_bytes); +} + +inline size_t iceberg_spill_merge_workspace(size_t spill_file_count, size_t spill_buffer_bytes, + size_t merge_limit_bytes) { + if (spill_file_count == 0 || spill_buffer_bytes == 0) { + return 0; + } + const size_t reader_bytes = iceberg_saturating_multiply(3, spill_buffer_bytes); + const size_t input_count = std::min( + spill_file_count, iceberg_spill_merge_fan_in(spill_buffer_bytes, merge_limit_bytes)); + // Each primed reader retains a serialized read buffer, parsed protobuf storage, and one + // deserialized cursor block; the merger additionally owns the block being emitted. + return iceberg_saturating_add(iceberg_saturating_multiply(input_count, reader_bytes), Review Comment: Fixed. Intermediate merge now budgets three output-sized buffers (merged Block, compressed PBlock, and serialized string), derives fan-in after reserving them, and expects 72 MiB. ########## be/src/format_v2/table/iceberg_reader.cpp: ########## @@ -81,61 +92,448 @@ static bool is_projected_iceberg_rowid(const format::ColumnDefinition& column) { return column.name == BeConsts::ICEBERG_ROWID_COL; } +static int iceberg_hex_value(char value) { + if (value >= '0' && value <= '9') { + return value - '0'; + } + if (value >= 'a' && value <= 'f') { + return value - 'a' + 10; + } + if (value >= 'A' && value <= 'F') { + return value - 'A' + 10; + } + return -1; +} + +static Status decode_iceberg_hex(std::string_view encoded, std::string* decoded) { + DORIS_CHECK(decoded != nullptr); + if ((encoded.size() & 1U) != 0) { + return Status::InvalidArgument("Invalid odd-length Iceberg binary default"); + } + decoded->resize(encoded.size() / 2); + for (size_t index = 0; index < encoded.size(); index += 2) { + const int high = iceberg_hex_value(encoded[index]); + const int low = iceberg_hex_value(encoded[index + 1]); + if (high < 0 || low < 0) { + return Status::InvalidArgument("Invalid hexadecimal Iceberg binary default"); + } + (*decoded)[index / 2] = static_cast<char>((high << 4) | low); + } + return Status::OK(); +} + +static Status decode_iceberg_json_binary(std::string_view encoded, std::string* decoded) { + DORIS_CHECK(decoded != nullptr); + const bool is_uuid = encoded.size() == 36 && encoded[8] == '-' && encoded[13] == '-' && + encoded[18] == '-' && encoded[23] == '-'; + if (!is_uuid) { + return decode_iceberg_hex(encoded, decoded); + } + + std::string uuid_hex; + uuid_hex.reserve(32); + for (size_t index = 0; index < encoded.size(); ++index) { + if (index != 8 && index != 13 && index != 18 && index != 23) { + uuid_hex.push_back(encoded[index]); + } + } + return decode_iceberg_hex(uuid_hex, decoded); +} + +static std::string iceberg_json_scalar_text(const rapidjson::Value& value) { + if (value.IsString()) { + return {value.GetString(), value.GetStringLength()}; + } + rapidjson::StringBuffer buffer; + rapidjson::Writer<rapidjson::StringBuffer> writer(buffer); + value.Accept(writer); + return {buffer.GetString(), buffer.GetSize()}; +} + +static void normalize_iceberg_json_timestamp(PrimitiveType primitive_type, std::string* value) { + if (primitive_type != TYPE_DATETIME && primitive_type != TYPE_DATETIMEV2 && + primitive_type != TYPE_TIMESTAMPTZ) { + return; + } + if (const size_t separator = value->find('T'); separator != std::string::npos) { + (*value)[separator] = ' '; + } + if (primitive_type == TYPE_TIMESTAMPTZ) { + return; + } + if (value->ends_with('Z')) { + value->pop_back(); + return; + } + const size_t time_start = value->find(' '); + if (time_start == std::string::npos) { + return; + } + const size_t offset = value->find_first_of("+-", time_start + 1); + if (offset != std::string::npos) { + value->erase(offset); + } +} + +static Status build_v2_null_default(const format::ColumnDefinition& field, + const DataTypePtr& data_type, Field* result) { + DORIS_CHECK(data_type != nullptr); + DORIS_CHECK(result != nullptr); + if (field.is_optional.has_value() && !*field.is_optional) { + return Status::InvalidArgument("Required Iceberg field '{}' has a null default", + field.name); + } + if (!data_type->is_nullable()) { + return Status::InternalError( + "Optional Iceberg field '{}' has a null default, but its Doris type '{}' is not " + "nullable", + field.name, data_type->get_name()); + } + *result = Field(); + return Status::OK(); +} + +static const format::ColumnDefinition* find_v2_struct_child(const format::ColumnDefinition& field, + const std::string& name) { + const auto exact_child = std::ranges::find_if( + field.children, [&](const auto& candidate) { return iequal(candidate.name, name); }); + if (exact_child != field.children.end()) { + return &*exact_child; + } + const auto aliased_child = std::ranges::find_if(field.children, [&](const auto& candidate) { + return std::ranges::any_of(candidate.name_mapping, + [&](const auto& alias) { return iequal(alias, name); }); + }); + return aliased_child == field.children.end() ? nullptr : &*aliased_child; +} + +static Status build_v2_initial_default_field(const format::ColumnDefinition& field, + const DataTypePtr& data_type, + std::deque<std::string>* binary_storage, + Field* result); + +static Status build_v2_json_default_field(const format::ColumnDefinition& field, + const DataTypePtr& data_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result); + +static Status build_v2_json_struct_default(const format::ColumnDefinition& field, + const DataTypePtr& value_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + if (!json_value.IsObject()) { + return Status::InvalidArgument("Invalid Iceberg struct default for field '{}'", field.name); + } + + const auto& struct_type = assert_cast<const DataTypeStruct&>(*value_type); + Struct struct_value; + struct_value.reserve(struct_type.get_elements().size()); + for (size_t index = 0; index < struct_type.get_elements().size(); ++index) { + const auto* child = find_v2_struct_child(field, struct_type.get_element_name(index)); + if (child == nullptr || !child->has_identifier_field_id()) { + return Status::InvalidArgument( + "Iceberg struct default for field '{}' has incomplete child metadata", + field.name); + } + + const std::string child_id = std::to_string(child->get_identifier_field_id()); + const auto member = json_value.FindMember(child_id.c_str()); + Field child_value; + if (member == json_value.MemberEnd()) { + RETURN_IF_ERROR(build_v2_initial_default_field(*child, struct_type.get_element(index), + binary_storage, &child_value)); + } else { + RETURN_IF_ERROR(build_v2_json_default_field(*child, struct_type.get_element(index), + member->value, binary_storage, + &child_value)); + } + struct_value.push_back(std::move(child_value)); + } + *result = Field::create_field<TYPE_STRUCT>(std::move(struct_value)); + return Status::OK(); +} + +// The child ColumnDefinition, recursively transported from the item TField, describes the element +// schema and its field-level default metadata. It cannot represent a particular list literal's +// length or per-position values, so the parent initial-default keeps those values in Iceberg's +// single-value JSON array. +static Status build_v2_json_array_default(const format::ColumnDefinition& field, + const DataTypePtr& value_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + if (!json_value.IsArray() || field.children.size() != 1) { + return Status::InvalidArgument("Invalid Iceberg list default for field '{}'", field.name); + } + + const auto& array_type = assert_cast<const DataTypeArray&>(*value_type); + Array array_value; + array_value.reserve(json_value.Size()); + for (const auto& json_element : json_value.GetArray()) { + Field element_value; + RETURN_IF_ERROR(build_v2_json_default_field(field.children.front(), + array_type.get_nested_type(), json_element, + binary_storage, &element_value)); + array_value.push_back(std::move(element_value)); + } + *result = Field::create_field<TYPE_ARRAY>(std::move(array_value)); + return Status::OK(); +} + +// The child ColumnDefinitions, recursively transported from the key/value TFields, describe entry +// schemas and field-level default metadata. They cannot represent the number, order, or concrete +// values of map entries, so the parent initial-default keeps the entries in Iceberg's single-value +// JSON key/value arrays. +static Status build_v2_json_map_default(const format::ColumnDefinition& field, + const DataTypePtr& value_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + if (!json_value.IsObject() || !json_value.HasMember("keys") || !json_value["keys"].IsArray() || + !json_value.HasMember("values") || !json_value["values"].IsArray() || + field.children.size() != 2) { + return Status::InvalidArgument("Invalid Iceberg map default for field '{}'", field.name); + } + const auto& keys = json_value["keys"]; + const auto& values = json_value["values"]; + if (keys.Size() != values.Size()) { + return Status::InvalidArgument( + "Iceberg map default for field '{}' has {} keys but {} values", field.name, + keys.Size(), values.Size()); + } + + const auto& map_type = assert_cast<const DataTypeMap&>(*value_type); + Array key_fields; + Array value_fields; + key_fields.reserve(keys.Size()); + value_fields.reserve(values.Size()); + for (rapidjson::SizeType index = 0; index < keys.Size(); ++index) { + Field key_value; + Field mapped_value; + RETURN_IF_ERROR(build_v2_json_default_field(field.children[0], map_type.get_key_type(), + keys[index], binary_storage, &key_value)); + RETURN_IF_ERROR(build_v2_json_default_field(field.children[1], map_type.get_value_type(), + values[index], binary_storage, &mapped_value)); + key_fields.push_back(std::move(key_value)); + value_fields.push_back(std::move(mapped_value)); + } + Map map_value; + map_value.push_back(Field::create_field<TYPE_ARRAY>(std::move(key_fields))); + map_value.push_back(Field::create_field<TYPE_ARRAY>(std::move(value_fields))); + *result = Field::create_field<TYPE_MAP>(std::move(map_value)); + return Status::OK(); +} + +static Status build_v2_json_scalar_default(const format::ColumnDefinition& field, + const DataTypePtr& value_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + const auto primitive_type = value_type->get_primitive_type(); + std::string serialized_value = iceberg_json_scalar_text(json_value); + const bool binary_like = + field.initial_default_value_is_base64 || primitive_type == TYPE_VARBINARY; + if (binary_like) { + if (!json_value.IsString()) { + return Status::InvalidArgument( + "Iceberg binary default for field '{}' is not a JSON string", field.name); + } + binary_storage->emplace_back(); + RETURN_IF_ERROR(decode_iceberg_json_binary(serialized_value, &binary_storage->back())); + if (primitive_type == TYPE_VARBINARY) { + *result = Field::create_field<TYPE_VARBINARY>(StringView(binary_storage->back())); + } else if (is_string_type(primitive_type)) { + *result = Field::create_field<TYPE_STRING>(binary_storage->back()); + } else { + return Status::InvalidArgument( + "Iceberg binary default for field '{}' has incompatible Doris type '{}'", + field.name, value_type->get_name()); + } + return Status::OK(); + } + + if (is_string_type(primitive_type)) { + if (!json_value.IsString()) { + return Status::InvalidArgument("Iceberg string default for field '{}' is not a string", + field.name); + } + *result = Field::create_field<TYPE_STRING>(std::move(serialized_value)); + return Status::OK(); + } + normalize_iceberg_json_timestamp(primitive_type, &serialized_value); + if (doris::iceberg::detail::parse_non_finite_default(primitive_type, serialized_value, + result)) { + return Status::OK(); + } + RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(serialized_value, *result)); + return Status::OK(); +} + +static Status build_v2_json_default_field(const format::ColumnDefinition& field, + const DataTypePtr& data_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + DORIS_CHECK(data_type != nullptr); + DORIS_CHECK(binary_storage != nullptr); + DORIS_CHECK(result != nullptr); + if (json_value.IsNull()) { + return build_v2_null_default(field, data_type, result); + } + + const auto value_type = remove_nullable(data_type); + switch (value_type->get_primitive_type()) { + case TYPE_STRUCT: + return build_v2_json_struct_default(field, value_type, json_value, binary_storage, result); + case TYPE_ARRAY: + return build_v2_json_array_default(field, value_type, json_value, binary_storage, result); + case TYPE_MAP: + return build_v2_json_map_default(field, value_type, json_value, binary_storage, result); + default: + return build_v2_json_scalar_default(field, value_type, json_value, binary_storage, result); + } +} + +static Status build_v2_initial_default_field(const format::ColumnDefinition& field, + const DataTypePtr& data_type, + std::deque<std::string>* binary_storage, + Field* result) { + DORIS_CHECK(data_type != nullptr); + DORIS_CHECK(binary_storage != nullptr); + DORIS_CHECK(result != nullptr); + if (!field.initial_default_value.has_value()) { + if (field.is_optional.has_value() && !*field.is_optional) { + return Status::InvalidArgument( + "Required Iceberg field '{}' is missing from the data file and has no initial " + "default", + field.name); + } + return build_v2_null_default(field, data_type, result); + } + + const auto value_type = remove_nullable(data_type); + const auto primitive_type = value_type->get_primitive_type(); + if (is_complex_type(primitive_type)) { + rapidjson::Document document; + document.Parse(field.initial_default_value->data(), field.initial_default_value->size()); + if (document.HasParseError()) { + return Status::InvalidArgument("Invalid Iceberg JSON initial default for field '{}'", + field.name); + } + return build_v2_json_default_field(field, data_type, document, binary_storage, result); + } + + if (field.initial_default_value_is_base64 || primitive_type == TYPE_VARBINARY) { + binary_storage->emplace_back(); + if (!base64_decode(*field.initial_default_value, &binary_storage->back())) { + return Status::InvalidArgument("Invalid Base64 Iceberg initial default for field {}", + field.name); + } + if (primitive_type == TYPE_VARBINARY) { + *result = Field::create_field<TYPE_VARBINARY>(StringView(binary_storage->back())); + } else if (is_string_type(primitive_type)) { + *result = Field::create_field<TYPE_STRING>(binary_storage->back()); + } else { + return Status::InvalidArgument( + "Base64 Iceberg initial default has incompatible Doris type {} for field {}", + data_type->get_name(), field.name); + } + return Status::OK(); + } + + if (doris::iceberg::detail::parse_non_finite_default(primitive_type, + *field.initial_default_value, result)) { + return Status::OK(); + } + RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(*field.initial_default_value, *result)); + return Status::OK(); +} + +static Status build_initial_default_literal(const format::ColumnDefinition& table_field, + VExprSPtr* literal) { + DORIS_CHECK(table_field.type != nullptr); + DORIS_CHECK(table_field.initial_default_value.has_value()); + DORIS_CHECK(literal != nullptr); + + std::deque<std::string> binary_storage; + Field initial_default; + RETURN_IF_ERROR(build_v2_initial_default_field(table_field, table_field.type, &binary_storage, + &initial_default)); + // VLiteral inserts the Field into an owning column before binary_storage is destroyed. + *literal = VLiteral::create_shared(table_field.type, initial_default); + return Status::OK(); +} + +Status prepare_iceberg_initial_default_exprs(format::ColumnDefinition* column) { + DORIS_CHECK(column != nullptr); + if (column->initial_default_value.has_value()) { + VExprSPtr literal; + RETURN_IF_ERROR(build_initial_default_literal(*column, &literal)); + column->default_expr = VExprContext::create_shared(std::move(literal)); + } + for (auto& child : column->children) { + RETURN_IF_ERROR(prepare_iceberg_initial_default_exprs(&child)); + } + return Status::OK(); +} + static Status build_missing_equality_delete_key_expr(const format::ColumnDefinition& table_field, const DataTypePtr& delete_key_type, + bool require_complete_metadata, VExprSPtr* key_expr) { DORIS_CHECK(delete_key_type != nullptr); DORIS_CHECK(key_expr != nullptr); if (!table_field.initial_default_value.has_value()) { + if (require_complete_metadata && !table_field.is_optional.has_value()) { + return Status::InvalidArgument( + "Iceberg equality delete field '{}' is missing optionality metadata", + table_field.name); + } + if (table_field.is_optional.has_value() && !*table_field.is_optional) { + return Status::InvalidArgument("Missing required field: {}", table_field.name); + } // A newly added optional field without an initial default is logically NULL in older // files. EqualityDeletePredicate treats NULL == NULL as a match. *key_expr = VLiteral::create_shared(make_nullable(delete_key_type), Field()); return Status::OK(); } VExprSPtr literal; - if (table_field.initial_default_value_is_base64 || - table_field.type->get_primitive_type() == TYPE_VARBINARY) { - // New FE versions mark every Iceberg UUID/BINARY/FIXED default as Base64 regardless of its - // Doris mapping. Keep the VARBINARY fallback for scan descriptors produced before that - // marker existed. Decode before parsing so STRING/CHAR and VARBINARY all compare against - // the raw bytes stored in equality-delete files. - std::string decoded_default; - if (!base64_decode(*table_field.initial_default_value, &decoded_default)) { - return Status::InvalidArgument("Invalid Base64 Iceberg initial default for field {}", - table_field.name); - } - if (table_field.type->get_primitive_type() == TYPE_VARBINARY) { - const auto initial_default = - Field::create_field<TYPE_VARBINARY>(StringView(decoded_default)); - // VLiteral must copy the borrowed StringView while decoded_default is alive; UUID and - // long FIXED defaults otherwise retain a pointer into freed decode storage. - literal = VLiteral::create_shared(table_field.type, initial_default); - } else { - DORIS_CHECK(is_string_type(table_field.type->get_primitive_type())); - literal = VLiteral::create_shared(table_field.type, - Field::create_field<TYPE_STRING>(decoded_default)); - } - } else { - // An added field's initial default is its logical value in every older data file that lacks - // the physical column. FE normalizes the string for the current Doris table type. - Field initial_default; - RETURN_IF_ERROR(table_field.type->get_serde()->from_fe_string( - *table_field.initial_default_value, initial_default)); - literal = VLiteral::create_shared(table_field.type, initial_default); - } - - DORIS_CHECK(literal != nullptr); + RETURN_IF_ERROR(build_initial_default_literal(table_field, &literal)); if (table_field.type->equals(*delete_key_type)) { *key_expr = std::move(literal); return Status::OK(); } auto cast_expr = Cast::create_shared(delete_key_type); - cast_expr->add_child(std::move(literal)); + cast_expr->add_child(literal); *key_expr = std::move(cast_expr); return Status::OK(); } +Status IcebergTableReader::annotate_projected_column(const TFileScanSlotInfo& slot_info, + format::ProjectedColumnBuildContext* context, + format::ColumnDefinition* column) const { + RETURN_IF_ERROR(format::TableReader::annotate_projected_column(slot_info, context, column)); + DORIS_CHECK(context != nullptr); + DORIS_CHECK(column != nullptr); + if (!context->schema_column.has_value()) { + return Status::OK(); + } + + auto& schema_column = *context->schema_column; + RETURN_IF_ERROR(prepare_iceberg_initial_default_exprs(&schema_column)); Review Comment: Fixed. Typed JSON decoding for complex initial defaults is now gated on scan semantics V2; V1 keeps the FE-provided generic expression. Added STRUCT, LIST, and MAP V1 coverage. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/rewrite/RewriteDataFileExecutor.java: ########## @@ -61,86 +63,113 @@ public RewriteDataFileExecutor(IcebergExternalTable dorisTable, */ public RewriteResult executeGroupsConcurrently(List<RewriteDataGroup> groups, long targetFileSizeBytes) throws UserException { - // Begin transaction - long transactionId = dorisTable.getCatalog().getTransactionManager().begin(); - IcebergTransaction transaction = (IcebergTransaction) dorisTable.getCatalog().getTransactionManager() - .getTransaction(transactionId); - MvccSnapshot targetSnapshot = dorisTable.loadSnapshot(Optional.empty(), Optional.empty()); - Table targetIcebergTable = ((IcebergMvccSnapshot) targetSnapshot).getSnapshotCacheValue() - .getIcebergTable().orElseThrow( - () -> new UserException("Iceberg rewrite target metadata is not available")); - transaction.beginRewrite(dorisTable, targetIcebergTable); - - // Register files to delete - for (RewriteDataGroup group : groups) { - transaction.updateRewriteFiles(Lists.newArrayList(group.getDataFiles())); - } - - // Create result collector and tasks + TransactionManager transactionManager = dorisTable.getCatalog().getTransactionManager(); + long transactionId = transactionManager.begin(); List<RewriteGroupTask> tasks = Lists.newArrayList(); - RewriteResultCollector resultCollector = new RewriteResultCollector(groups.size(), tasks); - - // Get available BE count once before creating tasks - // This avoids calling getBackendsNumber() in each task during multi-threaded execution. - // Use compute group from connect context to align with actual BE selection for queries. - int availableBeCount = getAvailableBeCount(); - - // Create tasks with callbacks - for (RewriteDataGroup group : groups) { - RewriteGroupTask task = new RewriteGroupTask( - group, - transactionId, - dorisTable, - targetSnapshot, - connectContext, - targetFileSizeBytes, - availableBeCount, - new RewriteGroupTask.RewriteResultCallback() { - @Override - public void onTaskCompleted(Long taskId) { - resultCollector.onTaskCompleted(taskId); - } - - @Override - public void onTaskFailed(Long taskId, Exception error) { - resultCollector.onTaskFailed(taskId, error); - } - }); - tasks.add(task); - } - - // Submit tasks to TransientTaskManager + boolean committed = false; try { - for (TransientTaskExecutor task : tasks) { - Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(task); + IcebergTransaction transaction = (IcebergTransaction) transactionManager + .getTransaction(transactionId); + MvccSnapshot targetSnapshot = dorisTable.loadSnapshot(Optional.empty(), Optional.empty()); + Table targetIcebergTable = ((IcebergMvccSnapshot) targetSnapshot).getSnapshotCacheValue() + .getIcebergTable().orElseThrow( + () -> new UserException("Iceberg rewrite target metadata is not available")); + transaction.beginRewrite(dorisTable, targetIcebergTable); + + for (RewriteDataGroup group : groups) { + transaction.updateRewriteFiles(Lists.newArrayList(group.getDataFiles())); } - } catch (JobException e) { - throw new UserException("Failed to submit rewrite tasks: " + e.getMessage(), e); - } - // Wait for all tasks to complete - waitForTasksCompletion(resultCollector, groups.size()); - - // Finish rewrite operation - transaction.finishRewrite(); + RewriteResultCollector resultCollector = new RewriteResultCollector(groups.size(), tasks); + int availableBeCount = getAvailableBeCount(); + for (RewriteDataGroup group : groups) { + RewriteGroupTask task = new RewriteGroupTask( + group, transactionId, dorisTable, targetSnapshot, connectContext, + targetFileSizeBytes, availableBeCount, + new RewriteGroupTask.RewriteResultCallback() { + @Override + public void onTaskCompleted(Long taskId) { + resultCollector.onTaskCompleted(taskId); + } + + @Override + public void onTaskFailed(Long taskId, Exception error) { + resultCollector.onTaskFailed(taskId, error); + } + }); + tasks.add(task); + } - // Collect statistics from transaction after all tasks are completed - int rewrittenDataFilesCount = groups.stream().mapToInt(group -> group.getDataFiles().size()).sum(); - // this should after finishRewrite - int addedDataFilesCount = transaction.getFilesToAddCount(); - long rewrittenBytesCount = groups.stream().mapToLong(group -> group.getTotalSize()).sum(); - int removedDeleteFilesCount = groups.stream().mapToInt(group -> group.getDeleteFileCount()).sum(); + try { + for (TransientTaskExecutor task : tasks) { + Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(task); + } + } catch (JobException e) { + throw new UserException("Failed to submit rewrite tasks: " + e.getMessage(), e); + } - commitAndInvalidate(transaction); + waitForTasksCompletion(resultCollector, groups.size()); + transaction.finishRewrite(); + + int rewrittenDataFilesCount = groups.stream() + .mapToInt(group -> group.getDataFiles().size()).sum(); + int addedDataFilesCount = transaction.getFilesToAddCount(); + long rewrittenBytesCount = groups.stream().mapToLong(group -> group.getTotalSize()).sum(); + int removedDeleteFilesCount = groups.stream() + .mapToInt(group -> group.getDeleteFileCount()).sum(); + + transactionManager.commit(transactionId); + committed = true; + invalidateTableCacheAfterCommit(); + return new RewriteResult(rewrittenDataFilesCount, addedDataFilesCount, + rewrittenBytesCount, removedDeleteFilesCount); + } finally { + if (!committed) { + cancelAndQuiesce(tasks); + // No task may update the transaction after rollback releases its rewrite fence. + transactionManager.rollback(transactionId); Review Comment: Fixed. Report attachment and transaction closing now share the transaction monitor, rewrite cleanup fences reports before cancellation and rollback, and a barrier test verifies a retained final-report reference is rejected after timed-out rollback begins. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
