[core] Clear cache when deleting the snapshot

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

@@ -252,7 +252,7 @@ public int expireUntil(long earliestId, long endExclusiveId) {
             if (expireConfig.isChangelogDecoupled()) {
                 commitChangelog(new Changelog(snapshot));
             }
-            snapshotManager.fileIO().deleteQuietly(snapshotManager.snapshotPath(id));
+            snapshotManager.deleteSnapshot(id);
         }
 
         writeEarliestHint(endExclusiveId);

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

@@ -122,7 +122,7 @@ private List<Snapshot> cleanSnapshotsDataFiles(Snapshot retainedSnapshot) {
             // Ignore the non-existent snapshots
             if (snapshotManager.snapshotExists(i)) {
                 toBeCleaned.add(snapshotManager.snapshot(i));
-                fileIO.deleteQuietly(snapshotManager.snapshotPath(i));
+                snapshotManager.deleteSnapshot(i);
             }
         }
 

paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java

@@ -183,6 +183,14 @@ public boolean snapshotExists(long snapshotId) {
         }
     }
 
+    public void deleteSnapshot(long snapshotId) {
+        Path path = snapshotPath(snapshotId);
+        if (cache != null) {
+            cache.invalidate(path);
+        }
+        fileIO().deleteQuietly(path);
+    }
+
     public boolean longLivedChangelogExists(long snapshotId) {
         Path path = longLivedChangelogPath(snapshotId);
         try {

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/SnapshotHintOperator.java

@@ -84,9 +84,7 @@ private void commitSnapshotHintInTargetTable(SnapshotManager targetTableSnapshot
             targetTableSnapshotManager.commitLatestHint(snapshotId);
             for (Snapshot snapshot : targetTableSnapshotManager.safelyGetAllSnapshots()) {
                 if (snapshot.id() != snapshotId) {
-                    targetTableSnapshotManager
-                            .fileIO()
-                            .deleteQuietly(targetTableSnapshotManager.snapshotPath(snapshot.id()));
+                    targetTableSnapshotManager.deleteSnapshot(snapshot.id());
                 }
             }
         }

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RollbackProcedureTest.scala

@@ -151,4 +151,12 @@ class RollbackProcedureTest extends PaimonSparkTestBase with StreamTest {
     }
   }
 
+  test("Paimon Procedure: rollback with cache") {
+    sql("CREATE TABLE T (id INT)")
+    sql("INSERT INTO T VALUES (1), (2), (3), (4)")
+    sql("DELETE FROM T WHERE id = 1")
+    sql("CALL sys.rollback(table => 'T', version => '1')")
+    sql("DELETE FROM T WHERE id = 1")
+    checkAnswer(sql("SELECT * FROM T ORDER BY id"), Seq(Row(2), Row(3), Row(4)))
+  }
 }