update

Author: luoluoyuyu

docs/Go-SDK/go-sdk-advanced-state-api-zh.md

@@ -192,6 +192,7 @@ Keyed API 对应 store 的 **ComplexKey**，有三个维度：
 
 | 工厂                                | 方法                                                                                                  | 返回                                      |
 |-----------------------------------|-----------------------------------------------------------------------------------------------------|-----------------------------------------|
+| KeyedValueStateFactory[V]         | `NewKeyedValue(primaryKey []byte, namespace []byte) (*KeyedValueState[V], error)`                   | 每个 (primaryKey, namespace) 一个 value 状态。 |
 | KeyedListStateFactory[V]          | `NewKeyedList(primaryKey []byte, namespace []byte) (*KeyedListState[V], error)`                     | 每个 (primaryKey, namespace) 一个 list 状态。  |
 | KeyedMapStateFactory[MK,MV]       | `NewKeyedMap(primaryKey []byte, mapName string) (*KeyedMapState[MK,MV], error)`                     | 每个 (primaryKey, mapName) 一个 map 状态。     |
 | KeyedPriorityQueueStateFactory[V] | `NewKeyedPriorityQueue(primaryKey []byte, namespace []byte) (*KeyedPriorityQueueState[V], error)`   | 每个 (primaryKey, namespace) 一个 PQ 状态。    |

docs/Go-SDK/go-sdk-advanced-state-api.md

@@ -192,6 +192,7 @@ The Keyed API maps onto the store’s **ComplexKey** with three dimensions:
 
 | Factory                           | Method                                                                                              | Returns                                        |
 |-----------------------------------|-----------------------------------------------------------------------------------------------------|------------------------------------------------|
+| KeyedValueStateFactory[V]         | `NewKeyedValue(primaryKey []byte, namespace []byte) (*KeyedValueState[V], error)`                   | One value state per (primaryKey, namespace).   |
 | KeyedListStateFactory[V]          | `NewKeyedList(primaryKey []byte, namespace []byte) (*KeyedListState[V], error)`                     | List state per (primaryKey, namespace).        |
 | KeyedMapStateFactory[MK,MV]       | `NewKeyedMap(primaryKey []byte, mapName string) (*KeyedMapState[MK,MV], error)`                     | Map state per (primaryKey, mapName).           |
 | KeyedPriorityQueueStateFactory[V] | `NewKeyedPriorityQueue(primaryKey []byte, namespace []byte) (*KeyedPriorityQueueState[V], error)`   | PQ state per (primaryKey, namespace).          |

docs/Python-SDK/python-sdk-advanced-state-api-zh.md

@@ -79,7 +79,7 @@
 | 概念            | API 参数                                   | 含义                                                      |
 |---------------|------------------------------------------|---------------------------------------------------------|
 | **key_group** | 创建工厂时的 `key_group`                       | **keyed 组**：标识该状态所属分区/组（如一组 “counters”，另一组 “sessions”）。 |
-| **key**       | 构造状态时的 `primary_key`（如 `KeyedValueState(factory, primary_key, namespace)`） | 当前记录的**流 key 的值**（如用户 ID、分区 key）。不同 key 对应不同状态。         |
+| **key**       | 工厂方法参数（如 `new_keyed_value(primary_key, namespace)`） | 当前记录的**流 key 的值**（如用户 ID、分区 key）。不同 key 对应不同状态。         |
 | **namespace** | 创建工厂时的 `namespace`（bytes）                | **有窗口时**为**窗口标识的 bytes**；**无窗口时**传**空 bytes**（如 `b""`）。 |
 
 ### 4.2 Keyed 工厂构造方法一览
@@ -97,7 +97,7 @@
 
 ### 4.3 KeyedValueState
 
-KeyedValueState 与 Go SDK 一致：工厂仅需 `key_group`（无 namespace）。工厂：`KeyedValueStateFactory.from_context(ctx, store_name, key_group, value_codec)` 或 `from_context_auto_codec(ctx, store_name, key_group, value_type=None)`。构造状态：`KeyedValueState(factory, primary_key, namespace)`，其中 namespace 可为 `state_name.encode("utf-8")`。状态方法：`update(value)`、`value()`（返回 `(value, found)`）、`clear()`。
+KeyedValueState 与 Go SDK 一致：工厂仅需 `key_group`（无 namespace）。工厂：`KeyedValueStateFactory.from_context(ctx, store_name, key_group, value_codec)` 或 `from_context_auto_codec(ctx, store_name, key_group, value_type=None)`。创建状态：`factory.new_keyed_value(primary_key, namespace)`（namespace 为 bytes，必填）。状态方法：`update(value)`、`value()`（返回 `(value, found)`）、`clear()`。
 
 ### 4.4 KeyedListState
 
