fall back to load snapshot from file system when external paimon table

Author: xiaohongbo

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

@@ -19,12 +19,14 @@
 package org.apache.paimon.catalog;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
 import org.apache.paimon.TableType;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.manifest.PartitionEntry;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.partition.Partition;
+import org.apache.paimon.rest.exceptions.NotImplementedException;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.schema.TableSchema;
@@ -323,6 +325,18 @@ private static List<Pair<Table, TableSnapshot>> toTableAndSnapshots(
                         snapshot = optional.get();
                     }
                 } catch (Catalog.TableNotExistException ignored) {
+                } catch (NotImplementedException e) {
+                    // does not support load external paimon table snapshot from rest server
+                    // construct TableSnapshot from local snapshot and table scan
+                    if (table instanceof FileStoreTable) {
+                        FileStoreTable fileStoreTable = (FileStoreTable) table;
+                        Snapshot lastSnapshot = table.latestSnapshot().orElse(null);
+                        if (lastSnapshot != null) {
+                            snapshot = tableSnapshotIdFromFileSystem(fileStoreTable, lastSnapshot);
+                        }
+                    } else {
+                        throw new NotImplementedException(e.getMessage());
+                    }
                 }
             }
             tableAndSnapshots.add(Pair.of(table, snapshot));
@@ -427,4 +441,29 @@ private static IcebergTable toIcebergTable(
                 .comment(schema.comment())
                 .build();
     }
+
+    private static TableSnapshot tableSnapshotIdFromFileSystem(
+            FileStoreTable fileStoreTable, Snapshot snapshot) {
+        Long totalRecordCount = snapshot.totalRecordCount();
+        long recordCount = totalRecordCount != null ? totalRecordCount : 0L;
+        long fileSizeInBytes = 0L;
+        long fileCount = 0L;
+        long lastFileCreationTime = 0L;
+
+        List<PartitionEntry> partitionEntries =
+                fileStoreTable.newSnapshotReader().withSnapshot(snapshot).partitionEntries();
+        if (partitionEntries != null) {
+            for (PartitionEntry entry : partitionEntries) {
+                if (entry != null) {
+                    fileSizeInBytes += entry.fileSizeInBytes();
+                    fileCount += entry.fileCount();
+                    lastFileCreationTime =
+                            Math.max(lastFileCreationTime, entry.lastFileCreationTime());
+                }
+            }
+        }
+
+        return new TableSnapshot(
+                snapshot, recordCount, fileSizeInBytes, fileCount, lastFileCreationTime);
+    }
 }

paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java

@@ -683,6 +683,19 @@ private MockResponse getDataTokenHandle(Identifier tableIdentifier) throws Excep
     }
 
     private MockResponse snapshotHandle(Identifier identifier) throws Exception {
+        if (!tableMetadataStore.containsKey(identifier.getFullName())) {
+            throw new Catalog.TableNotExistException(identifier);
+        }
+        TableMetadata tableMetadata = tableMetadataStore.get(identifier.getFullName());
+        if (tableMetadata.isExternal()) {
+            ErrorResponse response =
+                    new ErrorResponse(
+                            ErrorResponse.RESOURCE_TYPE_TABLE,
+                            identifier.getFullName(),
+                            "external paimon table does not support get table snapshot in rest server",
+                            501);
+            return mockResponse(response, 404);
+        }
         RESTResponse response;
         Optional<TableSnapshot> snapshotOptional =
                 Optional.ofNullable(tableLatestSnapshotStore.get(identifier.getFullName()));
@@ -714,17 +727,7 @@ private MockResponse listSnapshots(Identifier identifier) throws Exception {
     }
 
     private MockResponse loadSnapshot(Identifier identifier, String version) throws Exception {
-        if (!tableMetadataStore.containsKey(identifier.getFullName())) {
-            throw new Catalog.TableNotExistException(identifier);
-        }
-        TableMetadata tableMetadata = tableMetadataStore.get(identifier.getFullName());
-        if (tableMetadata.isExternal()) {
-            new ErrorResponse(
-                    ErrorResponse.RESOURCE_TYPE_TABLE,
-                    identifier.getFullName(),
-                    "external paimon table does not support load snapshot in rest server",
-                    403);
-        }
+
         FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
         SnapshotManager snapshotManager = table.snapshotManager();
         Snapshot snapshot = null;
@@ -2095,7 +2098,7 @@ private MockResponse commitSnapshot(
                     ErrorResponse.RESOURCE_TYPE_TABLE,
                     identifier.getFullName(),
                     "external paimon table does not support commit in rest server",
-                    403);
+                    501);
         }
         FileStoreTable table = getFileTable(identifier);
         if (!tableId.equals(table.catalogEnvironment().uuid())) {