github-actions[bot] commented on code in PR #66348:
URL: https://github.com/apache/doris/pull/66348#discussion_r3698246014
##########
be/src/io/fs/azure_obj_storage_client.cpp:
##########
@@ -216,35 +237,78 @@ ObjectStorageResponse
AzureObjStorageClient::put_object(const ObjectStoragePathO
ObjectStorageUploadResponse AzureObjStorageClient::upload_part(const
ObjectStoragePathOptions& opts,
std::string_view stream,
int part_num) {
- auto client = _client->GetBlockBlobClient(opts.key);
+ DCHECK(opts.upload_id.has_value());
+ auto client =
_client->GetBlockBlobClient(azure_multipart_temp_key(opts.key,
*opts.upload_id));
+ std::string block_id = azure_block_id(opts, part_num);
auto resp = do_azure_client_call(
[&]() {
Azure::Core::IO::MemoryBodyStream memory_body(
reinterpret_cast<const uint8_t*>(stream.data()),
stream.size());
// The blockId must be base64 encoded
SCOPED_BVAR_LATENCY(s3_bvar::s3_multi_part_upload_latency);
- client.StageBlock(base64_encode_part_num(part_num),
memory_body);
+ client.StageBlock(block_id, memory_body);
+ if (opts.deferred_completion) {
+ // During rolling upgrades an old FE still commits
deterministic IDs on the final blob.
+ Azure::Core::IO::MemoryBodyStream legacy_body(
+ reinterpret_cast<const uint8_t*>(stream.data()),
stream.size());
+ _client->GetBlockBlobClient(opts.key).StageBlock(
Review Comment:
[P2] Charge the byte limiter for the compatibility mirror.
`RateLimitedObjStorageClient::upload_part` reserves only `stream.size()` bytes,
but this second `StageBlock` sends the full stream again, so deferred Azure
writes can consume about twice the configured PUT bandwidth. Reserve/account
for both copies (subject to the existing clamp), or negotiate away the mirror
after the rolling window, and add a deferred Azure test that observes both
transfers in the byte quota.
##########
be/src/io/fs/azure_obj_storage_client.cpp:
##########
@@ -216,35 +237,78 @@ ObjectStorageResponse
AzureObjStorageClient::put_object(const ObjectStoragePathO
ObjectStorageUploadResponse AzureObjStorageClient::upload_part(const
ObjectStoragePathOptions& opts,
std::string_view stream,
int part_num) {
- auto client = _client->GetBlockBlobClient(opts.key);
+ DCHECK(opts.upload_id.has_value());
+ auto client =
_client->GetBlockBlobClient(azure_multipart_temp_key(opts.key,
*opts.upload_id));
+ std::string block_id = azure_block_id(opts, part_num);
auto resp = do_azure_client_call(
[&]() {
Azure::Core::IO::MemoryBodyStream memory_body(
reinterpret_cast<const uint8_t*>(stream.data()),
stream.size());
// The blockId must be base64 encoded
SCOPED_BVAR_LATENCY(s3_bvar::s3_multi_part_upload_latency);
- client.StageBlock(base64_encode_part_num(part_num),
memory_body);
+ client.StageBlock(block_id, memory_body);
+ if (opts.deferred_completion) {
+ // During rolling upgrades an old FE still commits
deterministic IDs on the final blob.
+ Azure::Core::IO::MemoryBodyStream legacy_body(
+ reinterpret_cast<const uint8_t*>(stream.data()),
stream.size());
+ _client->GetBlockBlobClient(opts.key).StageBlock(
+ legacy_azure_block_id(part_num), legacy_body);
+ }
},
opts, _tls_debug_context);
return ObjectStorageUploadResponse {
.resp = resp,
+ // Hive defers completion to FE, so the exact staged ID must cross
that boundary.
+ .etag = block_id,
};
}
ObjectStorageResponse AzureObjStorageClient::complete_multipart_upload(
const ObjectStoragePathOptions& opts,
const std::vector<ObjectCompleteMultiPart>& completed_parts) {
- auto client = _client->GetBlockBlobClient(opts.key);
+ DCHECK(opts.upload_id.has_value());
+ auto temp_client =
+ _client->GetBlockBlobClient(azure_multipart_temp_key(opts.key,
*opts.upload_id));
+ auto target_client = _client->GetBlockBlobClient(opts.key);
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); });
- return do_azure_client_call(
+ [&opts](const ObjectCompleteMultiPart& i) { return
azure_block_id(opts, i.part_num); });
+ auto response = do_azure_client_call(
[&]() {
SCOPED_BVAR_LATENCY(s3_bvar::s3_multi_part_upload_latency);
- client.CommitBlockList(string_block_ids);
+ // Per-upload temporary blobs keep Azure's blob-wide
staged-block namespace isolated.
+ temp_client.CommitBlockList(string_block_ids);
+ auto copy =
target_client.StartCopyFromUri(temp_client.GetUrl());
Review Comment:
[P1] Do not use asynchronous `Copy Blob` as an atomic promotion step. Azure
specifies that starting this operation overwrites an existing destination even
while the copy is still pending, and aborting a pending copy leaves a
zero-length destination. Thus after writer A publishes valid bytes, writer B
can start this copy, immediately make A unavailable, and then fail;
`PollUntilDone()` reports the failure but cannot restore A. The Java completion
path has the same publication gap. Use a publication protocol that preserves
the prior target until the new object is known complete (with an appropriate
same-key fence), and test a pending/failed second completion over an existing
target.
##########
be/src/runtime/runtime_state.cpp:
##########
@@ -52,12 +52,37 @@
#include "runtime/thread_context.h"
#include "storage/id_manager.h"
#include "storage/storage_engine.h"
+#include "util/thrift_util.h"
#include "util/timezone_utils.h"
#include "util/uid_util.h"
namespace doris {
using namespace ErrorCode;
+Status RuntimeState::add_iceberg_commit_datas(TIcebergCommitData
iceberg_commit_data) {
+ ThriftSerializer serializer(false, 256);
+ uint32_t serialized_size = 0;
+ uint8_t* buffer = nullptr;
+ RETURN_IF_ERROR(serializer.serialize(&iceberg_commit_data,
&serialized_size, &buffer));
+
+ constexpr size_t report_envelope_headroom = 1024 * 1024;
+ const size_t thrift_limit =
static_cast<size_t>(std::max(config::thrift_max_message_size, 0));
Review Comment:
[P1] Budget against the receiver's effective Thrift limit, not only this
BE's local setting. FE independently applies `Config.thrift_max_message_size`
to accepted sockets, so with FE at 32 MiB and BE at the 100 MiB default this
guard can accept tens of MiB that the final `reportExecStatus` cannot deliver.
By then successful writer close has cleared the file-cleanup lists, leaving the
uncommitted objects behind when the transport rejects the report.
Carry/negotiate the FE limit and use the smaller fragment-wide budget (with a
rolling fallback), and test an FE-lower-than-BE configuration.
##########
fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureObjStorage.java:
##########
@@ -244,15 +244,31 @@ public void completeMultipartUpload(String remotePath,
String uploadId,
List<UploadPartResult> parts) throws IOException {
try {
AzureUri uri = AzureUri.parse(remotePath);
- BlockBlobClient blockBlobClient =
getClient().getBlobContainerClient(uri.container())
- .getBlobClient(uri.key()).getBlockBlobClient();
+ BlobContainerClient containerClient =
getClient().getBlobContainerClient(uri.container());
List<String> blockIds = new ArrayList<>();
List<UploadPartResult> sorted = new ArrayList<>(parts);
sorted.sort((a, b) -> Integer.compare(a.partNumber(),
b.partNumber()));
+ boolean exactBlockIds = !sorted.isEmpty() && sorted.stream()
+ .allMatch(part -> part.etag() != null &&
!part.etag().isEmpty());
for (UploadPartResult part : sorted) {
- blockIds.add(toBlockId(part.partNumber()));
+ // A mixed-version upload must use one namespace consistently;
new BEs also stage legacy IDs.
+ blockIds.add(exactBlockIds ? part.etag() :
toBlockId(part.partNumber()));
+ }
+ String commitKey = exactBlockIds ? multipartTempKey(uri.key(),
uploadId) : uri.key();
+ BlobClient commitBlob = containerClient.getBlobClient(commitKey);
+ commitBlob.getBlockBlobClient().commitBlockList(blockIds);
Review Comment:
[P1] Keep the committed temporary blob outside every Hive scan prefix. This
key is `<target>.__doris_multipart/<uploadId>`, and Azure's flat prefix listing
exposes it as a regular file whose leaf name is the UUID, so Hive's `_`/`.`
leaf filter does not hide it. Once `commitBlockList` runs, concurrent scans can
read uncommitted rows; after the target copy they can read both objects, and
the swallowed delete failure (or a crash) leaves that duplicate indefinitely.
Use a staging namespace that table listings cannot reach, with recoverable
cleanup ownership, and test listing during copy and after an injected cleanup
failure.
##########
be/src/exec/operator/spill_iceberg_table_sink_operator.cpp:
##########
@@ -55,47 +55,61 @@ 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();
+ per_partition_reservations.reserve(active_writers->size());
+ for (const auto& writer : *active_writers) {
+ if (auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(writer.get())) {
+ auto reservation =
sort_writer->get_reserve_mem_size_components(state, eos);
+ per_partition_reservations.push_back(
+ {.retained_growth = reservation.retained_growth,
+ .transient_workspace = reservation.transient_workspace});
+ }
}
-
- return sort_writer->get_reserve_mem_size(state, eos);
+ // Column growth remains in every touched sorter, while sorting workspace
is reused by serial dispatch.
+ return iceberg_reserve_size(per_partition_reservations,
Review Comment:
[P1] Retain this reservation until the async writer consumes the queued
block. The pipeline task reserves this estimate only around `_sink->sink`, but
`AsyncResultWriter::sink` merely enqueues the block and returns;
`VIcebergTableWriter::write`/`FullSorter::append_block` allocate later on the
fragment-manager thread, after `DEFER_RELEASE_RESERVED` has already shrunk the
reservation. Parallel tasks can consequently reuse the same headroom, both pass
admission, and then exceed the hard limit concurrently. Transfer a reservation
token with each queued block and release/convert it after the async append,
with a deterministic two-task handoff test.
##########
be/src/io/fs/azure_obj_storage_client.cpp:
##########
@@ -216,35 +237,78 @@ ObjectStorageResponse
AzureObjStorageClient::put_object(const ObjectStoragePathO
ObjectStorageUploadResponse AzureObjStorageClient::upload_part(const
ObjectStoragePathOptions& opts,
std::string_view stream,
int part_num) {
- auto client = _client->GetBlockBlobClient(opts.key);
+ DCHECK(opts.upload_id.has_value());
+ auto client =
_client->GetBlockBlobClient(azure_multipart_temp_key(opts.key,
*opts.upload_id));
+ std::string block_id = azure_block_id(opts, part_num);
auto resp = do_azure_client_call(
[&]() {
Azure::Core::IO::MemoryBodyStream memory_body(
reinterpret_cast<const uint8_t*>(stream.data()),
stream.size());
// The blockId must be base64 encoded
SCOPED_BVAR_LATENCY(s3_bvar::s3_multi_part_upload_latency);
- client.StageBlock(base64_encode_part_num(part_num),
memory_body);
+ client.StageBlock(block_id, memory_body);
+ if (opts.deferred_completion) {
Review Comment:
[P1] Preserve same-key isolation in the old-FE compatibility path. Every
deferred writer also stages `legacy_azure_block_id(part_num)` on the shared
final blob, so if B stages part 1 after A, an old FE completing A selects B's
last-staged bytes for the same deterministic ID and reports A successful; an
aborted writer's legacy blocks can poison a later completion as well. The
per-upload temp blob protects only new-FE completion. Fence or reject
conflicting deferred same-key uploads during the rolling window instead of
mirroring into a shared block namespace, and test two interleaved new-BE
uploads completed by the legacy FE algorithm.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java:
##########
@@ -369,13 +369,15 @@ private IcebergWriteContext
buildWriteContext(ConnectorWriteHandle handle) {
// Carry it on the op-context so beginWrite anchors the RowDelta
baseSnapshotId at S_read, keeping
// the commit-time removeDeletes (option D) and BE's scan-time DV
union on one snapshot. -1 (no pin)
// preserves the legacy begin-time current snapshot.
- long readSnapshotId = handle.getTableHandle() instanceof
IcebergTableHandle
- ? ((IcebergTableHandle)
handle.getTableHandle()).getSnapshotId() : -1L;
+ IcebergTableHandle icebergHandle = handle.getTableHandle() instanceof
IcebergTableHandle
+ ? (IcebergTableHandle) handle.getTableHandle() : null;
+ long readSnapshotId = icebergHandle == null ? -1L :
icebergHandle.getSnapshotId();
+ boolean readSnapshotPinned = icebergHandle != null &&
icebergHandle.hasSnapshotPin();
Review Comment:
[P1] Preserve the target branch's exact pin when a statement reads multiple
references of this table. The write translator uses version-blind
`getSnapshotFromContext(targetTable)`, which intentionally returns empty for
two non-default refs. Thus `INSERT OVERWRITE t@branch('b1') SELECT ... FROM
t@branch('b1') JOIN t@branch('b2')` scans B1/C1 but reaches this line with
`readSnapshotPinned=false`; a concurrent b1 append before `beginWrite` becomes
the overwrite base and can be replaced without conflict. Carry the write
target's branch/version selector into the snapshot lookup (or carry its
resolved pin directly), and add this multi-reference scan-to-begin race.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,344 @@
+// 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.LinkedHashMap;
+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.
Review Comment:
[P2] Reject the unsafe cutoff before scanning reachable files.
`collectReachableFiles` recursively opens metadata and every unique data/delete
manifest, but the 24-hour check below depends only on `older_than` and the
clock. A recent or future cutoff is guaranteed to fail yet can still make a
long-history table perform the full synchronous FE scan first. Move the
retention check ahead of `collectReachableFiles`, and make the recent-cutoff
test assert that no manifest is opened.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,344 @@
+// 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.LinkedHashMap;
+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) {
+ List<String> ownedRoots = resolveOwnedRoots(table.location(),
table.properties());
+
+ 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);
+ }
+
+ return minimalOwnedRoots(ownedRoots);
+ }
+
+ static List<String> resolveOwnedRoots(String tableLocation, Map<String,
String> properties) {
Review Comment:
[P2] Preserve cleanup access across location-property changes. This list
contains only current data/metadata roots, so after moving writes from external
root A to B the default action no longer lists A and `location=A` is rejected,
even while retained metadata proves the table used A. Failed-write files or
obsolete metadata there become permanently uncleanable. Because prior use does
not prove exclusive ownership, derive only provably table-specific historical
candidates or provide a separately guarded expert override; test data/metadata
migrations plus a former root later shared by another table.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,344 @@
+// 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.LinkedHashMap;
+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));
Review Comment:
[P2] Bound the FE heap used by this reachability build.
`collectReachableFiles` first retains every live file as `String`; while that
set is still live, `ReachableIndex` creates a `FileIdentity` set and a second
path-string set for the same entries, and listing later retains every visited
object too. A valid large table can therefore OOM FE before comparison
completes even though each manifest is opened once. Build one compact index
directly and add a bounded/spillable or distributed path (or an explicit safe
preflight limit) with scale coverage.
##########
fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureObjStorage.java:
##########
@@ -244,15 +244,31 @@ public void completeMultipartUpload(String remotePath,
String uploadId,
List<UploadPartResult> parts) throws IOException {
try {
AzureUri uri = AzureUri.parse(remotePath);
- BlockBlobClient blockBlobClient =
getClient().getBlobContainerClient(uri.container())
- .getBlobClient(uri.key()).getBlockBlobClient();
+ BlobContainerClient containerClient =
getClient().getBlobContainerClient(uri.container());
List<String> blockIds = new ArrayList<>();
List<UploadPartResult> sorted = new ArrayList<>(parts);
sorted.sort((a, b) -> Integer.compare(a.partNumber(),
b.partNumber()));
+ boolean exactBlockIds = !sorted.isEmpty() && sorted.stream()
+ .allMatch(part -> part.etag() != null &&
!part.etag().isEmpty());
for (UploadPartResult part : sorted) {
- blockIds.add(toBlockId(part.partNumber()));
+ // A mixed-version upload must use one namespace consistently;
new BEs also stage legacy IDs.
+ blockIds.add(exactBlockIds ? part.etag() :
toBlockId(part.partNumber()));
+ }
+ String commitKey = exactBlockIds ? multipartTempKey(uri.key(),
uploadId) : uri.key();
+ BlobClient commitBlob = containerClient.getBlobClient(commitKey);
+ commitBlob.getBlockBlobClient().commitBlockList(blockIds);
+ if (exactBlockIds) {
+ BlobClient targetBlob =
containerClient.getBlobClient(uri.key());
+ // The temporary blob is the provider-visible writer fence;
only a completed copy publishes it.
+ targetBlob.beginCopy(commitBlob.getBlobUrl(),
null).waitForCompletion();
Review Comment:
[P1] Check the terminal copy status before deleting the temp blob. Azure's
`SyncPoller.waitForCompletion()` returns a final `PollResponse` for both
`SUCCESSFULLY_COMPLETED` and `FAILED`; it does not turn a failed copy status
into `BlobStorageException`. If the server-side copy reaches `FAILED`, this
code ignores that status, deletes the only committed temporary blob, and
returns success, so Hive/external commit can publish metadata for a missing or
incomplete target. Require `SUCCESSFULLY_COMPLETED` (and surface the
copy-status description) before cleanup, and add a failed-terminal-poll test
that keeps the temp blob and throws.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,344 @@
+// 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.LinkedHashMap;
+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) {
+ List<String> ownedRoots = resolveOwnedRoots(table.location(),
table.properties());
+
+ 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);
+ }
+
+ return minimalOwnedRoots(ownedRoots);
+ }
+
+ static List<String> resolveOwnedRoots(String tableLocation, Map<String,
String> properties) {
+ String tableRoot = normalizeLocation(tableLocation);
+ String dataRoot = normalizeLocation(resolveDataLocation(properties,
tableRoot));
+ List<String> configuredRoots = new ArrayList<>();
+ configuredRoots.add(tableRoot);
+ configuredRoots.add(dataRoot);
+ // Iceberg may place metadata outside both table and data roots, so it
remains an independent owned root.
+ String metadataRoot =
nonEmpty(properties.get(TableProperties.WRITE_METADATA_LOCATION));
+ if (metadataRoot != null) {
+ configuredRoots.add(normalizeLocation(metadataRoot));
+ }
+ return canonicalOwnedRoots(configuredRoots);
+ }
+
+ static List<String> minimalOwnedRoots(List<String> roots) {
+ List<String> canonicalRoots = canonicalOwnedRoots(roots);
+ List<String> minimal = new ArrayList<>();
+ for (String candidate : canonicalRoots) {
+ // Canonically equal aliases are deduplicated first, so only
strict containment removes a root.
+ if (canonicalRoots.stream().noneMatch(other ->
!sameFileIdentity(other, candidate)
+ && isWithinLocation(candidate, other))) {
+ minimal.add(candidate);
+ }
+ }
+ return minimal;
+ }
+
+ private static List<String> canonicalOwnedRoots(List<String> roots) {
+ Map<FileIdentity, String> byIdentity = new LinkedHashMap<>();
+ for (String root : roots) {
+ byIdentity.putIfAbsent(FileIdentity.of(root), root);
+ }
+ return new ArrayList<>(byIdentity.values());
+ }
+
+ private static String resolveDataLocation(Map<String, String> properties,
String tableRoot) {
Review Comment:
[P2] Handle the custom location-provider case explicitly. Iceberg gives
`write.location-provider.impl` precedence over every built-in location
property, and that provider can place data outside `table.location`; this
resolver ignores it, so the default action never scans those files and an
explicit provider root is rejected by the same incomplete allowlist. Obtain a
provider-specific owned-root contract, or fail with a precise
unsupported-layout error plus a guarded expert override, and add a
custom-provider test with external data.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,344 @@
+// 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.LinkedHashMap;
+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) {
+ List<String> ownedRoots = resolveOwnedRoots(table.location(),
table.properties());
+
+ 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);
+ }
+
+ return minimalOwnedRoots(ownedRoots);
+ }
+
+ static List<String> resolveOwnedRoots(String tableLocation, Map<String,
String> properties) {
+ String tableRoot = normalizeLocation(tableLocation);
+ String dataRoot = normalizeLocation(resolveDataLocation(properties,
tableRoot));
Review Comment:
[P1] Do not treat this raw external data location as exclusively
table-owned. Iceberg's standard `ObjectStoreLocationProvider` explicitly
supports multiple tables sharing one short `write.data.path`; it writes them
under `<root>/<hash>/<db>/<table>/...`. The default action lists the whole
root, but this table's reachable index knows nothing about its neighbors, so it
can classify another table's old live file as an orphan and delete it. Restrict
candidates to the exact table context encoded after the hash directories (or
fail closed where ownership cannot be proved), and add a two-table shared-root
deletion test.
--
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]
