github-actions[bot] commented on code in PR #66825: URL: https://github.com/apache/doris/pull/66825#discussion_r3802593683
########## be/src/exec/operator/iceberg_sorter_reserve_memory.h: ########## @@ -0,0 +1,156 @@ +// 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 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); Review Comment: [P1] Accumulate sorted destinations retained across partitions `transient_workspace` includes the destination created by `do_sort()`, but that destination is not transient across this loop: `add_sorted_block()` retains it in the partition sorter after dispatch returns. The same incoming block can then make a second partition take the pre-append rollover while the first destination remains live, yet this reduction reserves only one maximum. Under pressure the later retained run is allocated beyond the token. Separate retained destination growth from reusable scratch and accumulate it for every partition that this block can roll over. ########## be/src/exec/sink/writer/iceberg/viceberg_sort_writer.cpp: ########## @@ -103,92 +158,48 @@ Status VIcebergSortWriter::close(const Status& status) { } Status VIcebergSortWriter::_close_locked(const Status& status) { - // Track the actual internal status of operations performed during close. - // This is important because if intermediate operations (like do_sort()) fail, - // we need to propagate the actual error status to the underlying partition writer's - // close() call, rather than the original status parameter which could be OK. - Status internal_status = Status::OK(); - // Track the close status of the underlying partition writer. - // If _iceberg_partition_writer->close() fails (e.g., Parquet file flush error), - // we must propagate this error to the caller to avoid silent data loss. - Status close_status = Status::OK(); - - // Defer ensures the underlying partition writer is always closed and - // spill streams are cleaned up, regardless of whether intermediate operations succeed. - // Uses internal_status to propagate any errors that occurred during close operations. - Defer defer {[&]() { - // If any intermediate operation failed, pass that error to the partition writer; - // otherwise, pass the original status from the caller. - close_status = - _iceberg_partition_writer->close(internal_status.ok() ? status : internal_status); - if (!close_status.ok()) { - LOG(WARNING) << fmt::format("_iceberg_partition_writer close failed, reason: {}", - close_status.to_string()); - } - _cleanup_spill_streams(); - }}; - - // If the original status is already an error or the query is cancelled, - // skip all close operations and propagate the original error - if (!status.ok() || _runtime_state->is_cancelled()) { - return status; - } - - // If sorter was never initialized (e.g., no data was written), nothing to do - if (_sorter == nullptr) { - return Status::OK(); - } - - // Check if there is any remaining data in the sorter (either unsorted or already sorted blocks) - if (!_sorter->merge_sort_state()->unsorted_block()->empty() || - !_sorter->merge_sort_state()->get_sorted_block().empty()) { - if (_sorted_spill_files.empty()) { - // No spill has occurred, all data is in memory. - // Sort the remaining data, prepare for reading, and write to file. - internal_status = _sorter->do_sort(); - if (!internal_status.ok()) { - return internal_status; + Status internal_status = status; + if (status.ok() && !_runtime_state->is_cancelled()) { + internal_status = Status::OK(); + if (_sorter != nullptr && (!_sorter->merge_sort_state()->unsorted_block()->empty() || + !_sorter->merge_sort_state()->get_sorted_block().empty())) { + if (_sorted_spill_files.empty()) { + internal_status = _sorter->do_sort(); + if (internal_status.ok()) { + internal_status = _sorter->prepare_for_read(false); + } + if (internal_status.ok()) { + internal_status = _write_sorted_data(); + } + } else { + internal_status = _do_spill(); } - internal_status = _sorter->prepare_for_read(false); - if (!internal_status.ok()) { - return internal_status; - } - internal_status = _write_sorted_data(); - return internal_status; } - - // Some data has already been spilled to disk. - // Spill the remaining in-memory data to a new spill stream. - internal_status = _do_spill(); - if (!internal_status.ok()) { - return internal_status; + if (internal_status.ok() && !_sorted_spill_files.empty()) { + internal_status = _combine_files_output(); } } - // Merge all spilled streams using multi-way merge sort and output final sorted data to files - if (!_sorted_spill_files.empty()) { - internal_status = _combine_files_output(); - if (!internal_status.ok()) { - return internal_status; - } + // Form the return value only after the underlying close runs; a deferred assignment is too late. + Status close_status = + _iceberg_partition_writer->close(internal_status.ok() ? status : internal_status); + _cleanup_spill_streams(); + if (!internal_status.ok()) { + return internal_status; } - - // Return close_status if internal operations succeeded but the underlying - // partition writer's close() failed (e.g., file flush error). - // This prevents silent data loss where the caller thinks the write succeeded - // but the file was not properly closed. return close_status; } void VIcebergSortWriter::_update_spill_block_batch_row_count(const Block& block) { auto rows = block.rows(); - // Calculate average row size from the first non-empty block to determine - // the optimal batch size for spill operations - if (rows > 0 && 0 == _avg_row_bytes) { - _avg_row_bytes = std::max(1UL, block.bytes() / rows); + if (rows > 0) { + const size_t bytes = block.bytes(); + const size_t observed_row_bytes = bytes / rows + (bytes % rows != 0); + _avg_row_bytes = std::max<size_t>(_avg_row_bytes, std::max<size_t>(1, observed_row_bytes)); Review Comment: [P1] Do not use a block average as the oversized-row bound A block with one 64 MiB STRING/ARRAY/MAP and thousands of tiny rows records only a small average here. `_spill_block_batch_row_count` can then emit a block far above the 8 MiB spill budget, while final admission still sizes every cursor/output from that budget and this same average. The indivisible outlier is deserialized and copied without covered memory. Track a conservative maximum materialized row size, or enforce the byte limit on actual emitted blocks and feed their observed maximum into merge admission. ########## be/src/exec/operator/iceberg_sorter_reserve_memory.h: ########## @@ -0,0 +1,156 @@ +// 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 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<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); Review Comment: [P1] Include each spill reader's retained serialized buffers This budgets one deserialized block per cursor, but every cursor also owns a `SpillFileReader` whose `_read_buff` is reserved to the serialized block size and whose `_pb_block` retains the parsed protobuf. `VSortedRunMerger::prepare()` primes all suppliers and keeps those readers alive, so these allocations coexist for the full fan-in. With eight incompressible 8 MiB streams the read buffers alone add roughly another 64 MiB beyond this estimate. Include per-reader serialized/protobuf memory in the bound or reduce fan-in from the measured retained footprint. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java: ########## @@ -905,6 +955,7 @@ private void commitStaticPartitionOverwrite(List<WriteResult> pendingResults) { // Set partition filter to overwrite only matching partitions overwriteFiles = overwriteFiles.overwriteByRowFilter(partitionFilter); + overwriteFiles = validateOverwrite(overwriteFiles, partitionFilter); Review Comment: [P1] Bind nested identity predicates by their full schema path The filter passed into this new overwrite validation is built with `schema.findField(field.sourceId()).name()`. For `identity(payload.region)` that yields only `region`; Iceberg needs the bindable path from `schema.findColumnName(sourceId)`. With no top-level `region`, static overwrite fails after files are staged; with a colliding top-level field, overwrite/conflict detection targets the wrong data. `buildIdentityPartitionExpression()` has the same issue for RowDelta conflict filters. Resolve both predicates through the full path and fail closed when it is absent. ########## be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp: ########## @@ -351,9 +447,12 @@ Status VIcebergTableWriter::_write_prepared_block(Block& output_block) { transformed_block.insert( {std::move(col), result_type, iceberg_partition_columns.field().name()}); } else { + Block source_block; + source_block.insert( + _nested_partition_source(output_block, iceberg_partition_columns)); Review Comment: [P1] Reject or materialize static values for nested identity sources This dynamic branch correctly extracts `payload.region`, but the preceding static-value branch bypasses it. The FE's static partition binding substitutes only top-level target columns, so `PARTITION(region='X') SELECT payload(region='Y', ...)` leaves the row leaf as `Y`; the BE routes and labels the file with static `X` while writing the unchanged STRUCT. That violates the identity-partition invariant and makes pruning return or skip wrong rows. Reject nested static syntax until the FE can replace the leaf, or materialize and validate it before writing. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java: ########## @@ -298,6 +310,280 @@ 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, reachableSchemas(icebergTable, snapshot)); + } + + private boolean hasApplicableEqualityDeletes(TableScan scan) throws UserException { + Snapshot snapshot = scan.snapshot(); + if (snapshot == null) { + return false; + } + String equalityDeleteCount = snapshot.summary().get("total-equality-deletes"); + if (equalityDeleteCount != null) { + try { + // A positive snapshot total proves V2 semantics are required without opening every + // delete manifest; only old summaries that omit the counter need the fallback. + return Long.parseLong(equalityDeleteCount) > 0; + } catch (NumberFormatException ignored) { + // Fall through for non-standard summaries instead of weakening compatibility. + } + } + // Inspect only delete manifests, not data tasks: equality-delete semantics are snapshot-wide + // compatibility state even when the current predicate happens to prune their partitions. + for (ManifestFile manifest : snapshot.deleteManifests(icebergTable.io())) { + if (!manifest.hasAddedFiles() && !manifest.hasExistingFiles()) { + continue; + } + try (ManifestReader<DeleteFile> deletes = ManifestFiles.readDeleteManifest( + manifest, icebergTable.io(), icebergTable.specs())) { + for (DeleteFile delete : deletes) { + if (delete.content() == FileContent.EQUALITY_DELETES) { + return true; + } + } + } catch (IOException e) { + throw new UserException( + "Failed to inspect Iceberg delete manifest " + manifest.path(), e); + } + } + return false; + } + + private static boolean hasSmoothUpgradeSource(Iterable<Backend> backends) { + for (Backend backend : backends) { + if (backend.isSmoothUpgradeSrc()) { + return true; + } + } + return false; + } + + private Set<Integer> projectedFieldIds(Schema scanSchema) { + return projectedFieldIds(scanSchema, desc.getSlots()); + } + + @VisibleForTesting + static Set<Integer> projectedFieldIds(Schema scanSchema, Iterable<SlotDescriptor> slots) { + Set<Integer> projected = new HashSet<>(); + for (SlotDescriptor slot : slots) { + int fieldId = slot.getColumn().getUniqueId(); + // Stable Iceberg IDs prevent a dropped-and-readded name from selecting the wrong history. + NestedField field = fieldId >= 0 ? scanSchema.findField(fieldId) + : scanSchema.caseInsensitiveFindField(slot.getColumn().getName()); + if (field != null) { + collectProjectedFieldIds(field, slot.getType(), projected); + } + } + return projected; + } + + private static void collectProjectedFieldIds( + NestedField field, org.apache.doris.catalog.Type projectedType, + Set<Integer> projected) { + projected.add(field.fieldId()); + if (projectedType instanceof StructType && field.type().isStructType()) { + for (StructField projectedChild : ((StructType) projectedType).getFields()) { + NestedField icebergChild = field.type().asStructType().fields().stream() + .filter(child -> child.name().equalsIgnoreCase(projectedChild.getName())) + .findFirst().orElse(null); + if (icebergChild != null) { + collectProjectedFieldIds(icebergChild, projectedChild.getType(), projected); + } + } + } else if (projectedType instanceof ArrayType && field.type().isListType()) { + collectProjectedFieldIds(field.type().asListType().fields().get(0), + ((ArrayType) projectedType).getItemType(), projected); + } else if (projectedType instanceof MapType && field.type().isMapType()) { + collectProjectedFieldIds(field.type().asMapType().fields().get(0), + ((MapType) projectedType).getKeyType(), projected); + collectProjectedFieldIds(field.type().asMapType().fields().get(1), + ((MapType) projectedType).getValueType(), projected); + } + } + + @VisibleForTesting + static Iterable<Schema> reachableSchemas(Table table, Snapshot selectedSnapshot) { + Map<Integer, Schema> schemas = table.schemas(); + List<Schema> reachable = new ArrayList<>(); + Set<Long> visitedSnapshots = new HashSet<>(); + Set<Integer> visitedSchemaIds = new HashSet<>(); + Snapshot snapshot = selectedSnapshot; + while (snapshot != null && visitedSnapshots.add(snapshot.snapshotId())) { + Schema schema = schemas.get(snapshot.schemaId()); + if (schema != null && visitedSchemaIds.add(snapshot.schemaId())) { + reachable.add(schema); + } + Long parentId = snapshot.parentId(); + snapshot = parentId == null ? null : table.snapshot(parentId); + if (parentId != null && snapshot == null) { + // Expiration may remove the parent metadata while descendants still inherit its + // files, so an incomplete lineage must gate against every surviving schema. + for (Schema historicalSchema : schemas.values()) { + if (visitedSchemaIds.add(historicalSchema.schemaId())) { + reachable.add(historicalSchema); + } + } + } + } + // Only ancestors of the selected ref can have produced files visible to this scan. Review Comment: [P1] Include cherry-picked file provenance in the schema gate Walking only `parentId` misses non-fast-forward cherry-picks. Iceberg creates the new snapshot on the target lineage but imports files added by the source snapshot, so a live file written under an optional/default-sensitive source schema can be visible even though that schema is not a target ancestor. This can make `requiresV2` false and schedule a V1 BE for unsupported requiredness semantics. Derive relevant schema IDs from the selected snapshot's live manifests, or follow validated cherry-pick provenance in addition to parents. -- 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]
