github-actions[bot] commented on code in PR #66348:
URL: https://github.com/apache/doris/pull/66348#discussion_r3697997878
##########
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:
[P1] Do not recommit the prior block list while aborting. This SDK sends
these IDs as `Latest`, so if a failed pre-upgrade writer left an uncommitted
block with the same deterministic ID as a committed block, Azure selects the
staged bytes; staging does not change the ETag, so `IfMatch` still succeeds and
this abort silently rewrites the valid blob. Put Block List also resets omitted
metadata/properties. Please leave this writer's unique blocks to expire or use
explicit committed-version entries without mutating the object, and test a
committed legacy block plus an uncommitted same-ID replacement.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -1120,6 +1120,9 @@ private TableScan buildScan(Table table,
IcebergTableHandle handle, Optional<Con
/** Whether this table type's scan can actually honor Iceberg's
snapshot/ref selection APIs. */
private static boolean supportsSnapshotSelection(IcebergTableHandle
handle) {
+ if (!handle.hasSnapshotSelection()) {
Review Comment:
[P1] Preserve an explicit empty-table MVCC pin in the scan. A cached `-1`
hit can skip statement-table materialization while the table cache misses or is
disabled (for example, REST vended credentials). If S1 commits before scan
planning, the scan freshly loads S1; this returns false and `table.newScan()`
reads post-begin rows. A plain SELECT is not statement-consistent, and
INSERT-SELECT can diverge from the explicit-empty write/OCC handle. Please
short-circuit pinned empty to zero tasks and test cached-empty/first-commit
interleaving.
##########
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:
[P1] Do not treat UUID block IDs as isolated Azure upload sessions. These
IDs are still staged in the blob-wide namespace, and committing this writer's
list deletes every uncommitted block omitted from it. Thus if same-key writers
A and B stage blocks, A's completion (or abort recommit below) discards B's
blocks and B later fails despite its uploads succeeding; the FE-deferred
completion path has the same behavior. Please fence same-key writes with a
provider-visible conditional protocol or use per-writer temporary blobs, and
cover interleaved complete/abort cases.
##########
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:
[P1] Reserve memory for partitions this block will create. On the first
block (or any block introducing new partitions), `active_writers` is empty or
incomplete because the async consumer publishes a writer only after admission,
and an empty sorter also reports zero. The admitted block can then allocate
every per-partition selection and grow those sorters without
`_try_to_reserve_memory` covering them. This is separate from the existing
aggregation comments for already-active sorters. Please make admission
block-aware or conservatively cover not-yet-published sorters, and add a cold
many-partition test that verifies nonzero reservation before the first append.
##########
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:
[P2] Keep one root when the configured roots are canonically equal. For
example, `table.location = s3a://bucket/table` and `write.data.path =
s3://bucket/table` remain two raw strings, but `isWithin` considers each inside
the other, so this predicate drops both and the default cleanup silently lists
nothing. Please deduplicate by `FileIdentity` or remove only strictly nested
roots, and test alias-equivalent table/data locations.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteContext.java:
##########
@@ -81,12 +89,17 @@ Optional<String> getBranchName() {
/**
* The statement's READ snapshot id (the MVCC pin the scan used, S_read),
threaded from the write
- * handle in {@code planWrite}; {@code -1} = no pin (the legacy
fresh-current behavior). The
+ * handle in {@code planWrite}; {@code -1} means either no pin or an
explicitly empty read, as
+ * distinguished by {@link #isReadSnapshotPinned()}. The
* RowDelta path anchors {@code baseSnapshotId} at this snapshot so the
commit-time removeDeletes
* (option D) and the scan-time deletes BE unions into the new DV share
one snapshot — see
* {@link IcebergConnectorTransaction} [SHOULD-2] / Fix B.
*/
long getReadSnapshotId() {
return readSnapshotId;
}
+
+ boolean isReadSnapshotPinned() {
Review Comment:
[P1] Consume this explicit-empty distinction in the RowDelta begin guard
too. DELETE/UPDATE/MERGE still test only `readSnapshotId >= 0`; for a pinned
empty target they fall back to the writable table's current snapshot. If MERGE
scans the empty target and S1 commits before `beginWrite`, S1 becomes
`validateFromSnapshot` instead of a conflict, so `WHEN NOT MATCHED` rows
planned against emptiness can commit stale or duplicate results. Please
preserve an empty validation fence for RowDelta and add a
first-snapshot-between-scan-and-begin test.
##########
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<>();
Review Comment:
[P2] Include the configured Iceberg metadata root in `ownedRoots`. When
`write.metadata.path` points outside `table.location`, table operations create
metadata JSON, manifest-list, and manifest artifacts there; the default action
never lists that root, and an explicit `location` is rejected as unowned.
Failed or obsolete metadata files therefore cannot be cleaned. Please admit the
exact configured metadata root, deduplicate it canonically with table/data
roots, and test separate table, data, and metadata locations.
--
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]
