[rest] Introduce snapshot loading to REST Catalog (#5147)

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -168,7 +168,12 @@ private List<Path> createExternalPaths() {
 
     @Override
     public SnapshotManager snapshotManager() {
-        return new SnapshotManager(fileIO, options.path(), options.branch(), snapshotCache);
+        return new SnapshotManager(
+                fileIO,
+                options.path(),
+                options.branch(),
+                catalogEnvironment.snapshotLoader(),
+                snapshotCache);
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java

@@ -193,7 +193,11 @@ public static Table loadTable(
 
         CatalogEnvironment catalogEnv =
                 new CatalogEnvironment(
-                        identifier, metadata.uuid(), catalog.catalogLoader(), commitFactory);
+                        identifier,
+                        metadata.uuid(),
+                        catalog.catalogLoader(),
+                        commitFactory,
+                        catalog instanceof SupportsSnapshots);
         Path path = new Path(schema.options().get(PATH.key()));
         FileStoreTable table =
                 FileStoreTableFactory.create(dataFileIO.apply(path), path, schema, catalogEnv);

paimon-core/src/main/java/org/apache/paimon/catalog/SupportsSnapshots.java

@@ -0,0 +1,36 @@
+/*
+ * 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.catalog;
+
+import org.apache.paimon.Snapshot;
+
+import java.util.Optional;
+
+/** A {@link Catalog} supports loading table snapshots. */
+public interface SupportsSnapshots {
+
+    /**
+     * Return the snapshot of table identified by the given {@link Identifier}.
+     *
+     * @param identifier Path of the table
+     * @return The requested snapshot of the table
+     * @throws Catalog.TableNotExistException if the target does not exist
+     */
+    Optional<Snapshot> loadSnapshot(Identifier identifier) throws Catalog.TableNotExistException;
+}

paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java

@@ -25,6 +25,7 @@
 import org.apache.paimon.catalog.Database;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.catalog.PropertyChange;
+import org.apache.paimon.catalog.SupportsSnapshots;
 import org.apache.paimon.catalog.TableMetadata;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
@@ -58,6 +59,7 @@
 import org.apache.paimon.rest.responses.ErrorResponseResourceType;
 import org.apache.paimon.rest.responses.GetDatabaseResponse;
 import org.apache.paimon.rest.responses.GetTableResponse;
+import org.apache.paimon.rest.responses.GetTableSnapshotResponse;
 import org.apache.paimon.rest.responses.GetTableTokenResponse;
 import org.apache.paimon.rest.responses.GetViewResponse;
 import org.apache.paimon.rest.responses.ListDatabasesResponse;
@@ -83,6 +85,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ScheduledExecutorService;
 
@@ -100,7 +103,7 @@
 import static org.apache.paimon.utils.ThreadPoolUtils.createScheduledThreadPool;
 
 /** A catalog implementation for REST. */
-public class RESTCatalog implements Catalog {
+public class RESTCatalog implements Catalog, SupportsSnapshots {
 
     public static final String HEADER_PREFIX = "header.";
 
@@ -297,11 +300,44 @@ private FileIO fileIOFromOptions(Path path) {
         }
     }
 
-    protected GetTableTokenResponse loadTableToken(Identifier identifier) {
-        return client.get(
-                resourcePaths.tableToken(identifier.getDatabaseName(), identifier.getObjectName()),
-                GetTableTokenResponse.class,
-                restAuthFunction);
+    protected GetTableTokenResponse loadTableToken(Identifier identifier)
+            throws TableNotExistException {
+        GetTableTokenResponse response;
+        try {
+            response =
+                    client.get(
+                            resourcePaths.tableToken(
+                                    identifier.getDatabaseName(), identifier.getObjectName()),
+                            GetTableTokenResponse.class,
+                            restAuthFunction);
+        } catch (NoSuchResourceException e) {
+            throw new TableNotExistException(identifier);
+        } catch (ForbiddenException e) {
+            throw new TableNoPermissionException(identifier, e);
+        }
+        return response;
+    }
+
+    @Override
+    public Optional<Snapshot> loadSnapshot(Identifier identifier) throws TableNotExistException {
+        GetTableSnapshotResponse response;
+        try {
+            response =
+                    client.get(
+                            resourcePaths.tableSnapshot(
+                                    identifier.getDatabaseName(), identifier.getObjectName()),
+                            GetTableSnapshotResponse.class,
+                            restAuthFunction);
+        } catch (NoSuchResourceException e) {
+            if (e.resourceType() == ErrorResponseResourceType.SNAPSHOT) {
+                return Optional.empty();
+            }
+            throw new TableNotExistException(identifier);
+        } catch (ForbiddenException e) {
+            throw new TableNoPermissionException(identifier, e);
+        }
+
+        return Optional.of(response.getSnapshot());
     }
 
     public boolean commitSnapshot(Identifier identifier, Snapshot snapshot) {

paimon-core/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.rest;
 
+import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.fs.FileIO;
@@ -182,7 +183,11 @@ private boolean shouldRefresh() {
     private void refreshToken() {
         GetTableTokenResponse response;
         if (catalogInstance != null) {
-            response = catalogInstance.loadTableToken(identifier);
+            try {
+                response = catalogInstance.loadTableToken(identifier);
+            } catch (Catalog.TableNotExistException e) {
+                throw new RuntimeException(e);
+            }
         } else {
             try (RESTCatalog catalog = catalogLoader.load()) {
                 response = catalog.loadTableToken(identifier);

paimon-core/src/main/java/org/apache/paimon/rest/ResourcePaths.java

@@ -74,6 +74,10 @@ public String tableToken(String databaseName, String tableName) {
         return SLASH.join(V1, prefix, DATABASES, databaseName, TABLES, tableName, "token");
     }
 
+    public String tableSnapshot(String databaseName, String tableName) {
+        return SLASH.join(V1, prefix, DATABASES, databaseName, TABLES, tableName, "snapshot");
+    }
+
     public String partitions(String databaseName, String tableName) {
         return SLASH.join(V1, prefix, DATABASES, databaseName, TABLES, tableName, "partitions");
     }

paimon-core/src/main/java/org/apache/paimon/rest/responses/ErrorResponseResourceType.java

@@ -23,5 +23,6 @@ public enum ErrorResponseResourceType {
     DATABASE,
     TABLE,
     COLUMN,
-    VIEW
+    VIEW,
+    SNAPSHOT
 }

paimon-core/src/main/java/org/apache/paimon/rest/responses/GetTableSnapshotResponse.java

@@ -0,0 +1,47 @@
+/*
+ * 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.responses;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.rest.RESTResponse;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** Response for table snapshot. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class GetTableSnapshotResponse implements RESTResponse {
+
+    private static final String FIELD_SNAPSHOT = "snapshot";
+
+    @JsonProperty(FIELD_SNAPSHOT)
+    private final Snapshot snapshot;
+
+    @JsonCreator
+    public GetTableSnapshotResponse(@JsonProperty(FIELD_SNAPSHOT) Snapshot snapshot) {
+        this.snapshot = snapshot;
+    }
+
+    @JsonGetter(FIELD_SNAPSHOT)
+    public Snapshot getSnapshot() {
+        return snapshot;
+    }
+}

paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java

@@ -270,7 +270,8 @@ public TableSchema commitChanges(SchemaChange... changes) throws Exception {
     public TableSchema commitChanges(List<SchemaChange> changes)
             throws Catalog.TableNotExistException, Catalog.ColumnAlreadyExistException,
                     Catalog.ColumnNotExistException {
-        SnapshotManager snapshotManager = new SnapshotManager(fileIO, tableRoot, branch);
+        SnapshotManager snapshotManager =
+                new SnapshotManager(fileIO, tableRoot, branch, null, null);
         boolean hasSnapshots = (snapshotManager.latestSnapshotId() != null);
 
         while (true) {

paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java

@@ -22,6 +22,8 @@
 import org.apache.paimon.catalog.CatalogLoader;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.catalog.SnapshotCommit;
+import org.apache.paimon.tag.SnapshotLoaderImpl;
+import org.apache.paimon.utils.SnapshotLoader;
 import org.apache.paimon.utils.SnapshotManager;
 
 import javax.annotation.Nullable;
@@ -37,20 +39,23 @@ public class CatalogEnvironment implements Serializable {
     @Nullable private final String uuid;
     @Nullable private final CatalogLoader catalogLoader;
     @Nullable private final SnapshotCommit.Factory commitFactory;
+    private final boolean supportsSnapshots;
 
     public CatalogEnvironment(
             @Nullable Identifier identifier,
             @Nullable String uuid,
             @Nullable CatalogLoader catalogLoader,
-            @Nullable SnapshotCommit.Factory commitFactory) {
+            @Nullable SnapshotCommit.Factory commitFactory,
+            boolean supportsSnapshots) {
         this.identifier = identifier;
         this.uuid = uuid;
         this.catalogLoader = catalogLoader;
         this.commitFactory = commitFactory;
+        this.supportsSnapshots = supportsSnapshots;
     }
 
     public static CatalogEnvironment empty() {
-        return new CatalogEnvironment(null, null, null, null);
+        return new CatalogEnvironment(null, null, null, null, false);
     }
 
     @Nullable
@@ -79,6 +84,14 @@ public PartitionHandler partitionHandler() {
         return PartitionHandler.create(catalogLoader.load(), identifier);
     }
 
+    @Nullable
+    public SnapshotLoader snapshotLoader() {
+        if (catalogLoader == null || !supportsSnapshots) {
+            return null;
+        }
+        return new SnapshotLoaderImpl(catalogLoader, identifier);
+    }
+
     @VisibleForTesting
     public SnapshotCommit.Factory commitFactory() {
         return commitFactory;