Skip to content

[fix](iceberg) Run execute actions in catalog auth scope #67711

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Gabriel39 wants to merge 2 commits into from
Closed

[fix](iceberg) Run execute actions in catalog auth scope #67711

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bc8b16d
[fix](iceberg) Run execute actions in catalog auth scope
Gabriel39 Sep 9, 2026
e921629
[fix](iceberg) Fence execute actions to catalog generation
Gabriel39 Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -422,12 +422,19 @@ private void ensureCatalogGenerationStable(CatalogIf<?> catalog, IcebergMetadata
}
}

private static RuntimeException catalogGenerationMoved(NameMapping nameMapping) {
return new RuntimeException(String.format(
private static CatalogGenerationChangedException catalogGenerationMoved(NameMapping nameMapping) {
return new CatalogGenerationChangedException(String.format(
"Catalog %d was reset while acquiring iceberg table %s.%s, please retry.",
nameMapping.getCtlId(), nameMapping.getLocalDbName(), nameMapping.getLocalTblName()));
}

/** Signals that writable acquisition must restart before any metadata mutation begins. */
public static final class CatalogGenerationChangedException extends RuntimeException {
public CatalogGenerationChangedException(String message) {
super(message);
}
}

MetaCacheSizeEstimate prepareTableForCachePublication(
NameMapping nameMapping, IcebergTableCacheValue value) {
return value.prepareForCachePublication(nameMapping);
Expand Down
14 changes: 13 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/BaseIcebergAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,16 @@
import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
import org.apache.doris.datasource.iceberg.IcebergUtils;
import org.apache.doris.info.PartitionNamesInfo;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.plans.commands.execute.BaseExecuteAction;

import org.apache.iceberg.Table;

import java.util.Map;
import java.util.Objects;
import java.util.Optional;

/**
Expand All @@ -33,11 +38,13 @@
* functionality while inheriting common execution action behavior.
*/
public abstract class BaseIcebergAction extends BaseExecuteAction {
private final IcebergMetadataOps metadataOps;

protected BaseIcebergAction(String actionType, Map<String, String> properties,
Optional<PartitionNamesInfo> partitionNamesInfo,
Optional<Expression> whereCondition) {
Optional<Expression> whereCondition, IcebergMetadataOps metadataOps) {
super(actionType, properties, partitionNamesInfo, whereCondition);
this.metadataOps = Objects.requireNonNull(metadataOps, "metadataOps is null");
}

@Override
Expand Down Expand Up @@ -71,4 +78,9 @@ protected void validateIcebergAction() throws UserException {
// Default implementation does nothing.
}

protected final Table getWritableIcebergTable(TableIf table) {
// The expected ops fences lazy table acquisition to the authenticator generation selected at dispatch.
return IcebergUtils.getWritableIcebergTable((IcebergExternalTable) table, metadataOps);
}

}
9 changes: 5 additions & 4 deletions ...main/java/org/apache/doris/datasource/iceberg/action/IcebergCherrypickSnapshotAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
import org.apache.doris.common.ArgumentParsers;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
import org.apache.doris.info.PartitionNamesInfo;
import org.apache.doris.nereids.trees.expressions.Expression;

Expand All @@ -48,8 +48,9 @@ public class IcebergCherrypickSnapshotAction extends BaseIcebergAction {
public static final String SNAPSHOT_ID = "snapshot_id";

public IcebergCherrypickSnapshotAction(Map<String, String> properties,
Optional<PartitionNamesInfo> partitionNamesInfo, Optional<Expression> whereCondition) {
super("cherrypick_snapshot", properties, partitionNamesInfo, whereCondition);
Optional<PartitionNamesInfo> partitionNamesInfo, Optional<Expression> whereCondition,
IcebergMetadataOps metadataOps) {
super("cherrypick_snapshot", properties, partitionNamesInfo, whereCondition, metadataOps);
}

@Override
Expand All @@ -70,7 +71,7 @@ protected void validateIcebergAction() throws UserException {

@Override
protected List<String> executeAction(TableIf table) throws UserException {
Table icebergTable = ((IcebergExternalTable) table).getWritableIcebergTable();
Table icebergTable = getWritableIcebergTable(table);
Long sourceSnapshotId = namedArguments.getLong(SNAPSHOT_ID);

try {
Expand Down
42 changes: 33 additions & 9 deletions ...src/main/java/org/apache/doris/datasource/iceberg/action/IcebergExecuteActionFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,12 @@
package org.apache.doris.datasource.iceberg.action;

import org.apache.doris.common.DdlException;
import org.apache.doris.datasource.ExternalCatalog;
import org.apache.doris.datasource.hive.HMSExternalCatalog;
import org.apache.doris.datasource.iceberg.IcebergExternalCatalog;
import org.apache.doris.datasource.iceberg.IcebergExternalMetaCache.CatalogGenerationChangedException;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
import org.apache.doris.info.PartitionNamesInfo;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.plans.commands.execute.ExecuteAction;
Expand Down Expand Up @@ -59,41 +64,60 @@ public static ExecuteAction createAction(String actionType, Map<String, String>
Optional<PartitionNamesInfo> partitionNamesInfo,
Optional<Expression> whereCondition,
IcebergExternalTable table) throws DdlException {
IcebergMetadataOps metadataOps = getMetadataOps(table);

switch (actionType.toLowerCase()) {
case ROLLBACK_TO_SNAPSHOT:
return new IcebergRollbackToSnapshotAction(properties, partitionNamesInfo,
whereCondition);
whereCondition, metadataOps);
case ROLLBACK_TO_TIMESTAMP:
return new IcebergRollbackToTimestampAction(properties, partitionNamesInfo,
whereCondition);
whereCondition, metadataOps);
case SET_CURRENT_SNAPSHOT:
return new IcebergSetCurrentSnapshotAction(properties, partitionNamesInfo,
whereCondition);
whereCondition, metadataOps);
case CHERRYPICK_SNAPSHOT:
return new IcebergCherrypickSnapshotAction(properties, partitionNamesInfo,
whereCondition);
whereCondition, metadataOps);
case FAST_FORWARD:
return new IcebergFastForwardAction(properties, partitionNamesInfo,
whereCondition);
whereCondition, metadataOps);
case EXPIRE_SNAPSHOTS:
return new IcebergExpireSnapshotsAction(properties, partitionNamesInfo,
whereCondition);
whereCondition, metadataOps);
case REWRITE_DATA_FILES:
return new IcebergRewriteDataFilesAction(properties, partitionNamesInfo,
whereCondition);
whereCondition, metadataOps);
case PUBLISH_CHANGES:
return new IcebergPublishChangesAction(properties, partitionNamesInfo,
whereCondition);
whereCondition, metadataOps);
case REWRITE_MANIFESTS:
return new IcebergRewriteManifestsAction(properties, partitionNamesInfo,
whereCondition);
whereCondition, metadataOps);
default:
throw new DdlException("Unsupported Iceberg procedure: " + actionType
+ ". Supported procedures: " + String.join(", ", getSupportedActions()));
}
}

