github-actions[bot] commented on code in PR #66825:
URL: https://github.com/apache/doris/pull/66825#discussion_r4244425673
##########
be/src/exec/sink/writer/iceberg/viceberg_sort_writer.cpp:
##########
@@ -81,23 +82,70 @@ size_t VIcebergSortWriter::data_size() const {
return _sorter == nullptr ? 0 : _sorter->data_size();
}
-size_t VIcebergSortWriter::get_reserve_mem_size(RuntimeState* state, bool eos)
const {
+SorterReserveMemory
VIcebergSortWriter::get_reserve_mem_size_components(RuntimeState* state,
+ bool
eos) const {
std::lock_guard<std::mutex> lock(_sorter_mutex);
- size_t reserve_size = _sorter == nullptr ? 0 :
_sorter->get_reserve_mem_size(state, eos);
+ if (_sorter == nullptr) {
+ return {};
+ }
+ const size_t target = _target_file_size_bytes >= 0
+ ?
static_cast<size_t>(_target_file_size_bytes)
+ : std::numeric_limits<size_t>::max();
+ const auto& unsorted_block = _sorter->merge_sort_state()->unsorted_block();
+ const size_t incoming_rows = static_cast<size_t>(state->batch_size());
+ const size_t bytes_per_row =
+ unsorted_block->rows() == 0 ? 0 : unsorted_block->bytes() /
unsorted_block->rows();
+ const size_t incoming_bytes = iceberg_saturating_multiply(bytes_per_row,
incoming_rows);
+ auto reservation = _sorter->get_reserve_mem_size_components(state, eos,
incoming_rows,
+
incoming_bytes, target);
+ const bool reaches_target =
+ incoming_rows > 0 && target != std::numeric_limits<size_t>::max()
&&
+ iceberg_saturating_add(_sorter->data_size(), incoming_bytes) >=
target;
+ _include_merge_reservation(state, eos, reaches_target, &reservation);
+ return reservation;
+}
+
+SorterReserveMemory VIcebergSortWriter::get_reserve_mem_size_components(
+ RuntimeState* state, bool eos, size_t incoming_rows, size_t
incoming_bytes) const {
+ std::lock_guard<std::mutex> lock(_sorter_mutex);
+ if (_sorter == nullptr) {
+ return {};
+ }
+ const size_t target = _target_file_size_bytes >= 0
+ ?
static_cast<size_t>(_target_file_size_bytes)
+ : std::numeric_limits<size_t>::max();
+ auto reservation = _sorter->get_reserve_mem_size_components(state, eos,
incoming_rows,
+
incoming_bytes, target);
+ const bool reaches_target =
+ incoming_rows > 0 && target != std::numeric_limits<size_t>::max()
&&
+ iceberg_saturating_add(_sorter->data_size(), incoming_bytes) >=
target;
+ _include_merge_reservation(state, eos, reaches_target, &reservation);
+ return reservation;
+}
+
+void VIcebergSortWriter::_include_merge_reservation(RuntimeState* state, bool
eos,
+ bool reaches_target,
+ SorterReserveMemory*
reservation) const {
+ if (!eos && !reaches_target) {
+ return;
+ }
if (eos && !_sorted_spill_files.empty()) {
- const size_t buffer_size =
static_cast<size_t>(state->spill_buffer_size_bytes());
- const size_t merge_limit =
static_cast<size_t>(state->spill_sort_merge_mem_limit_bytes());
- const size_t max_fan_in = std::max<size_t>(2, merge_limit /
buffer_size);
- // Reservation is computed before sink(), so a non-empty EOS can add
one final spill run.
- const size_t selected_streams = std::min(_sorted_spill_files.size(),
max_fan_in - 1) + 1;
- // Every selected cursor eagerly owns one input block. A multiway
merge also builds a
- // separate output block while those inputs remain live.
- const size_t merge_buffer_count = selected_streams + (selected_streams
> 1 ? 1 : 0);
- DORIS_CHECK(buffer_size <= std::numeric_limits<size_t>::max() /
merge_buffer_count);
- const size_t merge_reserve_size = merge_buffer_count * buffer_size;
- reserve_size = std::max({reserve_size, merge_limit,
merge_reserve_size});
+ size_t spill_file_count = _sorted_spill_files.size();
+ if (_sorter->data_size() > 0) {
Review Comment:
[P1] Count the spill run created by a nonempty EOS block. If an earlier
spill reset `_sorter`, this check sees zero buffered bytes and reserves a
one-reader final merge. The EOS rows are then appended, and `_close_locked()`
spills them as a second run before merging; with default 8 MiB buffers, the
reserved merge workspace is 48 MiB but the two-reader workspace is 72 MiB.
Include a possible incoming EOS run even when the pre-sink sorter is empty, and
cover the one-to-two-run case.
##########
be/src/exec/sink/writer/paimon/paimon_table_writer.cpp:
##########
@@ -21,17 +21,56 @@
#include "common/logging.h"
#include "core/block/block.h"
#include "core/block/materialize_block.h"
+#include "exec/sink/writer/external_file_report_compatibility.h"
#include "exprs/vexpr_context.h"
#include "runtime/runtime_state.h"
namespace doris {
+PaimonPreparedCommitOwner::PaimonPreparedCommitOwner(std::unique_ptr<IPaimonWriter>
writer,
+
std::unique_ptr<IPaimonWriteBackend> backend)
+ : _writer(std::move(writer)), _backend(std::move(backend)) {}
+
+PaimonPreparedCommitOwner::~PaimonPreparedCommitOwner() {
+ _close();
+}
+
+Status PaimonPreparedCommitOwner::prepare_for_report() {
+ return _backend == nullptr ? Status::OK() :
_backend->prepare_close_for_commit();
+}
+
+void PaimonPreparedCommitOwner::finalize(ExternalFileReportOutcome outcome) {
+ if (_finalized || outcome == ExternalFileReportOutcome::AMBIGUOUS) {
+ return;
+ }
+ _finalized = true;
Review Comment:
[P1] Retain Paimon cleanup ownership when abort fails. After a rejected
final report, this sets `_finalized` before `_writer->abort()`; if the Paimon
committer fails to delete prepared files, the code only logs the error and
`_close()` discards the writer/backend. FE never accepted the commit messages,
so no owner can retry those files. Keep an abort-capable owner and prepared
messages until cleanup succeeds or reaches an authoritative terminal outcome
(including the Java `abortWriter()` state-clearing path), and exercise an abort
failure.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergCommitCoordinator.java:
##########
@@ -0,0 +1,90 @@
+// 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;
+
+import java.net.URI;
+import java.util.Locale;
+import java.util.concurrent.locks.StampedLock;
+
+/** Serializes destructive maintenance with Doris-coordinated commits for one
Iceberg table. */
+public final class IcebergCommitCoordinator {
+ private static final int STRIPE_COUNT = 1024;
+ private static final StampedLock[] TABLE_LOCKS = new
StampedLock[STRIPE_COUNT];
+
+ static {
+ for (int i = 0; i < TABLE_LOCKS.length; i++) {
+ TABLE_LOCKS[i] = new StampedLock();
+ }
+ }
+
+ private IcebergCommitCoordinator() {
+ }
+
+ public static Guard beginCommit(String tableLocation) {
+ StampedLock lock = lockFor(tableLocation);
+ return new Guard(lock, lock.readLock(), false);
+ }
+
+ public static Guard beginMaintenance(String tableLocation) {
+ StampedLock lock = lockFor(tableLocation);
+ return new Guard(lock, lock.writeLock(), true);
+ }
+
+ static StampedLock lockFor(String tableLocation) {
+ URI uri = URI.create(tableLocation).normalize();
+ String scheme = uri.getScheme() == null ? "" :
uri.getScheme().toLowerCase(Locale.ROOT);
+ if (scheme.equals("s3a") || scheme.equals("s3n")) {
+ scheme = "s3";
+ }
+ String authority = uri.getAuthority() == null
+ ? "" : uri.getAuthority().toLowerCase(Locale.ROOT);
+ String path = uri.getPath() == null ? "" : uri.getPath();
+ while (path.length() > 1 && path.endsWith("/")) {
+ path = path.substring(0, path.length() - 1);
+ }
+ String normalizedLocation = scheme + "://" + authority + path;
+ return TABLE_LOCKS[Math.floorMod(normalizedLocation.hashCode(),
STRIPE_COUNT)];
Review Comment:
[P2] Keep long maintenance work from blocking unrelated Iceberg tables.
`lockFor()` maps every table location to one of 1024 locks; distinct valid
locations such as `s3://bucket/Aa` and `s3://bucket/BB` have the same Java
hash. `remove_orphan_files` holds a write stamp through metadata reads, prefix
listing, and deletions, so an insert on the other table waits in
`beginCommit()` for the entire cleanup, with no cancellation-aware wait. Key
the guard by exact normalized location with a bounded lifecycle, or otherwise
avoid holding a shared stripe during the full external scan; cover colliding
locations in a concurrency test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,414 @@
+// 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.IcebergMetadataOps;
+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, IcebergMetadataOps
metadataOps) {
+ super("remove_orphan_files", properties, partitionNamesInfo,
whereCondition, metadataOps);
+ }
+
+ @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> executeIcebergAction(TableIf tableIf, Table table)
throws UserException {
+ // The base action retains this writable catalog generation through
listing and deletion.
+ 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)) {
+ // Preview must classify files against the latest metadata
generation just like deletion.
+ table.refresh();
+ return scanAndDeleteOrphans(table, olderThan, true);
+ }
+ // Refresh only after acquiring the commit fence, then keep the
same metadata generation
+ // stable until deletion finishes so an old imported file cannot
become reachable midway.
+ try (IcebergCommitCoordinator.Guard ignored =
+
IcebergCommitCoordinator.beginMaintenance(table.location())) {
+ table.refresh();
+ return scanAndDeleteOrphans(table, olderThan, false);
+ }
+ } catch (Exception e) {
+ throw new UserException("Failed to remove orphan files: " +
e.getMessage(), e);
+ }
+ }
+
+ private List<String> scanAndDeleteOrphans(Table table, long olderThan,
boolean dryRun)
+ throws IOException, UserException {
+ if (!(table.io() instanceof SupportsPrefixOperations)) {
+ throw new UserException("remove_orphan_files requires FileIO
prefix listing support");
+ }
+ if (!PropertyUtil.propertyAsBoolean(table.properties(),
TableProperties.GC_ENABLED,
+ TableProperties.GC_ENABLED_DEFAULT)) {
+ // A GC-disabled table may share files with another table, so no
destructive scan is safe.
+ throw new UserException("Cannot remove orphan files: Iceberg GC is
disabled");
+ }
+ List<ScanScope> scanScopes = resolveScanScopes(table);
+ ReachableIndex reachable = collectReachableFiles(table);
+ long orphanCount = 0;
+ long deletedCount = 0;
+ for (ScanScope scope : scanScopes) {
+ // Object stores use raw prefix matching, so the separator
excludes sibling prefixes.
+ String listingPrefix = scope.root.endsWith("/") ? scope.root :
scope.root + "/";
+ for (FileInfo file : ((SupportsPrefixOperations)
table.io()).listPrefix(listingPrefix)) {
+ // Unknown creation time cannot prove the file predates every
in-flight writer.
+ if (scope.owns(file.location()) && file.createdAtMillis() > 0
+ && file.createdAtMillis() < olderThan
+ && !isReachable(file.location(), reachable)) {
+ orphanCount++;
+ if (!dryRun) {
+ table.io().deleteFile(file.location());
+ deletedCount++;
+ }
+ }
+ }
+ }
+ return Lists.newArrayList(String.valueOf(orphanCount),
String.valueOf(deletedCount));
+ }
+
+ private List<ScanScope> resolveScanScopes(Table table) throws
UserException {
+ String tableRoot = normalizeLocation(table.location());
+ String requested = namedArguments.getString(LOCATION);
+ if (requested != null) {
+ String normalized = normalizeLocation(requested);
+ if (isWithinLocation(normalized, tableRoot)) {
+ return Lists.newArrayList(ScanScope.exclusive(normalized));
+ }
+ if (!namedArguments.getBoolean(ALLOW_UNSAFE_LOCATION)) {
+ throw new UserException(
+ "Cannot prove that location is owned by this table;
set allow_unsafe_location=true "
+ + "only after verifying the prefix is
exclusive to the table");
+ }
+ // This explicit escape hatch also covers historical roots after a
table-location migration.
+ return Lists.newArrayList(ScanScope.exclusive(normalized));
+ }
+ if
(nonEmpty(table.properties().get(TableProperties.WRITE_LOCATION_PROVIDER_IMPL))
!= null) {
+ throw new UserException(
+ "remove_orphan_files cannot infer ownership for a custom
write.location-provider.impl; "
+ + "provide location with
allow_unsafe_location=true after verifying exclusivity");
+ }
+ String metadataRoot =
nonEmpty(table.properties().get(TableProperties.WRITE_METADATA_LOCATION));
+ if (metadataRoot != null &&
!isWithinLocation(normalizeLocation(metadataRoot), tableRoot)) {
+ throw new UserException(
+ "Cannot prove that the configured external metadata
location is table-exclusive; "
+ + "provide location with
allow_unsafe_location=true after verifying exclusivity");
+ }
+ List<ScanScope> scopes = new ArrayList<>();
+ scopes.add(ScanScope.exclusive(tableRoot));
+ if
(Boolean.parseBoolean(table.properties().get(TableProperties.OBJECT_STORE_ENABLED)))
{
+ // Match Iceberg's ObjectStoreLocationProvider precedence exactly.
+ String objectRoot =
nonEmpty(table.properties().get(TableProperties.WRITE_DATA_LOCATION));
+ if (objectRoot == null) {
+ objectRoot =
nonEmpty(table.properties().get(TableProperties.OBJECT_STORE_PATH));
+ }
+ if (objectRoot == null) {
+ objectRoot =
nonEmpty(table.properties().get(TableProperties.WRITE_FOLDER_STORAGE_LOCATION));
+ }
+ if (objectRoot != null &&
!isWithinLocation(normalizeLocation(objectRoot), tableRoot)) {
+ // The hashed suffix is not a unique ownership key across
catalogs sharing a root.
+ throw new UserException(
+ "Cannot prove that the configured object-store root is
table-exclusive; "
+ + "provide location with
allow_unsafe_location=true after verifying exclusivity");
+ }
+ } else {
+ String externalDataRoot =
nonEmpty(table.properties().get(TableProperties.WRITE_DATA_LOCATION));
+ if (externalDataRoot == null) {
+ externalDataRoot = nonEmpty(
+
table.properties().get(TableProperties.WRITE_FOLDER_STORAGE_LOCATION));
+ }
+ if (externalDataRoot != null
+ && !isWithinLocation(normalizeLocation(externalDataRoot),
tableRoot)) {
+ throw new UserException(
+ "Cannot prove that the configured external data
location is table-exclusive; "
+ + "provide location with
allow_unsafe_location=true after verifying exclusivity");
+ }
+ }
+ return scopes;
+ }
+
+ private static String nonEmpty(String location) {
+ return location == null || location.isEmpty() ? null : location;
+ }
+
+ private ReachableIndex collectReachableFiles(Table table) throws
IOException, UserException {
+ ReachableIndex reachable = new ReachableIndex(MAX_REACHABLE_FILES,
MAX_REACHABLE_INDEX_BYTES);
+ reachable.addAll(ReachableFileUtil.metadataFileLocations(table, true));
Review Comment:
[P2] Bound metadata traversal before building the orphan index. This
`recursive=true` helper reads every older metadata JSON recursively before
`ReachableIndex.addAll()` can enforce the new 5M-entry/256 MiB limit. With
`write.metadata.previous-versions-max=1` and the default disabled post-commit
metadata deletion, a streaming table can accumulate thousands of one-link
metadata versions; this call can exhaust the FE stack or heap and prevent
cleanup. Iceberg's own orphan action uses the current metadata log without
recursion. Use that direct log or an iterative bounded walk, and cover a long
metadata chain.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java:
##########
@@ -341,6 +342,14 @@ public AbstractInsertExecutor initPlan(ConnectContext ctx,
StmtExecutor stmtExec
throw new AnalysisException("Insert plan failed. Could not get target
table lock.");
}
+ static void prepareMvccSnapshotsForPlanAttempt(StatementContext
statementContext, int attempt) {
+ if (attempt > 1) {
+ // MTMV injects snapshot fences before the first attempt; only a
failed planner attempt
+ // may be cleared so its stale metadata cannot leak into the retry.
+ statementContext.resetMvccSnapshots();
Review Comment:
[P1] Preserve MTMV's injected snapshots on planner retries. `MTMVTask`
installs the source snapshots used to choose refresh partitions before this
INSERT runs. A concurrent target ID/schema change can trigger attempt 2 here,
which clears those task fences and lets the retried scan use newer source data
for the previously chosen partitions. The earlier first-attempt reset was
fixed, but this retry-only path remains; retain or restore task-injected fences
across planner attempts and test a target-schema retry with a source commit.
--
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]
