Gabriel39 commented on code in PR #66348:
URL: https://github.com/apache/doris/pull/66348#discussion_r3712696523
##########
fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java:
##########
@@ -186,12 +186,37 @@ protected void
doProcessReportExecStatus(TReportExecStatusParams params, SingleF
}
}
- if (!fragmentTask.processReportExecStatus(params)) {
+ if (!fragmentTask.processReportExecStatus(params, () ->
acceptFinalReport(params))) {
+ if ((params.isSetHivePartitionUpdates() ||
params.isSetIcebergCommitDatas()
+ || params.isSetMcCommitDatas()) && !fragmentTask.isDone())
{
+ throw new IllegalStateException("External-file report was not
a completed fragment report");
Review Comment:
Fixed in 1a60ebae30. External commit vectors now flow through
RuntimeState::append_external_file_commit_data(), which returns immediately
unless the report is final. PipelineFragmentContext passes req.done for both
the fragment and task states, so periodic done=false reports carry no Hive,
Iceberg, or MaxCompute ownership data.
RuntimeStateIcebergCommitDataTest.PeriodicReportOmitsExternalCommitData
verifies that all three vectors are absent from a periodic report and present
in the final report; the related focused FE suites passed 30/30.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorTransaction.java:
##########
@@ -221,6 +221,8 @@ public void beginWrite(ConnectorSession session, String db,
String tableName, Hi
@Override
public void commit() {
+ // Object-store files remain unpublished until FE consumes one
completion record per file.
+ validateObjectStoreCommitRecords();
Review Comment:
Fixed in 1a60ebae30. HiveConnectorTransaction.commit() now self-rolls back
validation and classification failures before rethrowing, because the
transaction manager has already removed the connector at that point. Cleanup
failures are attached as suppressed failures. The rollback collector also skips
malformed cleanup records so one bad record cannot prevent valid provider
uploads from being aborted. The tests cover an incomplete/malformed validation
set while verifying the valid MPU is aborted, plus a classification failure
before HmsCommitter creation; HiveConnectorTransactionTest passed 17/17.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2548,8 +2557,19 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
params.__set_backend_id(_exec_env->cluster_info()->backend_id);
}
+ Status report_size_status = validate_report_exec_status_size(
+ params, req.runtime_state->coordinator_thrift_message_limit());
+ if (!report_size_status.ok()) {
+ if (req.done) {
+
req.runtime_state->finalize_iceberg_report_cleanup(IcebergReportOutcome::REJECTED);
Review Comment:
Fixed in 1a60ebae30. The shared report state is now external-file-wide
instead of Iceberg-specific. A successful or rollback-only Hive deferred upload
registers a cleanup callback that retains the provider client and exact upload
identity after S3FileWriter is CLOSED. Definite report rejection executes the
callbacks, acknowledgement clears them, and an ambiguous transport retains
them. The callback dispatches through
ObjStorageClient::abort_multipart_upload(), so it aborts an S3 MPU and releases
the corresponding Azure target lease. The provider callback and report-outcome
tests compile successfully, and the Azure extension suite passed 26/26.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergRemoveOrphanFilesAction.java:
##########
@@ -0,0 +1,415 @@
+// 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;
+import java.util.regex.Pattern;
+
+/** 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();
+ private static final int MAX_REACHABLE_FILES = 5_000_000;
+ public static final String OLDER_THAN = "older_than";
+ public static final String LOCATION = "location";
+ public static final String DRY_RUN = "dry_run";
+ public static final String ALLOW_UNSAFE_LOCATION = "allow_unsafe_location";
+
+ 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 to scan for
orphan files",
+ null, ArgumentParsers.nonEmptyString(LOCATION));
+ namedArguments.registerOptionalArgument(DRY_RUN, "Only count orphan
files", true,
+ ArgumentParsers.booleanValue(DRY_RUN));
+ namedArguments.registerOptionalArgument(ALLOW_UNSAFE_LOCATION,
+ "Allow an explicitly supplied location whose table ownership
cannot be proved",
+ false, ArgumentParsers.booleanValue(ALLOW_UNSAFE_LOCATION));
+ }
+
+ @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");
+ }
+ long olderThan = namedArguments.getLong(OLDER_THAN);
+ // Reject an unsafe cutoff before opening any metadata or manifest
file.
+ if (olderThan > System.currentTimeMillis() - MIN_RETENTION_MS) {
+ throw new DorisConnectorException("older_than must retain at least
24 hours of files");
+ }
+ List<ScanScope> scanScopes = resolveScanScopes(table);
+
+ try {
+ ReachableIndex reachable = collectReachableFiles(table);
+ long orphanCount = 0;
+ long deletedCount = 0;
+ boolean dryRun = namedArguments.getBoolean(DRY_RUN);
+ for (ScanScope scope : scanScopes) {
+ // Object stores use raw prefix matching, so the separator
excludes sibling prefixes.
+ String listingPrefix = scope.root.endsWith("/") ? scope.root :
scope.root + "/";
+ for (FileInfo file : ((SupportsPrefixOperations)
table.io()).listPrefix(listingPrefix)) {
+ if (scope.owns(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<ScanScope> resolveScanScopes(Table table) {
+ String tableRoot = normalizeLocation(table.location());
+ String requested = namedArguments.getString(LOCATION);
+ if (requested != null) {
+ String normalized = normalizeLocation(requested);
+ if (isWithin(normalized, tableRoot)) {
+ return Lists.newArrayList(ScanScope.exclusive(normalized));
+ }
+ if (!namedArguments.getBoolean(ALLOW_UNSAFE_LOCATION)) {
+ throw new DorisConnectorException(
+ "Cannot prove that location is owned by this table;
set allow_unsafe_location=true "
+ + "only after verifying the prefix is
exclusive to the table");
+ }
+ // This explicit escape hatch also covers historical roots after a
table-location migration.
+ return Lists.newArrayList(ScanScope.exclusive(normalized));
+ }
+ if
(nonEmpty(table.properties().get(TableProperties.WRITE_LOCATION_PROVIDER_IMPL))
!= null) {
+ throw new DorisConnectorException(
+ "remove_orphan_files cannot infer ownership for a custom
write.location-provider.impl; "
+ + "provide location with
allow_unsafe_location=true after verifying exclusivity");
+ }
+ String metadataRoot =
nonEmpty(table.properties().get(TableProperties.WRITE_METADATA_LOCATION));
+ if (metadataRoot != null && !isWithin(normalizeLocation(metadataRoot),
tableRoot)) {
+ throw new DorisConnectorException(
+ "Cannot prove that the configured external metadata
location is table-exclusive; "
+ + "provide location with
allow_unsafe_location=true after verifying exclusivity");
+ }
+ List<ScanScope> scopes = new ArrayList<>();
+ scopes.add(ScanScope.exclusive(tableRoot));
+ if
(Boolean.parseBoolean(table.properties().get(TableProperties.OBJECT_STORE_ENABLED)))
{
+ // Match Iceberg's ObjectStoreLocationProvider precedence exactly.
+ String objectRoot =
nonEmpty(table.properties().get(TableProperties.WRITE_DATA_LOCATION));
+ if (objectRoot == null) {
+ objectRoot =
nonEmpty(table.properties().get(TableProperties.OBJECT_STORE_PATH));
+ }
+ if (objectRoot == null) {
+ objectRoot =
nonEmpty(table.properties().get(TableProperties.WRITE_FOLDER_STORAGE_LOCATION));
+ }
+ if (objectRoot != null) {
+ String normalizedObjectRoot = normalizeLocation(objectRoot);
+ if (!isWithin(normalizedObjectRoot, tableRoot)) {
+ if (normalizedObjectRoot.startsWith(tableRoot)) {
+ // Iceberg omits table context for this raw-prefix
case, so ownership is not recoverable.
+ throw new DorisConnectorException(
+ "Cannot prove object-store ownership because
its path has the table location "
+ + "as a non-directory prefix; provide
a verified explicit location");
+ }
+ scopes.add(ScanScope.objectStore(normalizedObjectRoot,
tableRoot));
+ }
+ }
+ } else {
+ String externalDataRoot =
nonEmpty(table.properties().get(TableProperties.WRITE_DATA_LOCATION));
+ if (externalDataRoot == null) {
+ externalDataRoot = nonEmpty(
+
table.properties().get(TableProperties.WRITE_FOLDER_STORAGE_LOCATION));
+ }
+ if (externalDataRoot != null &&
!isWithin(normalizeLocation(externalDataRoot), tableRoot)) {
+ throw new DorisConnectorException(
+ "Cannot prove that the configured external data
location is table-exclusive; "
+ + "provide location with
allow_unsafe_location=true after verifying exclusivity");
+ }
+ }
+ return scopes;
+ }
+
+ private static String nonEmpty(String location) {
+ return location == null || location.isEmpty() ? null : location;
+ }
+
+ private boolean isWithin(String location, String root) {
+ return isWithinLocation(location, root);
+ }
+
+ private static boolean isWithinLocation(String location, String root) {
+ FileIdentity child = FileIdentity.of(location);
+ FileIdentity parent = FileIdentity.of(root);
+ String pathPrefix = parent.path.endsWith("/") ? parent.path :
parent.path + "/";
+ return child.scheme.equals(parent.scheme) &&
child.authority.equals(parent.authority)
+ && (child.path.equals(parent.path) ||
child.path.startsWith(pathPrefix));
+ }
+
+ private ReachableIndex collectReachableFiles(Table table) throws
IOException {
+ ReachableIndex reachable = new ReachableIndex(MAX_REACHABLE_FILES);
+ reachable.addAll(ReachableFileUtil.metadataFileLocations(table, true));
+ // Hadoop tables consult this live pointer even though it is not part
of the metadata log.
+ reachable.add(ReachableFileUtil.versionHintLocation(table));
+ Set<String> scannedDataManifests = new HashSet<>();
+ Set<String> scannedDeleteManifests = new HashSet<>();
+ reachable.addAll(ReachableFileUtil.manifestListLocations(table));
+ reachable.addAll(ReachableFileUtil.statisticsFilesLocations(table));
+ for (Snapshot snapshot : table.snapshots()) {
+ for (ManifestFile manifest : snapshot.allManifests(table.io())) {
+ reachable.add(manifest.path());
+ if (manifest.content() == ManifestContent.DATA) {
+ // Snapshots inherit manifests, so read each path once to
keep work linear.
+ if (scannedDataManifests.add(manifest.path())) {
+ try (ManifestReader<DataFile> dataFiles =
+ ManifestFiles.read(manifest, table.io(),
table.specs())) {
+ dataFiles.forEach(dataFile ->
reachable.add(dataFile.location()));
+ }
+ }
+ } else if (scannedDeleteManifests.add(manifest.path())) {
+ // A retained delete file may not apply to any current
data task, so read it directly.
+ try (ManifestReader<DeleteFile> deletes =
+ ManifestFiles.readDeleteManifest(manifest,
table.io(), table.specs())) {
+ deletes.forEach(delete ->
reachable.add(delete.location()));
+ }
+ }
+ }
+ }
+ return reachable;
+ }
+
+ private static boolean isReachable(String candidate, ReachableIndex
reachable) {
+ FileIdentity candidateIdentity = FileIdentity.of(candidate);
+ FileIdentity retainedIdentity =
reachable.byPath.get(candidateIdentity.path);
+ if (candidateIdentity.equals(retainedIdentity)) {
+ return true;
+ }
+ if (retainedIdentity != null) {
+ // A path collision across unknown providers/authorities cannot be
classified safely.
+ throw new DorisConnectorException(
+ "Cannot determine whether listed and reachable file
locations are equivalent");
+ }
+ return false;
+ }
+
+ static boolean sameFileIdentity(String first, String second) {
+ return FileIdentity.of(first).equals(FileIdentity.of(second));
+ }
+
+ private static final class FileIdentity {
+ private final String scheme;
+ private final String authority;
+ private final String path;
+
+ private FileIdentity(String scheme, String authority, String path) {
+ this.scheme = scheme;
+ this.authority = authority;
+ this.path = path;
+ }
+
+ private static FileIdentity of(String location) {
+ URI uri = URI.create(location).normalize();
+ String scheme = uri.getScheme();
+ scheme = scheme == null ? "" : scheme.toLowerCase(Locale.ROOT);
+ if (scheme.equals("s3a") || scheme.equals("s3n")) {
+ scheme = "s3";
+ }
+ String authority = uri.getAuthority();
+ authority = authority == null ? "" :
authority.toLowerCase(Locale.ROOT);
+ String path = uri.getPath();
+ return new FileIdentity(scheme, authority, path == null ? "" :
path);
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) {
+ return true;
+ }
+ if (!(other instanceof FileIdentity)) {
+ return false;
+ }
+ FileIdentity that = (FileIdentity) other;
+ return scheme.equals(that.scheme) &&
authority.equals(that.authority)
+ && path.equals(that.path);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(scheme, authority, path);
+ }
+ }
+
+ static void verifyReachableIndexLimit(Set<String> locations, int
maxEntries) {
+ ReachableIndex index = new ReachableIndex(maxEntries);
+ index.addAll(locations);
+ }
+
+ static boolean isOwnedObjectStorePath(String candidate, String
storageRoot, String tableLocation) {
+ return ScanScope.objectStore(normalizeLocation(storageRoot),
normalizeLocation(tableLocation))
+ .owns(candidate);
+ }
+
+ private static final class ReachableIndex {
+ private final Map<String, FileIdentity> byPath = new LinkedHashMap<>();
+ private final int maxEntries;
+
+ private ReachableIndex(int maxEntries) {
+ this.maxEntries = maxEntries;
+ }
+
+ private void addAll(Iterable<String> locations) {
+ locations.forEach(this::add);
+ }
+
+ private void add(String location) {
+ FileIdentity identity = FileIdentity.of(location);
+ FileIdentity existing = byPath.putIfAbsent(identity.path,
identity);
+ if (existing != null && !existing.equals(identity)) {
+ throw new DorisConnectorException(
+ "Cannot determine whether reachable file locations are
equivalent");
+ }
+ if (existing == null && byPath.size() > maxEntries) {
+ throw new DorisConnectorException(
+ "Reachable file index exceeds the safe in-memory limit
of " + maxEntries);
+ }
+ }
+ }
+
+ private static final class ScanScope {
+ private final String root;
+ private final Pattern ownedRelativePath;
+
+ private ScanScope(String root, Pattern ownedRelativePath) {
+ this.root = root;
+ this.ownedRelativePath = ownedRelativePath;
+ }
+
+ private static ScanScope exclusive(String root) {
+ return new ScanScope(root, null);
+ }
+
+ private static ScanScope objectStore(String root, String
tableLocation) {
+ URI tableUri = URI.create(tableLocation);
+ String[] segments = tableUri.getPath().split("/");
+ List<String> names = new ArrayList<>();
+ for (String segment : segments) {
+ if (!segment.isEmpty()) {
+ names.add(segment);
+ }
+ }
+ if (names.isEmpty()) {
+ throw new DorisConnectorException(
+ "Cannot infer an object-store table context from the
table location");
+ }
+ String context = names.size() > 1
+ ? names.get(names.size() - 2) + "/" +
names.get(names.size() - 1)
+ : names.get(names.size() - 1);
+ return new ScanScope(root, Pattern.compile(
+ "[01]{4}/[01]{4}/[01]{4}/[01]{4}/[01]{4}/"
+ + Pattern.quote(context) + "/.+"));
Review Comment:
Addressed by the current branch architecture. The hash-layout heuristic was
removed: an external data or metadata root outside the table-owned root now
fails closed unless the operator explicitly supplies a verified location with
allow_unsafe_location=true. The object-store regression constructs paths
through table.locationProvider().newDataLocation(), so it follows the Iceberg
dependency rather than duplicating a version-specific hash layout. There is
therefore no five-directory pattern left to correct.
--
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]
