github-actions[bot] commented on code in PR #66825:
URL: https://github.com/apache/doris/pull/66825#discussion_r3793865758
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2205,8 +2175,20 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
params.__set_backend_id(_exec_env->cluster_info()->backend_id);
}
+ Status report_size_status = validate_report_exec_status_size(
+ params, req.runtime_state->coordinator_thrift_message_limit());
+ if (!report_size_status.ok()) {
+ if (req.done) {
Review Comment:
[P1] Retain a cleanup owner for rejected Hive S3 reports
This pre-RPC rejection finalizes the report as REJECTED, but only the
Iceberg writers register rejected-report cleanup callbacks. Hive keeps its S3
upload IDs only in the final `THivePartitionUpdate`, and `S3FileWriter`
deliberately does not abort the MPU on destruction. When the combined final
report exceeds the limit, FE never receives those IDs and neither side owns
cleanup, leaving every staged upload indefinitely unless the bucket happens to
have a lifecycle rule. Register a Hive/S3 rejection cleanup handle, or enforce
the report budget before any upload is staged.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2235,14 +2217,40 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
rpc_status = Status::create<false>(res.status);
} catch (apache::thrift::TException& e) {
+ report_outcome_ambiguous = true;
rpc_status = Status::InternalError("ReportExecStatus() to {} failed:
{}",
PrintThriftNetworkAddress(req.coord_addr), e.what());
}
+ // Only Iceberg keeps BE rollback callbacks after close; the other vectors
remain compatible
+ // with coordinators that acknowledge acceptance through the RPC status
alone.
+ const bool requires_external_file_ack =
params.__isset.iceberg_commit_datas;
+ if (rpc_status.ok() && requires_external_file_ack &&
+ (!res.__isset.external_file_commit_data_accepted ||
+ !res.external_file_commit_data_accepted)) {
+ rpc_status = Status::InternalError(
+ "Coordinator did not accept ownership of the external-file
report");
+ }
+
if (!rpc_status.ok()) {
+ if (req.done && !report_outcome_ambiguous) {
Review Comment:
[P1] Mark the first report attempt ambiguous before retrying
A `TTransportException` from the first call can mean FE accepted the commit
data but its response was lost. The inner catch retries without setting
`report_outcome_ambiguous`; if the retry then returns an explicit error, this
branch treats the report as REJECTED and runs Iceberg rollback callbacks. FE
publishes acceptance before replying, but a restart or lost coordinator/cache
can make the retry reject after the table commit, so BE can delete files
already referenced by the committed snapshot. Mark ownership ambiguous as soon
as the first send has an uncertain outcome, and do not downgrade it on a later
rejection.
##########
be/src/format_v2/column_mapper.cpp:
##########
@@ -2204,15 +2163,15 @@ Status
TableColumnMapper::_create_mapping_for_column(const ColumnDefinition& tab
// Doris internal Iceberg row locator is never a physical Iceberg data
column. It is built
// from file path, row position and partition metadata for
delete/update/merge.
mapping->virtual_column_type = TableVirtualColumnType::ICEBERG_ROWID;
- } else if (table_column.initial_default_value.has_value()) {
- VExprContextSPtr initial_default;
- RETURN_IF_ERROR(build_initial_default_literal(table_column,
&initial_default));
- // Iceberg metadata is the authoritative logical value for files
written before the field
- // existed; the generic FE expression may still contain its Base64
transport text.
- _set_constant_mapping(mapping, std::move(initial_default));
} else if (table_column.default_expr != nullptr) {
Review Comment:
[P1] Preserve V1 binary initial-default decoding
For a semantics-V1 request, `IcebergTableReader` does not replace the
generic FE `default_expr` with a typed Iceberg literal; that happens only in
V2, while the request still carries the Base64 initial-default metadata.
Selecting `default_expr` first therefore makes a new BE materialize Base64 text
for UUID/BINARY/FIXED fields instead of the logical bytes. The existing
`IcebergInitialDefaultMetadataOverridesGenericBinaryDefaultExpr` test
constructs exactly this V1 state and expects the decoded 16-byte value. Please
preserve the metadata decode precedence for V1, or prepare the typed binary
literal for both semantics versions.
##########
be/src/format_v2/column_mapper.cpp:
##########
@@ -2770,10 +2729,24 @@ Status TableColumnMapper::_create_direct_mapping(const
ColumnDefinition& table_c
child_mapping.file_type = table_child.type;
child_mapping.variant_access_paths =
table_child.variant_access_paths;
child_mapping.filter_conversion =
FilterConversionType::FINALIZE_ONLY;
- // A missing nested field still has its Iceberg
initial-default value in every row
- // written before the field was added; carry it into recursive
materialization.
- RETURN_IF_ERROR(build_initial_default_column(
- table_child, &child_mapping.initial_default_column));
+ if (table_child.default_expr != nullptr) {
+ const auto* literal =
+ dynamic_cast<const
VLiteral*>(table_child.default_expr->root().get());
+ if (literal == nullptr) {
+ return Status::InvalidArgument(
+ "Missing typed initial-default literal for
table field '{}'",
+ table_child.name);
+ }
+ // Keep the literal's owning column alive for recursive
materialization.
+ child_mapping.initial_default_column =
literal->get_column_ptr();
+ } else if (table_child.initial_default_value.has_value()) {
Review Comment:
[P1] Carry typed defaults into projected nested children
V2 prepares typed default expressions recursively on
`build_context.schema_column`, but `FileScannerV2` then rebuilds the actual
projected child tree through `AccessPathParser`. Its metadata inheritance
copies the raw initial value and Base64 bit, not the prepared `default_expr`,
so a missing nested child reaches this new branch and fails with `Missing typed
initial-default expression` instead of materializing its value. The existing
`ProjectedStructFillsMissingChildWithBinaryInitialDefault` test is a direct
witness. Copy the prepared literal into projected children, or prepare the
completed projected tree after access-path expansion.
##########
be/src/format_v2/column_mapper.cpp:
##########
@@ -2204,15 +2163,15 @@ Status
TableColumnMapper::_create_mapping_for_column(const ColumnDefinition& tab
// Doris internal Iceberg row locator is never a physical Iceberg data
column. It is built
// from file path, row position and partition metadata for
delete/update/merge.
mapping->virtual_column_type = TableVirtualColumnType::ICEBERG_ROWID;
- } else if (table_column.initial_default_value.has_value()) {
- VExprContextSPtr initial_default;
- RETURN_IF_ERROR(build_initial_default_literal(table_column,
&initial_default));
- // Iceberg metadata is the authoritative logical value for files
written before the field
- // existed; the generic FE expression may still contain its Base64
transport text.
- _set_constant_mapping(mapping, std::move(initial_default));
} else if (table_column.default_expr != nullptr) {
- // Missing schema-evolution column with an explicit default expression.
+ // The table-format reader supplies a typed literal so complex and
binary defaults stay exact.
_set_constant_mapping(mapping, table_column.default_expr);
+ } else if (table_column.initial_default_value.has_value()) {
+ return Status::InvalidArgument(
+ "Missing typed initial-default expression for table field
'{}'", table_column.name);
+ } else if (_options.reject_missing_required_field &&
table_column.is_optional.has_value() &&
Review Comment:
[P1] Transport Iceberg requiredness instead of Doris nullability
This rejection requires `is_optional` to be explicitly false, but
`IcebergUtils.parseSchema` constructs every top-level Iceberg `Column` with
`allowNull=true`, and `ExternalUtil` serializes that presentation property as
`TField.is_optional`. Projected nested children lose `is_optional` again in
`AccessPathParser`. Consequently the exact V2 history cases selected for
missing-required-field rejection arrive here as true/unset and keep the generic
NULL fallback, silently changing Iceberg semantics. Serialize
`NestedField.isOptional()` recursively from the Iceberg schema and preserve it
through access-path pruning.
##########
be/src/format/transformer/iceberg_partition_function.cpp:
##########
@@ -196,8 +249,13 @@ Status
IcebergInsertPartitionFunction::_compute_hashes_with_transform(
if (_partition_fields[i].transformer == nullptr) {
return Status::InternalError("Merge partitioning transform is not
initialized");
}
+ ColumnWithTypeAndName source = block->get_by_position(results[i]);
+ if (!_partition_fields[i].source_field_path.empty()) {
Review Comment:
[P1] Resolve the nested leaf type before creating its transform
This extraction happens in `get_partitions`, but `open()` has already
constructed the transformer with `field.expr_ctx->root()->data_type()`. For a
nested source that is the root STRUCT, and `PartitionColumnTransforms::create`
rejects STRUCT for bucket/truncate/time transforms before this code can reach
the primitive leaf. The added regression uses `bucket(4,
payload.nested.count)`, so its merge partitioner fails during open. Resolve
`source_field_path` through the root type before constructing the transformer,
and cover a nested non-identity transform in the BE unit test.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java:
##########
@@ -2587,87 +2588,108 @@ public void
updateFragmentExecStatus(TReportExecStatusParams params) {
}
PipelineExecContext ctx =
pipelineExecContexts.get(Pair.of(params.getFragmentId(),
params.getBackendId()));
- if (ctx == null || !ctx.updatePipelineStatus(params)) {
+ boolean hasExternalCommitData = params.isSetHivePartitionUpdates()
+ || params.isSetIcebergCommitDatas() ||
params.isSetMcCommitDatas()
+ || params.isSetPaimonCommitMessages();
+ if (ctx == null) {
+ if (hasExternalCommitData) {
+ throw new IllegalStateException("Missing fragment handler for
external-file report");
+ }
+ return false;
+ }
+ if (!ctx.updatePipelineStatus(params)) {
+ if (hasExternalCommitData && !ctx.done) {
Review Comment:
[P1] Accept legacy non-final commit vectors during rollout
Legacy BEs attach any accumulated Hive/Iceberg/MC/Paimon vectors to periodic
reports and retain them for the eventual final resend. If one parallel writer
closes or rolls over while the fragment is still running,
`updatePipelineStatus` returns false and this new check throws;
`QeProcessorImpl` returns `INTERNAL_ERROR`, so the old BE cancels the write
instead of sending its final report. Ignore or buffer legacy non-final vectors
while returning ordinary OK, or exclude old BEs before any external write
starts. The same compatibility handling is needed in `LoadProcessor`.
##########
be/src/format_v2/table/iceberg_reader.cpp:
##########
@@ -81,61 +92,450 @@ 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());
Review Comment:
[P1] Send the single-value JSON representation this parser expects
The new complex path parses Iceberg single-value JSON and its helpers
require struct members keyed by field ID, list arrays, and map `keys`/`values`.
FE still builds `initial_default_value` with
`Transforms.identity(type).toHumanString(type, value)`, whose nested
representation is not `SingleValueParser` JSON. A real FE-produced
struct/list/map default therefore fails parsing or has the wrong shape before
it can be materialized. Use Iceberg's single-value JSON serializer for complex
defaults while preserving the existing scalar/Base64 contract, and add an
FE-to-BE round-trip test.
##########
be/src/format_v2/table/iceberg_reader.cpp:
##########
@@ -81,61 +92,450 @@ 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;
Review Comment:
[P1] Preserve nested binary source types in complex defaults
A binary-like JSON scalar is decoded only when this child has its own
Base64-default marker or maps to VARBINARY. For a UUID/BINARY/FIXED element or
value inside a complex parent's default, FE marks only the parent default; the
child has no separate marker. With varbinary mapping disabled, that child maps
to STRING/CHAR and takes the plain string branch, returning Iceberg's textual
JSON spelling instead of the logical bytes. Transport the recursive Iceberg
source type (or an equivalent binary marker) independently of whether the child
owns an initial default, then decode all nested binary-like values.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/HiveTableSink.java:
##########
@@ -93,6 +93,9 @@ public String getExplainString(String prefix, TExplainLevel
explainLevel) {
public void bindDataSink(Optional<InsertCommandContext> insertCtx)
throws AnalysisException {
THiveTableSink tSink = new THiveTableSink();
+ // The legacy planner uses the same deferred Azure protocol as the
connector planner, so the BE
+ // must not fall back to publishing blocks before FE has durably
accepted the commit records.
+ tSink.setSupportsDeferredAzureMultipart(true);
Review Comment:
[P1] Gate deferred Azure writes on backend capability
This flag is sent unconditionally, but an old BE ignores it and uses the
legacy Azure path, which returns neither an upload token nor the exact staged
block IDs. The new FE then reaches
`HMSTransaction.validateObjectStoreCommitRecords` only after the whole write
was staged and rejects the transaction because there is no completion record.
Thus new-FE/old-BE rolling clusters cannot complete Azure Hive writes. Gate
eligible backends on this protocol (or add a capability handshake) before
opening the upload; final validation should remain only a fail-closed backstop.
##########
be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp:
##########
@@ -121,24 +129,106 @@ std::vector<VIcebergTableWriter::IcebergPartitionColumn>
VIcebergTableWriter::_to_iceberg_partition_columns() {
std::vector<IcebergPartitionColumn> partition_columns;
- std::unordered_map<int, int> id_to_column_idx;
- id_to_column_idx.reserve(_schema->columns().size());
- for (int i = 0; i < _schema->columns().size(); i++) {
- id_to_column_idx[_schema->columns()[i].field_id()] = i;
- }
for (const auto& partition_field : _partition_spec->fields()) {
- int column_idx = id_to_column_idx[partition_field.source_id()];
+ const auto* field_path =
_schema->find_field_path(partition_field.source_id());
+ if (field_path == nullptr || field_path->empty()) {
Review Comment:
[P1] Gate nested partition sources away from old writers
This PR teaches the new writer to resolve a schema-wide nested source ID,
but a new FE can still schedule an old BE for the same table. The authoritative
old writer indexed only top-level IDs and used `operator[]`; an unknown nested
ID therefore default-inserted column index 0 and wrote rows under the wrong
partition transform. There is no nested-partition writer capability or
smooth-upgrade gate, so ordinary INSERTs and merge insert images can silently
stage mispartitioned files. Reject such writes while an eligible old BE
remains, or schedule them only on backends advertising the new writer behavior.
--
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]