private static IcebergMetadataOps getMetadataOps(IcebergExternalTable table) throws DdlException {
ExternalCatalog catalog = table.getCatalog();
IcebergMetadataOps metadataOps;
if (catalog instanceof HMSExternalCatalog) {
metadataOps = ((HMSExternalCatalog) catalog).getIcebergMetadataOps();
} else if (catalog instanceof IcebergExternalCatalog) {
metadataOps = (IcebergMetadataOps) catalog.getMetadataOps();
} else {
throw new DdlException("Unsupported catalog type for Iceberg execute action: "
+ catalog.getClass().getSimpleName());
}
if (metadataOps == null) {
throw new CatalogGenerationChangedException(
"Catalog was reset while preparing Iceberg execute action");
}
return metadataOps;
}

/**
* Get supported Iceberg procedure names.
*
Expand Down
8 changes: 4 additions & 4 deletions ...rc/main/java/org/apache/doris/datasource/iceberg/action/IcebergExpireSnapshotsAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
import org.apache.doris.common.ArgumentParsers;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
import org.apache.doris.info.PartitionNamesInfo;
import org.apache.doris.nereids.trees.expressions.Expression;

Expand Down Expand Up @@ -69,8 +69,8 @@ public class IcebergExpireSnapshotsAction extends BaseIcebergAction {

public IcebergExpireSnapshotsAction(Map<String, String> properties,
Optional<PartitionNamesInfo> partitionNamesInfo,
Optional<Expression> whereCondition) {
super("expire_snapshots", properties, partitionNamesInfo, whereCondition);
Optional<Expression> whereCondition, IcebergMetadataOps metadataOps) {
super("expire_snapshots", properties, partitionNamesInfo, whereCondition, metadataOps);
}

@Override
Expand Down Expand Up @@ -149,7 +149,7 @@ protected void validateIcebergAction() throws UserException {

@Override
protected List<String> executeAction(TableIf table) throws UserException {
Table icebergTable = ((IcebergExternalTable) table).getWritableIcebergTable();
Table icebergTable = getWritableIcebergTable(table);

// Parse parameters
String olderThan = namedArguments.getString(OLDER_THAN);
Expand Down
8 changes: 4 additions & 4 deletions ...re/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergFastForwardAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
import org.apache.doris.common.ArgumentParsers;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
import org.apache.doris.info.PartitionNamesInfo;
import org.apache.doris.nereids.trees.expressions.Expression;

Expand All @@ -46,8 +46,8 @@ public class IcebergFastForwardAction extends BaseIcebergAction {

public IcebergFastForwardAction(Map<String, String> properties,
Optional<PartitionNamesInfo> partitionNamesInfo,
Optional<Expression> whereCondition) {
super("fast_forward", properties, partitionNamesInfo, whereCondition);
Optional<Expression> whereCondition, IcebergMetadataOps metadataOps) {
super("fast_forward", properties, partitionNamesInfo, whereCondition, metadataOps);
}

@Override
Expand All @@ -70,7 +70,7 @@ protected void validateIcebergAction() throws UserException {

@Override
protected List<String> executeAction(TableIf table) throws UserException {
Table icebergTable = ((IcebergExternalTable) table).getWritableIcebergTable();
Table icebergTable = getWritableIcebergTable(table);

String sourceBranch = namedArguments.getString(BRANCH);
String desBranch = namedArguments.getString(TO);
Expand Down
9 changes: 5 additions & 4 deletions ...src/main/java/org/apache/doris/datasource/iceberg/action/IcebergPublishChangesAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
import org.apache.doris.common.ArgumentParsers;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
import org.apache.doris.info.PartitionNamesInfo;
import org.apache.doris.nereids.trees.expressions.Expression;

Expand All @@ -47,8 +47,9 @@ public class IcebergPublishChangesAction extends BaseIcebergAction {
private static final String WAP_ID_PROP = "wap.id";

public IcebergPublishChangesAction(Map<String, String> properties,
Optional<PartitionNamesInfo> partitionNamesInfo, Optional<Expression> whereCondition) {
super("publish_changes", properties, partitionNamesInfo, whereCondition);
Optional<PartitionNamesInfo> partitionNamesInfo, Optional<Expression> whereCondition,
IcebergMetadataOps metadataOps) {
super("publish_changes", properties, partitionNamesInfo, whereCondition, metadataOps);
}

@Override
Expand All @@ -66,7 +67,7 @@ protected void validateIcebergAction() throws UserException {

@Override
protected List<String> executeAction(TableIf table) throws UserException {
Table icebergTable = ((IcebergExternalTable) table).getWritableIcebergTable();
Table icebergTable = getWritableIcebergTable(table);
String targetWapId = namedArguments.getString(WAP_ID);

// Find the target WAP snapshot
Expand Down
5 changes: 3 additions & 2 deletions ...c/main/java/org/apache/doris/datasource/iceberg/action/IcebergRewriteDataFilesAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.apache.doris.common.ArgumentParsers;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
import org.apache.doris.datasource.iceberg.IcebergUtils;
import org.apache.doris.datasource.iceberg.rewrite.RewriteDataFileExecutor;
import org.apache.doris.datasource.iceberg.rewrite.RewriteDataFilePlanner;
Expand Down Expand Up @@ -75,8 +76,8 @@ public class IcebergRewriteDataFilesAction extends BaseIcebergAction {

public IcebergRewriteDataFilesAction(Map<String, String> properties,
Optional<PartitionNamesInfo> partitionNamesInfo,
Optional<Expression> whereCondition) {
super("rewrite_data_files", properties, partitionNamesInfo, whereCondition);
Optional<Expression> whereCondition, IcebergMetadataOps metadataOps) {
super("rewrite_data_files", properties, partitionNamesInfo, whereCondition, metadataOps);

@github-actions github-actions Bot Sep 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Keep rewrite_data_files on the captured generation

This is the one mutating action that never consumes the generation fence added by this constructor: executeAction still uses the ordinary cached IcebergUtils.getIcebergTable, and RewriteDataFileExecutor later reads catalog.getTransactionManager() twice. If a reset occurs after dispatch captures ops/authenticator A, the rewrite can plan or create its transaction from generation B while the command's outer doAs is still A; a reset between the manager reads can also lose the newly allocated transaction. Its final transaction.commitTransaction() therefore retains the cross-generation authentication problem this PR is fixing, and its broad catch prevents a pre-mutation generation sentinel from reaching the retry loop. Please carry the captured ops/table/transaction manager through the rewrite end-to-end, with the fence outside the catch, and add a deterministic reset test. This is distinct from the earlier command thread because that fix relies on fenced writable acquisition, which this action bypasses.

}

/**
Expand Down
8 changes: 4 additions & 4 deletions ...c/main/java/org/apache/doris/datasource/iceberg/action/IcebergRewriteManifestsAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
import org.apache.doris.common.ArgumentParsers;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
import org.apache.doris.datasource.iceberg.rewrite.RewriteManifestExecutor;
import org.apache.doris.info.PartitionNamesInfo;
import org.apache.doris.nereids.trees.expressions.Expression;
Expand All @@ -47,8 +47,8 @@ public class IcebergRewriteManifestsAction extends BaseIcebergAction {

public IcebergRewriteManifestsAction(Map<String, String> properties,
Optional<PartitionNamesInfo> partitionNamesInfo,
Optional<Expression> whereCondition) {
super("rewrite_manifests", properties, partitionNamesInfo, whereCondition);
Optional<Expression> whereCondition, IcebergMetadataOps metadataOps) {
super("rewrite_manifests", properties, partitionNamesInfo, whereCondition, metadataOps);
}

@Override
Expand All @@ -67,8 +67,8 @@ protected void validateIcebergAction() throws UserException {

@Override
protected List<String> executeAction(TableIf table) throws UserException {
Table icebergTable = getWritableIcebergTable(table);
try {
Table icebergTable = ((IcebergExternalTable) table).getWritableIcebergTable();
Snapshot current = icebergTable.currentSnapshot();
if (current == null) {
// No current snapshot means the table is empty, no manifests to rewrite
Expand Down
8 changes: 4 additions & 4 deletions ...main/java/org/apache/doris/datasource/iceberg/action/IcebergRollbackToSnapshotAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
import org.apache.doris.common.ArgumentParsers;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
import org.apache.doris.info.PartitionNamesInfo;
import org.apache.doris.nereids.trees.expressions.Expression;

Expand All @@ -46,8 +46,8 @@ public class IcebergRollbackToSnapshotAction extends BaseIcebergAction {

public IcebergRollbackToSnapshotAction(Map<String, String> properties,
Optional<PartitionNamesInfo> partitionNamesInfo,
Optional<Expression> whereCondition) {
super("rollback_to_snapshot", properties, partitionNamesInfo, whereCondition);
Optional<Expression> whereCondition, IcebergMetadataOps metadataOps) {
super("rollback_to_snapshot", properties, partitionNamesInfo, whereCondition, metadataOps);
}

@Override
Expand All @@ -68,7 +68,7 @@ protected void validateIcebergAction() throws UserException {

@Override
protected List<String> executeAction(TableIf table) throws UserException {
Table icebergTable = ((IcebergExternalTable) table).getWritableIcebergTable();
Table icebergTable = getWritableIcebergTable(table);
Long targetSnapshotId = namedArguments.getLong(SNAPSHOT_ID);

Snapshot targetSnapshot = icebergTable.snapshot(targetSnapshotId);
Expand Down
8 changes: 4 additions & 4 deletions ...ain/java/org/apache/doris/datasource/iceberg/action/IcebergRollbackToTimestampAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
import org.apache.doris.common.UserException;
import org.apache.doris.common.util.TimeUtils;
import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
import org.apache.doris.info.PartitionNamesInfo;
import org.apache.doris.nereids.trees.expressions.Expression;

Expand All @@ -49,8 +49,8 @@ public class IcebergRollbackToTimestampAction extends BaseIcebergAction {

public IcebergRollbackToTimestampAction(Map<String, String> properties,
Optional<PartitionNamesInfo> partitionNamesInfo,
Optional<Expression> whereCondition) {
super("rollback_to_timestamp", properties, partitionNamesInfo, whereCondition);
Optional<Expression> whereCondition, IcebergMetadataOps metadataOps) {
super("rollback_to_timestamp", properties, partitionNamesInfo, whereCondition, metadataOps);
}

@Override
Expand Down Expand Up @@ -96,7 +96,7 @@ protected void validateIcebergAction() throws UserException {

@Override
protected List<String> executeAction(TableIf table) throws UserException {
Table icebergTable = ((IcebergExternalTable) table).getWritableIcebergTable();
Table icebergTable = getWritableIcebergTable(table);

String timestampStr = namedArguments.getString(TIMESTAMP);

Expand Down
Loading
Loading