add cache

Author: JingsongLi

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

@@ -121,6 +121,12 @@ public void setManifestCache(SegmentsCache<Path> manifestCache) {
         store().setManifestCache(manifestCache);
     }
 
+    @Nullable
+    @Override
+    public SegmentsCache<Path> getManifestCache() {
+        return manifestCache;
+    }
+
     @Override
     public void setSnapshotCache(Cache<Path, Snapshot> cache) {
         this.snapshotCache = cache;

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

@@ -138,6 +138,12 @@ public void setManifestCache(SegmentsCache<Path> manifestCache) {
         wrapped.setManifestCache(manifestCache);
     }
 
+    @Nullable
+    @Override
+    public SegmentsCache<Path> getManifestCache() {
+        return wrapped.getManifestCache();
+    }
+
     @Override
     public void setSnapshotCache(Cache<Path, Snapshot> cache) {
         wrapped.setSnapshotCache(cache);

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

@@ -56,6 +56,9 @@ public interface FileStoreTable extends DataTable {
 
     void setManifestCache(SegmentsCache<Path> manifestCache);
 
+    @Nullable
+    SegmentsCache<Path> getManifestCache();
+
     void setSnapshotCache(Cache<Path, Snapshot> cache);
 
     void setStatsCache(Cache<String, Statistics> cache);

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java

@@ -36,6 +36,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 
 import static org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
 import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
 
 /**
@@ -53,6 +54,7 @@ public class TableWriteCoordinator {
 
     public TableWriteCoordinator(FileStoreTable table) {
         this.table = table;
+        checkNotNull(table.getManifestCache());
         this.latestCommittedIdentifiers = new ConcurrentHashMap<>();
         this.scan = table.store().newScan();
         if (table.coreOptions().manifestDeleteFileDropStats()) {

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinatorTest.java

@@ -26,12 +26,14 @@
 import org.apache.paimon.table.TableTestBase;
 import org.apache.paimon.types.DataTypes;
 
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
 import static org.apache.paimon.data.BinaryRow.EMPTY_ROW;
 import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 class TableWriteCoordinatorTest extends TableTestBase {
 
@@ -63,4 +65,15 @@ public void testLatestIdentifierAndScan(boolean initSnapshot) throws Exception {
         assertThat(scan.snapshot().id()).isEqualTo(latest.id());
         assertThat(scan.extractDataFiles().size()).isEqualTo(initSnapshot ? 2 : 1);
     }
+
+    @Test
+    public void testNoManifestCache() throws Exception {
+        Identifier identifier = new Identifier("db", "table");
+        catalog.createDatabase("db", false);
+        createTable(identifier);
+        FileStoreTable table = getTable(identifier);
+        table.setManifestCache(null);
+        assertThatThrownBy(() -> new TableWriteCoordinator(table))
+                .isInstanceOf(NullPointerException.class);
+    }
 }