github-actions[bot] commented on code in PR #66348:
URL: https://github.com/apache/doris/pull/66348#discussion_r3697704515
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java:
##########
@@ -304,8 +303,10 @@ private void applyBeginGuards(IcebergWriteContext ctx,
String tableName) {
throw new IllegalArgumentException(branchName
+ " is a tag, not a branch. Tags cannot be targets
for producing snapshots");
}
+ this.baseSnapshotId = op == WriteOperation.OVERWRITE ?
branchRef.snapshotId() : null;
} else {
this.branchName = null;
+ this.baseSnapshotId = op == WriteOperation.OVERWRITE ?
getSnapshotIdIfPresent(table) : null;
Review Comment:
[P1] Anchor overwrite OCC at the statement's read state. The write handle
already carries the target's MVCC pin, but this replaces it with the head
observed at beginWrite. For a self-overwrite that scanned S1, an append can
commit S2 before beginWrite; validation then starts at S2, treats that file as
pre-existing, and the replacement derived from S1 can delete it. Please
preserve a pinned snapshot or explicit empty-read state (distinct from no pin),
resolve it against the target branch, and add scan-to-begin conflicts for each
overwrite shape.
##########
be/src/exec/operator/spill_iceberg_table_sink_operator.cpp:
##########
@@ -55,47 +64,57 @@ size_t
SpillIcebergTableSinkLocalState::get_reserve_mem_size(RuntimeState* state
if (!_writer) {
return 0;
}
- auto current_writer = _writer->current_writer();
- auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(current_writer.get());
- if (!sort_writer) {
- return 0;
+ std::vector<size_t> per_partition_reservations;
+ auto active_writers = _writer->active_writers();
+ per_partition_reservations.reserve(active_writers->size());
+ for (const auto& writer : *active_writers) {
+ if (auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(writer.get())) {
+
per_partition_reservations.push_back(sort_writer->get_reserve_mem_size(state,
eos));
+ }
}
-
- return sort_writer->get_reserve_mem_size(state, eos);
+ // One input block is partitioned among writers and consumed serially, so
their full-batch estimates overlap.
+ return bounded_iceberg_reserve_size(per_partition_reservations);
Review Comment:
[P1] Include cumulative retained growth across touched sorters. The writers
consume a split block serially, but each FullSorter can reallocate its unsorted
columns and retain that larger capacity. Growth from earlier partitions is
still live while later partitions append, so taking only the maximum can admit
a block whose total persistent growth exceeds the reservation and hit the hard
memory limit instead of spilling first. Please separate cumulative
per-partition capacity growth from maximum transient workspace (or reserve from
actual selected row counts), and test multiple near-capacity sorters touched by
one block.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,266 @@
+// 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.connector.iceberg.action;
+
+import org.apache.doris.connector.api.ConnectorColumn;
+import org.apache.doris.connector.api.ConnectorSession;
+import org.apache.doris.connector.api.ConnectorType;
+import org.apache.doris.connector.api.DorisConnectorException;
+import org.apache.doris.connector.api.pushdown.ConnectorPredicate;
+import org.apache.doris.foundation.util.ArgumentParsers;
+
+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.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+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();
+ public static final String OLDER_THAN = "older_than";
+ public static final String LOCATION = "location";
+ public static final String DRY_RUN = "dry_run";
+
+ public IcebergRemoveOrphanFilesAction(Map<String, String> properties,
List<String> partitionNames,
+ ConnectorPredicate whereCondition) {
+ super("remove_orphan_files", properties, partitionNames,
whereCondition);
+ }
+
+ @Override
+ protected void registerIcebergArguments() {
+ namedArguments.registerRequiredArgument(OLDER_THAN, "Creation time
cutoff in milliseconds",
+ ArgumentParsers.nonNegativeLong(OLDER_THAN));
+ namedArguments.registerOptionalArgument(LOCATION, "Prefix within the
table location",
+ null, ArgumentParsers.nonEmptyString(LOCATION));
+ namedArguments.registerOptionalArgument(DRY_RUN, "Only count orphan
files", true,
+ ArgumentParsers.booleanValue(DRY_RUN));
+ }
+
+ @Override
+ protected void validateIcebergAction() {
+ validateNoPartitions();
+ validateNoWhereCondition();
+ String location = namedArguments.getString(LOCATION);
+ if (location != null) {
+ try {
+ normalizeLocation(location);
+ } catch (IllegalArgumentException e) {
+ throw new DorisConnectorException("Invalid location URI: " +
location, e);
+ }
+ }
+ }
+
+ @Override
+ protected List<String> executeAction(Table table, ConnectorSession
session) {
+ if (!(table.io() instanceof SupportsPrefixOperations)) {
+ throw new DorisConnectorException("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 DorisConnectorException("Cannot remove orphan files:
Iceberg GC is disabled");
+ }
+ String tableLocation = normalizeLocation(table.location());
+ String scanLocation = namedArguments.getString(LOCATION);
+ scanLocation = scanLocation == null ? tableLocation :
normalizeLocation(scanLocation);
+ // Normalize dot segments before the containment check so local FileIO
paths cannot escape the table root.
+ if (!scanLocation.equals(tableLocation) &&
!scanLocation.startsWith(tableLocation + "/")) {
Review Comment:
[P2] Allow cleanup of configured table-owned data roots. Doris writes
Iceberg data to WRITE_DATA_LOCATION (and the supported object-store/folder
fallbacks), which may be outside table.location(). The default call never lists
that prefix, and this check rejects an explicit call for it, so failed writes
there leave orphans that this procedure cannot remove. Please admit the exact
configured table-owned roots while keeping arbitrary prefixes fail-closed, and
test distinct metadata/data prefixes.
##########
be/src/io/fs/azure_obj_storage_client.cpp:
##########
@@ -64,10 +65,13 @@ std::string to_lower_ascii(std::string_view input) {
return lowered;
}
-auto base64_encode_part_num(int part_num) {
- uint8_t buf[4];
- doris::encode_fixed32_le(buf, static_cast<uint32_t>(part_num));
- return Aws::Utils::HashingUtils::Base64Encode({buf, sizeof(buf)});
+std::string azure_block_id(const doris::io::ObjectStoragePathOptions& opts,
int part_num) {
+ DCHECK(opts.upload_id.has_value());
+ // Azure requires every block ID for one blob to have the same decoded
length.
+ std::string raw_id = fmt::format("{}:{:010}", *opts.upload_id, part_num);
Review Comment:
[P1] Preserve Azure block IDs across deferred Hive completion. This stages
each block as Base64(UUID:part), but the Hive pending-upload payload carries
only the UUID and part numbers, while FE Azure completion ignores uploadId and
reconstructs Base64(little-endian int) IDs. It therefore asks Azure to commit
block IDs that BE never staged, so Azure Hive writes fail at commit. Please
make FE derive the UUID-prefixed IDs with rolling-version compatibility, or
carry the exact staged IDs, and cover the BE-to-FE completion path.
--
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]
