github-actions[bot] commented on code in PR #66825: URL: https://github.com/apache/doris/pull/66825#discussion_r3801930066
########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRemoveOrphanFilesAction.java: ########## @@ -0,0 +1,412 @@ +// 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.IcebergCommitCoordinator; +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; + private static final long MAX_REACHABLE_INDEX_BYTES = 256L * 1024 * 1024; + 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(); + 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"); + } + + try { + if (namedArguments.getBoolean(DRY_RUN)) { Review Comment: [P2] Refresh before computing the dry-run candidates `table` comes from `IcebergExternalMetaCache`, but this default dry-run is the only branch that never calls `refresh()`. If another engine commits an already-old staged object before the action, prefix listing sees the object while the cached reachable metadata can omit it, so the preview reports a currently referenced file as orphan; the destructive invocation then refreshes and can produce a different set. Refresh before the preview scan too, and cover a cached generation that predates a commit of an old staged file. ########## be/src/exec/sort/sorter.cpp: ########## @@ -184,35 +199,81 @@ bool FullSorter::has_enough_capacity(Block* input_block, Block* unsorted_block) } size_t FullSorter::get_reserve_mem_size(RuntimeState* state, bool eos) const { - size_t size_to_reserve = 0; + return get_reserve_mem_size_components(state, eos).total(); +} + +SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState* state, + bool eos) const { + return get_reserve_mem_size_components(state, eos, std::numeric_limits<size_t>::max()); +} + +SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState* state, bool eos, + size_t sort_threshold_bytes) const { + const auto rows = _state->unsorted_block()->rows(); + const auto bytes = _state->unsorted_block()->bytes(); + const auto bytes_per_row = rows == 0 ? 0 : bytes / rows; + return get_reserve_mem_size_components( + state, eos, state->batch_size(), + saturating_multiply_size(bytes_per_row, state->batch_size()), sort_threshold_bytes); +} + +SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState* state, bool eos, + size_t incoming_rows, + size_t incoming_bytes) const { + return get_reserve_mem_size_components(state, eos, incoming_rows, incoming_bytes, + std::numeric_limits<size_t>::max()); +} + +SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState* state, bool eos, + size_t incoming_rows, + size_t incoming_bytes, + size_t sort_threshold_bytes) const { + SorterReserveMemory reserve; const auto rows = _state->unsorted_block()->rows(); if (rows != 0) { const auto bytes = _state->unsorted_block()->bytes(); const auto allocated_bytes = _state->unsorted_block()->allocated_bytes(); - const auto bytes_per_row = bytes / rows; - const auto estimated_size_of_next_block = bytes_per_row * state->batch_size(); - auto new_block_bytes = estimated_size_of_next_block + bytes; - auto new_rows = rows + state->batch_size(); + auto new_block_bytes = saturating_add_size(bytes, incoming_bytes); + auto new_rows = saturating_add_size(rows, incoming_rows); // If the new size is greater than 85% of allocalted bytes, it maybe need to realloc. - if ((new_block_bytes * 100 / allocated_bytes) >= 85) { - size_to_reserve += (size_t)(allocated_bytes * 1.15); + const auto growth_threshold = static_cast<size_t>( + (static_cast<unsigned __int128>(allocated_bytes) * 85 + 99) / 100); + const size_t growth_trigger_bytes = growth_threshold > bytes ? growth_threshold - bytes : 0; + if (incoming_rows > 0 && growth_trigger_bytes <= incoming_bytes) { + reserve.retained_growth = static_cast<size_t>(std::min<unsigned __int128>( + (static_cast<unsigned __int128>(allocated_bytes) * 115 + 99) / 100, + std::numeric_limits<size_t>::max())); + reserve.retained_growth_trigger_bytes = growth_trigger_bytes; } - auto sort = new_rows > _buffered_block_size || new_block_bytes > _buffered_block_bytes; + // Iceberg close forces every nonempty pending run to sort at EOS, even when the generic + // append thresholds are not reached, so admission must cover that final allocation too. + // The reservation must mirror every caller-side rollover that immediately invokes do_sort(). + // After the retained-capacity threshold, append_block may sort before inserting when any + // individual column is full. Admission lacks the post-projection column distribution, so + // every nonempty append at that boundary must conservatively reserve the transient sort. + const bool may_rollover_before_append = incoming_rows > 0 && _reach_limit(); + auto sort = may_rollover_before_append || (eos && new_rows > 0) || Review Comment: [P1] Reserve the non-spill merge output at EOS This predicate reserves the pending run's sort, but close subsequently calls `_write_sorted_data()`, where `FullSorter::get_next()` materializes up to `RuntimeState::batch_size()` rows while the merge cursors still retain the sorted runs. Capacity/target rollovers can leave several wide runs and an empty or tiny final run, so the EOS reservation regains no reliable headroom before allocating that output block; `Block::clear_column_data()` also retains column capacity. The spill-merger batch bound does not cover this non-spill path. Include a bounded merge-output block in the EOS transient reservation (or bound `_write_sorted_data()` to admitted bytes) and add a wide-row test with prior runs plus a tiny/empty final run. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/rewrite/RewriteGroupTask.java: ########## @@ -186,6 +210,11 @@ private void executeGroup(ConnectContext taskConnectContext, StatementBase taskParsedStmt) throws Exception { // Step 1: Create stmt executor stmtExecutor = new StmtExecutor(taskConnectContext, taskParsedStmt); + if (isCanceled.get()) { Review Comment: [P1] Keep cancellation sticky through rewrite planning This recheck only closes cancellation before `initPlan()`. If timeout cancellation arrives during planning, `cancel()` sees a non-null `StmtExecutor` but its coordinator is not installed yet; `StmtExecutor.cancel()` stores no pending state, and there is no post-plan recheck. The task can therefore enter `executeSingleInsert()` after the parent has canceled it, while `cancelAndQuiesce()` waits without a deadline and continues holding the rewrite transaction fence. Add a race-free cancellation handoff/post-plan check (and bounded cleanup), with `initPlan()` paused before coordinator publication. -- 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]
