github-actions[bot] commented on code in PR #66825:
URL: https://github.com/apache/doris/pull/66825#discussion_r3801246683
##########
be/src/runtime/runtime_state.cpp:
##########
@@ -53,13 +53,113 @@
#include "runtime/thread_context.h"
#include "storage/id_manager.h"
#include "storage/storage_engine.h"
+#include "util/thrift_util.h"
#include "util/timezone_utils.h"
#include "util/uid_util.h"
namespace doris {
#include "common/compile_check_begin.h"
using namespace ErrorCode;
+Status RuntimeState::add_iceberg_commit_datas(TIcebergCommitData
iceberg_commit_data) {
+ ThriftSerializer serializer(false, 256);
+ uint32_t serialized_size = 0;
+ uint8_t* buffer = nullptr;
+ RETURN_IF_ERROR(serializer.serialize(&iceberg_commit_data,
&serialized_size, &buffer));
+
+ // This is an early per-vector guard only; the assembled RPC is measured
again before send.
+ constexpr size_t report_envelope_headroom = 1024 * 1024;
+ const size_t thrift_limit = coordinator_thrift_message_limit();
+ const size_t commit_data_limit =
+ thrift_limit > report_envelope_headroom ? thrift_limit -
report_envelope_headroom : 0;
+ std::lock_guard<std::mutex>
budget_lock(_external_file_report_state->mutex);
+ // Parallel task states share this budget because FE receives their
vectors in one fragment report.
+ if (_external_file_report_state->iceberg_serialized_bytes +
serialized_size + sizeof(uint32_t) >
+ commit_data_limit) {
+ return Status::InternalError(
+ "Iceberg commit metadata exceeds the Thrift report limit;
reduce output file "
+ "count");
+ }
+ std::lock_guard<std::mutex> data_lock(_iceberg_commit_datas_mutex);
+ _external_file_report_state->iceberg_serialized_bytes += serialized_size +
sizeof(uint32_t);
+ _iceberg_commit_datas.emplace_back(std::move(iceberg_commit_data));
+ return Status::OK();
+}
+
+size_t RuntimeState::coordinator_thrift_message_limit() const {
+ int32_t effective_thrift_limit = std::max(config::thrift_max_message_size,
0);
+ if (_query_options.__isset.coordinator_thrift_max_message_size &&
+ _query_options.coordinator_thrift_max_message_size > 0) {
+ // An older FE omits this field; otherwise the receiver's smaller
limit is authoritative.
+ effective_thrift_limit = std::min(effective_thrift_limit,
+
_query_options.coordinator_thrift_max_message_size);
+ }
+ return static_cast<size_t>(effective_thrift_limit);
+}
+
+void RuntimeState::append_external_file_commit_data(TReportExecStatusParams*
params,
+ bool final_report) const {
+ if (!final_report) {
+ // Ownership-bearing commit vectors must only appear in the final
report that transfers them.
+ return;
+ }
+ if (auto updates = hive_partition_updates(); !updates.empty()) {
+ params->__isset.hive_partition_updates = true;
+
params->hive_partition_updates.insert(params->hive_partition_updates.end(),
updates.begin(),
+ updates.end());
+ }
+ append_iceberg_commit_datas(¶ms->iceberg_commit_datas);
+ if (!params->iceberg_commit_datas.empty()) {
+ params->__isset.iceberg_commit_datas = true;
+ }
+ if (auto commit_datas = mc_commit_datas(); !commit_datas.empty()) {
+ params->__isset.mc_commit_datas = true;
+ params->mc_commit_datas.insert(params->mc_commit_datas.end(),
commit_datas.begin(),
+ commit_datas.end());
+ }
+ if (auto commit_messages = paimon_commit_messages();
!commit_messages.empty()) {
+ // branch-4.1 still carries Paimon commit messages in the shared
external-file report.
+ params->__isset.paimon_commit_messages = true;
+
params->paimon_commit_messages.insert(params->paimon_commit_messages.end(),
+ commit_messages.begin(),
commit_messages.end());
+ }
+}
+
+void
RuntimeState::add_rejected_external_file_report_cleanup(std::function<void()>
cleanup) {
+ add_external_file_report_finalizer(
+ [cleanup = std::move(cleanup)](ExternalFileReportOutcome outcome) {
+ if (outcome == ExternalFileReportOutcome::REJECTED) {
+ cleanup();
+ }
+ });
+}
+
+void RuntimeState::add_external_file_report_finalizer(
+ std::function<void(ExternalFileReportOutcome)> finalizer) {
+ std::lock_guard lock(_external_file_report_state->mutex);
+
_external_file_report_state->report_finalizers.emplace_back(std::move(finalizer));
Review Comment:
[P1] Finalize cleanup owners that register after cancellation
A cancelled pipeline task does not wait for
`AsyncResultWriter::process_block()`: `TaskScheduler` closes it,
`AsyncWriterSink::close()` only calls `force_close()`, and
`_close_fragment_instance()` can send the final report immediately. For Hive
S3, the async thread registers the MPU rejection callback only after
`_file_format_transformer->close()`. If the report callback reaches
`finalize_external_file_report_cleanup(REJECTED)` first, it swaps the current
vector; this later `emplace_back` has no terminal outcome to observe and the
upload ID also missed the report, so the MPU is left without an abort owner.
Please either join async shutdown before the terminal report or persist the
terminal outcome and invoke late registrations immediately; add a cancellation
barrier test that lets the report reject before Hive close registers its
callback.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java:
##########
@@ -264,6 +264,8 @@ public AbstractInsertExecutor initPlan(ConnectContext ctx,
StmtExecutor stmtExec
int retryTimes = 0;
ctx.getStatementContext().setIsInsert(true);
while (++retryTimes <
Math.max(ctx.getSessionVariable().dmlPlanRetryTimes, 3)) {
+ // Each attempt must repin MVCC metadata or it can reuse the
schema that triggered the retry.
+ ctx.getStatementContext().resetMvccSnapshots();
Review Comment:
[P1] Keep MTMV's injected snapshots on the first attempt
`MTMVTask.exec()` fills the `StatementContext` with the MVCC snapshots
captured in `beforeMTMVRefresh()`, then `UpdateMvByPartitionCommand` reaches
this ordinary `InsertIntoTableCommand`. This unconditional reset runs before
attempt 1, erasing those fences before any base relation is bound. A base-table
commit between refresh partition selection and planning can therefore make the
refresh read a newer or mixed generation than the captured snapshot set. Move
retry cleanup after a failed attempt or distinguish planner-acquired snapshots
from caller-injected fences; add a refresh test that changes a base snapshot
after injection and verifies planning remains pinned.
##########
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:
[P1] Prefer current nested names before aliases when building default types
`prepare_iceberg_initial_default_exprs()` consumes the schema tree built by
`build_schema_column_from_external_field()`, whose nested STRUCT lookup
performs a single element-order pass over current names and aliases. If complex
child id 11 is renamed `old` to `renamed` and a reordered id 12 now owns name
`old`, resolving id 11 can hit id 12 through id 11's alias before reaching
`renamed`. Primitive types are restored from Thrift, but complex types retain
that wrong fallback, so a valid complex initial default is parsed as the
sibling's STRUCT/ARRAY/MAP type and fails or is mis-materialized. Match the
root's exact-name-first precedence (or bind by field ID) and cover both
reordered colliding complex children.
##########
be/src/exec/operator/spill_iceberg_table_sink_operator.cpp:
##########
@@ -53,51 +128,75 @@ bool SpillIcebergTableSinkLocalState::is_blockable() const
{
return true;
}
-size_t SpillIcebergTableSinkLocalState::get_reserve_mem_size(RuntimeState*
state, bool eos) {
+size_t SpillIcebergTableSinkLocalState::get_reserve_mem_size(RuntimeState*
state, bool eos,
+ const Block*
block) {
if (!_writer) {
return 0;
}
- auto current_writer = _writer->current_writer();
- auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(current_writer.get());
- if (!sort_writer) {
- return 0;
+ std::vector<IcebergSorterReserveMemory> per_partition_reservations;
+ const size_t incoming_rows = block == nullptr ? 0 : block->rows();
+ const size_t incoming_bytes = block == nullptr ? 0 :
block->allocated_bytes();
+ auto active_writers = _writer->active_writers();
+ per_partition_reservations.reserve(active_writers->size());
+ for (const auto& writer : *active_writers) {
+ if (auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(writer.get())) {
+ auto reservation = sort_writer->get_reserve_mem_size_components(
+ state, eos, incoming_rows, incoming_bytes);
+ per_partition_reservations.push_back(
+ {.retained_growth = reservation.retained_growth,
+ .retained_growth_trigger_bytes =
reservation.retained_growth_trigger_bytes,
+ .transient_workspace = reservation.transient_workspace});
+ }
}
-
- return sort_writer->get_reserve_mem_size(state, eos);
+ // Column growth remains in every touched sorter, while sorting workspace
is reused by serial dispatch.
+ // The final queued item may contain rows and also owns the reservation
used by async finish().
+ const size_t incoming_reserve =
+ block == nullptr ? state->minimum_operator_memory_required_bytes()
+ : iceberg_cold_writer_reserve_size(
+ *block,
state->minimum_operator_memory_required_bytes());
+ return iceberg_reserve_size(per_partition_reservations, incoming_reserve,
incoming_rows,
+ incoming_bytes);
}
size_t SpillIcebergTableSinkLocalState::get_revocable_mem_size(RuntimeState*
state) const {
if (!_writer) {
return 0;
}
- auto current_writer = _writer->current_writer();
- auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(current_writer.get());
- if (!sort_writer) {
- return 0;
+ size_t revocable_size = 0;
+ // Retain the published container while the async writer may replace the
current snapshot.
+ auto active_writers = _writer->active_writers();
+ for (const auto& writer : *active_writers) {
+ if (auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(writer.get())) {
+ revocable_size += sort_writer->data_size();
+ }
}
-
- return sort_writer->data_size();
+ return revocable_size;
}
Status SpillIcebergTableSinkLocalState::revoke_memory(RuntimeState* state) {
RETURN_IF_CANCELLED(state);
if (!_writer) {
return Status::OK();
}
- auto current_writer = _writer->current_writer();
- auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(current_writer.get());
- if (!sort_writer) {
- return Status::OK();
+ std::shared_ptr<IPartitionWriterBase> largest_writer;
+ size_t largest_size = 0;
+ // Retain the snapshot while the async writer may publish a replacement.
+ auto active_writers = _writer->active_writers();
+ for (const auto& writer : *active_writers) {
+ if (auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(writer.get())) {
+ size_t size = sort_writer->data_size();
+ if (size > largest_size) {
+ largest_size = size;
+ largest_writer = writer;
+ }
+ }
}
-
- auto exception_catch_func = [current_writer, sort_writer]() {
- auto status = [&]() {
- RETURN_IF_CATCH_EXCEPTION({ return sort_writer->trigger_spill();
});
- }();
- return status;
- };
-
- return run_spill_task(state, exception_catch_func);
+ if (largest_writer != nullptr) {
+ // Drain one largest partition per revocation to avoid launching O(P)
spill tasks.
+ auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(largest_writer.get());
+ RETURN_IF_CATCH_EXCEPTION({
RETURN_IF_ERROR(sort_writer->trigger_spill()); });
Review Comment:
[P1] Serialize revocation with whole-table async dispatch
`get_revocable_size()` can expose this sink while its pipeline task is
blocked but `AsyncResultWriter` is still dispatching one queued block. This
loop samples each writer under only that writer's mutex, then `trigger_spill()`
locks just the selected writer. After it samples B, async dispatch can start a
target flush/sort on A; the revoker can then sort B concurrently. Yet admission
combines per-partition transient workspace with `max`, explicitly assuming
serial dispatch, so the second sorted destination/permutation is unreserved and
can OOM during memory pressure. Add a table-writer-wide exclusion or quiescence
boundary around selection and spill versus dispatch, with a multi-partition
barrier test.
--
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]
