github-actions[bot] commented on code in PR #66825:
URL: https://github.com/apache/doris/pull/66825#discussion_r3794720950
##########
be/src/exec/operator/spill_iceberg_table_sink_operator.cpp:
##########
@@ -26,12 +27,27 @@
namespace doris {
#include "common/compile_check_begin.h"
+size_t iceberg_cold_writer_reserve_size(const Block& block, size_t
writer_workspace_bytes) {
+ const size_t block_bytes = block.allocated_bytes();
+ const size_t row_index_bytes =
+ std::min(std::numeric_limits<size_t>::max() / sizeof(size_t),
block.rows()) *
+ sizeof(size_t);
+ const size_t dispatch_copies = block_bytes >
std::numeric_limits<size_t>::max() / 4
Review Comment:
[P1] Size cold admission from the partitions it creates
The fixed four-copy estimate does not cover the buffers retained in every
newly created partition sorter. Each one-row ColumnString selection is copied
into a distinct sorter; PaddedPODArray allocates at least 4096 bytes for chars
and offsets, so 128 retained sorter buffers total about 1 MiB while this helper
returns about 280 KiB. Only one selected block is temporary at a time, but all
sorter copies coexist. Derive the bound from touched partitions and column
minimum capacities.
##########
be/src/runtime/runtime_state.cpp:
##########
@@ -53,13 +53,106 @@
#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) {
Review Comment:
[P1] Retain a Paimon abort owner until the final report is accepted
After prepare_commit succeeds, PaimonTableWriter closes and resets the only
writer/backend that can call abort, leaving only serialized messages in
RuntimeState. This new final-only transport means a size/client/submission
rejection can run the shared rejection cleanup before FE ever sees those
messages, but that cleanup has handlers only for Iceberg and Hive; FE rollback
then has no payloads to abort. Register a Paimon rejection callback that
retains the abort handle/messages until ACK, or otherwise keep an abort-capable
owner across the report transfer.
##########
regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_complex_evolution.groovy:
##########
@@ -166,6 +183,13 @@ suite("test_iceberg_write_complex_evolution",
group by spec_id
order by spec_id
"""
+ order_qt_complex_nested_partition_pruning """
Review Comment:
[P1] Regenerate the golden results for this expanded suite
This suite now retains id 6 and adds this result query, but the checked-in
output only adds id 6 to complex_current. complex_children still stops at id 5,
complex_nulls omits id 6 even though its complex columns are NULL, and there is
no complex_nested_partition_pruning section for the expected ids 4 and 6.
Regenerate the output file, including the changed partition-spec totals, so the
suite can pass.
##########
be/src/exec/operator/iceberg_sorter_reserve_memory.h:
##########
@@ -0,0 +1,132 @@
+// 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 <limits>
+#include <vector>
+
+namespace doris {
+
+class Block;
+
+struct IcebergSorterReserveMemory {
+ size_t retained_growth = 0;
+ size_t retained_growth_trigger_bytes = 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 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<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(retained_growth, 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_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 max_fan_in = std::max<size_t>(2, merge_limit_bytes /
spill_buffer_bytes);
+ const size_t input_count = std::min(spill_file_count, max_fan_in);
+ const size_t max_size = std::numeric_limits<size_t>::max();
+ const size_t input_bytes = input_count > max_size / spill_buffer_bytes
+ ? max_size
+ : input_count * spill_buffer_bytes;
+ // VSortedRunMerger materializes one block per input cursor plus the block
being emitted.
Review Comment:
[P1] Reserve the final merger's actual output batch
This helper budgets one spill_buffer_bytes block for output, but the final
merger is constructed with RuntimeState::batch_size() and its multi-stream path
materializes up to that many rows while input cursor blocks remain live. With
64 KiB rows and the default 4062-row batch, the output alone is about 254 MiB
rather than the reserved 8 MiB. Size the output from observed row width and the
final row count, or use the spill-bounded batch size for final merging as well.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -298,6 +306,214 @@ void
checkVariantBackendCompatibilityForCurrentScan(Iterable<Backend> backends)
checkVariantBackendCompatibility(projectsVariant, backends);
}
+ private boolean requiresIcebergScanSemanticsV2() throws UserException {
+ if (isSystemTable) {
+ // position_deletes already has its stricter dedicated
mixed-version gate.
+ return false;
+ }
+ TableScan scan = createTableScan();
+ Snapshot snapshot = scan.snapshot();
+ if (snapshot == null) {
+ return false;
+ }
+ if (hasApplicableEqualityDeletes(scan)) {
+ return true;
+ }
+ Schema scanSchema = scan.schema();
+ Set<Integer> projectedFieldIds = projectedFieldIds(scanSchema);
+ Set<Integer> topLevelIds = new HashSet<>();
+ for (NestedField field : scanSchema.columns()) {
+ topLevelIds.add(field.fieldId());
+ }
+ Map<Integer, NestedField> fieldsById =
TypeUtil.indexById(scanSchema.asStruct());
+ for (Integer fieldId : projectedFieldIds) {
+ NestedField field = fieldsById.get(fieldId);
+ if (field.initialDefault() != null
+ && (!topLevelIds.contains(field.fieldId()) ||
field.type().isNestedType())) {
+ return true;
+ }
+ }
+ if (hasProjectedNameAliasCollision(scanSchema, projectedFieldIds,
extractNameMapping())) {
+ return true;
+ }
+ return schemaHistoryRequiresMissingRequiredFieldRejection(
+ scanSchema, projectedFieldIds,
icebergTable.schemas().values());
+ }
+
+ private boolean hasApplicableEqualityDeletes(TableScan scan) throws
UserException {
+ Snapshot snapshot = scan.snapshot();
+ if (snapshot == null ||
"0".equals(snapshot.summary().get("total-equality-deletes"))) {
Review Comment:
[P2] Use a positive equality-delete summary before opening manifests
A present positive total-equality-deletes summary already proves that V2
semantics are required, but this path still opens remote delete manifests and
walks entries until it rediscovers an equality delete. During a rolling upgrade
every scan can therefore perform metadata I/O proportional to preceding delete
metadata only to reject an old backend. Return true immediately for a positive
summary; reserve manifest inspection (and cache it by frozen
generation/snapshot) for the missing-summary fallback.
##########
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,
Review Comment:
[P1] Handle non-finite defaults before building the generic FE expression
A real scan never reaches this parser for a top-level FLOAT/DOUBLE default
of NaN or infinity: FileScanNode first parses column.getDefaultValueSql(), and
numeric defaults are emitted unquoted, so these spellings become unresolved
column references and analysis fails. The helper test bypasses that FE path.
Construct and transport a typed FE literal for these values, or bypass the
generic expression and then gate old BEs; cover the full FE-to-BE scan path.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,377 @@
+// 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.
+
+package org.apache.doris.datasource.iceberg.action;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ArgumentParsers;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.iceberg.IcebergExternalTable;
+import org.apache.doris.info.PartitionNamesInfo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+
+import com.google.common.collect.Lists;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.ManifestContent;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.ReachableFileUtil;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.SupportsPrefixOperations;
+import org.apache.iceberg.util.PropertyUtil;
+
+import java.io.IOException;
+import java.net.URI;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+/** Safely lists or deletes old files that are unreachable from every retained
snapshot. */
+public class IcebergRemoveOrphanFilesAction extends BaseIcebergAction {
+ private static final long MIN_RETENTION_MS =
Duration.ofHours(24).toMillis();
+ private static final int MAX_REACHABLE_FILES = 5_000_000;
+ public static final String OLDER_THAN = "older_than";
+ public static final String LOCATION = "location";
+ public static final String DRY_RUN = "dry_run";
+ public static final String ALLOW_UNSAFE_LOCATION = "allow_unsafe_location";
+
+ public IcebergRemoveOrphanFilesAction(Map<String, String> properties,
+ Optional<PartitionNamesInfo> partitionNamesInfo,
+ Optional<Expression> whereCondition) {
+ super("remove_orphan_files", properties, partitionNamesInfo,
whereCondition);
+ }
+
+ @Override
+ protected void registerIcebergArguments() {
+ namedArguments.registerRequiredArgument(OLDER_THAN, "Creation time
cutoff in milliseconds",
+ ArgumentParsers.nonNegativeLong(OLDER_THAN));
+ namedArguments.registerOptionalArgument(LOCATION, "Prefix to scan for
orphan files",
+ null, ArgumentParsers.nonEmptyString(LOCATION));
+ namedArguments.registerOptionalArgument(DRY_RUN, "Only count orphan
files", true,
+ ArgumentParsers.booleanValue(DRY_RUN));
+ namedArguments.registerOptionalArgument(ALLOW_UNSAFE_LOCATION,
+ "Allow an explicitly supplied location whose table ownership
cannot be proved",
+ false, ArgumentParsers.booleanValue(ALLOW_UNSAFE_LOCATION));
+ }
+
+ @Override
+ protected void validateIcebergAction() throws UserException {
+ validateNoPartitions();
+ validateNoWhereCondition();
+ String location = namedArguments.getString(LOCATION);
+ if (location != null) {
+ try {
+ normalizeLocation(location);
+ } catch (IllegalArgumentException e) {
+ throw new AnalysisException("Invalid location URI: " +
location, e);
+ }
+ }
+ }
+
+ @Override
+ protected List<String> executeAction(TableIf tableIf) throws UserException
{
+ Table table = ((IcebergExternalTable) tableIf).getIcebergTable();
+ if (!(table.io() instanceof SupportsPrefixOperations)) {
+ throw new UserException("remove_orphan_files requires FileIO
prefix listing support");
+ }
+ if (!PropertyUtil.propertyAsBoolean(table.properties(),
TableProperties.GC_ENABLED,
+ TableProperties.GC_ENABLED_DEFAULT)) {
+ // A GC-disabled table may share files with another table, so no
destructive scan is safe.
+ throw new UserException("Cannot remove orphan files: Iceberg GC is
disabled");
+ }
+ long olderThan = namedArguments.getLong(OLDER_THAN);
+ // Reject an unsafe cutoff before opening any metadata or manifest
file.
+ if (olderThan > System.currentTimeMillis() - MIN_RETENTION_MS) {
+ throw new UserException("older_than must retain at least 24 hours
of files");
+ }
+ List<ScanScope> scanScopes = resolveScanScopes(table);
+
+ try {
+ ReachableIndex reachable = collectReachableFiles(table);
+ long orphanCount = 0;
+ long deletedCount = 0;
+ boolean dryRun = namedArguments.getBoolean(DRY_RUN);
+ for (ScanScope scope : scanScopes) {
+ // Object stores use raw prefix matching, so the separator
excludes sibling prefixes.
+ String listingPrefix = scope.root.endsWith("/") ? scope.root :
scope.root + "/";
+ for (FileInfo file : ((SupportsPrefixOperations)
table.io()).listPrefix(listingPrefix)) {
+ // Unknown creation time cannot prove the file predates
every in-flight writer.
+ if (scope.owns(file.location()) && file.createdAtMillis()
> 0
+ && file.createdAtMillis() < olderThan
+ && !isReachable(file.location(), reachable)) {
+ orphanCount++;
+ if (!dryRun) {
+ table.io().deleteFile(file.location());
Review Comment:
[P1] Fence orphan deletion against concurrent commits
The reachable index is a one-time metadata snapshot. A concurrent add/import
can commit a pre-existing object whose creation time is already below olderThan
after that snapshot is built; this loop will still classify it as unreachable
and physically delete a file referenced by the new snapshot. The 24-hour guard
only protects newly created in-flight files. Hold commit exclusion for the
destructive phase, or otherwise guarantee the metadata generation cannot change
through reachability validation and deletion.
--
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]
