Gabriel39 commented on code in PR #66348:
URL: https://github.com/apache/doris/pull/66348#discussion_r3698477508
##########
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:
Fixed by moving MVCC pinning from the version-blind translation point to
bindDataSink, where the exact target branch is available. The sink constructs
branch-aware TableScanParams, looks up that exact StatementContext pin, and
passes the pinned table handle to planWrite. The binding test asserts the audit
branch selector and pinned handle, and the row-level DML test verifies the
late-bound handle reaches the provider.
##########
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:
Fixed by building one path-keyed ReachableIndex directly instead of
retaining the full String set plus identity and path sets. Listing no longer
keeps a visited-file set, manifest paths are retained only for read
deduplication, and the reachable index has an explicit 5,000,000-entry safety
cap with coverage.
##########
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:
Fixed by validating the 24-hour retention fence before resolving scan scopes
or collecting any reachable metadata. The recent-cutoff test wraps FileIO and
now asserts that zero manifests are opened.
--
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]
