This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new e5406443893 [fix](iceberg) Keep the Kerberos identity on Iceberg
procedure worker threads (#68902)
e5406443893 is described below
commit e54064438936dd408a3577cc6a975f04a6e69f05
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Mon Oct 12 14:27:53 2026 +0800
[fix](iceberg) Keep the Kerberos identity on Iceberg procedure worker
threads (#68902)
### What problem does this PR solve?
Issue Number: None
Related PR: #67695
Problem Summary:
**In short.** This ports the tests of branch-4.1 #67695, which made
Iceberg maintenance actions run inside the catalog's authentication
scope, and fixes what the ported Kerberos case found on master: on a
Kerberos-secured catalog, `expire_snapshots` committed its new metadata
and then failed while Iceberg read the expired snapshots' manifest lists
on its worker threads, which do not carry the catalog's Kerberos
identity. `cherrypick_snapshot`, `publish_changes` and
`rewrite_data_files` planning open files on those threads too.
**Background.** On branch-4.x an action loaded the table under the
catalog's authentication context and then read and committed metadata
after that context had ended, so on a Kerberized HDFS catalog the commit
could run under the FE process identity. #67695 captures the action's
metadata operations and the catalog's authenticator as one generation,
runs table acquisition and the commit in that scope, and retries once
when the catalog is reset before the mutation starts.
On master the actions live in `fe-connector-iceberg`:
`ConnectorExecuteAction` calls `IcebergProcedureOps.execute`, which runs
the table load, the procedure body and its commit inside one
`ConnectorContext.executeAuthenticated`. Each call takes its catalog ops
and its authenticator from the same connector instance, and the catalog
generation it starts with is leased until the call ends
(`IcebergCatalogResourceTracker`), so a table from a newer generation
can never meet an older authenticator, and the generation fence and
retry of #67695 have nothing to port.
That `doAs` covers only the calling thread, though. Iceberg reads and
writes metadata files on its shared worker pool, and those threads run
without the plugin's Kerberos login. For scans and writes the connector
already wraps the table's `FileIO` so that every file is opened inside
the catalog's `doAs` (`IcebergScanPlanProvider.wrapTableForScan`,
`IcebergConnectorTransaction.openTransaction`); a file opened that way
keeps its authenticated connection on any thread. Procedures were handed
the raw table.
**The problem, and what it cost.**
1. On a Kerberos-secured HDFS catalog, a procedure fails as soon as
Iceberg opens a metadata file on a worker thread. The first CI run of
this PR hit it: `test_iceberg_hadoop_catalog_kerberos` failed at
`expire_snapshots` with `Failed to expire snapshots: Failed to open
input stream for file: hdfs://.../metadata/snap-....avro`.
`RemoveSnapshots` had already committed the new metadata, so the
snapshots were gone while their files stayed on HDFS, the statement
reported an error, and the FE did not refresh the table.
| Procedure on a Kerberos HDFS catalog | Opened on a worker thread |
Before | After |
|---|---|---|---|
| `expire_snapshots`, when something expires | the expired snapshots'
manifest lists and manifests, after the commit | commits, then fails |
succeeds |
| `cherrypick_snapshot` / `publish_changes`, when main has moved on |
the manifest of the picked files | fails | succeeds |
| `rewrite_data_files` on a table with more than one manifest | the
manifests, while planning | fails | succeeds |
2. Two of #67695's tests have no counterpart on master:
| branch-4.x test | master |
|---|---|
|
`ExecuteActionCommandTest.executesActionWithinCatalogAuthenticationScope`
| `IcebergProcedureOpsTest` only asserts that one auth scope is entered.
Loading the table inside `executeAuthenticated` and running the body
after it returns keeps every existing case green. |
| `test_iceberg_hadoop_catalog_kerberos`: `rollback_to_snapshot` and
`expire_snapshots` on the Kerberized HDFS catalog | The suite creates,
inserts and counts, but runs no action. |
**How this PR fixes it.**
- `IcebergProcedureOps.wrapTableForProcedure`: the procedure body and
the rewrite planner get the table with its `FileIO` wrapped like the
scan and write paths. Non-Kerberos catalogs keep the loaded table
unchanged.
- `IcebergAuthenticatedFileIO.withPrefixOperations`: unlike the scan and
write wrapper, the procedure's `FileIO` keeps the delegate's prefix
listing, inside the same `doAs`, because `remove_orphan_files` rejects a
`FileIO` without it. Bulk deletes stay hidden, so deletes go file by
file through the authenticated `deleteFile`; `HadoopFileIO`'s own bulk
delete would run on its private pool, outside any `doAs`. One visible
effect: `expire_snapshots`' `max_concurrent_deletes` used to be ignored
with a warning on these catalogs, because the raw `HadoopFileIO`
supports bulk deletes, and now takes effect.
- Tests:
- `IcebergProcedureOpsKerberosIoTest` (new) runs procedures on a local
Hadoop table through an authenticator that marks only the thread inside
its `doAs`, the way a Kerberos `Subject` is invisible to pool threads,
and a `FileIO` that records every call made outside it:
`expire_snapshots` after the rollback sequence of the Kerberos suite,
`cherrypick_snapshot` after main has moved on, `remove_orphan_files`,
and `rewrite_data_files` planning on a two-manifest table. A
non-Kerberos catalog keeps the loaded table.
- `IcebergAuthenticatedFileIOTest`: prefix listing and deletion run
inside `doAs`, bulk deletes are not exposed, and a delegate without
prefix listing gets none.
- From #67695:
`IcebergProcedureOpsTest.commitsTheProcedureInsideTheAuthScope` reads
the table's current snapshot just before `executeAuthenticated` leaves
its scope and asserts the rollback has already been committed (the test
`RecordingConnectorContext` gains a hook that runs at that point);
`test_iceberg_hadoop_catalog_kerberos` gains the two steps of #67695
unchanged: a second insert, `rollback_to_snapshot` to the first snapshot
(two rows back to one), a third insert, and
`expire_snapshots("retain_last" = "1")` (three snapshots down to one).
`ExecuteActionCommandTest.retriesRollbackOnCatalogAuthenticationGenerationChange`
is not ported: master has no generation retry to test.
Not changed: `ConnectorExecuteAction` refreshes the table only when the
procedure returns, so a failure after Iceberg's commit still skips the
refresh. branch-4.x's `ExecuteActionCommand` behaves the same way
(`logRefreshTable` runs only after the action returns); with this fix
the Kerberos case no longer fails after the commit.
**Results.** Each change is pinned by a test that fails without it:
| Change undone | Result |
|---|---|
| `execute` hands the action the loaded table | the `expire_snapshots`
case records six opens on `iceberg-worker-pool` threads outside the
`doAs` (three `snap-*.avro` manifest lists, three manifests), the CI
failure; the `cherrypick_snapshot` case records the new manifest's
`newOutputFile` |
| the rewrite planner gets the loaded table | the planning case records
two manifest opens on worker threads |
| the plain `IcebergAuthenticatedFileIO` instead of
`withPrefixOperations` | `remove_orphan_files` fails with "requires
FileIO prefix listing support" |
| the procedure body runs after `executeAuthenticated` returns |
`commitsTheProcedureInsideTheAuthScope` fails |
The fe-connector-iceberg tests (1507) pass. The External pipeline runs
`test_iceberg_hadoop_catalog_kerberos`.
Co-authored-by: Gabriel <[email protected]>
---
.../iceberg/IcebergAuthenticatedFileIO.java | 48 ++-
.../connector/iceberg/IcebergProcedureOps.java | 44 ++-
.../connector/iceberg/IcebergScanPlanning.java | 14 +-
.../iceberg/IcebergAuthenticatedFileIOTest.java | 60 +++-
.../iceberg/IcebergProcedureOpsKerberosIoTest.java | 338 +++++++++++++++++++++
.../connector/iceberg/IcebergProcedureOpsTest.java | 22 ++
.../iceberg/IcebergRestScanPlanningTest.java | 42 +++
.../iceberg/RecordingConnectorContext.java | 9 +-
.../test_iceberg_hadoop_catalog_kerberos.groovy | 71 ++++-
9 files changed, 634 insertions(+), 14 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergAuthenticatedFileIO.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergAuthenticatedFileIO.java
index bfefb188fb7..c612f448a7c 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergAuthenticatedFileIO.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergAuthenticatedFileIO.java
@@ -20,8 +20,10 @@ package org.apache.doris.connector.iceberg;
import org.apache.doris.kerberos.HadoopAuthenticator;
import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.FileInfo;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.SupportsPrefixOperations;
import java.io.IOException;
import java.io.UncheckedIOException;
@@ -54,9 +56,10 @@ import java.util.Objects;
* <p>The bundled {@code HadoopFileIO} additionally implements {@code
DelegateFileIO} (bulk / prefix ops), used only
* by maintenance actions (orphan-file cleanup), never by the append/rewrite
commit path exercised here; a caller
* that probes for those interfaces simply falls back to per-file operations,
which route through the wrapped
- * primitives above. Kept a plain {@link FileIO} deliberately to avoid
coupling to that optional surface.
+ * primitives above. Kept a plain {@link FileIO} deliberately to avoid
coupling to that optional surface; the
+ * procedure path, which does need prefix listing, uses {@link
#withPrefixOperations}.
*/
-final class IcebergAuthenticatedFileIO implements FileIO {
+class IcebergAuthenticatedFileIO implements FileIO {
private final FileIO delegate;
private final HadoopAuthenticator authenticator;
@@ -66,6 +69,20 @@ final class IcebergAuthenticatedFileIO implements FileIO {
this.authenticator = Objects.requireNonNull(authenticator,
"authenticator");
}
+ /**
+ * Wraps {@code delegate} for an {@code ALTER TABLE EXECUTE} procedure.
When the delegate lists and deletes by
+ * prefix ({@code HadoopFileIO} does), so does the result, inside the same
{@code doAs}:
+ * {@code remove_orphan_files} rejects a FileIO without prefix listing.
Bulk deletes stay hidden, so callers
+ * delete file by file through {@link #deleteFile}, which authenticates on
any thread; {@code HadoopFileIO}'s own
+ * bulk delete would run on its private pool, outside any {@code doAs}.
+ */
+ static IcebergAuthenticatedFileIO withPrefixOperations(FileIO delegate,
HadoopAuthenticator authenticator) {
+ if (delegate instanceof SupportsPrefixOperations) {
+ return new PrefixOperations((SupportsPrefixOperations) delegate,
authenticator);
+ }
+ return new IcebergAuthenticatedFileIO(delegate, authenticator);
+ }
+
@Override
public InputFile newInputFile(String path) {
return doAs(() -> delegate.newInputFile(path));
@@ -104,11 +121,36 @@ final class IcebergAuthenticatedFileIO implements FileIO {
delegate.close();
}
- private <T> T doAs(PrivilegedExceptionAction<T> action) {
+ <T> T doAs(PrivilegedExceptionAction<T> action) {
try {
return authenticator.doAs(action);
} catch (IOException e) {
throw new UncheckedIOException(e);
}
}
+
+ private static final class PrefixOperations extends
IcebergAuthenticatedFileIO
+ implements SupportsPrefixOperations {
+
+ private final SupportsPrefixOperations prefixDelegate;
+
+ private PrefixOperations(SupportsPrefixOperations delegate,
HadoopAuthenticator authenticator) {
+ super(delegate, authenticator);
+ this.prefixDelegate = delegate;
+ }
+
+ @Override
+ public Iterable<FileInfo> listPrefix(String prefix) {
+ // Like newInputFile: HadoopFileIO resolves the FileSystem here
and the returned iterable lists through it.
+ return doAs(() -> prefixDelegate.listPrefix(prefix));
+ }
+
+ @Override
+ public void deletePrefix(String prefix) {
+ doAs(() -> {
+ prefixDelegate.deletePrefix(prefix);
+ return null;
+ });
+ }
+ }
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergProcedureOps.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergProcedureOps.java
index e0a19bd2e5e..1b77d762060 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergProcedureOps.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergProcedureOps.java
@@ -32,6 +32,11 @@ import
org.apache.doris.connector.spi.procedure.ConnectorRewriteGroup;
import org.apache.doris.connector.spi.procedure.ConnectorRewriteStatistics;
import org.apache.doris.connector.spi.procedure.ProcedureExecutionMode;
import org.apache.doris.connector.spi.pushdown.ConnectorPredicate;
+import org.apache.doris.kerberos.HadoopAuthenticator;
+
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
import java.time.ZoneId;
import java.util.Arrays;
@@ -203,7 +208,7 @@ public class IcebergProcedureOps implements
ConnectorProcedureOps {
try {
result = context.executeAuthenticated(() ->
ops.withTable(handle.getDbName(),
handle.getTableName(),
- table -> action.execute(table, session)));
+ table ->
action.execute(wrapTableForProcedure(table), session)));
} catch (DorisConnectorException e) {
throw e;
} catch (Exception e) {
@@ -237,8 +242,11 @@ public class IcebergProcedureOps implements
ConnectorProcedureOps {
groups = ops.withTable(handle.getDbName(), handle.getTableName(),
planner::planAndOrganizeTasks);
} else {
try {
- groups = context.executeAuthenticated(() -> ops.withTable(
- handle.getDbName(), handle.getTableName(),
planner::planAndOrganizeTasks));
+ // A server-planned table (RESTTable) keeps its own newScan(),
which asks the REST server for the
+ // file scan tasks; rebuilding it as a plain BaseTable would
plan from local manifests instead.
+ groups = context.executeAuthenticated(() ->
ops.withTable(handle.getDbName(), handle.getTableName(),
+ table ->
planner.planAndOrganizeTasks(IcebergScanPlanning.requiresServerSideScanPlanning(table)
+ ? table : wrapTableForProcedure(table))));
} catch (DorisConnectorException e) {
throw e;
} catch (Exception e) {
@@ -249,6 +257,36 @@ public class IcebergProcedureOps implements
ConnectorProcedureOps {
return
groups.stream().map(IcebergProcedureOps::toConnectorRewriteGroup).collect(Collectors.toList());
}
+ /**
+ * Routes the procedure table's {@code io()} through the plugin-side
Kerberos {@code doAs} — the procedure-path
+ * mirror of {@code IcebergScanPlanProvider.wrapTableForScan} and
+ * {@code IcebergConnectorTransaction.openTransaction}. The {@code
executeAuthenticated} scope around the body
+ * covers only the calling thread, while iceberg reads and writes metadata
files on its shared worker pool:
+ * {@code expire_snapshots} reads the expired snapshots' manifest lists
there after committing (CI:
+ * test_iceberg_hadoop_catalog_kerberos failed "Failed to open input
stream for file: .../snap-*.avro"),
+ * {@code cherrypick_snapshot} writes the picked files' manifest there,
and rewrite planning reads the manifests
+ * of a multi-manifest table there. The wrapped FileIO keeps prefix
listing for {@code remove_orphan_files}
+ * ({@link IcebergAuthenticatedFileIO#withPrefixOperations}). Rewrite
planning skips the wrap for a server-planned
+ * table, whose scan tasks come from the REST server ({@link
#planInAuthScopeUntracked}).
+ *
+ * <p>Non-Kerberos catalogs ({@code getPluginAuthenticator() == null}) and
offline tests (plain context /
+ * non-{@link BaseTable} fakes) get the table unchanged. Package-private
for unit testing.
+ */
+ Table wrapTableForProcedure(Table table) {
+ if (!(context instanceof TcclPinningConnectorContext) || !(table
instanceof BaseTable)) {
+ return table;
+ }
+ HadoopAuthenticator auth = ((TcclPinningConnectorContext)
context).getPluginAuthenticator();
+ if (auth == null) {
+ return table;
+ }
+ BaseTable baseTable = (BaseTable) table;
+ TableOperations rawOps = baseTable.operations();
+ return new BaseTable(new IcebergAuthenticatedTableOperations(rawOps,
+ IcebergAuthenticatedFileIO.withPrefixOperations(rawOps.io(),
auth)),
+ table.name(), baseTable.reporter());
+ }
+
private <T> T withCatalogLease(Supplier<T> operation) {
if (resourceTracker == null) {
return operation.get();
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanning.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanning.java
index bf48244ff84..dfef09e818b 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanning.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanning.java
@@ -26,11 +26,17 @@ import org.apache.iceberg.Table;
final class IcebergScanPlanning {
private IcebergScanPlanning() {}
+ /** Whether the SDK table plans its scans on the catalog server ({@code
RESTTable}), not from local manifests. */
+ static boolean requiresServerSideScanPlanning(Table table) {
+ // RESTTable overrides this to false; plain BaseTable inherits true.
+ return table instanceof SupportsDistributedScanPlanning
+ && !((SupportsDistributedScanPlanning)
table).allowDistributedPlanning();
+ }
+
static void rejectServerSideScanPlanning(Table table, String tableName) {
- if (table instanceof SupportsDistributedScanPlanning
- && !((SupportsDistributedScanPlanning)
table).allowDistributedPlanning()) {
- // RESTTable overrides this to false; plain BaseTable inherits
true. Doris cannot consume
- // server tasks/residuals or scan-scoped credentials yet, so
reject before local manifest I/O.
+ if (requiresServerSideScanPlanning(table)) {
+ // Doris cannot consume server tasks/residuals or scan-scoped
credentials yet, so reject before
+ // local manifest I/O.
throw new DorisConnectorException("Iceberg server-side scan
planning is not supported for table "
+ tableName + "; configure the REST catalog to use
client-side scan planning");
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergAuthenticatedFileIOTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergAuthenticatedFileIOTest.java
index 352b477f53c..bbabe19075b 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergAuthenticatedFileIOTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergAuthenticatedFileIOTest.java
@@ -20,9 +20,13 @@ package org.apache.doris.connector.iceberg;
import org.apache.doris.kerberos.HadoopAuthenticator;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.iceberg.io.DelegateFileIO;
import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.FileInfo;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.SupportsBulkOperations;
+import org.apache.iceberg.io.SupportsPrefixOperations;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -65,6 +69,35 @@ public class IcebergAuthenticatedFileIOTest {
Assertions.assertFalse(auth.inDoAs, "doAs must have exited after each
call");
}
+ @Test
+ public void withPrefixOperationsListsAndDeletesByPrefixInsideDoAs() {
+ RecordingAuthenticator auth = new RecordingAuthenticator();
+ PrefixRecordingFileIO delegate = new PrefixRecordingFileIO(auth);
+
+ IcebergAuthenticatedFileIO io =
IcebergAuthenticatedFileIO.withPrefixOperations(delegate, auth);
+
+ // remove_orphan_files rejects a FileIO without prefix listing.
+ Assertions.assertTrue(io instanceof SupportsPrefixOperations);
+ // HadoopFileIO's bulk delete runs on its own pool, outside any doAs,
so callers must delete one by one.
+ Assertions.assertFalse(io instanceof SupportsBulkOperations, "bulk
deletes must stay hidden");
+ ((SupportsPrefixOperations) io).listPrefix("hdfs://nn/t/");
+ ((SupportsPrefixOperations) io).deletePrefix("hdfs://nn/t/tmp/");
+ io.newInputFile("hdfs://nn/t/in");
+ Assertions.assertEquals(3, auth.doAsCount);
+ Assertions.assertEquals(Arrays.asList("listPrefix", "deletePrefix",
"newInputFile"), delegate.reachedInDoAs);
+ }
+
+ @Test
+ public void withPrefixOperationsAddsNoPrefixListingToAPlainDelegate() {
+ RecordingAuthenticator auth = new RecordingAuthenticator();
+
+ IcebergAuthenticatedFileIO io =
+ IcebergAuthenticatedFileIO.withPrefixOperations(new
RecordingFileIO(auth), auth);
+
+ Assertions.assertFalse(io instanceof SupportsPrefixOperations,
+ "the wrapper must not claim prefix listing its delegate cannot
do");
+ }
+
@Test
public void nonAuthMethodsForwardWithoutDoAs() {
RecordingAuthenticator auth = new RecordingAuthenticator();
@@ -105,7 +138,7 @@ public class IcebergAuthenticatedFileIOTest {
}
/** FileIO double: records the name of each factory call together with
whether it ran inside a doAs. */
- private static final class RecordingFileIO implements FileIO {
+ private static class RecordingFileIO implements FileIO {
private final RecordingAuthenticator auth;
final List<String> reachedInDoAs = new ArrayList<>();
boolean closed;
@@ -114,7 +147,7 @@ public class IcebergAuthenticatedFileIOTest {
this.auth = auth;
}
- private void record(String op) {
+ void record(String op) {
// Only record when observed inside the authenticator's doAs — the
property under test.
if (auth.inDoAs) {
reachedInDoAs.add(op);
@@ -150,4 +183,27 @@ public class IcebergAuthenticatedFileIOTest {
closed = true;
}
}
+
+ /** Prefix-capable double shaped like {@code HadoopFileIO}, which also
offers bulk deletes. */
+ private static final class PrefixRecordingFileIO extends RecordingFileIO
implements DelegateFileIO {
+ PrefixRecordingFileIO(RecordingAuthenticator auth) {
+ super(auth);
+ }
+
+ @Override
+ public Iterable<FileInfo> listPrefix(String prefix) {
+ record("listPrefix");
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void deletePrefix(String prefix) {
+ record("deletePrefix");
+ }
+
+ @Override
+ public void deleteFiles(Iterable<String> pathsToDelete) {
+ record("deleteFiles");
+ }
+ }
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergProcedureOpsKerberosIoTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergProcedureOpsKerberosIoTest.java
new file mode 100644
index 00000000000..d4d78c10e1a
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergProcedureOpsKerberosIoTest.java
@@ -0,0 +1,338 @@
+// 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;
+
+import org.apache.doris.connector.spi.procedure.ConnectorProcedureResult;
+import org.apache.doris.connector.spi.procedure.ConnectorRewriteGroup;
+import org.apache.doris.kerberos.HadoopAuthenticator;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.hadoop.HadoopTableOperations;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.FileInfo;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.SupportsPrefixOperations;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.LockManagers;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.FileTime;
+import java.security.PrivilegedExceptionAction;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Guards the Kerberos I/O seam of {@code ALTER TABLE ... EXECUTE}: every
metadata file a procedure opens, lists or
+ * deletes must be reached inside the plugin-side {@code doAs}, including the
files iceberg opens on its shared
+ * worker pool.
+ *
+ * <p>WHY it matters: {@link IcebergProcedureOps} runs the procedure inside
{@code executeAuthenticated}, whose
+ * {@code doAs} covers only the calling thread. {@code expire_snapshots}
commits the new metadata and then reads the
+ * expired snapshots' manifest lists on the worker pool; through the raw table
FileIO those reads reach secured HDFS
+ * without Kerberos credentials (CI: test_iceberg_hadoop_catalog_kerberos
failed "Failed to open input stream for
+ * file: .../snap-*.avro" on expire_snapshots after rollback_to_snapshot). The
authenticator here marks only the
+ * thread that entered {@code doAs}, as a Kerberos {@code Subject} is
invisible to pool threads, and the table's
+ * FileIO records every call made outside that mark.
+ *
+ * <p>No Mockito: real {@code HadoopTables} tables in a temp directory and
recording doubles, mirroring
+ * {@link IcebergScanPlanProviderKerberosScanIoTest}.
+ */
+public class IcebergProcedureOpsKerberosIoTest {
+
+ private static final Schema SCHEMA = new
Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()));
+ private static final IcebergTableHandle HANDLE = new
IcebergTableHandle("db1", "t1");
+
+ @Test
+ public void
expireSnapshotsReadsExpiredManifestListsInsidePluginDoAs(@TempDir Path temp)
throws Exception {
+ // The CI sequence: the rollback leaves the second snapshot outside
main's ancestry, so the expiry walks the
+ // reachable files and reads every expired manifest list on the worker
pool.
+ Table table = createTable(temp);
+ appendFile(table, temp.resolve("s1.parquet"));
+ long firstSnapshot = table.currentSnapshot().snapshotId();
+ appendFile(table, temp.resolve("s2.parquet"));
+ table.manageSnapshots().rollbackTo(firstSnapshot).commit();
+ appendFile(table, temp.resolve("s3.parquet"));
+ ThreadMarkingAuthenticator auth = new ThreadMarkingAuthenticator();
+ DoAsCheckingFileIO io = new DoAsCheckingFileIO(auth);
+
+ kerberosProcedureOps(withFileIO(table, io), auth).execute(null, HANDLE,
+ "expire_snapshots", ImmutableMap.of("retain_last", "1"), null,
Collections.emptyList());
+
+ table.refresh();
+ Assertions.assertEquals(1,
ImmutableList.copyOf(table.snapshots()).size());
+ // Only the second snapshot's data file is left unreferenced, and the
cleanup deleted it.
+ Assertions.assertFalse(Files.exists(temp.resolve("s2.parquet")));
+ Assertions.assertTrue(Files.exists(temp.resolve("s1.parquet")));
+ Assertions.assertTrue(Files.exists(temp.resolve("s3.parquet")));
+ Assertions.assertTrue(io.workerThreadCalls.get() > 0,
+ "the cleanup must have opened files on iceberg's worker pool,
or this test proves nothing");
+ // MUTATION: handing the action the loaded table instead of
wrapTableForProcedure(table) -> the worker-pool
+ // manifest-list reads run outside the doAs -> red (the CI failure).
+ Assertions.assertEquals(Collections.emptyList(), io.callsOutsideDoAs);
+ }
+
+ @Test
+ public void cherrypickSnapshotWritesItsManifestInsidePluginDoAs(@TempDir
Path temp) throws Exception {
+ // Main moves on after the append is staged, so the cherry-pick cannot
fast-forward: iceberg writes the
+ // picked files into a new manifest on its worker pool.
+ Table table = createTable(temp);
+ appendFile(table, temp.resolve("a.parquet"));
+ long mainSnapshot = table.currentSnapshot().snapshotId();
+
table.newAppend().appendFile(dataFile(createOldFile(temp.resolve("staged.parquet")))).stageOnly().commit();
+ table.refresh();
+ long stagedSnapshot =
ImmutableList.copyOf(table.snapshots()).stream().mapToLong(Snapshot::snapshotId)
+ .filter(id -> id != mainSnapshot).findFirst().getAsLong();
+ appendFile(table, temp.resolve("b.parquet"));
+ ThreadMarkingAuthenticator auth = new ThreadMarkingAuthenticator();
+ DoAsCheckingFileIO io = new DoAsCheckingFileIO(auth);
+
+ kerberosProcedureOps(withFileIO(table, io), auth).execute(null,
HANDLE, "cherrypick_snapshot",
+ ImmutableMap.of("snapshot_id",
String.valueOf(stagedSnapshot)), null, Collections.emptyList());
+
+ table.refresh();
+ Assertions.assertEquals("3",
table.currentSnapshot().summary().get("total-data-files"));
+ Assertions.assertTrue(io.workerThreadCalls.get() > 0,
+ "the cherry-pick must have written its manifest on iceberg's
worker pool, or this test proves nothing");
+ // MUTATION: handing the action the loaded table instead of
wrapTableForProcedure(table) -> the worker-pool
+ // manifest write runs outside the doAs -> red.
+ Assertions.assertEquals(Collections.emptyList(), io.callsOutsideDoAs);
+ }
+
+ @Test
+ public void removeOrphanFilesListsAndDeletesInsidePluginDoAs(@TempDir Path
temp) throws Exception {
+ Table table = createTable(temp);
+ appendFile(table, temp.resolve("table/data/live.parquet"));
+ Path orphan = createOldFile(temp.resolve("table/data/orphan.parquet"));
+ ThreadMarkingAuthenticator auth = new ThreadMarkingAuthenticator();
+ DoAsCheckingFileIO io = new DoAsCheckingFileIO(auth);
+
+ // MUTATION: wrapping with a FileIO that drops the delegate's prefix
listing -> "remove_orphan_files
+ // requires FileIO prefix listing support" -> red.
+ ConnectorProcedureResult result =
kerberosProcedureOps(withFileIO(table, io), auth).execute(null, HANDLE,
+ "remove_orphan_files", ImmutableMap.of(
+ "older_than",
String.valueOf(System.currentTimeMillis() - Duration.ofHours(24).toMillis()),
+ "dry_run", "false"),
+ null, Collections.emptyList());
+
+ Assertions.assertEquals(ImmutableList.of("1", "1"),
result.getRows().get(0));
+ Assertions.assertFalse(Files.exists(orphan));
+
Assertions.assertTrue(Files.exists(temp.resolve("table/data/live.parquet")));
+ Assertions.assertTrue(io.listPrefixCalls.get() > 0, "the orphan scan
must list through the table FileIO");
+ Assertions.assertEquals(Collections.emptyList(), io.callsOutsideDoAs);
+ }
+
+ @Test
+ public void rewritePlanningReadsManifestsInsidePluginDoAs(@TempDir Path
temp) throws Exception {
+ // Two fast appends leave two data manifests, so iceberg plans the
scan on its worker pool.
+ Table table = createTable(temp);
+ appendFile(table, temp.resolve("a.parquet"));
+ appendFile(table, temp.resolve("b.parquet"));
+ ThreadMarkingAuthenticator auth = new ThreadMarkingAuthenticator();
+ DoAsCheckingFileIO io = new DoAsCheckingFileIO(auth);
+
+ List<ConnectorRewriteGroup> groups =
kerberosProcedureOps(withFileIO(table, io), auth).planRewrite(null, HANDLE,
+ "rewrite_data_files", ImmutableMap.of("rewrite-all", "true"),
null, Collections.emptyList());
+
+ Assertions.assertEquals(1, groups.size());
+ Assertions.assertEquals(2, groups.get(0).getDataFileCount());
+ Assertions.assertTrue(io.workerThreadCalls.get() > 0,
+ "planning must have read manifests on iceberg's worker pool,
or this test proves nothing");
+ // MUTATION: planning the loaded table instead of
wrapTableForProcedure(table) -> the worker-pool manifest
+ // reads run outside the doAs -> red.
+ Assertions.assertEquals(Collections.emptyList(), io.callsOutsideDoAs);
+ }
+
+ @Test
+ public void nonKerberosProcedureKeepsTheLoadedTable(@TempDir Path temp) {
+ Table table = createTable(temp);
+ // A null plugin authenticator is a non-Kerberos catalog; a plain
context is an offline test or fe-core fake.
+ IcebergProcedureOps nonKerberos = new
IcebergProcedureOps(Collections.emptyMap(), opsReturning(table),
+ new TcclPinningConnectorContext(new
RecordingConnectorContext(), getClass().getClassLoader(),
+ () -> null));
+ IcebergProcedureOps plainContext = new
IcebergProcedureOps(Collections.emptyMap(), opsReturning(table),
+ new RecordingConnectorContext());
+
+ Assertions.assertSame(table, nonKerberos.wrapTableForProcedure(table));
+ Assertions.assertSame(table,
plainContext.wrapTableForProcedure(table));
+ }
+
+ private static Table createTable(Path temp) {
+ return new HadoopTables(new Configuration()).create(SCHEMA,
PartitionSpec.unpartitioned(),
+ Collections.emptyMap(),
temp.resolve("table").toUri().toString());
+ }
+
+ private static void appendFile(Table table, Path path) throws IOException {
+
table.newFastAppend().appendFile(dataFile(createOldFile(path))).commit();
+ }
+
+ private static DataFile dataFile(Path path) {
+ return DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath(path.toUri().toString())
+ .withFileSizeInBytes(1)
+ .withRecordCount(1)
+ .build();
+ }
+
+ private static Path createOldFile(Path path) throws IOException {
+ Files.createDirectories(path.getParent());
+ Files.write(path, new byte[] {1});
+ Files.setLastModifiedTime(path, FileTime.fromMillis(1));
+ return path;
+ }
+
+ /** The same table, reopened over {@code io} so that every metadata file
goes through it. */
+ private static Table withFileIO(Table table, DoAsCheckingFileIO io) {
+ HadoopTableOperations ops = new HadoopTableOperations(new
org.apache.hadoop.fs.Path(table.location()), io,
+ new Configuration(), LockManagers.defaultLockManager()) {
+ };
+ return new BaseTable(ops, table.name());
+ }
+
+ private static IcebergProcedureOps kerberosProcedureOps(Table table,
ThreadMarkingAuthenticator auth) {
+ TcclPinningConnectorContext context = new TcclPinningConnectorContext(
+ new RecordingConnectorContext(),
IcebergProcedureOpsKerberosIoTest.class.getClassLoader(), () -> auth);
+ return new IcebergProcedureOps(Collections.emptyMap(),
opsReturning(table), context);
+ }
+
+ private static RecordingIcebergCatalogOps opsReturning(Table table) {
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.table = table;
+ return ops;
+ }
+
+ /**
+ * Runs the action inline and marks only the current thread as inside the
doAs, as a Kerberos {@code Subject}
+ * entered on one thread is invisible to iceberg's pool threads.
+ */
+ private static final class ThreadMarkingAuthenticator implements
HadoopAuthenticator {
+ private final ThreadLocal<Boolean> inDoAs = ThreadLocal.withInitial(()
-> false);
+
+ boolean inDoAs() {
+ return inDoAs.get();
+ }
+
+ @Override
+ public UserGroupInformation getUGI() {
+ throw new UnsupportedOperationException("doAs is overridden; no
real UGI in this test");
+ }
+
+ @Override
+ public <T> T doAs(PrivilegedExceptionAction<T> action) throws
IOException {
+ boolean outer = inDoAs.get();
+ inDoAs.set(true);
+ try {
+ return action.run();
+ } catch (IOException | RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new IOException(e);
+ } finally {
+ inDoAs.set(outer);
+ }
+ }
+ }
+
+ /**
+ * {@code HadoopFileIO} that records each call made outside the
authenticator's doAs and counts the calls made
+ * off the test thread.
+ */
+ private static final class DoAsCheckingFileIO implements
SupportsPrefixOperations {
+ private final HadoopFileIO delegate = new HadoopFileIO(new
Configuration());
+ private final ThreadMarkingAuthenticator auth;
+ private final Thread testThread = Thread.currentThread();
+ final List<String> callsOutsideDoAs = Collections.synchronizedList(new
ArrayList<>());
+ final AtomicInteger workerThreadCalls = new AtomicInteger();
+ final AtomicInteger listPrefixCalls = new AtomicInteger();
+
+ DoAsCheckingFileIO(ThreadMarkingAuthenticator auth) {
+ this.auth = auth;
+ }
+
+ private void check(String op, String path) {
+ if (Thread.currentThread() != testThread) {
+ workerThreadCalls.incrementAndGet();
+ }
+ if (!auth.inDoAs()) {
+ callsOutsideDoAs.add(op + " " + path + " on " +
Thread.currentThread().getName());
+ }
+ }
+
+ @Override
+ public InputFile newInputFile(String path) {
+ check("newInputFile", path);
+ return delegate.newInputFile(path);
+ }
+
+ @Override
+ public InputFile newInputFile(String path, long length) {
+ check("newInputFile", path);
+ return delegate.newInputFile(path, length);
+ }
+
+ @Override
+ public OutputFile newOutputFile(String path) {
+ check("newOutputFile", path);
+ return delegate.newOutputFile(path);
+ }
+
+ @Override
+ public void deleteFile(String path) {
+ check("deleteFile", path);
+ delegate.deleteFile(path);
+ }
+
+ @Override
+ public Iterable<FileInfo> listPrefix(String prefix) {
+ check("listPrefix", prefix);
+ listPrefixCalls.incrementAndGet();
+ return delegate.listPrefix(prefix);
+ }
+
+ @Override
+ public void deletePrefix(String prefix) {
+ check("deletePrefix", prefix);
+ delegate.deletePrefix(prefix);
+ }
+
+ @Override
+ public Map<String, String> properties() {
+ return delegate.properties();
+ }
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergProcedureOpsTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergProcedureOpsTest.java
index fd7e4035738..4370ccbb367 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergProcedureOpsTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergProcedureOpsTest.java
@@ -45,6 +45,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
/**
@@ -298,6 +299,27 @@ public class IcebergProcedureOpsTest {
Assertions.assertEquals(1, ops.tableCleanupCount, "procedure execution
releases its operation-owned table");
}
+ @Test
+ public void commitsTheProcedureInsideTheAuthScope() {
+ // On a Kerberized catalog the metadata commit must run under the
catalog's identity, so the procedure has
+ // committed before the auth scope ends, not after the scope hands the
loaded table back.
+ // MUTATION: loading the table inside executeAuthenticated but running
the body outside it -> red.
+ InMemoryCatalog catalog = catalogWithTwoSnapshots();
+ TableIdentifier id = TableIdentifier.of("db1", "t");
+ long snap1 = catalog.loadTable(id).history().get(0).snapshotId();
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.table = catalog.loadTable(id);
+ RecordingConnectorContext ctx = new RecordingConnectorContext();
+ AtomicLong snapshotAtScopeExit = new AtomicLong();
+ ctx.beforeScopeExit = () ->
snapshotAtScopeExit.set(catalog.loadTable(id).currentSnapshot().snapshotId());
+ IcebergProcedureOps procOps = new
IcebergProcedureOps(Collections.emptyMap(), ops, ctx);
+
+ procOps.execute(SESSION, new IcebergTableHandle("db1", "t"),
"rollback_to_snapshot",
+ ImmutableMap.of("snapshot_id", String.valueOf(snap1)), null,
Collections.emptyList());
+
+ Assertions.assertEquals(snap1, snapshotAtScopeExit.get(), "the
rollback is committed inside the auth scope");
+ }
+
@Test
public void threadsSessionToTimestampBody() {
InMemoryCatalog catalog = catalogWithTwoSnapshots();
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergRestScanPlanningTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergRestScanPlanningTest.java
index bb48030e317..fe0de66d869 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergRestScanPlanningTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergRestScanPlanningTest.java
@@ -24,8 +24,10 @@ import org.apache.doris.connector.spi.ConnectorSession;
import org.apache.doris.connector.spi.ConnectorStatementScope;
import org.apache.doris.connector.spi.DorisConnectorException;
import org.apache.doris.connector.spi.scan.ConnectorScanRequest;
+import org.apache.doris.kerberos.HadoopAuthenticator;
import com.sun.net.httpserver.HttpServer;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.DataFiles;
import org.apache.iceberg.PartitionSpec;
@@ -52,6 +54,7 @@ import org.junit.jupiter.api.function.Executable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
+import java.security.PrivilegedExceptionAction;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
@@ -202,6 +205,26 @@ public class IcebergRestScanPlanningTest {
}
}
+ @Test
+ public void kerberosRewritePlanningAsksTheServerForTasks() throws
Exception {
+ try (RestFixture fixture = new RestFixture(true, true, true)) {
+ Table table = fixture.load();
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.table = table;
+ IcebergProcedureOps procedureOps = new
IcebergProcedureOps(Map.of(), ops, new TcclPinningConnectorContext(
+ new RecordingConnectorContext(),
getClass().getClassLoader(), InlineAuthenticator::new));
+
+ // The fixture answers every plan request with HTTP 500, so
planning fails; the test is where it went.
+ Assertions.assertThrows(DorisConnectorException.class, () ->
procedureOps.planRewrite(null,
+ new IcebergTableHandle("db", "t"), "rewrite_data_files",
Map.of("rewrite-all", "true"), null,
+ List.of()));
+ // MUTATION: planning wrapTableForProcedure(table) -> the plain
BaseTable plans from local manifests ->
+ // the recording FileIO fails the read and no plan request reaches
the server -> red.
+ Assertions.assertEquals(1, fixture.planRequests.get());
+ Assertions.assertEquals(0, fixture.io(table).reads.get());
+ }
+ }
+
private static void reject(Executable action) {
DorisConnectorException failure =
Assertions.assertThrows(DorisConnectorException.class, action);
Assertions.assertEquals("Iceberg server-side scan planning is not
supported for table db.t; "
@@ -280,6 +303,25 @@ public class IcebergRestScanPlanningTest {
}
}
+ /** A Kerberos-catalog stand-in: runs the action inline, without a real
UGI. */
+ private static final class InlineAuthenticator implements
HadoopAuthenticator {
+ @Override
+ public UserGroupInformation getUGI() {
+ throw new UnsupportedOperationException("doAs is overridden; no
real UGI in this test");
+ }
+
+ @Override
+ public <T> T doAs(PrivilegedExceptionAction<T> action) throws
IOException {
+ try {
+ return action.run();
+ } catch (IOException | RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+ }
+
public static class RecordingFileIO implements FileIO {
private final AtomicInteger reads = new AtomicInteger();
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/RecordingConnectorContext.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/RecordingConnectorContext.java
index 238637b7f35..a2d2240af72 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/RecordingConnectorContext.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/RecordingConnectorContext.java
@@ -56,6 +56,9 @@ final class RecordingConnectorContext implements
ConnectorContext, ConnectorStor
int authCount;
boolean failAuth;
+ /** Runs when a task has finished but {@link #executeAuthenticated} has
not left its scope yet, so a test can
+ * see what the task did inside the scope. */
+ Runnable beforeScopeExit;
/** Storage properties the fake returns from {@link
#getStorageProperties()} — the typed fe-filesystem
* seam both the scan and (design S3) the write path derive their
BE-canonical static creds from via
@@ -168,7 +171,11 @@ final class RecordingConnectorContext implements
ConnectorContext, ConnectorStor
// Deliberately do NOT call task -> the wrapped seam call must not
run.
throw new RuntimeException("auth failed");
}
- return task.call();
+ T result = task.call();
+ if (beforeScopeExit != null) {
+ beforeScopeExit.run();
+ }
+ return result;
}
/** Locations the connector asked the engine to clean (B1 managed-location
cleanup). */
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 9844499284d..77a0926a6ac 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
@@ -97,6 +97,75 @@ suite("test_iceberg_hadoop_catalog_kerberos", "p0,external")
{
'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
+ // COUNT(1) can be answered from the manifests alone; read the rows so
both retained data files are opened.
+ def rowsAfterExpire = sql """select col2, col7 from ${test_tbl_name} order
by col2"""
+ assert rowsAfterExpire == [[123, 'example text'], [456, 'another example']]
+
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]
