[state]: state api refactor. (#374)

* state api refactor.

* [state]: fix code style.

* [state]: change cstore test.

* fix ut.

* disable time-consuming test in native cstore.

* fix code style.

* fix comments.

* fix comments.

Author: tanghaodong25

geaflow-cstore/src/engine/store_engine.rs

@@ -737,6 +737,7 @@ mod tests {
     }
 
     #[test]
+    #[ignore]
     fn test_engine_compact_local() {
         log_util::try_init(LogType::ConsoleAndFile, LogLevel::Debug, 0);
 

geaflow-cstore/tests/test_engine_compact_without_fo.rs

@@ -34,6 +34,7 @@ const COMPACT_ITERATIONS: [u32; 4] = [10000, 50000, 100000, 200000];
 const COMPACT_TEST_ITERATIONS: u32 = 10;
 
 #[test]
+#[ignore]
 fn test_engine_compact_without_fo() {
     log_util::try_init(LogType::ConsoleAndFile, LogLevel::Debug, 0);
     let mut rng: IsaacRng = SeedableRng::from_entropy();

geaflow-cstore/tests/test_table_with_large_data.rs

@@ -204,6 +204,7 @@ fn test_table(test_mode: &TestMode) {
 }
 
 #[test]
+#[ignore]
 fn test_table_local_mode() {
     test_table(&TestMode::Local);
 }

geaflow-cstore/tests/test_two_store_sf10.rs

@@ -24,6 +24,7 @@ use rustc_hash::FxHashMap;
 use crate::log_util::info;
 
 #[test]
+#[ignore]
 fn test_two_store_with_ldbc_sf10_source() {
     log_util::try_init(LogType::ConsoleAndFile, LogLevel::Debug, 0);
     info!("start test_two_store_with_ldbc_sf10_source");

geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/com/antgroup/geaflow/cluster/system/MemoryClusterMetaKVStore.java

@@ -17,8 +17,8 @@
 import com.antgroup.geaflow.state.DataModel;
 import com.antgroup.geaflow.state.StoreType;
 import com.antgroup.geaflow.store.IStoreBuilder;
+import com.antgroup.geaflow.store.api.StoreBuilderFactory;
 import com.antgroup.geaflow.store.api.key.IKVStore;
-import com.antgroup.geaflow.store.api.key.StoreBuilderFactory;
 import com.antgroup.geaflow.store.context.StoreContext;
 
 public class MemoryClusterMetaKVStore<V> implements IClusterMetaKVStore<String, V> {

geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/com/antgroup/geaflow/cluster/system/RocksdbClusterMetaKVStore.java

@@ -17,8 +17,8 @@
 import com.antgroup.geaflow.state.DataModel;
 import com.antgroup.geaflow.state.StoreType;
 import com.antgroup.geaflow.store.IStoreBuilder;
-import com.antgroup.geaflow.store.api.key.IKVStore;
-import com.antgroup.geaflow.store.api.key.StoreBuilderFactory;
+import com.antgroup.geaflow.store.api.StoreBuilderFactory;
+import com.antgroup.geaflow.store.api.key.IKVStatefulStore;
 import com.antgroup.geaflow.store.context.StoreContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -29,15 +29,16 @@ public class RocksdbClusterMetaKVStore<K, V> implements IClusterMetaKVStore<K, V
 
     private static final Integer DEFAULT_VERSION = 1;
 
-    private IKVStore<K, Object> kvStore;
+    private IKVStatefulStore<K, Object> kvStore;
     private transient long version;
     private String name;
 
     @Override
     public void init(StoreContext storeContext) {
         IStoreBuilder builder = StoreBuilderFactory.build(StoreType.ROCKSDB.name());
         this.name = storeContext.getName();
-        kvStore = (IKVStore<K, Object>) builder.getStore(DataModel.KV, storeContext.getConfig());
+        kvStore = (IKVStatefulStore<K, Object>) builder.getStore(DataModel.KV,
+            storeContext.getConfig());
         kvStore.init(storeContext);
 
         // recovery

geaflow/geaflow-core/geaflow-engine/geaflow-highavailability/src/main/java/com/antgroup/geaflow/ha/service/RedisHAService.java

@@ -20,7 +20,7 @@
 import com.antgroup.geaflow.state.StoreType;
 import com.antgroup.geaflow.state.serializer.DefaultKVSerializer;
 import com.antgroup.geaflow.store.IStoreBuilder;
-import com.antgroup.geaflow.store.api.key.StoreBuilderFactory;
+import com.antgroup.geaflow.store.api.StoreBuilderFactory;
 import com.antgroup.geaflow.store.context.StoreContext;
 import com.antgroup.geaflow.store.redis.KVRedisStore;
 

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-api/src/main/java/com/antgroup/geaflow/dsl/connector/api/function/OffsetStore.java

@@ -28,8 +28,8 @@
 import com.antgroup.geaflow.state.serializer.DefaultKVSerializer;
 import com.antgroup.geaflow.state.serializer.IKVSerializer;
 import com.antgroup.geaflow.store.IStoreBuilder;
+import com.antgroup.geaflow.store.api.StoreBuilderFactory;
 import com.antgroup.geaflow.store.api.key.IKVStore;
-import com.antgroup.geaflow.store.api.key.StoreBuilderFactory;
 import com.antgroup.geaflow.store.context.StoreContext;
 import com.antgroup.geaflow.store.rocksdb.RocksdbStoreBuilder;
 import java.nio.charset.StandardCharsets;

geaflow/geaflow-metrics/geaflow-stats-metrics/src/main/java/com/antgroup/geaflow/stats/sink/AsyncKvStoreWriter.java

@@ -25,8 +25,8 @@
 import com.antgroup.geaflow.state.DataModel;
 import com.antgroup.geaflow.state.serializer.DefaultKVSerializer;
 import com.antgroup.geaflow.store.IStoreBuilder;
+import com.antgroup.geaflow.store.api.StoreBuilderFactory;
 import com.antgroup.geaflow.store.api.key.IKVStore;
-import com.antgroup.geaflow.store.api.key.StoreBuilderFactory;
 import com.antgroup.geaflow.store.context.StoreContext;
 import java.util.LinkedList;
 import java.util.Queue;

geaflow/geaflow-metrics/geaflow-stats-metrics/src/main/java/com/antgroup/geaflow/stats/sink/SyncKvStoreWriter.java

@@ -22,8 +22,8 @@
 import com.antgroup.geaflow.state.DataModel;
 import com.antgroup.geaflow.state.serializer.DefaultKVSerializer;
 import com.antgroup.geaflow.store.IStoreBuilder;
+import com.antgroup.geaflow.store.api.StoreBuilderFactory;
 import com.antgroup.geaflow.store.api.key.IKVStore;
-import com.antgroup.geaflow.store.api.key.StoreBuilderFactory;
 import com.antgroup.geaflow.store.context.StoreContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

geaflow/geaflow-plugins/geaflow-store/README.md

@@ -0,0 +1,130 @@
+# GeaFlow Store API
+
+GeaFlow Store API provides a unified storage interface for managing state data in graph computing.
+This document introduces how to use and implement these storage interfaces.
+
+## Core Interfaces
+
+GeaFlow Store API includes the following core interfaces:
+
+### IStoreBuilder
+
+Interface for building storage instances. Main methods:
+
+- `getStoreDesc()`: Get storage description
+- `build()`: Build storage instance
+
+### IBaseStore
+
+The base interface for all storage implementations, defining basic storage operations:
+
+- `init()`: Initialize storage
+- `flush()`: Flush data to storage
+- `close()`: Close storage connection
+
+### IStatefulStore
+
+State management storage interface, inheriting from IBaseStore, providing stateful management
+operations:
+
+- `archive()`: Archive current state
+- `recovery()`: Recover state from specified position
+- `recoveryLatest()`: Recover latest version state
+- `compact()`: Compress/merge historical states
+- `drop()`: Drop all data
+
+### IGraphStore
+
+Graph data storage interface, inheriting from IStatefulStore, used for storing graph vertices and
+edges:
+
+- `addVertex()`: Add vertex
+- `getVertex()`: Get vertex
+- `addEdge()`: Add edge
+- `getEdges()`: Get edges
+- `getOneDegreeGraph()`: Get one degree Graph.
+
+## Usage Examples
+
+### Basic Graph Storage Example
+
+```java
+// 1. Create StoreBuilder
+IStoreBuilder builder = new MemoryStoreBuilder();
+
+// 2. Build storage instance
+IGraphStore graphStore = (IGraphStore) builder.build();
+
+// 3. Initialize storage
+graphStore.init(context);
+
+// 4. Use storage
+// Add vertex
+graphStore.addVertex(vertex);
+
+// Get vertex
+IVertex vertex = graphStore.getVertex(vertexId);
+
+// Add edge
+graphStore.addEdge(edge);
+
+// Get edges
+List<IEdge> edges = graphStore.getEdges(vertexId);
+
+// 5. Close storage
+graphStore.close();
+```
+
+### State Management Example
+
+```java
+// 1. Create stateful storage instance
+IStatefulStore statefulStore = new MemoryStatefulStore();
+
+// 2. Initialize
+statefulStore.init(context);
+
+// 3. State management operations
+// Archive current state
+statefulStore.archive();
+
+// Recover from specific version
+statefulStore.recovery(version);
+
+// Compact historical states
+statefulStore.compact(version);
+
+// 4. Close storage
+statefulStore.close();
+```
+
+## Implementing Custom Storage
+
+To implement custom storage, you need to:
+
+1. Implement `IStoreBuilder` interface to create storage builder
+2. Implement corresponding storage interfaces based on requirements:
+    - Basic storage: implement `IBaseStore`
+    - State management storage: implement `IStatefulStore`
+    - Graph data storage: implement `IGraphStore`
+3. Configure SPI service in `resources/META-INF/services`
+
+For examples, refer to implementations in `geaflow-store-memory` module:
+
+- `MemoryStoreBuilder`
+- `StaticGraphMemoryStore`
+
+## Configuration
+
+Specify storage implementation and related parameters through configuration:
+
+```java
+Configuration conf = new Configuration();
+// Specify storage type
+conf.put(StoreConfig.STORE_TYPE, "memory");
+
+// State management configurations
+conf.put(StoreConfig.STATE_RETENTION_TIME, "24h");  // State retention time
+conf.put(StoreConfig.COMPACT_INTERVAL, "6h");       // State compaction interval
+
+// Other storage related configurations

geaflow/geaflow-plugins/geaflow-store/geaflow-store-api/src/main/java/com/antgroup/geaflow/store/AbstractBaseStore.java

@@ -27,37 +27,12 @@ public interface IBaseStore {
     void init(StoreContext storeContext);
 
     /**
-     * archive current store data for persistence.
-     */
-    void archive(long checkpointId);
-
-    /**
-     * recovery the store data from persistent storage.
-     */
-    void recovery(long checkpointId);
-
-    /**
-     * recovery the latest store data.
-     */
-    long recoveryLatest();
-
-    /**
-     * compact the store data.
-     */
-    void compact();
-
-    /**
-     * flush the store data to disk or remote storage.
+     * flush memory data to disk.
      */
     void flush();
 
     /**
-     * close the store handler and ll other used resources.
+     * close the store handler and all other used resources.
      */
     void close();
-
-    /**
-     * delete the disk data.
-     */
-    void drop();
 }

geaflow/geaflow-plugins/geaflow-store/geaflow-store-api/src/main/java/com/antgroup/geaflow/store/IBaseStore.java

@@ -0,0 +1,32 @@
+package com.antgroup.geaflow.store;
+
+/**
+ * IStateful store is stateful, which means it ensure data HA and can be recovered.
+ */
+public interface IStatefulStore extends IBaseStore {
+
+    /**
+     * make a snapshot and ensure data HA.
+     */
+    void archive(long checkpointId);
+
+    /**
+     * recover the store data.
+     */
+    void recovery(long checkpointId);
+
+    /**
+     * recover the latest store data.
+     */
+    long recoveryLatest();
+
+    /**
+     * trigger manual store data compaction.
+     */
+    void compact();
+
+    /**
+     * delete the store data.
+     */
+    void drop();
+}