This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 625d0d4e81 [core] Carry table UUID and writer identity in catalog
rollback (#10520)
625d0d4e81 is described below
commit 625d0d4e815d3897358a35728e8650d636a3e340
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 13 09:27:35 2026 +0800
[core] Carry table UUID and writer identity in catalog rollback (#10520)
---
docs/static/rest-catalog-open-api.yaml | 11 +++
.../main/java/org/apache/paimon/rest/RESTApi.java | 26 ++---
.../paimon/rest/requests/RollbackTableRequest.java | 31 +++++-
.../java/org/apache/paimon/AbstractFileStore.java | 2 +-
.../org/apache/paimon/catalog/AbstractCatalog.java | 7 +-
.../java/org/apache/paimon/catalog/Catalog.java | 29 +++---
.../org/apache/paimon/catalog/DelegateCatalog.java | 9 +-
.../java/org/apache/paimon/rest/RESTCatalog.java | 9 +-
.../apache/paimon/table/CatalogEnvironment.java | 17 +++-
.../org/apache/paimon/tag/SnapshotLoaderImpl.java | 15 ++-
.../org/apache/paimon/rest/MockRESTMessage.java | 4 +-
.../org/apache/paimon/rest/RESTCatalogTest.java | 5 +-
.../apache/paimon/rest/RollbackProtocolTest.java | 107 +++++++++++++++++++++
.../paimon/table/CatalogEnvironmentTest.java | 91 ++++++++++++++++++
.../table/FallbackReadFileStoreTableTest.java | 52 ++++++++++
.../apache/paimon/tag/SnapshotLoaderImplTest.java | 88 +++++++++++++++++
16 files changed, 450 insertions(+), 53 deletions(-)
diff --git a/docs/static/rest-catalog-open-api.yaml
b/docs/static/rest-catalog-open-api.yaml
index a52be685d0..559a433024 100644
--- a/docs/static/rest-catalog-open-api.yaml
+++ b/docs/static/rest-catalog-open-api.yaml
@@ -3514,12 +3514,23 @@ components:
$ref: '#/components/schemas/PartitionStatistics'
RollbackTableRequest:
type: object
+ description: >-
+ tableId identifies the expected table incarnation independently of
lock ownership.
+ For rollback within a writer's commit lock, also supply commitUser and
fromSnapshot so
+ the server can validate the expected head and owner. Omit commitUser
when no writer
+ lock is used, and omit tableId only when its UUID is unavailable.
properties:
instant:
$ref: '#/components/schemas/Instant'
fromSnapshot:
type: [ integer, "null" ]
format: int64
+ tableId:
+ type: [ string, "null" ]
+ description: Expected table or branch UUID, independently of
commitUser.
+ commitUser:
+ type: [ string, "null" ]
+ description: Exact owner of the writer's lock.
RollbackSchemaRequest:
type: object
required:
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
index 77a871d400..c454f26341 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
@@ -738,30 +738,24 @@ public class RESTApi {
* Rollback instant for table.
*
* @param identifier database name and table name.
- * @param instant instant to rollback
- * @throws NoSuchResourceException Exception thrown on HTTP 404 means the
table or the snapshot
- * or the tag not exists
- * @throws ForbiddenException Exception thrown on HTTP 403 means don't
have the permission for
- * this table
- */
- public void rollbackTo(Identifier identifier, Instant instant) {
- rollbackTo(identifier, instant, null);
- }
-
- /**
- * Rollback instant for table.
- *
- * @param identifier database name and table name.
+ * @param tableUuid expected table incarnation, or null when its UUID is
unavailable
* @param instant instant to rollback
* @param fromSnapshot snapshot from, success only occurs when the latest
snapshot is this
* snapshot.
+ * @param commitUser owner of the writer lock, or null without a writer
lock
* @throws NoSuchResourceException Exception thrown on HTTP 404 means the
table or the snapshot
* or the tag not exists
* @throws ForbiddenException Exception thrown on HTTP 403 means don't
have the permission for
* this table
*/
- public void rollbackTo(Identifier identifier, Instant instant, @Nullable
Long fromSnapshot) {
- RollbackTableRequest request = new RollbackTableRequest(instant,
fromSnapshot);
+ public void rollbackTo(
+ Identifier identifier,
+ @Nullable String tableUuid,
+ Instant instant,
+ @Nullable Long fromSnapshot,
+ @Nullable String commitUser) {
+ RollbackTableRequest request =
+ new RollbackTableRequest(instant, fromSnapshot, tableUuid,
commitUser);
client.post(
resourcePaths.rollbackTable(
identifier.getDatabaseName(),
identifier.getObjectName()),
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/requests/RollbackTableRequest.java
b/paimon-api/src/main/java/org/apache/paimon/rest/requests/RollbackTableRequest.java
index 9c8ca25c66..77fc97049c 100644
---
a/paimon-api/src/main/java/org/apache/paimon/rest/requests/RollbackTableRequest.java
+++
b/paimon-api/src/main/java/org/apache/paimon/rest/requests/RollbackTableRequest.java
@@ -29,6 +29,8 @@ import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonPro
import javax.annotation.Nullable;
+import java.beans.ConstructorProperties;
+
/** Request for rollback table. */
@JsonIgnoreProperties(ignoreUnknown = true)
public class RollbackTableRequest implements RESTRequest {
@@ -44,12 +46,39 @@ public class RollbackTableRequest implements RESTRequest {
@Nullable
private final Long fromSnapshot;
+ @JsonProperty("tableId")
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ @Nullable
+ private final String tableId;
+
+ @JsonProperty("commitUser")
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ @Nullable
+ private final String commitUser;
+
@JsonCreator
+ @ConstructorProperties({"instant", "fromSnapshot", "tableId",
"commitUser"})
public RollbackTableRequest(
@JsonProperty(FIELD_INSTANT) Instant instant,
- @JsonProperty(FIELD_FROM_SNAPSHOT) @Nullable Long fromSnapshot) {
+ @JsonProperty(FIELD_FROM_SNAPSHOT) @Nullable Long fromSnapshot,
+ @JsonProperty("tableId") @Nullable String tableId,
+ @JsonProperty("commitUser") @Nullable String commitUser) {
this.instant = instant;
this.fromSnapshot = fromSnapshot;
+ this.tableId = tableId;
+ this.commitUser = commitUser;
+ }
+
+ @JsonGetter("tableId")
+ @Nullable
+ public String getTableId() {
+ return tableId;
+ }
+
+ @JsonGetter("commitUser")
+ @Nullable
+ public String getCommitUser() {
+ return commitUser;
}
@JsonGetter(FIELD_INSTANT)
diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
index fe381cbaae..8dd60a7603 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -307,7 +307,7 @@ abstract class AbstractFileStore<T> implements FileStore<T>
{
snapshotManager,
scanner);
CommitRollback rollback = null;
- TableRollback tableRollback =
catalogEnvironment.catalogTableRollback();
+ TableRollback tableRollback =
catalogEnvironment.catalogTableRollback(null, null);
if (tableRollback != null) {
rollback = new CommitRollback(tableRollback);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
index 08f28ce9fc..d9503d43d9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -712,7 +712,12 @@ public abstract class AbstractCatalog implements Catalog {
}
@Override
- public void rollbackTo(Identifier identifier, Instant instant, @Nullable
Long fromSnapshot)
+ public void rollbackTo(
+ Identifier identifier,
+ @Nullable String tableUuid,
+ Instant instant,
+ @Nullable Long fromSnapshot,
+ @Nullable String commitUser)
throws Catalog.TableNotExistException {
throw new UnsupportedOperationException();
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
index 2f61c526ff..02e318a13e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -715,7 +715,7 @@ public interface Catalog extends AutoCloseable {
* <ul>
* <li>{@link #commitSnapshot(Identifier, String, String, Snapshot,
List)}.
* <li>{@link #loadSnapshot(Identifier)}.
- * <li>{@link #rollbackTo(Identifier, Instant)}.
+ * <li>{@link #rollbackTo(Identifier, String, Instant, Long, String)}.
* <li>{@link #createBranch(Identifier, String, String)}.
* <li>{@link #dropBranch(Identifier, String)}.
* <li>{@link #renameBranch(Identifier, String, String)}.
@@ -840,31 +840,24 @@ public interface Catalog extends AutoCloseable {
}
/**
- * rollback table by the given {@link Identifier} and instant.
- *
- * @param identifier path of the table
- * @param instant like snapshotId or tagName
- * @throws Catalog.TableNotExistException if the table does not exist
- * @throws UnsupportedOperationException if the catalog does not {@link
- * #supportsVersionManagement()}
- */
- default void rollbackTo(Identifier identifier, Instant instant)
- throws Catalog.TableNotExistException {
- rollbackTo(identifier, instant, null);
- }
-
- /**
- * rollback table by the given {@link Identifier} and instant.
+ * Rollback table by the given {@link Identifier} and instant.
*
* @param identifier path of the table
+ * @param tableUuid expected table incarnation, or null when its UUID is
unavailable
* @param instant like snapshotId or tagName
* @param fromSnapshot snapshot from, success only occurs when the latest
snapshot is this
- * snapshot.
+ * snapshot
+ * @param commitUser owner of the writer's lock, or null when no writer
lock is used
* @throws Catalog.TableNotExistException if the table does not exist
* @throws UnsupportedOperationException if the catalog does not {@link
* #supportsVersionManagement()}
*/
- void rollbackTo(Identifier identifier, Instant instant, @Nullable Long
fromSnapshot)
+ void rollbackTo(
+ Identifier identifier,
+ @Nullable String tableUuid,
+ Instant instant,
+ @Nullable Long fromSnapshot,
+ @Nullable String commitUser)
throws Catalog.TableNotExistException;
/**
diff --git
a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
index 41052b8baf..89159783fb 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
@@ -249,9 +249,14 @@ public abstract class DelegateCatalog implements Catalog {
}
@Override
- public void rollbackTo(Identifier identifier, Instant instant, @Nullable
Long fromSnapshot)
+ public void rollbackTo(
+ Identifier identifier,
+ @Nullable String tableUuid,
+ Instant instant,
+ @Nullable Long fromSnapshot,
+ @Nullable String commitUser)
throws Catalog.TableNotExistException {
- wrapped.rollbackTo(identifier, instant, fromSnapshot);
+ wrapped.rollbackTo(identifier, tableUuid, instant, fromSnapshot,
commitUser);
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
index 61ad72db0d..56939c3a61 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
@@ -540,10 +540,15 @@ public class RESTCatalog implements Catalog {
}
@Override
- public void rollbackTo(Identifier identifier, Instant instant, @Nullable
Long fromSnapshot)
+ public void rollbackTo(
+ Identifier identifier,
+ @Nullable String tableUuid,
+ Instant instant,
+ @Nullable Long fromSnapshot,
+ @Nullable String commitUser)
throws Catalog.TableNotExistException {
try {
- api.rollbackTo(identifier, instant, fromSnapshot);
+ api.rollbackTo(identifier, tableUuid, instant, fromSnapshot,
commitUser);
} catch (NoSuchResourceException e) {
if (StringUtils.equals(e.resourceType(),
ErrorResponse.RESOURCE_TYPE_SNAPSHOT)) {
throw new IllegalArgumentException(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
index d35c50706d..5988ee128f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
@@ -180,13 +180,22 @@ public class CatalogEnvironment implements Serializable {
lockContext, lockIdentifier, uuid, commitUser,
tableOptions);
}
+ /** Create rollback with the table incarnation and optional writer lock
context. */
@Nullable
- public TableRollback catalogTableRollback() {
+ public TableRollback catalogTableRollback(
+ @Nullable String branch, @Nullable String commitUser) {
if (catalogLoader != null && supportsVersionManagement) {
Catalog catalog = catalogLoader.load();
+ Identifier rollbackIdentifier =
+ branch == null
+ ? identifier
+ : new Identifier(
+ identifier.getDatabaseName(),
+ identifier.getTableName(),
+ branch);
return (instant, fromSnapshot) -> {
try {
- catalog.rollbackTo(identifier, instant, fromSnapshot);
+ catalog.rollbackTo(rollbackIdentifier, uuid, instant,
fromSnapshot, commitUser);
} catch (Catalog.TableNotExistException e) {
throw new RuntimeException(e);
}
@@ -215,7 +224,7 @@ public class CatalogEnvironment implements Serializable {
if (catalogLoader == null) {
return null;
}
- return new SnapshotLoaderImpl(catalogLoader, identifier);
+ return new SnapshotLoaderImpl(catalogLoader, identifier, uuid);
}
@Nullable
@@ -280,7 +289,7 @@ public class CatalogEnvironment implements Serializable {
public CatalogEnvironment copy(Identifier identifier) {
return new CatalogEnvironment(
identifier,
- uuid,
+ identifier.equals(this.identifier) ? uuid : null,
catalogLoader,
lockFactory,
lockContext,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/tag/SnapshotLoaderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/tag/SnapshotLoaderImpl.java
index 53e7f54697..8a47e88b1b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/tag/SnapshotLoaderImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/tag/SnapshotLoaderImpl.java
@@ -26,6 +26,8 @@ import org.apache.paimon.table.Instant;
import org.apache.paimon.table.TableSnapshot;
import org.apache.paimon.utils.SnapshotLoader;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.Optional;
@@ -34,10 +36,13 @@ public class SnapshotLoaderImpl implements SnapshotLoader {
private final CatalogLoader catalogLoader;
private final Identifier identifier;
+ @Nullable private final String uuid;
- public SnapshotLoaderImpl(CatalogLoader catalogLoader, Identifier
identifier) {
+ public SnapshotLoaderImpl(
+ CatalogLoader catalogLoader, Identifier identifier, @Nullable
String uuid) {
this.catalogLoader = catalogLoader;
this.identifier = identifier;
+ this.uuid = uuid;
}
@Override
@@ -54,7 +59,7 @@ public class SnapshotLoaderImpl implements SnapshotLoader {
@Override
public void rollback(Instant instant) throws IOException {
try (Catalog catalog = catalogLoader.load()) {
- catalog.rollbackTo(identifier, instant);
+ catalog.rollbackTo(identifier, uuid, instant, null, null);
} catch (RuntimeException e) {
throw e;
} catch (Exception e) {
@@ -64,8 +69,10 @@ public class SnapshotLoaderImpl implements SnapshotLoader {
@Override
public SnapshotLoader copyWithBranch(String branch) {
+ Identifier branchIdentifier =
+ new Identifier(identifier.getDatabaseName(),
identifier.getTableName(), branch);
+ // A different branch can have its own UUID. Its identity is unknown
in this view.
return new SnapshotLoaderImpl(
- catalogLoader,
- new Identifier(identifier.getDatabaseName(),
identifier.getTableName(), branch));
+ catalogLoader, branchIdentifier,
branchIdentifier.equals(identifier) ? uuid : null);
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java
b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java
index 4d4f101bd6..dafeae833a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java
@@ -274,11 +274,11 @@ public class MockRESTMessage {
}
public static RollbackTableRequest rollbackTableRequestBySnapshot(long
snapshotId) {
- return new RollbackTableRequest(Instant.snapshot(snapshotId), null);
+ return new RollbackTableRequest(Instant.snapshot(snapshotId), null,
null, null);
}
public static RollbackTableRequest rollbackTableRequestByTag(String
tagName) {
- return new RollbackTableRequest(Instant.tag(tagName), null);
+ return new RollbackTableRequest(Instant.tag(tagName), null, null,
null);
}
public static AlterViewRequest alterViewRequest() {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
index 05f62688eb..67106e1616 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
@@ -2377,9 +2377,10 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
assertThat(tagSnapshot).isEqualTo(restCatalog.loadSnapshot(identifier).get().snapshot());
// rollback to snapshot 2 from snapshot
- assertThatThrownBy(() -> catalog.rollbackTo(identifier,
Instant.snapshot(2L), 4L))
+ assertThatThrownBy(
+ () -> catalog.rollbackTo(identifier, null,
Instant.snapshot(2L), 4L, null))
.hasMessageContaining("Latest snapshot 3 is not 4");
- catalog.rollbackTo(identifier, Instant.snapshot(2L), 3L);
+ catalog.rollbackTo(identifier, null, Instant.snapshot(2L), 3L, null);
assertThat(table.latestSnapshot().get().id()).isEqualTo(2);
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RollbackProtocolTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RollbackProtocolTest.java
new file mode 100644
index 0000000000..b26aad6899
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RollbackProtocolTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.paimon.rest;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogLoader;
+import org.apache.paimon.catalog.DelegateCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.rest.requests.RollbackTableRequest;
+import org.apache.paimon.table.Instant;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.RecordedRequest;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests rollback routing and optional writer context through a delegated
REST catalog. */
+class RollbackProtocolTest {
+ @ParameterizedTest
+ @CsvSource({
+ "owned,snapshot", "unowned,snapshot", "without-uuid,snapshot",
+ "owned,tag", "unowned,tag", "without-uuid,tag"
+ })
+ void testRollbackCarriesTableUuidIndependentlyOfWriterContext(String
context, String kind)
+ throws Exception {
+ try (MockWebServer server = new MockWebServer()) {
+ server.start();
+ Options options = new Options();
+ options.set(RESTCatalogOptions.URI, server.url("/").toString());
+ options.set(RESTCatalogOptions.TOKEN_PROVIDER, "bear");
+ options.set(RESTCatalogOptions.TOKEN, "token");
+ options.set(RESTCatalogInternalOptions.PREFIX, "catalog");
+ try (Catalog catalog =
+ new DelegateCatalog(new
RESTCatalog(CatalogContext.create(options), false)) {
+ @Override
+ public CatalogLoader catalogLoader() {
+ return wrapped.catalogLoader();
+ }
+ }) {
+ Identifier identifier = new Identifier("db", "table", "dev");
+ Instant instant =
+ "snapshot".equals(kind) ? Instant.snapshot(1L) :
Instant.tag("tag-1");
+ server.enqueue(
+ new MockResponse()
+ .setResponseCode(200)
+ .setHeader("Content-Type", "application/json")
+ .setBody("{}"));
+ String tableUuid = "without-uuid".equals(context) ? null :
"table-id";
+ String commitUser = "owned".equals(context) ? "Writer-1" :
null;
+ catalog.rollbackTo(identifier, tableUuid, instant, 2L,
commitUser);
+ RecordedRequest recorded = server.takeRequest(10,
TimeUnit.SECONDS);
+ assertThat(recorded.getMethod()).isEqualTo("POST");
+ assertThat(recorded.getPath())
+
.isEqualTo("/v1/catalog/databases/db/tables/table%24branch_dev/rollback");
+ String body = recorded.getBody().readUtf8();
+ ObjectNode payload = RESTApi.fromJson(body, ObjectNode.class);
+
assertThat(payload.get("instant").get("type").asText()).isEqualTo(kind);
+ RollbackTableRequest request = RESTApi.fromJson(body,
RollbackTableRequest.class);
+ assertThat(request.getFromSnapshot()).isEqualTo(2L);
+ if ("snapshot".equals(kind)) {
+ assertThat(((Instant.SnapshotInstant)
request.getInstant()).getSnapshotId())
+ .isEqualTo(1L);
+ } else {
+ assertThat(((Instant.TagInstant)
request.getInstant()).getTagName())
+ .isEqualTo("tag-1");
+ }
+ if (tableUuid != null) {
+ assertThat(request.getTableId()).isEqualTo(tableUuid);
+ } else {
+ assertThat(payload.has("tableId")).isFalse();
+ }
+ if (commitUser != null) {
+ assertThat(request.getCommitUser()).isEqualTo(commitUser);
+ } else {
+ assertThat(payload.has("commitUser")).isFalse();
+ }
+ assertThat(payload.size())
+ .isEqualTo(2 + (tableUuid == null ? 0 : 1) +
(commitUser == null ? 0 : 1));
+ }
+ }
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java
index c8c5f4bb63..1b18b7039b 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java
@@ -42,6 +42,7 @@ import org.apache.paimon.utils.SnapshotManager;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
@@ -64,6 +65,96 @@ class CatalogEnvironmentTest {
private static final String READ_VIA_OPTION = RESTApi.HEADER_PREFIX +
RESTApi.READ_VIA_HEADER;
+ @ParameterizedTest
+ @ValueSource(strings = {"main", "dev"})
+ void testCatalogTableRollbackUsesRuntimeBranchAndOwner(String branch)
throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ CatalogEnvironment environment =
+ new CatalogEnvironment(
+ new Identifier("db", "table", "original"),
+ "table-id",
+ () -> catalog,
+ null,
+ null,
+ null,
+ true,
+ false);
+ Instant instant = Instant.snapshot(1L);
+ environment.catalogTableRollback(branch, "writer").rollbackTo(instant,
2L);
+ verify(catalog)
+ .rollbackTo(
+ new Identifier("db", "table", branch), "table-id",
instant, 2L, "writer");
+ }
+
+ @ParameterizedTest
+ @CsvSource({"main,table-id", "dev,table-id", "main,", "dev,"})
+ void testCatalogTableRollbackWithoutOwnerPreservesTableUuid(String branch,
String tableUuid)
+ throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ CatalogEnvironment environment =
+ new CatalogEnvironment(
+ new Identifier("db", "table", branch),
+ tableUuid,
+ () -> catalog,
+ null,
+ null,
+ null,
+ true,
+ false);
+ Instant instant = Instant.tag("tag-1");
+ environment.catalogTableRollback(null, null).rollbackTo(instant, null);
+ verify(catalog)
+ .rollbackTo(new Identifier("db", "table", branch), tableUuid,
instant, null, null);
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "main,main,table-id,table-id",
+ "dev,dev,branch-id,branch-id",
+ "main,dev,table-id,",
+ "dev,main,branch-id,",
+ "main,main,,",
+ "main,dev,,"
+ })
+ void testCopiedEnvironmentRollbackUsesTargetIdentity(
+ String sourceBranch, String targetBranch, String sourceUuid,
String expectedUuid)
+ throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Identifier sourceIdentifier =
+ "main".equals(sourceBranch)
+ ? Identifier.create("db", "table")
+ : new Identifier("db", "table", sourceBranch);
+ CatalogEnvironment environment =
+ new CatalogEnvironment(
+ sourceIdentifier, sourceUuid, () -> catalog, null,
null, null, true, false);
+ Identifier targetIdentifier = new Identifier("db", "table",
targetBranch);
+ CatalogEnvironment copied = environment.copy(targetIdentifier);
+
+ Instant snapshot = Instant.snapshot(1L);
+ copied.catalogTableRollback(null, null).rollbackTo(snapshot, 2L);
+ verify(catalog).rollbackTo(targetIdentifier, expectedUuid, snapshot,
2L, null);
+
+ Instant tag = Instant.tag("checkpoint");
+ copied.snapshotLoader().rollback(tag);
+ verify(catalog).rollbackTo(targetIdentifier, expectedUuid, tag, null,
null);
+ assertThat(environment.uuid()).isEqualTo(sourceUuid);
+ }
+
+ @Test
+ void testCatalogTableRollbackWithoutCatalogReturnsNull() {
+ assertThat(CatalogEnvironment.empty().catalogTableRollback(null,
null)).isNull();
+ }
+
+ @Test
+ void testCatalogTableRollbackWithoutVersionManagementReturnsNull() {
+ Catalog catalog = mock(Catalog.class);
+ CatalogEnvironment environment =
+ new CatalogEnvironment(null, null, () -> catalog, null, null,
null, false, false);
+ assertThat(environment.catalogTableRollback(null, null)).isNull();
+ assertThat(environment.catalogTableRollback("dev", "writer")).isNull();
+ verifyNoInteractions(catalog);
+ }
+
@Test
void testDependencyReadContextForRestCatalog() {
Identifier root = Identifier.create("db", "root$branch_dev");
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java
index 4e139d80c5..78ac14e268 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java
@@ -19,6 +19,7 @@
package org.apache.paimon.table;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.catalog.TableQueryAuthResult;
import org.apache.paimon.data.BinaryRow;
@@ -544,6 +545,57 @@ public class FallbackReadFileStoreTableTest {
assertThat(switchedId.getDatabaseName()).isEqualTo("mydb");
assertThat(switchedId.getBranchName()).isEqualTo(branchName);
assertThat(switchedId.getObjectName()).isEqualTo("mytable$branch_bc");
+ assertThat(switched.catalogEnvironment().uuid()).isNull();
+ assertThat(mainTable.catalogEnvironment().uuid()).isEqualTo("uuid-1");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"scan.fallback-branch", "scan.primary-branch"})
+ void testBranchViewsRollbackWithoutSourceUuid(String branchOption) throws
Exception {
+ FileStoreTable localTable = createTable();
+ localTable.createBranch("bc");
+ Catalog catalog = Mockito.mock(Catalog.class);
+ Identifier mainId = Identifier.create("mydb", "mytable");
+ CatalogEnvironment environment =
+ new CatalogEnvironment(
+ mainId, "main-uuid", () -> catalog, null, null, null,
true, false);
+ Options options = new Options();
+ options.set(branchOption, "bc");
+ FallbackReadFileStoreTable table =
+ (FallbackReadFileStoreTable)
+ FileStoreTableFactory.create(
+ fileIO, tablePath, localTable.schema(),
options, environment);
+ Identifier branchId = new Identifier("mydb", "mytable", "bc");
+
+ table.other().rollbackTo(1L);
+ Mockito.verify(catalog)
+ .rollbackTo(
+ Mockito.eq(branchId),
+ Mockito.isNull(),
+ Mockito.argThat(
+ instant ->
+ instant instanceof
Instant.SnapshotInstant
+ && ((Instant.SnapshotInstant)
instant)
+
.getSnapshotId()
+ == 1L),
+ Mockito.isNull(),
+ Mockito.isNull());
+
+ table.switchToBranch("bc").rollbackTo("checkpoint");
+ Mockito.verify(catalog)
+ .rollbackTo(
+ Mockito.eq(branchId),
+ Mockito.isNull(),
+ Mockito.argThat(
+ instant ->
+ instant instanceof Instant.TagInstant
+ && "checkpoint"
+ .equals(
+
((Instant.TagInstant) instant)
+
.getTagName())),
+ Mockito.isNull(),
+ Mockito.isNull());
+ assertThat(table.catalogEnvironment().uuid()).isEqualTo("main-uuid");
}
private void writeDataIntoTable(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/tag/SnapshotLoaderImplTest.java
b/paimon-core/src/test/java/org/apache/paimon/tag/SnapshotLoaderImplTest.java
new file mode 100644
index 0000000000..772564281b
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/tag/SnapshotLoaderImplTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.paimon.tag;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.CatalogEnvironment;
+import org.apache.paimon.table.Instant;
+import org.apache.paimon.utils.SnapshotLoader;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+/** Tests table incarnation propagation through catalog snapshot loaders. */
+class SnapshotLoaderImplTest {
+ @ParameterizedTest
+ @CsvSource({"main,main-uuid", "dev,branch-uuid", "main,", "dev,"})
+ void testRollbackPreservesTableUuid(String branch, String uuid) throws
Exception {
+ Catalog catalog = mock(Catalog.class);
+ Identifier identifier = new Identifier("db", "table", branch);
+ CatalogEnvironment environment = environment(catalog, identifier,
uuid);
+ Instant instant = Instant.tag("checkpoint");
+
+ environment.snapshotLoader().rollback(instant);
+
+ verify(catalog).rollbackTo(identifier, uuid, instant, null, null);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"main", "dev"})
+ void testCopyWithinBranchKeepsTableUuid(String branch) throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ // The main branch also has an implicit null spelling.
+ Identifier identifier =
+ "main".equals(branch)
+ ? Identifier.create("db", "table")
+ : new Identifier("db", "table", branch);
+ SnapshotLoader loader = environment(catalog, identifier,
"table-uuid").snapshotLoader();
+ Instant instant = Instant.snapshot(1L);
+
+ loader.copyWithBranch(branch).rollback(instant);
+
+ verify(catalog)
+ .rollbackTo(
+ new Identifier("db", "table", branch), "table-uuid",
instant, null, null);
+ }
+
+ @ParameterizedTest
+ @CsvSource({"main,dev", "dev,main"})
+ void testCopyToAnotherBranchDoesNotReuseSourceUuid(String source, String
target)
+ throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ SnapshotLoader loader =
+ environment(catalog, new Identifier("db", "table", source),
"source-uuid")
+ .snapshotLoader();
+ Instant instant = Instant.snapshot(1L);
+
+ loader.copyWithBranch(target).rollback(instant);
+
+ verify(catalog)
+ .rollbackTo(new Identifier("db", "table", target), null,
instant, null, null);
+ }
+
+ private CatalogEnvironment environment(Catalog catalog, Identifier
identifier, String uuid) {
+ return new CatalogEnvironment(
+ identifier, uuid, () -> catalog, null, null, null, true,
false);
+ }
+}
