update

Author: luoluoyuyu

docs/Go-SDK/go-sdk-guide-zh.md

@@ -301,9 +301,18 @@ if err != nil {
 
 ---
 
-## 七、目录结构参考
+## 七、高级状态 API（进阶文档）
 
-**go-sdk**：
+本指南仅覆盖**低阶 go-sdk**（Driver、Context、Store、目录结构）。**高级状态 API**（Codec、ValueState、ListState、MapState、PriorityQueueState、AggregatingState、ReducingState、Keyed\* 工厂与用法）由独立库 **go-sdk-advanced** 提供，完整说明、Codec 约定、构造函数表与示例均在进阶文档中：
+
+- **[Go SDK — 高级状态 API](go-sdk-advanced-state-api-zh.md)**（中文）
+- [Go SDK — Advanced State API](go-sdk-advanced-state-api.md)（英文）
+
+---
+
+## 八、目录结构参考
+
+**低阶库 go-sdk**：
 
 ```text
 go-sdk/
@@ -325,4 +334,14 @@ go-sdk/
 └── bindings/         # wit-bindgen-go 生成的 Go 代码（make bindings）
 ```
 
+**高阶库 go-sdk-advanced**（依赖 go-sdk，含 Codec 与全部状态类型）：
+
+```text
+go-sdk-advanced/
+├── go.mod            # require go-sdk
+├── codec/            # Codec[T]、DefaultCodecFor、内置与 JSON codec
+├── structures/       # ValueState、ListState、MapState、PriorityQueue、Aggregating、Reducing
+└── keyed/            # Keyed 状态工厂（value、list、map、PQ、aggregating、reducing）
+```
+
 更多示例与 SQL 操作见 [examples/go-processor/README.md](../../examples/go-processor/README.md)、[SQL CLI 指南](../sql-cli-guide-zh.md)。

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

@@ -21,7 +21,7 @@
 
 # Python SDK — 高级状态 API
 
-本文档介绍 Python SDK 的**高级状态 API**：基于底层 KvStore 的带类型状态抽象（ValueState、ListState、MapState 等），通过 **codec** 序列化，并支持按主键的 **keyed state**。设计与 [Go SDK 高级状态 API](../Go-SDK/go-sdk-guide.md#7-advanced-state-api) 对齐。
+本文档介绍 Python SDK 的**高级状态 API**：基于底层 KvStore 的带类型状态抽象（ValueState、ListState、MapState 等），通过 **codec** 序列化，并支持按主键的 **keyed state**。
 
 **两个独立库：** 高级状态 API 由 **functionstream-api-advanced** 提供，依赖低阶 **functionstream-api**。安装：`pip install functionstream-api functionstream-api-advanced`。使用时从 `fs_api_advanced` 导入 Codec、ValueState、ListState、MapState 等。
 
@@ -34,7 +34,7 @@
 
 ## 1. 概述
 
-当需要结构化状态（单值、列表、Map、优先队列、聚合、归约）而不想手写字节编码或 key 布局时，可使用高级状态 API。创建方式有两种：通过**运行时的 Context**（如使用 functionstream-runtime 时 `ctx.getOrCreateValueState(...)`）或通过状态类型上的**类型级构造方法**（推荐，便于复用，与 Go SDK 用法一致）。
+当需要结构化状态（单值、列表、Map、优先队列、聚合、归约）而不想手写字节编码或 key 布局时，可使用高级状态 API。创建方式有两种：通过**运行时的 Context**（如使用 functionstream-runtime 时 `ctx.getOrCreateValueState(...)`）或通过状态类型上的**类型级构造方法**（推荐，便于复用）。
 
 ---
 
@@ -44,7 +44,7 @@
 
 使用 **functionstream-api-advanced** 时，运行时的 Context 实现（如 functionstream-runtime 的 WitContext）会提供 `getOrCreateValueState(store_name, codec)`、`getOrCreateValueStateAutoCodec(store_name)` 以及 ListState、MapState、PriorityQueueState、AggregatingState、ReducingState 与所有 Keyed\* 工厂的对应方法，内部委托给下面所述的类型级 `from_context` / `from_context_auto_codec`。
 
-### 2.2 通过状态类型（推荐，与 Go SDK 一致）
+### 2.2 通过状态类型（推荐）
 
 每种状态类型和 keyed 工厂提供：
 
@@ -95,9 +95,15 @@
 
 也可使用 Context 的 `ctx.getOrCreateKeyed*Factory(...)` 方法，其内部会委托给上述构造方法。
 
+### 4.3 KeyedValueState
+
+KeyedValueState 只需 **value codec**，不要求有序。工厂创建状态：`factory.new_keyed_value(primary_key, state_name="")`，得到 `KeyedValueState[V]`。状态方法：`update(value)`、`value()`（返回 `Optional[V]`）、`clear()`。主键由创建时传入的 `primary_key`（bytes）固定。
+
 ---
 
-## 5. 示例：使用 from_context_auto_codec 的 ValueState
+## 5. 示例
+
+### 5.1 ValueState（from_context_auto_codec）
 
 从 **fs_api_advanced** 导入 ValueState（Codec、ListState、MapState 等同此包）：
 
@@ -108,18 +114,41 @@ from fs_api_advanced import ValueState
 class CounterProcessor(FSProcessorDriver):
     def process(self, ctx: Context, source_id: int, data: bytes):
         state = ValueState.from_context_auto_codec(ctx, "my-store")
-        cur, found = state.value()
-        if not found or cur is None:
+        cur = state.value()
+        if cur is None:
+            cur = 0
+        state.update(cur + 1)
+        ctx.emit(str(cur + 1).encode(), 0)
+```
+
+### 5.2 KeyedValueState（keyed 算子）
+
+流按 key 分区时，在 `init` 中创建工厂，在 `process` 中按当前记录的 `primary_key` 取状态，再 `update(value)` / `value()` / `clear()`：
+
+```python
+from fs_api import FSProcessorDriver, Context
+from fs_api_advanced import KeyedValueStateFactory
+
+class KeyedCounterProcessor(FSProcessorDriver):
+    def init(self, ctx: Context, config: dict):
+        self._factory = KeyedValueStateFactory.from_context_auto_codec(
+            ctx, "counters", b"", b"by_key", value_type=int
+        )
+
+    def process(self, ctx: Context, source_id: int, data: bytes):
+        primary_key = data[:8]
+        state = self._factory.new_keyed_value(primary_key, "count")
+        cur = state.value()
+        if cur is None:
             cur = 0
         state.update(cur + 1)
         ctx.emit(str(cur + 1).encode(), 0)
 ```
 
-其他状态类型用法相同：按上表使用 `XxxState.from_context(ctx, store_name, ...)` 或 `XxxState.from_context_auto_codec(ctx, store_name)`。
+其他状态类型按上表使用 `XxxState.from_context(ctx, store_name, ...)` 或 `XxxState.from_context_auto_codec(ctx, store_name)`。
 
 ---
 
 ## 6. 参见
 
 - [Python SDK 指南](python-sdk-guide-zh.md) — fs_api、fs_client 及 Context/KvStore 基础用法。
-- [Go SDK 指南 — 高级状态 API](../Go-SDK/go-sdk-guide.md#7-advanced-state-api) — Go SDK 中的等价 API。

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

@@ -21,7 +21,7 @@
 
 # Python SDK — Advanced State API
 
-This document describes the **high-level state API** for the Python SDK: typed state abstractions (ValueState, ListState, MapState, etc.) built on top of the low-level KvStore, with serialization via **codecs** and optional **keyed state** per primary key. The design aligns with the [Go SDK Advanced State API](../Go-SDK/go-sdk-guide.md#7-advanced-state-api).
+This document describes the **high-level state API** for the Python SDK: typed state abstractions (ValueState, ListState, MapState, etc.) built on top of the low-level KvStore, with serialization via **codecs** and optional **keyed state** per primary key.
 
 **Two separate libraries:** The advanced state API is provided by **functionstream-api-advanced**, which depends on the low-level **functionstream-api**. Install with: `pip install functionstream-api functionstream-api-advanced`. Import Codec, ValueState, ListState, MapState, etc. from `fs_api_advanced`.
 
@@ -34,7 +34,7 @@ This document describes the **high-level state API** for the Python SDK: typed s
 
 ## 1. Overview
 
-Use the advanced state API when you need structured state (single value, list, map, priority queue, aggregation, reduction) without manual byte encoding or key layout. You can create state either from the **runtime Context** (e.g. `ctx.getOrCreateValueState(...)` when using functionstream-runtime) or via **type-level constructors** on the state class (recommended for clarity and reuse, same pattern as the Go SDK).
+Use the advanced state API when you need structured state (single value, list, map, priority queue, aggregation, reduction) without manual byte encoding or key layout. You can create state either from the **runtime Context** (e.g. `ctx.getOrCreateValueState(...)` when using functionstream-runtime) or via **type-level constructors** on the state class (recommended for clarity and reuse).
 
 ---
 
@@ -44,7 +44,7 @@ Use the advanced state API when you need structured state (single value, list, m
 
 When using **functionstream-api-advanced**, the runtime Context implementation (e.g. WitContext in functionstream-runtime) provides `getOrCreateValueState(store_name, codec)`, `getOrCreateValueStateAutoCodec(store_name)`, and the same pattern for ListState, MapState, PriorityQueueState, AggregatingState, ReducingState, and all Keyed\* factories; these delegate to the type-level `from_context` / `from_context_auto_codec` methods below.
 
-### 2.2 From the state type (recommended, same as Go SDK)
+### 2.2 From the state type (recommended)
 
 Each state type and keyed factory provides:
 
@@ -95,9 +95,15 @@ All of the above can also be obtained via the corresponding `ctx.getOrCreate*` m
 
 You can also use the corresponding `ctx.getOrCreateKeyed*Factory(...)` methods, which delegate to these constructors.
 
+### 4.3 KeyedValueState
+
+KeyedValueState requires only a **value codec**; no ordered codec. Create state from the factory with `factory.new_keyed_value(primary_key, state_name="")`, yielding `KeyedValueState[V]`. State methods: `update(value)`, `value()` (returns `Optional[V]`), `clear()`. The primary key is fixed at creation time via `primary_key` (bytes).
+
 ---
 
-## 5. Example: ValueState with from_context_auto_codec
+## 5. Examples
+
+### 5.1 ValueState (from_context_auto_codec)
 
 Import ValueState from **fs_api_advanced** (Codec, ListState, MapState, etc. are in the same package):
 
@@ -108,8 +114,32 @@ from fs_api_advanced import ValueState
 class CounterProcessor(FSProcessorDriver):
     def process(self, ctx: Context, source_id: int, data: bytes):
         state = ValueState.from_context_auto_codec(ctx, "my-store")
-        cur, found = state.value()
-        if not found or cur is None:
+        cur = state.value()
+        if cur is None:
+            cur = 0
+        state.update(cur + 1)
+        ctx.emit(str(cur + 1).encode(), 0)
+```
+
+### 5.2 KeyedValueState (keyed operator)
+
+When the stream is partitioned by key, create the factory in `init` and obtain state per record’s `primary_key` in `process`, then use `update(value)` / `value()` / `clear()`:
+
+```python
+from fs_api import FSProcessorDriver, Context
+from fs_api_advanced import KeyedValueStateFactory
+
+class KeyedCounterProcessor(FSProcessorDriver):
+    def init(self, ctx: Context, config: dict):
+        self._factory = KeyedValueStateFactory.from_context_auto_codec(
+            ctx, "counters", b"", b"by_key", value_type=int
+        )
+
+    def process(self, ctx: Context, source_id: int, data: bytes):
+        primary_key = data[:8]
+        state = self._factory.new_keyed_value(primary_key, "count")
+        cur = state.value()
+        if cur is None:
             cur = 0
         state.update(cur + 1)
         ctx.emit(str(cur + 1).encode(), 0)
@@ -122,4 +152,3 @@ Same pattern for other state types: use `XxxState.from_context(ctx, store_name,
 ## 6. See also
 
 - [Python SDK Guide](python-sdk-guide.md) — main guide for fs_api, fs_client, and basic Context/KvStore usage.
-- [Go SDK Guide — Advanced State API](../Go-SDK/go-sdk-guide.md#7-advanced-state-api) — equivalent API in the Go SDK.

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

@@ -15,9 +15,6 @@
 from fs_api.store import ComplexKey, KvError, KvStore
 from fs_api_advanced.codec import Codec, PickleCodec, default_codec_for
 
-from ._keyed_common import ensure_ordered_key_codec
-
-K = TypeVar("K")
 V = TypeVar("V")
 
 
@@ -64,65 +61,68 @@ def from_context_auto_codec(
         key_group: bytes,
         value_type: Optional[type] = None,
     ) -> "KeyedValueStateFactory[V]":
-        """Create a KeyedValueStateFactory with default codec from context and store name."""
+        """Create a KeyedValueStateFactory with default value codec from context and store name."""
         store = ctx.getOrCreateKVStore(store_name)
         codec = default_codec_for(value_type) if value_type is not None else PickleCodec()
         return cls(store, namespace, key_group, codec)
 
-    def new_value(self, key_codec: Codec[K]) -> "KeyedValueState[K, V]":
-        ensure_ordered_key_codec(key_codec, "keyed value")
+    def new_keyed_value(self, primary_key: bytes, state_name: str = "") -> "KeyedValueState[V]":
         return KeyedValueState(
             self._store,
             self._namespace,
-            key_codec,
-            self._value_codec,
             self._key_group,
+            self._value_codec,
+            primary_key,
+            state_name,
         )
 
 
-class KeyedValueState(Generic[K, V]):
+class KeyedValueState(Generic[V]):
     def __init__(
         self,
         store: KvStore,
         namespace: bytes,
-        key_codec: Codec[K],
-        value_codec: Codec[V],
         key_group: bytes,
+        value_codec: Codec[V],
+        primary_key: bytes,
+        state_name: str = "",
     ):
         if namespace is None:
             raise KvError("keyed value state namespace must not be None")
         if key_group is None:
             raise KvError("keyed value state key_group must not be None")
         if value_codec is None:
             raise KvError("keyed value state value_codec must not be None")
+        if primary_key is None:
+            raise KvError("keyed value state primary_key must not be None")
         self._store = store
         self._namespace = namespace
         self._key_group = key_group
-        self._key_codec = key_codec
         self._value_codec = value_codec
-        ensure_ordered_key_codec(key_codec, "keyed value")
+        self._primary_key = primary_key
+        self._state_name = state_name
 
-    def _build_ck(self, key: K) -> ComplexKey:
+    def _build_ck(self) -> ComplexKey:
         return ComplexKey(
             key_group=self._key_group,
-            key=self._key_codec.encode(key),
+            key=self._primary_key,
             namespace=self._namespace,
-            user_key=b"",
+            user_key=self._state_name.encode("utf-8") if self._state_name else b"",
         )
 
-    def set(self, key: K, value: V) -> None:
-        ck = self._build_ck(key)
+    def update(self, value: V) -> None:
+        ck = self._build_ck()
         self._store.put(ck, self._value_codec.encode(value))
 
-    def get(self, key: K) -> Optional[V]:
-        ck = self._build_ck(key)
+    def value(self) -> Optional[V]:
+        ck = self._build_ck()
         raw = self._store.get(ck)
         if raw is None:
             return None
         return self._value_codec.decode(raw)
 
-    def delete(self, key: K) -> None:
-        self._store.delete(self._build_ck(key))
+    def clear(self) -> None:
+        self._store.delete(self._build_ck())
 
 
 __all__ = ["KeyedValueState", "KeyedValueStateFactory"]

python/functionstream-api-advanced/src/fs_api_advanced/structures/value_state.py

@@ -53,7 +53,7 @@ def from_context_auto_codec(cls, ctx: Any, store_name: str) -> "ValueState[T]":
     def update(self, value: T) -> None:
         self._store.put(self._ck, self._codec.encode(value))
 
-    def value(self) -> Tuple[Optional[T], bool]:
+    def value(self) -> Optional[T]:
         raw = self._store.get(self._ck)
         if raw is None:
             return (None, False)