github-actions[bot] commented on code in PR #66825:
URL: https://github.com/apache/doris/pull/66825#discussion_r3841511699
##########
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());
Review Comment:
[P1] Allow V1 void fields after their source column is dropped
Iceberg V1 removes a partition field by retaining it in the current spec
with a `void` transform. After that, the source column may also be deleted, so
this field can legitimately have no path in the current schema. This
unconditional lookup now throws before the existing
`VoidPartitionColumnTransform` can produce its all-NULL value, making every
later INSERT fail during writer initialization. Please special-case `void`
before resolving the source path (while keeping the strict check for
value-consuming transforms), and cover remove-partition-field ->
drop-source-column -> INSERT on a V1 table.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java:
##########
@@ -784,6 +792,26 @@ static void
validateVariantWriteBackendCompatibility(List<Column> columns, Itera
}
}
+ @VisibleForTesting
+ static void validateNestedPartitionWriteBackendCompatibility(
+ PartitionSpec spec, Schema schema, Iterable<Backend> backends)
throws AnalysisException {
+ Set<Integer> topLevelIds = schema.columns().stream()
+ .map(Types.NestedField::fieldId).collect(Collectors.toSet());
+ boolean hasNestedSource = spec.fields().stream()
+ .anyMatch(field -> !topLevelIds.contains(field.sourceId()));
Review Comment:
[P2] Do not classify a dropped V1 void source as nested
A V1 partition field removed through evolution remains in the current spec
as `void`, and its source column may subsequently be deleted. Such a field is
absent from `topLevelIds`, but it is not a live nested source and old writers
do not need nested-path support to emit its all-NULL value. This predicate
therefore blocks otherwise compatible INSERTs for the whole smooth-upgrade
window. Filter out `field.transform().isVoid()` before applying the
nested-source gate, and include the dropped-source V1 case in the compatibility
test.
##########
be/src/exec/operator/spill_iceberg_table_sink_operator.cpp:
##########
@@ -26,12 +34,79 @@
namespace doris {
#include "common/compile_check_begin.h"
+namespace {
+constexpr size_t MIN_POD_ARRAY_CAPACITY = 4096;
+
+// Admission runs before a writer may allocate; model the structural minimum
recursively so a
+// single oversized value cannot be cloned and multiplied by the
cold-partition fan-out.
+size_t minimum_selected_column_capacity(const IColumn& column) {
+ if (const auto* constant = check_and_get_column<ColumnConst>(column)) {
+ return minimum_selected_column_capacity(constant->get_data_column());
+ }
+ if (const auto* nullable = check_and_get_column<ColumnNullable>(column)) {
+ return iceberg_saturating_add(
+ MIN_POD_ARRAY_CAPACITY,
+
minimum_selected_column_capacity(nullable->get_nested_column()));
+ }
+ if (const auto* array = check_and_get_column<ColumnArray>(column)) {
+ return iceberg_saturating_add(MIN_POD_ARRAY_CAPACITY,
+
minimum_selected_column_capacity(array->get_data()));
+ }
+ if (const auto* map = check_and_get_column<ColumnMap>(column)) {
+ size_t capacity = iceberg_saturating_add(MIN_POD_ARRAY_CAPACITY,
+
minimum_selected_column_capacity(map->get_keys()));
+ return iceberg_saturating_add(capacity,
+
minimum_selected_column_capacity(map->get_values()));
+ }
+ if (const auto* structure = check_and_get_column<ColumnStruct>(column)) {
+ size_t capacity = 0;
+ for (const auto& child : structure->get_columns()) {
+ capacity = iceberg_saturating_add(capacity,
minimum_selected_column_capacity(*child));
+ }
+ return capacity;
+ }
+ if (check_and_get_column<ColumnString>(column) != nullptr) {
+ return 2 * MIN_POD_ARRAY_CAPACITY;
+ }
+ return MIN_POD_ARRAY_CAPACITY;
+}
+} // namespace
+
+size_t iceberg_cold_writer_reserve_size(const Block& block, size_t
writer_workspace_bytes) {
+ const size_t block_bytes = block.allocated_bytes();
Review Comment:
[P1] Reserve for sink-expression materialization
This estimator sees the upstream block, but the async Iceberg writer
evaluates `_vec_output_expr_ctxs` and materializes constants only after this
reservation is transferred. A sink expression such as a large constant
STRING/ARRAY/MAP is one payload here and then expands to `rows * payload` in
`materialize_block_inplace()`, so the four-copy input estimate can admit tens
of MiB before the writer allocates GiB under the token. This is distinct from
the earlier estimator-allocation issue because the missing allocation happens
after admission. Make the bound projection-aware without materializing first
(or keep constants lazy until charged), and add a hard-limit regression with
many rows and a large constant sink expression.
--
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]
