This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new a8344169095 branch-4.1: [fix](iceberg) Run execute actions in catalog
auth scope (#67695)
a8344169095 is described below
commit a8344169095efbb27f305ab675a40b9f199ac756
Author: Gabriel <[email protected]>
AuthorDate: Thu Sep 10 11:19:09 2026 +0800
branch-4.1: [fix](iceberg) Run execute actions in catalog auth scope
(#67695)
## What problem does this PR solve?
Iceberg maintenance actions can load a table under the catalog
authentication context but execute metadata reads and commits after that
context has ended. On authenticated filesystems, actions such as
`expire_snapshots` and `rollback_to_snapshot` can therefore commit under
the FE process identity or splice a writable table from a new catalog
generation into a stale authentication scope.
## What is changed and how does it work?
- Capture the action's metadata operations and execution authenticator
atomically under the catalog reset monitor.
- Execute writable table acquisition and metadata mutation in that
generation's authentication scope.
- Fence writable acquisition with the captured metadata operations. If
the catalog resets before mutation starts, rebuild the action and retry
once under the new generation; commit failures are never retried.
- Preserve checked `UserException` handling across the Hadoop
authentication boundary.
- Add a deterministic two-authenticator reset test that verifies
generation A cannot load or commit after generation B replaces it.
- Cover both `rollback_to_snapshot` and `expire_snapshots` in the
Kerberos Iceberg/HDFS regression.
## Check List
- [x] Unit tests
- [x] FE Checkstyle
- [x] Regression script syntax validation
## Tests
- `ExecuteActionCommandTest`: 2 passed
- Catalog generation fence tests: 2 passed
- FE Checkstyle: 0 violations
- Kerberos regression Groovy syntax validation: passed
- Full Kerberos E2E was not run locally because it requires the
dedicated external Docker environment.
---
.../iceberg/IcebergExternalMetaCache.java | 11 +-
.../iceberg/action/BaseIcebergAction.java | 14 +-
.../action/IcebergCherrypickSnapshotAction.java | 9 +-
.../action/IcebergExecuteActionFactory.java | 42 +++-
.../action/IcebergExpireSnapshotsAction.java | 8 +-
.../iceberg/action/IcebergFastForwardAction.java | 8 +-
.../action/IcebergPublishChangesAction.java | 9 +-
.../action/IcebergRewriteDataFilesAction.java | 5 +-
.../action/IcebergRewriteManifestsAction.java | 8 +-
.../action/IcebergRollbackToSnapshotAction.java | 8 +-
.../action/IcebergRollbackToTimestampAction.java | 8 +-
.../action/IcebergSetCurrentSnapshotAction.java | 8 +-
.../trees/plans/commands/ExecuteActionCommand.java | 75 ++++++-
.../plans/commands/ExecuteActionCommandTest.java | 245 +++++++++++++++++++++
.../test_iceberg_hadoop_catalog_kerberos.groovy | 68 +++++-
15 files changed, 470 insertions(+), 56 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java
index b58e659abfc..1cd7c5684f6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java
@@ -422,12 +422,19 @@ public class IcebergExternalMetaCache extends
AbstractExternalMetaCache {
}
}
- 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);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/BaseIcebergAction.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/BaseIcebergAction.java
index b54d7509f00..550c8331ae2 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/BaseIcebergAction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/BaseIcebergAction.java
@@ -20,11 +20,16 @@ package org.apache.doris.datasource.iceberg.action;
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;
/**
@@ -33,11 +38,13 @@ import java.util.Optional;
* 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
@@ -71,4 +78,9 @@ public abstract class BaseIcebergAction extends
BaseExecuteAction {
// 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);
+ }
+
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergCherrypickSnapshotAction.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergCherrypickSnapshotAction.java
index 94924514a58..bf3be0c8559 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergCherrypickSnapshotAction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergCherrypickSnapshotAction.java
@@ -24,7 +24,7 @@ import org.apache.doris.catalog.Type;
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;
@@ -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
@@ -70,7 +71,7 @@ public class IcebergCherrypickSnapshotAction extends
BaseIcebergAction {
@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 {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergExecuteActionFactory.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergExecuteActionFactory.java
index 0d09a9ef35c..db3eb223707 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergExecuteActionFactory.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergExecuteActionFactory.java
@@ -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;
@@ -59,41 +64,60 @@ public class IcebergExecuteActionFactory {
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.
*
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergExpireSnapshotsAction.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergExpireSnapshotsAction.java
index 82a93022354..f2653f39216 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergExpireSnapshotsAction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergExpireSnapshotsAction.java
@@ -25,7 +25,7 @@ import org.apache.doris.common.AnalysisException;
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;
@@ -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
@@ -149,7 +149,7 @@ public class IcebergExpireSnapshotsAction extends
BaseIcebergAction {
@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);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergFastForwardAction.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergFastForwardAction.java
index cd746a7dbe6..82a0d746537 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergFastForwardAction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergFastForwardAction.java
@@ -24,7 +24,7 @@ import org.apache.doris.catalog.Type;
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;
@@ -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
@@ -70,7 +70,7 @@ public class IcebergFastForwardAction extends
BaseIcebergAction {
@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);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergPublishChangesAction.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergPublishChangesAction.java
index bf3f116d1cb..7fd7fbb710a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergPublishChangesAction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergPublishChangesAction.java
@@ -24,7 +24,7 @@ import org.apache.doris.catalog.Type;
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;
@@ -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
@@ -66,7 +67,7 @@ public class IcebergPublishChangesAction extends
BaseIcebergAction {
@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
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRewriteDataFilesAction.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRewriteDataFilesAction.java
index a22397a146b..940431d04c5 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRewriteDataFilesAction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRewriteDataFilesAction.java
@@ -23,6 +23,7 @@ import org.apache.doris.catalog.Type;
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;
@@ -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);
}
/**
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRewriteManifestsAction.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRewriteManifestsAction.java
index dce45c27296..9af02f29c39 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRewriteManifestsAction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRewriteManifestsAction.java
@@ -23,7 +23,7 @@ import org.apache.doris.catalog.Type;
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;
@@ -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
@@ -67,8 +67,8 @@ public class IcebergRewriteManifestsAction extends
BaseIcebergAction {
@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
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRollbackToSnapshotAction.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRollbackToSnapshotAction.java
index a5609f83439..c8fef0e9529 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRollbackToSnapshotAction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRollbackToSnapshotAction.java
@@ -24,7 +24,7 @@ import org.apache.doris.catalog.Type;
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;
@@ -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
@@ -68,7 +68,7 @@ public class IcebergRollbackToSnapshotAction extends
BaseIcebergAction {
@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);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRollbackToTimestampAction.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRollbackToTimestampAction.java
index de7e2a68079..948e27ec245 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRollbackToTimestampAction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergRollbackToTimestampAction.java
@@ -24,7 +24,7 @@ import org.apache.doris.catalog.Type;
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;
@@ -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
@@ -96,7 +96,7 @@ public class IcebergRollbackToTimestampAction extends
BaseIcebergAction {
@Override
protected List<String> executeAction(TableIf table) throws UserException {
- Table icebergTable = ((IcebergExternalTable)
table).getWritableIcebergTable();
+ Table icebergTable = getWritableIcebergTable(table);
String timestampStr = namedArguments.getString(TIMESTAMP);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergSetCurrentSnapshotAction.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergSetCurrentSnapshotAction.java
index 5b2c5bd220e..c16ef354673 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergSetCurrentSnapshotAction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/action/IcebergSetCurrentSnapshotAction.java
@@ -25,7 +25,7 @@ import org.apache.doris.common.AnalysisException;
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;
@@ -48,8 +48,8 @@ public class IcebergSetCurrentSnapshotAction extends
BaseIcebergAction {
public IcebergSetCurrentSnapshotAction(Map<String, String> properties,
Optional<PartitionNamesInfo> partitionNamesInfo,
- Optional<Expression> whereCondition) {
- super("set_current_snapshot", properties, partitionNamesInfo,
whereCondition);
+ Optional<Expression> whereCondition, IcebergMetadataOps
metadataOps) {
+ super("set_current_snapshot", properties, partitionNamesInfo,
whereCondition, metadataOps);
}
@Override
@@ -87,7 +87,7 @@ public class IcebergSetCurrentSnapshotAction extends
BaseIcebergAction {
@Override
protected List<String> executeAction(TableIf table) throws UserException {
- Table icebergTable = ((IcebergExternalTable)
table).getWritableIcebergTable();
+ Table icebergTable = getWritableIcebergTable(table);
Snapshot previousSnapshot = icebergTable.currentSnapshot();
Long previousSnapshotId = previousSnapshot != null ?
previousSnapshot.snapshotId() : null;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExecuteActionCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExecuteActionCommand.java
index edb3a059078..dd7e4b412bf 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExecuteActionCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExecuteActionCommand.java
@@ -24,9 +24,12 @@ import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.UserException;
+import org.apache.doris.common.security.authentication.ExecutionAuthenticator;
import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.datasource.ExternalCatalog;
import org.apache.doris.datasource.ExternalObjectLog;
import org.apache.doris.datasource.ExternalTable;
+import
org.apache.doris.datasource.iceberg.IcebergExternalMetaCache.CatalogGenerationChangedException;
import org.apache.doris.info.PartitionNamesInfo;
import org.apache.doris.info.TableNameInfo;
import org.apache.doris.nereids.trees.expressions.Expression;
@@ -47,6 +50,7 @@ import java.util.Optional;
* [WHERE condition]
*/
public class ExecuteActionCommand extends Command implements ForwardWithSync {
+ private static final int MAX_CATALOG_GENERATION_RETRIES = 1;
private final TableNameInfo tableNameInfo;
private final String actionName;
private final Map<String, String> properties;
@@ -96,15 +100,7 @@ public class ExecuteActionCommand extends Command
implements ForwardWithSync {
}
try {
- ExecuteAction action = ExecuteActionFactory.createAction(
- actionName, properties, partitionNamesInfo,
whereCondition, table);
-
- if (!action.isSupported(table)) {
- throw new AnalysisException("Action '" + actionName + "' is
not supported for this table engine");
- }
-
- action.validate(tableNameInfo, ctx.getCurrentUserIdentity());
- ResultSet resultSet = action.execute(table);
+ ResultSet resultSet = executeWithCatalogGenerationRetry(ctx,
table);
logRefreshTable(table, System.currentTimeMillis());
if (resultSet != null) {
executor.sendResultSet(resultSet);
@@ -144,6 +140,54 @@ public class ExecuteActionCommand extends Command
implements ForwardWithSync {
return whereCondition;
}
+ private ResultSet executeWithCatalogGenerationRetry(ConnectContext ctx,
TableIf table) throws Exception {
+ ExternalTable externalTable = (ExternalTable) table;
+ ExternalCatalog catalog = externalTable.getCatalog();
+ for (int retry = 0; ; retry++) {
+ try {
+ ExecuteAction action;
+ ExecutionAuthenticator authenticator;
+ synchronized (catalog) {
+ // Reset also holds this monitor, so the action's metadata
ops and
+ // authenticator form one generation.
+ catalog.makeSureInitialized();
+ action = ExecuteActionFactory.createAction(
+ actionName, properties, partitionNamesInfo,
whereCondition, table);
+ authenticator = catalog.getExecutionAuthenticator();
+ }
+ if (!action.isSupported(table)) {
+ throw new AnalysisException("Action '" + actionName + "'
is not supported for this table engine");
+ }
+ action.validate(tableNameInfo, ctx.getCurrentUserIdentity());
+ return executeAuthenticated(action, externalTable,
authenticator);
+ } catch (CatalogGenerationChangedException e) {
+ // A generation fence fails before mutation, so rebuilding the
action is safe;
+ // never retry commit errors.
+ if (retry >= MAX_CATALOG_GENERATION_RETRIES) {
+ throw new UserException(e.getMessage(), e);
+ }
+ }
+ }
+ }
+
+ private ResultSet executeAuthenticated(ExecuteAction action, ExternalTable
table,
+ ExecutionAuthenticator authenticator) throws Exception {
+ try {
+ // Iceberg tables retain filesystem configuration, not the
caller's UGI, so loading the table and
+ // committing its metadata must stay within one catalog
authentication scope.
+ return authenticator.execute(() -> {
+ try {
+ return action.execute(table);
+ } catch (UserException e) {
+ // Hadoop doAs obscures checked exceptions, so carry this
one across as a runtime exception.
+ throw new AuthenticatedActionException(e);
+ }
+ });
+ } catch (AuthenticatedActionException e) {
+ throw e.getUserException();
+ }
+ }
+
/**
* Log refresh table to make follow fe metadata cache refresh.
*
@@ -164,4 +208,17 @@ public class ExecuteActionCommand extends Command
implements ForwardWithSync {
throw new UserException("Unsupported table type: " +
table.getClass().getName() + " for refresh table");
}
}
+
+ private static final class AuthenticatedActionException extends
RuntimeException {
+ private final UserException userException;
+
+ private AuthenticatedActionException(UserException userException) {
+ super(userException);
+ this.userException = userException;
+ }
+
+ private UserException getUserException() {
+ return userException;
+ }
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExecuteActionCommandTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExecuteActionCommandTest.java
new file mode 100644
index 00000000000..2ed4d7a6cdf
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExecuteActionCommandTest.java
@@ -0,0 +1,245 @@
+// 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.nereids.trees.plans.commands;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.security.authentication.ExecutionAuthenticator;
+import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.datasource.ExternalMetaCacheMgr;
+import org.apache.doris.datasource.NameMapping;
+import org.apache.doris.datasource.iceberg.IcebergExternalCatalog;
+import org.apache.doris.datasource.iceberg.IcebergExternalMetaCache;
+import org.apache.doris.datasource.iceberg.IcebergExternalTable;
+import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
+import org.apache.doris.info.TableNameInfo;
+import org.apache.doris.mysql.privilege.AccessControllerManager;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.trees.plans.commands.execute.ExecuteAction;
+import
org.apache.doris.nereids.trees.plans.commands.execute.ExecuteActionFactory;
+import org.apache.doris.persist.EditLog;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.StmtExecutor;
+
+import org.apache.iceberg.ManageSnapshots;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+class ExecuteActionCommandTest {
+
+ @Test
+ void executesActionWithinCatalogAuthenticationScope() throws Exception {
+ Env env = Mockito.mock(Env.class);
+ CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
+ CatalogIf<?> catalog = Mockito.mock(CatalogIf.class);
+ DatabaseIf<?> database = Mockito.mock(DatabaseIf.class);
+ IcebergExternalCatalog externalCatalog =
Mockito.mock(IcebergExternalCatalog.class);
+ IcebergExternalTable table = Mockito.mock(IcebergExternalTable.class);
+ ExecuteAction action = Mockito.mock(ExecuteAction.class);
+ TableNameInfo tableName = Mockito.mock(TableNameInfo.class);
+ ConnectContext context = Mockito.mock(ConnectContext.class);
+ StmtExecutor executor = Mockito.mock(StmtExecutor.class);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ RecordingAuthenticator authenticator = new RecordingAuthenticator();
+ AtomicBoolean actionExecutedInScope = new AtomicBoolean();
+
+ Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+ Mockito.when(tableName.getCtl()).thenReturn("iceberg");
+ Mockito.when(tableName.getDb()).thenReturn("db");
+ Mockito.when(tableName.getTbl()).thenReturn("tbl");
+ Mockito.when(catalogMgr.getCatalog("iceberg")).thenReturn(catalog);
+ Mockito.doReturn(database).when(catalog).getDbNullable("db");
+ Mockito.doReturn(table).when(database).getTableNullable("tbl");
+ Mockito.when(table.getCatalog()).thenReturn(externalCatalog);
+ Mockito.when(table.getDbName()).thenReturn("db");
+ Mockito.when(table.getName()).thenReturn("tbl");
+
Mockito.when(externalCatalog.getExecutionAuthenticator()).thenReturn(authenticator);
+ Mockito.when(action.isSupported(table)).thenReturn(true);
+ Mockito.when(action.execute(table)).thenAnswer(invocation -> {
+ actionExecutedInScope.set(authenticator.inScope);
+ return null;
+ });
+
+ ExecuteActionCommand command = new ExecuteActionCommand(tableName,
"expire_snapshots",
+ Collections.emptyMap(), Optional.empty(), Optional.empty());
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<ExecuteActionFactory> factoryMock =
Mockito.mockStatic(ExecuteActionFactory.class)) {
+ envMock.when(Env::getCurrentEnv).thenReturn(env);
+ factoryMock.when(() -> ExecuteActionFactory.createAction(
+ "expire_snapshots", Collections.emptyMap(),
Optional.empty(), Optional.empty(), table))
+ .thenReturn(action);
+
+ command.run(context, executor);
+ }
+
+ Assertions.assertEquals(1, authenticator.executionCount);
+ Assertions.assertTrue(actionExecutedInScope.get());
+ }
+
+ @Test
+ void retriesRollbackOnCatalogAuthenticationGenerationChange() throws
Exception {
+ Env env = Mockito.mock(Env.class);
+ CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
+ CatalogIf<?> commandCatalog = Mockito.mock(CatalogIf.class);
+ DatabaseIf<?> database = Mockito.mock(DatabaseIf.class);
+ IcebergExternalCatalog externalCatalog =
Mockito.mock(IcebergExternalCatalog.class);
+ IcebergExternalTable table = Mockito.mock(IcebergExternalTable.class);
+ IcebergMetadataOps generationOneOps =
Mockito.mock(IcebergMetadataOps.class);
+ IcebergMetadataOps generationTwoOps =
Mockito.mock(IcebergMetadataOps.class);
+ Table generationTwoTable = Mockito.mock(Table.class);
+ Snapshot targetSnapshot = Mockito.mock(Snapshot.class);
+ Snapshot previousSnapshot = Mockito.mock(Snapshot.class);
+ ManageSnapshots manageSnapshots = Mockito.mock(ManageSnapshots.class);
+ TableNameInfo tableName = Mockito.mock(TableNameInfo.class);
+ ConnectContext context = Mockito.mock(ConnectContext.class);
+ UserIdentity userIdentity = Mockito.mock(UserIdentity.class);
+ StmtExecutor executor = Mockito.mock(StmtExecutor.class);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ AccessControllerManager accessManager =
Mockito.mock(AccessControllerManager.class);
+ ExternalMetaCacheMgr externalMetaCacheMgr =
Mockito.mock(ExternalMetaCacheMgr.class);
+ AtomicReference<ExecutionAuthenticator> currentAuthenticator = new
AtomicReference<>();
+ AtomicReference<IcebergMetadataOps> currentOps = new
AtomicReference<>(generationOneOps);
+ AtomicReference<ExecutionAuthenticator> activeAuthenticator = new
AtomicReference<>();
+ RecordingAuthenticator generationTwo = new
RecordingAuthenticator(activeAuthenticator, null);
+ RecordingAuthenticator generationOne = new
RecordingAuthenticator(activeAuthenticator, () -> {
+ currentOps.set(generationTwoOps);
+ currentAuthenticator.set(generationTwo);
+ });
+ currentAuthenticator.set(generationOne);
+ Map<String, String> properties =
Collections.singletonMap("snapshot_id", "123");
+ NameMapping mapping = new NameMapping(1L, "test_db", "test_table",
"remote_db", "remote_table");
+ ExecutorService cacheExecutor = Executors.newSingleThreadExecutor();
+ IcebergExternalMetaCache cache = new
IcebergExternalMetaCache(cacheExecutor) {
+ @Override
+ protected CatalogIf<?> getCatalog(long catalogId) {
+ return externalCatalog;
+ }
+ };
+
+ Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+ Mockito.when(env.getAccessManager()).thenReturn(accessManager);
+
Mockito.when(env.getExtMetaCacheMgr()).thenReturn(externalMetaCacheMgr);
+ Mockito.when(tableName.getCtl()).thenReturn("iceberg_catalog");
+ Mockito.when(tableName.getDb()).thenReturn("test_db");
+ Mockito.when(tableName.getTbl()).thenReturn("test_table");
+
Mockito.when(catalogMgr.getCatalog("iceberg_catalog")).thenReturn(commandCatalog);
+
Mockito.doReturn(database).when(commandCatalog).getDbNullable("test_db");
+ Mockito.doReturn(table).when(database).getTableNullable("test_table");
+ Mockito.when(table.getCatalog()).thenReturn(externalCatalog);
+ Mockito.when(externalCatalog.getId()).thenReturn(1L);
+ Mockito.when(table.getDbName()).thenReturn("test_db");
+ Mockito.when(table.getName()).thenReturn("test_table");
+ Mockito.when(table.getOrBuildNameMapping()).thenReturn(mapping);
+
Mockito.when(context.getCurrentUserIdentity()).thenReturn(userIdentity);
+
Mockito.when(accessManager.checkTblPriv(Mockito.nullable(ConnectContext.class),
+ Mockito.anyString(), Mockito.anyString(), Mockito.anyString(),
Mockito.eq(PrivPredicate.ALTER)))
+ .thenReturn(true);
+ Mockito.when(externalCatalog.getExecutionAuthenticator()).thenAnswer(
+ invocation -> currentAuthenticator.get());
+ Mockito.when(externalCatalog.getMetadataOps()).thenAnswer(invocation
-> currentOps.get());
+
Mockito.when(generationOneOps.getExecutionAuthenticator()).thenReturn(generationOne);
+
Mockito.when(generationTwoOps.getExecutionAuthenticator()).thenReturn(generationTwo);
+ Mockito.when(generationTwoOps.loadTable("remote_db",
"remote_table")).thenAnswer(invocation -> {
+ Assertions.assertSame(generationTwo, activeAuthenticator.get());
+ return generationTwoTable;
+ });
+
Mockito.when(generationTwoTable.snapshot(123L)).thenReturn(targetSnapshot);
+
Mockito.when(generationTwoTable.currentSnapshot()).thenReturn(previousSnapshot);
+ Mockito.when(previousSnapshot.snapshotId()).thenReturn(456L);
+
Mockito.when(generationTwoTable.manageSnapshots()).thenReturn(manageSnapshots);
+
Mockito.when(manageSnapshots.rollbackTo(123L)).thenReturn(manageSnapshots);
+ ExecuteActionCommand command = new ExecuteActionCommand(tableName,
"rollback_to_snapshot",
+ properties, Optional.empty(), Optional.empty());
+
+ try {
+ cache.initCatalog(1L, Collections.emptyMap());
+ Mockito.when(externalMetaCacheMgr.iceberg(1L)).thenReturn(cache);
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class)) {
+ envMock.when(Env::getCurrentEnv).thenReturn(env);
+
+ command.run(context, executor);
+ }
+
+ Assertions.assertEquals(1, generationOne.executionCount);
+ Assertions.assertEquals(2, generationTwo.executionCount);
+ Mockito.verify(generationOneOps,
Mockito.never()).loadTable(Mockito.anyString(), Mockito.anyString());
+ Mockito.verify(generationTwoOps).loadTable("remote_db",
"remote_table");
+ Mockito.verify(manageSnapshots).commit();
+ } finally {
+ cache.close();
+ cacheExecutor.shutdownNow();
+ }
+ }
+
+ private static class RecordingAuthenticator implements
ExecutionAuthenticator {
+ private boolean inScope;
+ private int executionCount;
+ private final AtomicReference<ExecutionAuthenticator>
activeAuthenticator;
+ private final Runnable beforeTask;
+
+ private RecordingAuthenticator() {
+ this(null, null);
+ }
+
+ private RecordingAuthenticator(AtomicReference<ExecutionAuthenticator>
activeAuthenticator,
+ Runnable beforeTask) {
+ this.activeAuthenticator = activeAuthenticator;
+ this.beforeTask = beforeTask;
+ }
+
+ @Override
+ public <T> T execute(Callable<T> task) throws Exception {
+ executionCount++;
+ inScope = true;
+ ExecutionAuthenticator previousAuthenticator = null;
+ if (activeAuthenticator != null) {
+ previousAuthenticator = activeAuthenticator.get();
+ activeAuthenticator.set(this);
+ }
+ try {
+ if (beforeTask != null) {
+ beforeTask.run();
+ }
+ return task.call();
+ } finally {
+ inScope = false;
+ if (activeAuthenticator != null) {
+ activeAuthenticator.set(previousAuthenticator);
+ }
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/kerberos/test_iceberg_hadoop_catalog_kerberos.groovy
b/regression-test/suites/external_table_p0/kerberos/test_iceberg_hadoop_catalog_kerberos.groovy
index 484b698d27c..01cfa5d9961 100644
---
a/regression-test/suites/external_table_p0/kerberos/test_iceberg_hadoop_catalog_kerberos.groovy
+++
b/regression-test/suites/external_table_p0/kerberos/test_iceberg_hadoop_catalog_kerberos.groovy
@@ -98,6 +98,72 @@ suite("test_iceberg_hadoop_catalog_kerberos",
"p0,external,kerberos,external_doc
'partition_val2'
);
"""
+ def snapshotsAfterFirstInsert = sql """
+ select snapshot_id from ${test_tbl_name}\$snapshots order by
committed_at
+ """
+ assert snapshotsAfterFirstInsert.size() == 1
+ String firstSnapshotId = String.valueOf(snapshotsAfterFirstInsert[0][0])
+
+ sql """
+ insert into ${test_tbl_name} values (
+ '2024-05-27 12:34:56',
+ false,
+ 456,
+ 4567890123456,
+ 45.67,
+ 89.012,
+ 45678.9012,
+ 'another example',
+ '2024-05-27',
+ '2024-05-27 14:00:00',
+ 'partition_val1',
+ 'partition_val2'
+ );
+ """
+ def dataBeforeRollback = sql """select count(1) from ${test_tbl_name}"""
+ assert dataBeforeRollback.get(0).get(0) == 2
+
+ // Rollback must commit metadata with the same Kerberos identity used to
acquire the table.
+ sql """
+ alter table ${catalog_name}.${database_name}.${test_tbl_name}
+ execute rollback_to_snapshot("snapshot_id" = "${firstSnapshotId}")
+ """
+ def dataAfterRollback = sql """select count(1) from ${test_tbl_name}"""
+ assert dataAfterRollback.get(0).get(0) == 1
+
+ sql """
+ insert into ${test_tbl_name} values (
+ '2024-05-27 12:34:56',
+ false,
+ 456,
+ 4567890123456,
+ 45.67,
+ 89.012,
+ 45678.9012,
+ 'another example',
+ '2024-05-27',
+ '2024-05-27 14:00:00',
+ 'partition_val1',
+ 'partition_val2'
+ );
+ """
+
+ def snapshotsBeforeExpire = sql """
+ select snapshot_id from ${test_tbl_name}\$snapshots order by
committed_at
+ """
+ assert snapshotsBeforeExpire.size() == 3
+
+ // The action must reuse the catalog's Kerberos identity while committing
new metadata.
+ sql """
+ alter table ${catalog_name}.${database_name}.${test_tbl_name}
+ execute expire_snapshots("retain_last" = "1")
+ """
+
+ def snapshotsAfterExpire = sql """
+ select snapshot_id from ${test_tbl_name}\$snapshots order by
committed_at
+ """
+ assert snapshotsAfterExpire.size() == 1
+
def dataResult = sql """select count(1) from ${test_tbl_name} """
- assert dataResult.get(0).get(0) == 1
+ assert dataResult.get(0).get(0) == 2
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
