github-actions[bot] commented on code in PR #66825:
URL: https://github.com/apache/doris/pull/66825#discussion_r3837918754
##########
be/test/exec/sort/full_sort_test.cpp:
##########
@@ -94,6 +102,36 @@ TEST_F(FullSorterTest, test_full_sorter2) {
std::cout << sorter->get_reserve_mem_size(&_state, false) << std::endl;
}
+TEST_F(FullSorterTest, EosReservationIncludesForcedSortBelowAppendThresholds) {
+ sorter = FullSorter::create_unique(ordering_expr_ctxs, -1, 0, &pool,
is_asc_order, nulls_first,
+ *row_desc, &_state, nullptr);
+ Block block = ColumnHelper::create_block<DataTypeInt64>({10, 9, 8, 7, 6,
5, 4, 3, 2, 1});
+ const size_t buffered_bytes = block.bytes();
+ const size_t buffered_rows = block.rows();
+ ASSERT_TRUE(sorter->append_block(&block).ok());
+
+ const auto reservation = sorter->get_reserve_mem_size_components(&_state,
true, 0, 0);
+
+ EXPECT_GE(reservation.transient_workspace,
Review Comment:
[P1] Assert the EOS destination in its retained component
`get_reserve_mem_size_components()` deliberately records the sorted block
bytes in `retained_sorted_destination` and only the permutation scratch in
`transient_workspace`. For this ten-row input the assertion asks
`transient_workspace` for both values, so the new test deterministically fails
even though `reservation.total()` contains them. Please assert the two fields
separately, or compare `reservation.total()` with their sum.
##########
be/src/exec/operator/iceberg_sorter_reserve_memory.h:
##########
@@ -0,0 +1,187 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <algorithm>
+#include <functional>
+#include <limits>
+#include <vector>
+
+namespace doris {
+
+class Block;
+
+struct IcebergSorterReserveMemory {
+ size_t retained_growth = 0;
+ size_t retained_growth_trigger_bytes = 0;
+ size_t retained_sorted_destination = 0;
+ size_t transient_workspace = 0;
+};
+
+inline size_t iceberg_saturating_add(size_t lhs, size_t rhs) {
+ return std::min(std::numeric_limits<size_t>::max() - lhs, rhs) + lhs;
+}
+
+inline size_t iceberg_saturating_multiply(size_t lhs, size_t rhs) {
+ return lhs != 0 && rhs > std::numeric_limits<size_t>::max() / lhs
+ ? std::numeric_limits<size_t>::max()
+ : lhs * rhs;
+}
+
+inline size_t bounded_iceberg_reserve_size(
+ const std::vector<IcebergSorterReserveMemory>&
per_partition_reservations,
+ size_t incoming_rows = std::numeric_limits<size_t>::max(),
+ size_t incoming_bytes = std::numeric_limits<size_t>::max()) {
+ size_t transient_workspace = 0;
+ for (const auto& reservation : per_partition_reservations) {
+ transient_workspace = std::max(transient_workspace,
reservation.transient_workspace);
+ }
+
+ std::vector<size_t> sorted_destinations;
+ sorted_destinations.reserve(per_partition_reservations.size());
+ for (const auto& reservation : per_partition_reservations) {
+ if (reservation.retained_sorted_destination > 0) {
+
sorted_destinations.push_back(reservation.retained_sorted_destination);
+ }
+ }
+ std::sort(sorted_destinations.begin(), sorted_destinations.end(),
std::greater<>());
+ // A row can touch only one partition, but each destination survives
serial dispatch. At EOS
+ // every nonempty sorter is closed, so a zero-row final item must retain
all destinations.
+ const size_t destination_count = incoming_rows == 0
Review Comment:
[P1] Reserve the extra forced-close sort on non-empty EOS
The all-destinations special case applies only when the final item has zero
rows. With a non-empty EOS item, destinations produced by partitions touched
during dispatch remain retained, and `VIcebergTableWriter::close()` can then
force-sort one untouched nonempty partition before draining it.
`min(incoming_rows, sorted_destinations.size())` covers only the touched
destinations, so that additional close-time destination is unreserved and can
cross the query/process limit under pressure. Please model the touched retained
destinations plus the largest serial forced-close destination, and add a
one-row EOS test with another pending partition.
##########
be/src/exec/operator/iceberg_sorter_reserve_memory.h:
##########
@@ -0,0 +1,187 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <algorithm>
+#include <functional>
+#include <limits>
+#include <vector>
+
+namespace doris {
+
+class Block;
+
+struct IcebergSorterReserveMemory {
+ size_t retained_growth = 0;
+ size_t retained_growth_trigger_bytes = 0;
+ size_t retained_sorted_destination = 0;
+ size_t transient_workspace = 0;
+};
+
+inline size_t iceberg_saturating_add(size_t lhs, size_t rhs) {
+ return std::min(std::numeric_limits<size_t>::max() - lhs, rhs) + lhs;
+}
+
+inline size_t iceberg_saturating_multiply(size_t lhs, size_t rhs) {
+ return lhs != 0 && rhs > std::numeric_limits<size_t>::max() / lhs
+ ? std::numeric_limits<size_t>::max()
+ : lhs * rhs;
+}
+
+inline size_t bounded_iceberg_reserve_size(
+ const std::vector<IcebergSorterReserveMemory>&
per_partition_reservations,
+ size_t incoming_rows = std::numeric_limits<size_t>::max(),
+ size_t incoming_bytes = std::numeric_limits<size_t>::max()) {
+ size_t transient_workspace = 0;
+ for (const auto& reservation : per_partition_reservations) {
+ transient_workspace = std::max(transient_workspace,
reservation.transient_workspace);
+ }
+
+ std::vector<size_t> sorted_destinations;
+ sorted_destinations.reserve(per_partition_reservations.size());
+ for (const auto& reservation : per_partition_reservations) {
+ if (reservation.retained_sorted_destination > 0) {
+
sorted_destinations.push_back(reservation.retained_sorted_destination);
+ }
+ }
+ std::sort(sorted_destinations.begin(), sorted_destinations.end(),
std::greater<>());
+ // A row can touch only one partition, but each destination survives
serial dispatch. At EOS
+ // every nonempty sorter is closed, so a zero-row final item must retain
all destinations.
+ const size_t destination_count = incoming_rows == 0
+ ? sorted_destinations.size()
+ : std::min(incoming_rows,
sorted_destinations.size());
+ size_t retained_sorted_destinations = 0;
+ for (size_t i = 0; i < destination_count; ++i) {
+ retained_sorted_destinations =
+ iceberg_saturating_add(retained_sorted_destinations,
sorted_destinations[i]);
+ }
+
+ std::vector<const IcebergSorterReserveMemory*> growth_candidates;
+ growth_candidates.reserve(per_partition_reservations.size());
+ for (const auto& reservation : per_partition_reservations) {
+ if (reservation.retained_growth > 0) {
+ growth_candidates.push_back(&reservation);
+ }
+ }
+
+ std::sort(growth_candidates.begin(), growth_candidates.end(),
+ [](const auto* lhs, const auto* rhs) {
+ return lhs->retained_growth > rhs->retained_growth;
+ });
+ size_t row_bound = 0;
+ for (size_t i = 0; i < std::min(incoming_rows, growth_candidates.size());
++i) {
+ row_bound = iceberg_saturating_add(row_bound,
growth_candidates[i]->retained_growth);
+ }
+
+ size_t byte_bound = 0;
+ std::vector<const IcebergSorterReserveMemory*> positive_trigger_candidates;
+ positive_trigger_candidates.reserve(growth_candidates.size());
+ for (const auto* reservation : growth_candidates) {
+ if (reservation->retained_growth_trigger_bytes == 0) {
+ byte_bound = iceberg_saturating_add(byte_bound,
reservation->retained_growth);
+ } else {
+ positive_trigger_candidates.push_back(reservation);
+ }
+ }
+ std::sort(positive_trigger_candidates.begin(),
positive_trigger_candidates.end(),
+ [](const auto* lhs, const auto* rhs) {
+ return static_cast<unsigned __int128>(lhs->retained_growth) *
+ rhs->retained_growth_trigger_bytes >
+ static_cast<unsigned __int128>(rhs->retained_growth) *
+ lhs->retained_growth_trigger_bytes;
+ });
+ size_t remaining_bytes = incoming_bytes;
+ for (const auto* reservation : positive_trigger_candidates) {
+ if (reservation->retained_growth_trigger_bytes <= remaining_bytes) {
+ byte_bound = iceberg_saturating_add(byte_bound,
reservation->retained_growth);
+ remaining_bytes -= reservation->retained_growth_trigger_bytes;
+ continue;
+ }
+ const auto numerator =
+ static_cast<unsigned __int128>(reservation->retained_growth) *
remaining_bytes +
+ reservation->retained_growth_trigger_bytes - 1;
+ const auto fractional_growth =
+ std::min<unsigned __int128>(numerator /
reservation->retained_growth_trigger_bytes,
+
std::numeric_limits<size_t>::max());
+ byte_bound = iceberg_saturating_add(byte_bound,
static_cast<size_t>(fractional_growth));
+ break;
+ }
+
+ // A block's rows and bytes are divided across partition sorters. The two
fractional-relaxation
+ // bounds avoid charging the complete input block to every active
partition while remaining safe.
+ const size_t retained_growth = std::min(row_bound, byte_bound);
+ return iceberg_saturating_add(
+ iceberg_saturating_add(retained_growth,
retained_sorted_destinations),
+ transient_workspace);
+}
+
+inline size_t iceberg_reserve_size(
+ const std::vector<IcebergSorterReserveMemory>&
per_partition_reservations,
+ size_t incoming_block_reserve, size_t incoming_rows =
std::numeric_limits<size_t>::max(),
+ size_t incoming_bytes = std::numeric_limits<size_t>::max()) {
+ size_t sorter_reserve =
+ bounded_iceberg_reserve_size(per_partition_reservations,
incoming_rows, incoming_bytes);
+ // The incoming block creates cold partition writers before they can
appear in the published snapshot.
+ return iceberg_saturating_add(sorter_reserve, incoming_block_reserve);
+}
+
+size_t iceberg_cold_writer_reserve_size(const Block& block, size_t
writer_workspace_bytes);
+
+inline size_t iceberg_spill_merge_fan_in(size_t spill_buffer_bytes, size_t
merge_limit_bytes) {
+ if (spill_buffer_bytes == 0) {
+ return 0;
+ }
+ const size_t reader_bytes = iceberg_saturating_multiply(3,
spill_buffer_bytes);
+ const size_t available_reader_bytes =
+ merge_limit_bytes > spill_buffer_bytes ? merge_limit_bytes -
spill_buffer_bytes : 0;
+ return std::max<size_t>(2, reader_bytes == 0 ? 0 : available_reader_bytes
/ reader_bytes);
+}
+
+inline size_t iceberg_spill_merge_workspace(size_t spill_file_count, size_t
spill_buffer_bytes,
+ size_t merge_limit_bytes) {
+ if (spill_file_count == 0 || spill_buffer_bytes == 0) {
+ return 0;
+ }
+ const size_t reader_bytes = iceberg_saturating_multiply(3,
spill_buffer_bytes);
+ const size_t input_count = std::min(
+ spill_file_count, iceberg_spill_merge_fan_in(spill_buffer_bytes,
merge_limit_bytes));
+ // Each primed reader retains a serialized read buffer, parsed protobuf
storage, and one
+ // deserialized cursor block; the merger additionally owns the block being
emitted.
+ return iceberg_saturating_add(iceberg_saturating_multiply(input_count,
reader_bytes),
Review Comment:
[P1] Include all intermediate-merge output copies
This adds only one output buffer, but `_do_intermediate_merge()` keeps the
merged `Block` alive while `SpillFileWriter::_write_internal()` builds a
compressed `PBlock` and then `SerializeToString()` creates a second serialized
buffer before the protobuf is destroyed. Those three output-sized allocations
coexist with every primed reader. With the default 8 MiB buffer and fan-in 2,
the helper admits 56 MiB although the conservative incompressible peak is 72
MiB, which is also why the new integration test expects 72. Please budget all
three output buffers (and use that footprint when deriving fan-in), or remove
the overlapping copies.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/rewrite/RewriteDataFileExecutor.java:
##########
@@ -61,86 +63,113 @@ public RewriteDataFileExecutor(IcebergExternalTable
dorisTable,
*/
public RewriteResult executeGroupsConcurrently(List<RewriteDataGroup>
groups, long targetFileSizeBytes)
throws UserException {
- // Begin transaction
- long transactionId =
dorisTable.getCatalog().getTransactionManager().begin();
- IcebergTransaction transaction = (IcebergTransaction)
dorisTable.getCatalog().getTransactionManager()
- .getTransaction(transactionId);
- MvccSnapshot targetSnapshot =
dorisTable.loadSnapshot(Optional.empty(), Optional.empty());
- Table targetIcebergTable = ((IcebergMvccSnapshot)
targetSnapshot).getSnapshotCacheValue()
- .getIcebergTable().orElseThrow(
- () -> new UserException("Iceberg rewrite target
metadata is not available"));
- transaction.beginRewrite(dorisTable, targetIcebergTable);
-
- // Register files to delete
- for (RewriteDataGroup group : groups) {
-
transaction.updateRewriteFiles(Lists.newArrayList(group.getDataFiles()));
- }
-
- // Create result collector and tasks
+ TransactionManager transactionManager =
dorisTable.getCatalog().getTransactionManager();
+ long transactionId = transactionManager.begin();
List<RewriteGroupTask> tasks = Lists.newArrayList();
- RewriteResultCollector resultCollector = new
RewriteResultCollector(groups.size(), tasks);
-
- // Get available BE count once before creating tasks
- // This avoids calling getBackendsNumber() in each task during
multi-threaded execution.
- // Use compute group from connect context to align with actual BE
selection for queries.
- int availableBeCount = getAvailableBeCount();
-
- // Create tasks with callbacks
- for (RewriteDataGroup group : groups) {
- RewriteGroupTask task = new RewriteGroupTask(
- group,
- transactionId,
- dorisTable,
- targetSnapshot,
- connectContext,
- targetFileSizeBytes,
- availableBeCount,
- new RewriteGroupTask.RewriteResultCallback() {
- @Override
- public void onTaskCompleted(Long taskId) {
- resultCollector.onTaskCompleted(taskId);
- }
-
- @Override
- public void onTaskFailed(Long taskId, Exception error)
{
- resultCollector.onTaskFailed(taskId, error);
- }
- });
- tasks.add(task);
- }
-
- // Submit tasks to TransientTaskManager
+ boolean committed = false;
try {
- for (TransientTaskExecutor task : tasks) {
-
Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(task);
+ IcebergTransaction transaction = (IcebergTransaction)
transactionManager
+ .getTransaction(transactionId);
+ MvccSnapshot targetSnapshot =
dorisTable.loadSnapshot(Optional.empty(), Optional.empty());
+ Table targetIcebergTable = ((IcebergMvccSnapshot)
targetSnapshot).getSnapshotCacheValue()
+ .getIcebergTable().orElseThrow(
+ () -> new UserException("Iceberg rewrite target
metadata is not available"));
+ transaction.beginRewrite(dorisTable, targetIcebergTable);
+
+ for (RewriteDataGroup group : groups) {
+
transaction.updateRewriteFiles(Lists.newArrayList(group.getDataFiles()));
}
- } catch (JobException e) {
- throw new UserException("Failed to submit rewrite tasks: " +
e.getMessage(), e);
- }
- // Wait for all tasks to complete
- waitForTasksCompletion(resultCollector, groups.size());
-
- // Finish rewrite operation
- transaction.finishRewrite();
+ RewriteResultCollector resultCollector = new
RewriteResultCollector(groups.size(), tasks);
+ int availableBeCount = getAvailableBeCount();
+ for (RewriteDataGroup group : groups) {
+ RewriteGroupTask task = new RewriteGroupTask(
+ group, transactionId, dorisTable, targetSnapshot,
connectContext,
+ targetFileSizeBytes, availableBeCount,
+ new RewriteGroupTask.RewriteResultCallback() {
+ @Override
+ public void onTaskCompleted(Long taskId) {
+ resultCollector.onTaskCompleted(taskId);
+ }
+
+ @Override
+ public void onTaskFailed(Long taskId, Exception
error) {
+ resultCollector.onTaskFailed(taskId, error);
+ }
+ });
+ tasks.add(task);
+ }
- // Collect statistics from transaction after all tasks are completed
- int rewrittenDataFilesCount = groups.stream().mapToInt(group ->
group.getDataFiles().size()).sum();
- // this should after finishRewrite
- int addedDataFilesCount = transaction.getFilesToAddCount();
- long rewrittenBytesCount = groups.stream().mapToLong(group ->
group.getTotalSize()).sum();
- int removedDeleteFilesCount = groups.stream().mapToInt(group ->
group.getDeleteFileCount()).sum();
+ try {
+ for (TransientTaskExecutor task : tasks) {
+
Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(task);
+ }
+ } catch (JobException e) {
+ throw new UserException("Failed to submit rewrite tasks: " +
e.getMessage(), e);
+ }
- commitAndInvalidate(transaction);
+ waitForTasksCompletion(resultCollector, groups.size());
+ transaction.finishRewrite();
+
+ int rewrittenDataFilesCount = groups.stream()
+ .mapToInt(group -> group.getDataFiles().size()).sum();
+ int addedDataFilesCount = transaction.getFilesToAddCount();
+ long rewrittenBytesCount = groups.stream().mapToLong(group ->
group.getTotalSize()).sum();
+ int removedDeleteFilesCount = groups.stream()
+ .mapToInt(group -> group.getDeleteFileCount()).sum();
+
+ transactionManager.commit(transactionId);
+ committed = true;
+ invalidateTableCacheAfterCommit();
+ return new RewriteResult(rewrittenDataFilesCount,
addedDataFilesCount,
+ rewrittenBytesCount, removedDeleteFilesCount);
+ } finally {
+ if (!committed) {
+ cancelAndQuiesce(tasks);
+ // No task may update the transaction after rollback releases
its rewrite fence.
+ transactionManager.rollback(transactionId);
Review Comment:
[P1] Reject final reports once rewrite rollback starts
`cancelAndQuiesce()` can time out with a report handler still live, yet this
immediately rolls back and removes the transaction. A handler that already
obtained the `IcebergTransaction` from `getTxnById()` can then resume and call
`updateIcebergCommitData()` on that detached object; it has no closing-state
check, so Coordinator returns accepted and QeProcessor publishes the ACK. BE
consequently relinquishes cleanup even though the transaction can no longer
commit those staged files and rollback does not delete them. Please make
transaction closing and report attachment atomic, reject appends after closing
begins, and add a barrier test that resumes a final report after timeout
rollback.
##########
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] Preserve V1 complex-default decoding on new BEs
This unconditionally prepares typed defaults for every semantics-V1-or-newer
descriptor. The preceding FE version sends V1 and serializes complex defaults
with `Transforms.identity(...).toHumanString(...)`, while
`build_v2_initial_default_field()` now parses every complex value as Iceberg
single-value JSON. During an FE-first or mixed rollout, a new BE receiving that
V1 descriptor therefore fails projected-column setup instead of preserving V1
behavior. Please gate the complex JSON decoder on semantics V2 (with a
V1-compatible path or encoding discriminator) and add an old-FE-V1 to new-BE
STRUCT/LIST/MAP 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]
