[Fix][zeta][Connector-paimon] Fix paimon run from checkpoint bug

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/handler/PaimonSaveModeHandler.java

@@ -53,6 +53,9 @@ public void handleSchemaSaveMode() {
         TablePath tablePath = catalogTable.getTablePath();
         Table paimonTable = ((PaimonCatalog) catalog).getPaimonTable(tablePath);
         // load paimon table and set it into paimon sink
-        this.supportLoadTable.setLoadTable(paimonTable);
+        Table loadTable = this.supportLoadTable.getLoadTable();
+        if (loadTable == null || this.schemaSaveMode == SchemaSaveMode.RECREATE_SCHEMA) {
+            this.supportLoadTable.setLoadTable(paimonTable);
+        }
     }
 }

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSink.java

@@ -29,6 +29,7 @@
 import org.apache.seatunnel.api.sink.SupportSaveMode;
 import org.apache.seatunnel.api.sink.SupportSchemaEvolutionSink;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.TablePath;
 import org.apache.seatunnel.api.table.schema.SchemaChangeType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.connectors.seatunnel.paimon.catalog.PaimonCatalog;
@@ -80,6 +81,18 @@ public PaimonSink(ReadonlyConfig readonlyConfig, CatalogTable catalogTable) {
         this.paimonSinkConfig = new PaimonSinkConfig(readonlyConfig);
         this.catalogTable = catalogTable;
         this.paimonHadoopConfiguration = PaimonSecurityContext.loadHadoopConfig(paimonSinkConfig);
+        try (PaimonCatalog paimonCatalog = PaimonCatalog.loadPaimonCatalog(readonlyConfig)) {
+            paimonCatalog.open();
+            boolean databaseExists =
+                    paimonCatalog.databaseExists(this.paimonSinkConfig.getNamespace());
+            if (databaseExists) {
+                TablePath tablePath = catalogTable.getTablePath();
+                boolean tableExists = paimonCatalog.tableExists(tablePath);
+                if (tableExists) {
+                    this.paimonTable = paimonCatalog.getPaimonTable(tablePath);
+                }
+            }
+        }
     }
 
     @Override
@@ -152,6 +165,11 @@ public void setLoadTable(Table table) {
         this.paimonTable = table;
     }
 
+    @Override
+    public Table getLoadTable() {
+        return this.paimonTable;
+    }
+
     @Override
     public Optional<CatalogTable> getWriteCatalogTable() {
         return Optional.ofNullable(catalogTable);

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/SupportLoadTable.java

@@ -19,4 +19,6 @@
 
 public interface SupportLoadTable<T> {
     void setLoadTable(T table);
+
+    T getLoadTable();
 }