Gabriel39 commented on code in PR #66348:
URL: https://github.com/apache/doris/pull/66348#discussion_r3698114972
##########
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:
Fixed in 399b6fe356. Azure uploads now stage exact UUID-prefixed IDs on
per-upload temporary blobs. New-BE/old-FE rolling compatibility is preserved by
additionally staging deterministic legacy IDs for deferred completion, while
the new FE falls back consistently when block IDs are missing.
##########
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:
Fixed in 481123549a and 399b6fe356. The statement read snapshot is threaded
into write validation, including an explicitly empty -1 snapshot. The added
tests verify that overwrite and RowDelta validation reject concurrent commits
after the statement snapshot.
##########
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:
Fixed in 481123549a and 399b6fe356. Orphan cleanup now treats configured
data-location fallbacks and write.metadata.path as owned roots, canonicalizes
scheme aliases, and scans each non-nested root independently.
##########
be/src/exec/operator/spill_iceberg_table_sink_operator.cpp:
##########
@@ -55,47 +55,60 @@ 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<IcebergSorterReserveMemory> per_partition_reservations;
+ auto active_writers = _writer->active_writers();
Review Comment:
Fixed in 399b6fe356. The reserve now includes a conservative
incoming-block/cold-writer allowance even when no sorter has reached the
published snapshot. A regression test covers the first-block case with no
published partition writer.
##########
be/src/io/fs/azure_obj_storage_client.cpp:
##########
@@ -238,7 +247,7 @@ ObjectStorageResponse
AzureObjStorageClient::complete_multipart_upload(
std::vector<std::string> string_block_ids;
std::ranges::transform(
completed_parts, std::back_inserter(string_block_ids),
- [](const ObjectCompleteMultiPart& i) { return
base64_encode_part_num(i.part_num); });
+ [&opts](const ObjectCompleteMultiPart& i) { return
azure_block_id(opts, i.part_num); });
Review Comment:
Fixed in 399b6fe356. Each upload now stages blocks on a per-upload temporary
blob, so Azure's blob-wide staged-block namespace is isolated. Completion
publishes by copying the committed temporary blob to the target; the same-key
integration test verifies independent writers.
##########
be/src/io/fs/azure_obj_storage_client.cpp:
##########
@@ -247,6 +256,41 @@ ObjectStorageResponse
AzureObjStorageClient::complete_multipart_upload(
opts, _tls_debug_context);
}
+ObjectStorageResponse AzureObjStorageClient::abort_multipart_upload(
+ const ObjectStoragePathOptions& opts) {
+ auto client = _client->GetBlockBlobClient(opts.key);
+ auto response = do_azure_client_call(
+ [&]() {
+ GetBlockListOptions get_options;
+ get_options.ListType = Models::BlockListType::All;
+ auto block_list = client.GetBlockList(get_options);
+ const bool has_committed_blob =
azure_block_list_has_committed_blob(
+ block_list.Value.CommittedBlocks.size(),
block_list.Value.ETag.HasValue());
+ if (!has_committed_blob) {
+ // Uncommitted blocks are invisible and expire without
deleting a racing commit.
+ return;
+ }
+ if (block_list.Value.CommittedBlocks.empty()) {
+ // Azure cannot selectively discard staged blocks without
replacing Put Blob content.
+ return;
+ }
+ std::vector<std::string> committed_ids;
+ committed_ids.reserve(block_list.Value.CommittedBlocks.size());
+ std::ranges::transform(block_list.Value.CommittedBlocks,
+ std::back_inserter(committed_ids),
+ [](const Models::BlobBlock& block) {
return block.Name; });
+ CommitBlockListOptions commit_options;
+ commit_options.AccessConditions.IfMatch =
block_list.Value.ETag;
+ // Recommitting only the old IDs discards this writer's unique
staged blocks.
+ client.CommitBlockList(committed_ids, commit_options);
Review Comment:
Fixed in 399b6fe356. Abort deletes only this writer's temporary blob and
never reads or recommits the final blob. Legacy blocks staged on the final blob
are left to expire, avoiding replacement of committed content with same-ID
staged blocks.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,317 @@
+// 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.ArrayList;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+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");
+ }
+ List<String> scanLocations = resolveScanLocations(table);
+
+ try {
+ ReachableIndex reachable = new
ReachableIndex(collectReachableFiles(table));
+ long orphanCount = 0;
+ long deletedCount = 0;
+ long olderThan = namedArguments.getLong(OLDER_THAN);
+ // The SQL procedure needs a retention fence because concurrent
uploads are not reachable until commit.
+ if (olderThan > System.currentTimeMillis() - MIN_RETENTION_MS) {
+ throw new DorisConnectorException(
+ "older_than must retain at least 24 hours of files");
+ }
+ boolean dryRun = namedArguments.getBoolean(DRY_RUN);
+ Set<String> visitedFiles = new HashSet<>();
+ for (String scanLocation : scanLocations) {
+ // Object stores use raw prefix matching, so the separator
excludes sibling prefixes.
+ String listingPrefix = scanLocation.endsWith("/") ?
scanLocation : scanLocation + "/";
+ for (FileInfo file : ((SupportsPrefixOperations)
table.io()).listPrefix(listingPrefix)) {
+ if (visitedFiles.add(file.location()) &&
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));
+ } catch (Exception e) {
+ throw new DorisConnectorException("Failed to remove orphan files:
" + e.getMessage(), e);
+ }
+ }
+
+ private List<String> resolveScanLocations(Table table) {
+ String tableRoot = normalizeLocation(table.location());
+ String dataRoot = normalizeLocation(resolveDataLocation(table,
tableRoot));
+ Set<String> ownedRoots = new LinkedHashSet<>();
+ ownedRoots.add(tableRoot);
+ ownedRoots.add(dataRoot);
+
+ String requested = namedArguments.getString(LOCATION);
+ if (requested != null) {
+ String normalized = normalizeLocation(requested);
+ boolean owned = ownedRoots.stream().anyMatch(root ->
isWithin(normalized, root));
+ if (!owned) {
+ throw new DorisConnectorException(
+ "location must be within an Iceberg table-owned
metadata or data location");
+ }
+ return Lists.newArrayList(normalized);
+ }
+
+ List<String> roots = new ArrayList<>();
+ for (String candidate : ownedRoots) {
+ // Avoid listing a nested default data directory twice when the
table root already covers it.
+ if (ownedRoots.stream().noneMatch(other ->
!other.equals(candidate) && isWithin(candidate, other))) {
Review Comment:
Fixed in 399b6fe356. Canonically equal roots are deduplicated before strict
nesting minimization, so s3, s3a, and s3n aliases retain one scan root. A
regression test covers the alias-equal case.
--
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]
