Gabriel39 commented on code in PR #66825:
URL: https://github.com/apache/doris/pull/66825#discussion_r3802172872
##########
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:
Fixed in e70d5659c3c. Non-spill merge output now uses the same byte budget
covered by EOS admission, and row width is sampled before FullSorter consumes
the input block. A wide prior run plus tiny tail regression test covers the
rollover case.
##########
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:
Fixed in e70d5659c3c. Dry-run now refreshes the Iceberg table before
building the reachable-file index, with a regression test verifying refresh
occurs before the scan.
##########
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:
Fixed in e70d5659c3c. The planned coordinator is published before
cancellation is rechecked, making cancellation sticky across the planning
handoff. Cleanup also uses a shared deadline, and the regression test covers
cancellation while planning has no coordinator.
--
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]
