github-actions[bot] commented on code in PR #66348:
URL: https://github.com/apache/doris/pull/66348#discussion_r3695317514
##########
be/src/exec/operator/spill_iceberg_table_sink_operator.cpp:
##########
@@ -55,47 +55,50 @@ 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;
+ size_t reserve_size = 0;
+ for (const auto& writer : *_writer->active_writers()) {
Review Comment:
[P1] Do not reserve a full input batch per partition
Each `FullSorter::get_reserve_mem_size` assumes that sorter may receive
`state->batch_size()` rows, but this loop sums that full-batch estimate across
every active partition even though there is only one incoming block and its
rows are divided among them. At the default 128-partition limit this can
request memory for roughly 128 blocks; when that false reservation fails,
`revoke_memory` spills only one small partition and the same block repeatedly
pauses. Please bound the aggregate to the actual one-block input (or reserve
after dispatch using each selected row count) and test the many-partition case.
##########
be/src/exec/operator/spill_iceberg_table_sink_operator.cpp:
##########
@@ -55,47 +55,50 @@ 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;
+ size_t reserve_size = 0;
+ for (const auto& writer : *_writer->active_writers()) {
+ if (auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(writer.get())) {
+ reserve_size += sort_writer->get_reserve_mem_size(state, eos);
+ }
}
-
- return sort_writer->get_reserve_mem_size(state, eos);
+ return reserve_size;
}
size_t SpillIcebergTableSinkLocalState::get_revocable_mem_size(RuntimeState*
state) const {
if (!_writer) {
return 0;
}
- auto current_writer = _writer->current_writer();
- auto* sort_writer =
dynamic_cast<VIcebergSortWriter*>(current_writer.get());
- if (!sort_writer) {
- return 0;
+ size_t revocable_size = 0;
+ for (const auto& writer : *_writer->active_writers()) {
Review Comment:
[P1] Retain the snapshot owner while iterating
`active_writers()` returns a `shared_ptr` by value, but this C++20 range-for
dereferences that temporary directly. The temporary owner is destroyed before
iterator initialization; if the async writer then publishes a replacement
snapshot, the old vector can be freed while the revocation thread is traversing
it. The same lifetime bug appears in the loops at lines 59 and 87. Please first
store the snapshot in a named `shared_ptr` and iterate through that retained
owner, and cover publication versus traversal concurrency.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2514,16 +2514,14 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
}
}
}
- if (auto icd = req.runtime_state->iceberg_commit_datas(); !icd.empty()) {
+
req.runtime_state->append_iceberg_commit_datas(¶ms.iceberg_commit_datas);
+ if (!params.iceberg_commit_datas.empty()) {
params.__isset.iceberg_commit_datas = true;
- params.iceberg_commit_datas.insert(params.iceberg_commit_datas.end(),
icd.begin(),
- icd.end());
} else if (!req.runtime_states.empty()) {
for (auto* rs : req.runtime_states) {
- if (auto rs_icd = rs->iceberg_commit_datas(); !rs_icd.empty()) {
+ rs->append_iceberg_commit_datas(¶ms.iceberg_commit_datas);
Review Comment:
[P1] Enforce one budget for the aggregated report
The new guard is per task `RuntimeState`, but this loop concatenates every
task's accepted vector into one `TReportExecStatusParams`. With a 100 MiB
limit, two parallel sink tasks can each accept about 60 MiB and then send a
roughly 120 MiB RPC, which FE rejects. By this point successful writers have
cleared their local cleanup lists, so the failed report also strands the
uncommitted objects. Please enforce a shared fragment/report budget before
cleanup ownership is released and add a multi-task aggregate-limit test.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,163 @@
+// 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.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+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.io.CloseableIterable;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.SupportsPrefixOperations;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** Safely lists or deletes old files that are unreachable from every retained
snapshot. */
+public class IcebergRemoveOrphanFilesAction extends BaseIcebergAction {
+ 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");
+ }
+ 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 + "/")) {
+ throw new DorisConnectorException("location must be within the
Iceberg table location");
+ }
+
+ try {
+ Set<String> reachable = collectReachableFiles(table);
+ long orphanCount = 0;
+ long deletedCount = 0;
+ long olderThan = namedArguments.getLong(OLDER_THAN);
+ boolean dryRun = namedArguments.getBoolean(DRY_RUN);
+ // Object stores use raw prefix matching, so the separator
prevents "table_backup" siblings
+ // from being treated as children of "table".
+ String listingPrefix = scanLocation.endsWith("/") ? scanLocation :
scanLocation + "/";
+ for (FileInfo file : ((SupportsPrefixOperations)
table.io()).listPrefix(listingPrefix)) {
+ if (file.createdAtMillis() < olderThan &&
!reachable.contains(file.location())) {
Review Comment:
[P1] Compare canonical file identities before deleting
The membership test compares raw locations. If retained metadata references
`s3a://bucket/key` while prefix listing returns `s3://bucket/key` for the same
object (or an equivalent authority differs), this branch treats the live file
as orphaned and deletes it. Iceberg's 1.10.1 action canonicalizes both sides,
treats s3/s3a/s3n as equivalent by default, and fails closed on unresolved
prefix mismatches. Please adopt equivalent identity handling and add
alias/mismatch tests before enabling deletion.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,163 @@
+// 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.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+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.io.CloseableIterable;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.SupportsPrefixOperations;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** Safely lists or deletes old files that are unreachable from every retained
snapshot. */
+public class IcebergRemoveOrphanFilesAction extends BaseIcebergAction {
+ 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) {
Review Comment:
[P1] Honor Iceberg's GC-disabled safety fence
This destructive path never checks `gc.enabled`. Iceberg uses
`gc.enabled=false` specifically when files may be shared with another table,
and its 1.10.1 orphan-file action refuses to run in that state because deletion
can corrupt the other table. With `dry_run=false`, this implementation proceeds
directly to `deleteFile`. Please reject execution when GC is disabled (using
the Iceberg property/default) and add an execution test proving no deletion
occurs.
##########
be/src/io/fs/s3_file_writer.cpp:
##########
@@ -78,20 +78,51 @@ S3FileWriter::~S3FileWriter() {
// For thread safety
std::ignore = _async_close_pack->future.get();
_async_close_pack = nullptr;
+ } else if (state() == State::OPENED) {
+ WARN_IF_ERROR(abort(), "failed to abort unfinished S3 writer");
} else {
// Consider one situation where the file writer is destructed after it
submit at least one async task
// without calling close(), then there exists one occasion where the
async task is executed right after
// the correspoding S3 file writer is already destructed
_wait_until_finish(fmt::format("wait s3 file {} upload to be finished",
_obj_storage_path_opts.path.native()));
}
- // We won't do S3 abort operation in BE, we let s3 service do it own.
if (state() == State::OPENED && !_failed) {
s3_bytes_written_total << _bytes_appended;
}
s3_file_being_written << -1;
}
+Status S3FileWriter::abort() {
+ if (state() == State::CLOSED) {
+ return Status::OK();
+ }
+ if (state() == State::ASYNC_CLOSING) {
+ return Status::InternalError("cannot abort an asynchronously closing
S3 writer");
+ }
+ RETURN_IF_ERROR(_abort_impl());
+ _state = State::CLOSED;
+ return Status::OK();
+}
+
+Status S3FileWriter::_abort_impl() {
+ _wait_until_finish(
+ fmt::format("wait s3 file {} before abort",
_obj_storage_path_opts.path.native()));
+ _pending_buf.reset();
Review Comment:
[P2] Implement abort for Azure staged blocks
Azure uses this writer too, but `create_multipart_upload` intentionally
returns no upload ID and `upload_part` stages blocks. This guard therefore
skips provider cleanup, returns OK, and lets `abort()` mark the writer CLOSED
even after Azure blocks were staged; generic upload/destructor failures have no
later path deletion, so those blocks remain until provider GC. Please make
abandonment provider-aware (for example, delete/discard the uncommitted blob
through an Azure override), preserve cleanup failures, and add an Azure failure
test.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,163 @@
+// 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.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+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.io.CloseableIterable;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.SupportsPrefixOperations;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** Safely lists or deletes old files that are unreachable from every retained
snapshot. */
+public class IcebergRemoveOrphanFilesAction extends BaseIcebergAction {
+ 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");
+ }
+ 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 + "/")) {
+ throw new DorisConnectorException("location must be within the
Iceberg table location");
+ }
+
+ try {
+ Set<String> reachable = collectReachableFiles(table);
+ long orphanCount = 0;
+ long deletedCount = 0;
+ long olderThan = namedArguments.getLong(OLDER_THAN);
+ boolean dryRun = namedArguments.getBoolean(DRY_RUN);
+ // Object stores use raw prefix matching, so the separator
prevents "table_backup" siblings
+ // from being treated as children of "table".
+ String listingPrefix = scanLocation.endsWith("/") ? scanLocation :
scanLocation + "/";
+ for (FileInfo file : ((SupportsPrefixOperations)
table.io()).listPrefix(listingPrefix)) {
+ if (file.createdAtMillis() < olderThan &&
!reachable.contains(file.location())) {
+ 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 Set<String> collectReachableFiles(Table table) throws IOException {
+ Set<String> reachable = new
HashSet<>(ReachableFileUtil.metadataFileLocations(table, true));
Review Comment:
[P2] Keep the live version hint reachable
`metadataFileLocations(table, true)` does not include Hadoop Iceberg's
`metadata/version-hint.text`; Iceberg exposes that path separately via
`ReachableFileUtil.versionHintLocation(table)` and its own orphan action
explicitly adds it. `HadoopFileIO` supports prefix listing, so a table-root run
with a cutoff newer than an old hint will classify and delete this live
metadata file. Please add the hint to the reachable identities and cover it in
a non-dry-run test.
##########
be/src/io/fs/rate_limited_obj_storage_client.cpp:
##########
@@ -73,6 +73,15 @@ ObjectStorageResponse
RateLimitedObjStorageClient::complete_multipart_upload(
return _inner->complete_multipart_upload(opts, completed_parts);
}
+ObjectStorageResponse RateLimitedObjStorageClient::abort_multipart_upload(
+ const ObjectStoragePathOptions& opts) {
+ S3RateLimitGuard guard(S3RateLimitType::PUT, 0);
Review Comment:
[P2] Ensure cleanup bypasses or reserves the PUT limit
This abort uses the same hard PUT request-count bucket as
create/upload/complete. When that bucket is what caused the write failure, the
immediately following abort is rejected before `_inner` is called;
`S3FileWriter::close` only warns, transitions to its terminal close state, and
the destructor no longer retries, leaving the multipart upload behind. Please
give mandatory abort cleanup guaranteed provider admission (or retain a
retryable cleanup state) and cover the exhausted-limit path rather than only an
unwrapped successful abort.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,163 @@
+// 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.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+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.io.CloseableIterable;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.SupportsPrefixOperations;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** Safely lists or deletes old files that are unreachable from every retained
snapshot. */
+public class IcebergRemoveOrphanFilesAction extends BaseIcebergAction {
+ 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");
+ }
+ 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 + "/")) {
+ throw new DorisConnectorException("location must be within the
Iceberg table location");
+ }
+
+ try {
+ Set<String> reachable = collectReachableFiles(table);
+ long orphanCount = 0;
+ long deletedCount = 0;
+ long olderThan = namedArguments.getLong(OLDER_THAN);
Review Comment:
[P1] Reject unsafe recent orphan cutoffs
`older_than` is accepted as any nonnegative value, including now or a future
timestamp. A concurrent writer's newly uploaded file is not yet reachable from
table snapshots, so this action can list and delete it and the writer can then
commit metadata pointing to the missing object. Iceberg's 1.10.1 SQL procedure
rejects intervals shorter than 24 hours for exactly this race, reserving
arbitrary cutoffs for its expert Action API. Please enforce the SQL safety
interval (or an explicit expert-only override) and test a recent uncommitted
file.
--
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]