@@ -147,7 +147,7 @@ class CounterProcessor(FSProcessorDriver):
 
 ```python
 from fs_api import FSProcessorDriver, Context
-from fs_api_advanced import KeyedValueState, KeyedValueStateFactory
+from fs_api_advanced import KeyedValueStateFactory
 
 class KeyedCounterProcessor(FSProcessorDriver):
     def init(self, ctx: Context, config: dict):
@@ -157,7 +157,7 @@ class KeyedCounterProcessor(FSProcessorDriver):
 
     def process(self, ctx: Context, source_id: int, data: bytes):
         primary_key = data[:8]
-        state = KeyedValueState(self._factory, primary_key, "count".encode("utf-8"))
+        state = self._factory.new_keyed_value(primary_key, b"count")
         cur, found = state.value()
         if not found:
             cur = 0

docs/Python-SDK/python-sdk-advanced-state-api.md

@@ -79,7 +79,7 @@ All of the above can also be obtained via the corresponding `ctx.getOrCreate*` m
 | Term          | API parameter                                                         | Meaning                                                                                                                                    |
 |---------------|-----------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------|
 | **key_group** | `key_group` when creating the factory                                 | The **keyed group**: identifies which keyed partition/group this state belongs to (e.g. one group for "counters", another for "sessions"). |
-| **key**       | `primary_key` when constructing state (e.g. `KeyedValueState(factory, primary_key, namespace)`) | The **value of the stream key** for the current record (e.g. user ID, partition key). Each distinct key value gets isolated state.         |
+| **key**       | Argument to factory methods (e.g. `new_keyed_value(primary_key, namespace)`) | The **value of the stream key** for the current record (e.g. user ID, partition key). Each distinct key value gets isolated state.         |
 | **namespace** | `namespace` (bytes) when creating the factory                         | **If a window function is present**, use the **window identifier as bytes**. **Without windows**, pass **empty bytes** (e.g. `b""`).       |
 
 ### 4.2 Factory constructor summary (keyed)
@@ -97,7 +97,7 @@ You can also use the corresponding `ctx.getOrCreateKeyed*Factory(...)` methods,
 
 ### 4.3 KeyedValueState
 
-KeyedValueState aligns with the Go SDK: the factory takes only `key_group` (no namespace). Factory: `KeyedValueStateFactory.from_context(ctx, store_name, key_group, value_codec)` or `from_context_auto_codec(ctx, store_name, key_group, value_type=None)`. Construct state: `KeyedValueState(factory, primary_key, namespace)` with e.g. `namespace = state_name.encode("utf-8")`. State methods: `update(value)`, `value()` (returns `(value, found)`), `clear()`.
+KeyedValueState aligns with the Go SDK: the factory takes only `key_group` (no namespace). Factory: `KeyedValueStateFactory.from_context(ctx, store_name, key_group, value_codec)` or `from_context_auto_codec(ctx, store_name, key_group, value_type=None)`. Create state: `factory.new_keyed_value(primary_key, namespace)` (namespace is bytes, required). State methods: `update(value)`, `value()` (returns `(value, found)`), `clear()`.
 
 ### 4.4 KeyedListState
 
@@ -147,7 +147,7 @@ When the stream is partitioned by key, create the factory in `init` and obtain s
 
 ```python
 from fs_api import FSProcessorDriver, Context
-from fs_api_advanced import KeyedValueState, KeyedValueStateFactory
+from fs_api_advanced import KeyedValueStateFactory
 
 class KeyedCounterProcessor(FSProcessorDriver):
     def init(self, ctx: Context, config: dict):
@@ -157,7 +157,7 @@ class KeyedCounterProcessor(FSProcessorDriver):
 
     def process(self, ctx: Context, source_id: int, data: bytes):
         primary_key = data[:8]
-        state = KeyedValueState(self._factory, primary_key, "count".encode("utf-8"))
+        state = self._factory.new_keyed_value(primary_key, b"count")
         cur, found = state.value()
         if not found:
             cur = 0

go-sdk-advanced/keyed/keyed_list_state.go

@@ -154,7 +154,9 @@ func (s *KeyedListState[V]) Get() ([]V, error) {
 
 // Update replaces the list with the given values (one Put with batch payload).
 func (s *KeyedListState[V]) Update(values []V) error {
-	s.Clear()
+	if err := s.Clear(); err != nil {
+		return err
+	}
 	payload, err := s.serializeBatch(values)
 	if err != nil {
 		return err

go-sdk-advanced/keyed/keyed_value_state.go

@@ -73,6 +73,18 @@ func newKeyedValueStateFactory[V any](
 	}, nil
 }
 
+// NewKeyedValue creates a KeyedValueState for the given primary key and namespace.
+func (f *KeyedValueStateFactory[V]) NewKeyedValue(primaryKey []byte, namespace []byte) (*KeyedValueState[V], error) {
+	if primaryKey == nil || namespace == nil {
+		return nil, api.NewError(api.ErrStoreInternal, "primary key and namespace are required")
+	}
+	return &KeyedValueState[V]{
+		factory:    f,
+		primaryKey: common.DupBytes(primaryKey),
+		namespace:  common.DupBytes(namespace),
+	}, nil
+}
+
 type KeyedValueState[V any] struct {
 	factory    *KeyedValueStateFactory[V]
 	primaryKey []byte

python/functionstream-api-advanced/src/fs_api_advanced/keyed/keyed_value_state.py

@@ -19,7 +19,7 @@
 
 
 class KeyedValueStateFactory(Generic[V]):
-    """Factory for keyed value state. Create from context with key_group; construct KeyedValueState(factory, primary_key, namespace) per (primary_key, namespace)."""
+    """Factory for keyed value state. Create from context with key_group; obtain state via new_keyed_value(primary_key, namespace)."""
 
     def __init__(
         self,
@@ -64,6 +64,16 @@ def from_context_auto_codec(
         codec = default_codec_for(value_type)
         return cls(store, key_group, codec)
 
+    def new_keyed_value(
+        self, primary_key: bytes, namespace: bytes
+    ) -> "KeyedValueState[V]":
+        """Create a KeyedValueState for the given primary key and namespace."""
+        if primary_key is None:
+            raise KvError("keyed value state primary_key must not be None")
+        if namespace is None:
+            raise KvError("keyed value state namespace is required")
+        return KeyedValueState(self, primary_key, namespace)
+
 
 class KeyedValueState(Generic[V]):
     """Value state for one (primary_key, namespace). update(value), value() -> (value, found), clear()."""