feat: implement advanced state API for Go and Python

README-zh.md

@@ -206,7 +206,8 @@ function-stream-<version>/
 | [Function 任务配置规范](docs/function-configuration-zh.md) | 任务定义规范        |
 | [SQL CLI 交互式管理指南](docs/sql-cli-guide-zh.md)          | 交互式管理指南       |
 | [Function 管理与开发指南](docs/function-development-zh.md)  | 管理与开发指南       |
-| [Python SDK 开发与交互指南](docs/python-sdk-guide-zh.md)    | Python SDK 指南 |
+| [Go SDK 开发与交互指南](docs/Go-SDK/go-sdk-guide-zh.md)     | Go SDK 指南        |
+| [Python SDK 开发与交互指南](docs/Python-SDK/python-sdk-guide-zh.md) | Python SDK 指南 |
 
 ## 配置
 

README.md

@@ -205,7 +205,8 @@ We provide a robust shell script to manage the server process, capable of handli
 | [Function Configuration](docs/function-configuration.md) | Task Definition Specification     |
 | [SQL CLI Guide](docs/sql-cli-guide.md)                   | Interactive Management Guide      |
 | [Function Development](docs/function-development.md)     | Management & Development Guide    |
-| [Python SDK Guide](docs/python-sdk-guide.md)             | Python SDK Guide                  |
+| [Go SDK Guide](docs/Go-SDK/go-sdk-guide.md)              | Go SDK Guide                      |
+| [Python SDK Guide](docs/Python-SDK/python-sdk-guide.md)   | Python SDK Guide                  |
 
 ## Configuration
 

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

@@ -263,7 +263,7 @@ create function with (
 );
 ```
 
-Configure `name`, `type: processor`, `input-groups`, and `outputs` (e.g. Kafka) in config.yaml. See [Function Configuration](function-configuration.md) and [examples/go-processor/README.md](../examples/go-processor/README.md).
+Configure `name`, `type: processor`, `input-groups`, and `outputs` (e.g. Kafka) in config.yaml. See [Function Configuration](../function-configuration.md) and [examples/go-processor/README.md](../../examples/go-processor/README.md).
 
 ---
 
@@ -301,7 +301,22 @@ if err != nil {
 
 ---
 
-## 7. Directory Layout
+## 7. Advanced State API
+
+The Go SDK provides a **typed, high-level state API** on top of the low-level `Store`: ValueState, ListState, MapState, PriorityQueueState, AggregatingState, ReducingState, and Keyed\* factories, with serialization via **codecs**. Use it when you need structured state (single value, list, map, priority queue, aggregation, reduction) without manual byte encoding or key layout.
+
+Full reference, codec contract, keyGroup/primaryKey/namespace semantics, constructor tables, and examples are in a dedicated document:
+
+- **[Go SDK — Advanced State API](go-sdk-advanced-state-api.md)**
+
+Quick reference:
+
+- **Non-keyed:** `structures.NewValueStateFromContext(ctx, storeName, codec)` or `structures.NewValueStateFromContextAutoCodec[T](ctx, storeName)`; same pattern for ListState, MapState, PriorityQueueState, AggregatingState, ReducingState.
+- **Keyed:** Create a factory with `keyed.NewKeyedXxxFromContext(ctx, storeName, keyGroup, ...)`, then obtain per-key state with `factory.NewKeyedValue(primaryKey, stateName)` etc. Use **keyGroup** for the keyed group, **primaryKey** for the stream key, **namespace** for window bytes (or empty).
+
+---
+
+## 8. Directory Layout
 
 ```text
 go-sdk/
@@ -317,8 +332,13 @@ go-sdk/
 │   ├── runtime.go
 │   ├── context.go
 │   └── store.go
+├── state/            # Advanced state API
+│   ├── structures/   # ValueState, ListState, MapState, PriorityQueue, Aggregating, Reducing
+│   ├── keyed/        # Keyed state factories (value, list, map, PQ, aggregating, reducing)
+│   ├── codec/        # Codec[T], DefaultCodecFor, built-in and JSON codecs
+│   └── common/       # Shared helpers (e.g. DupBytes)
 ├── wit/              # processor.wit and deps (make wit)
 └── bindings/         # Generated by wit-bindgen-go (make bindings)
 ```
 
-For more examples and SQL operations, see [examples/go-processor/README.md](../examples/go-processor/README.md) and the [SQL CLI Guide](sql-cli-guide.md).
+For more examples and SQL operations, see [examples/go-processor/README.md](../../examples/go-processor/README.md) and the [SQL CLI Guide](../sql-cli-guide.md).

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

@@ -0,0 +1,115 @@
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+        http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+
+# 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) 对齐。
+
+---
+
+## 1. 概述
+
+当需要结构化状态（单值、列表、Map、优先队列、聚合、归约）而不想手写字节编码或 key 布局时，可使用高级状态 API。创建方式有两种：通过 **Context**（如 `ctx.getOrCreateValueState(...)`）或通过状态类型上的**类型级构造方法**（推荐，便于复用，与 Go SDK 用法一致）。
+
+---
+
+## 2. 创建状态的两种方式
+
+### 2.1 通过 Context（getOrCreate\*）
+
+`Context` 提供 `getOrCreateValueState(store_name, codec)`、`getOrCreateValueStateAutoCodec(store_name)` 等方法，以及 ListState、MapState、PriorityQueueState、AggregatingState、ReducingState 与所有 Keyed\* 工厂的对应方法。运行时实现会委托给下面所述的类型级 `from_context` / `from_context_auto_codec`。
+
+### 2.2 通过状态类型（推荐，与 Go SDK 一致）
+
+每种状态类型和 keyed 工厂提供：
+
+- **带 codec：** `XxxState.from_context(ctx, store_name, codec, ...)`
+- **AutoCodec：** `XxxState.from_context_auto_codec(ctx, store_name)` 或带可选类型参数，由 SDK 使用默认 codec（如 PickleCodec，或 Map key / PQ 元素所需的有序 codec）。
+
+状态实例是轻量的；可在每次 `process` 中创建，或在 driver 中（如 `init`）缓存。同一 store 名称对应同一底层 store。
+
+---
+
+## 3. 非 Keyed 状态 — 构造方法一览
+
+| 状态类型 | 带 codec | AutoCodec |
+|----------|----------|-----------|
+| ValueState | `ValueState.from_context(ctx, store_name, codec)` | `ValueState.from_context_auto_codec(ctx, store_name)` |
+| ListState | `ListState.from_context(ctx, store_name, codec)` | `ListState.from_context_auto_codec(ctx, store_name)` |
+| MapState | `MapState.from_context(ctx, store_name, key_codec, value_codec)` 或 `MapState.from_context_auto_key_codec(ctx, store_name, value_codec)` | — |
+| PriorityQueueState | `PriorityQueueState.from_context(ctx, store_name, codec)` | `PriorityQueueState.from_context_auto_codec(ctx, store_name)` |
+| AggregatingState | `AggregatingState.from_context(ctx, store_name, acc_codec, agg_func)` | `AggregatingState.from_context_auto_codec(ctx, store_name, agg_func)` |
+| ReducingState | `ReducingState.from_context(ctx, store_name, value_codec, reduce_func)` | `ReducingState.from_context_auto_codec(ctx, store_name, reduce_func)` |
+
+以上均可通过 Context 的 `ctx.getOrCreate*` 方法获得（如 `ctx.getOrCreateValueState(store_name, codec)`），其内部会委托给上述构造方法。
+
+---
+
+## 4. Keyed 状态 — 工厂与 key_group / key / namespace
+
+**Keyed 状态面向 keyed 算子。** 流按 key 分区（如 keyBy）时，每个 key 拥有独立状态。可先获取一次**工厂**（通过 context、store 名称、**namespace** 和 **key_group**），再按**主键**（当前记录的流 key）创建状态。
+
+### 4.1 key_group、key（主键）与 namespace
+
+| 概念 | API 参数 | 含义 |
+|------|----------|------|
+| **key_group** | 创建工厂时的 `key_group` | **keyed 组**：标识该状态所属分区/组（如一组 “counters”，另一组 “sessions”）。 |
+| **key** | 工厂方法参数（如 `new_keyed_value(primary_key)`） | 当前记录的**流 key 的值**（如用户 ID、分区 key）。不同 key 对应不同状态。 |
+| **namespace** | 创建工厂时的 `namespace`（bytes） | **有窗口时**为**窗口标识的 bytes**；**无窗口时**传**空 bytes**（如 `b""`）。 |
+
+### 4.2 Keyed 工厂构造方法一览
+
+| 工厂 | 带 codec | AutoCodec |
+|------|----------|-----------|
+| KeyedValueStateFactory | `KeyedValueStateFactory.from_context(ctx, store_name, namespace, key_group, value_codec)` | `KeyedValueStateFactory.from_context_auto_codec(ctx, store_name, namespace, key_group, value_type=None)` |
+| KeyedListStateFactory | `KeyedListStateFactory.from_context(ctx, store_name, namespace, key_group, value_codec)` | `KeyedListStateFactory.from_context_auto_codec(ctx, store_name, namespace, key_group, value_type=None)` |
+| KeyedMapStateFactory | `KeyedMapStateFactory.from_context(ctx, store_name, namespace, key_group, key_codec, value_codec)` | `KeyedMapStateFactory.from_context_auto_codec(ctx, store_name, namespace, key_group, value_codec)` |
+| KeyedPriorityQueueStateFactory | `KeyedPriorityQueueStateFactory.from_context(ctx, store_name, namespace, key_group, item_codec)` | `KeyedPriorityQueueStateFactory.from_context_auto_codec(ctx, store_name, namespace, key_group, item_type=None)` |
+| KeyedAggregatingStateFactory | `KeyedAggregatingStateFactory.from_context(ctx, store_name, namespace, key_group, acc_codec, agg_func)` | `KeyedAggregatingStateFactory.from_context_auto_codec(ctx, store_name, namespace, key_group, agg_func, acc_type=None)` |
+| KeyedReducingStateFactory | `KeyedReducingStateFactory.from_context(ctx, store_name, namespace, key_group, value_codec, reduce_func)` | `KeyedReducingStateFactory.from_context_auto_codec(ctx, store_name, namespace, key_group, reduce_func, value_type=None)` |
+
+也可使用 Context 的 `ctx.getOrCreateKeyed*Factory(...)` 方法，其内部会委托给上述构造方法。
+
+---
+
+## 5. 示例：使用 from_context_auto_codec 的 ValueState
+
+```python
+from fs_api import FSProcessorDriver, Context
+from fs_api.store import ValueState
+
+class CounterProcessor(FSProcessorDriver):
+    def process(self, ctx: Context, source_id: int, data: bytes):
+        # 每条消息创建一次状态（或在 init 中缓存）
+        state = ValueState.from_context_auto_codec(ctx, "my-store")
+        cur, _ = state.value() or (0, False)
+        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)`。
+
+---
+
+## 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

@@ -0,0 +1,115 @@
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+        http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+
+# 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).
+
+---
+
+## 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 **Context** (e.g. `ctx.getOrCreateValueState(...)`) or via **type-level constructors** on the state class (recommended for clarity and reuse, same pattern as the Go SDK).
+
+---
+
+## 2. Creating State: Two Ways
+
+### 2.1 From Context (getOrCreate\*)
+
+`Context` defines methods such as `getOrCreateValueState(store_name, codec)`, `getOrCreateValueStateAutoCodec(store_name)`, and the same pattern for ListState, MapState, PriorityQueueState, AggregatingState, ReducingState, and all Keyed\* factories. The runtime implementation delegates to the type-level `from_context` / `from_context_auto_codec` methods below.
+
+### 2.2 From the state type (recommended, same as Go SDK)
+
+Each state type and keyed factory provides:
+
+- **With codec:** `XxxState.from_context(ctx, store_name, codec, ...)` — you pass the codec(s).
+- **AutoCodec:** `XxxState.from_context_auto_codec(ctx, store_name)` or with optional type hint — the SDK uses a default codec (e.g. `PickleCodec`, or ordered codecs for map key / PQ element where required).
+
+State instances are lightweight; you may create them per message in `process` or cache in the driver (e.g. in `init`). Same store name yields the same underlying store.
+
+---
+
+## 3. Non-Keyed State — Constructor Summary
+
+| State | With codec | AutoCodec |
+|-------|------------|-----------|
+| ValueState | `ValueState.from_context(ctx, store_name, codec)` | `ValueState.from_context_auto_codec(ctx, store_name)` |
+| ListState | `ListState.from_context(ctx, store_name, codec)` | `ListState.from_context_auto_codec(ctx, store_name)` |
+| MapState | `MapState.from_context(ctx, store_name, key_codec, value_codec)` or `MapState.from_context_auto_key_codec(ctx, store_name, value_codec)` | — |
+| PriorityQueueState | `PriorityQueueState.from_context(ctx, store_name, codec)` | `PriorityQueueState.from_context_auto_codec(ctx, store_name)` |
+| AggregatingState | `AggregatingState.from_context(ctx, store_name, acc_codec, agg_func)` | `AggregatingState.from_context_auto_codec(ctx, store_name, agg_func)` |
+| ReducingState | `ReducingState.from_context(ctx, store_name, value_codec, reduce_func)` | `ReducingState.from_context_auto_codec(ctx, store_name, reduce_func)` |
+
+All of the above can also be obtained via the corresponding `ctx.getOrCreate*` methods (e.g. `ctx.getOrCreateValueState(store_name, codec)`), which delegate to these constructors.
+
+---
+
+## 4. Keyed State — Factories and keyGroup / key / namespace
+
+**Keyed state is for keyed operators.** When the stream is partitioned by a key (e.g. after keyBy), each key gets isolated state. You obtain a **factory** once (from context, store name, **namespace**, and **key_group**), then create state **per primary key** (the stream key for the current record).
+
+### 4.1 keyGroup, key (primaryKey), and namespace
+
+| 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** | The argument to factory methods (e.g. `new_keyed_value(primary_key)`) | 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)
+
+| Factory | With codec | AutoCodec |
+|---------|------------|-----------|
+| KeyedValueStateFactory | `KeyedValueStateFactory.from_context(ctx, store_name, namespace, key_group, value_codec)` | `KeyedValueStateFactory.from_context_auto_codec(ctx, store_name, namespace, key_group, value_type=None)` |
+| KeyedListStateFactory | `KeyedListStateFactory.from_context(ctx, store_name, namespace, key_group, value_codec)` | `KeyedListStateFactory.from_context_auto_codec(ctx, store_name, namespace, key_group, value_type=None)` |
+| KeyedMapStateFactory | `KeyedMapStateFactory.from_context(ctx, store_name, namespace, key_group, key_codec, value_codec)` | `KeyedMapStateFactory.from_context_auto_codec(ctx, store_name, namespace, key_group, value_codec)` |
+| KeyedPriorityQueueStateFactory | `KeyedPriorityQueueStateFactory.from_context(ctx, store_name, namespace, key_group, item_codec)` | `KeyedPriorityQueueStateFactory.from_context_auto_codec(ctx, store_name, namespace, key_group, item_type=None)` |
+| KeyedAggregatingStateFactory | `KeyedAggregatingStateFactory.from_context(ctx, store_name, namespace, key_group, acc_codec, agg_func)` | `KeyedAggregatingStateFactory.from_context_auto_codec(ctx, store_name, namespace, key_group, agg_func, acc_type=None)` |
+| KeyedReducingStateFactory | `KeyedReducingStateFactory.from_context(ctx, store_name, namespace, key_group, value_codec, reduce_func)` | `KeyedReducingStateFactory.from_context_auto_codec(ctx, store_name, namespace, key_group, reduce_func, value_type=None)` |
+
+You can also use the corresponding `ctx.getOrCreateKeyed*Factory(...)` methods, which delegate to these constructors.
+
+---
+
+## 5. Example: ValueState with from_context_auto_codec
+
+```python
+from fs_api import FSProcessorDriver, Context
+from fs_api.store import ValueState
+
+class CounterProcessor(FSProcessorDriver):
+    def process(self, ctx: Context, source_id: int, data: bytes):
+        # Create state per message (or cache in init)
+        state = ValueState.from_context_auto_codec(ctx, "my-store")
+        cur, _ = state.value() or (0, False)
+        state.update(cur + 1)
+        ctx.emit(str(cur + 1).encode(), 0)
+```
+
+Same pattern for other state types: use `XxxState.from_context(ctx, store_name, ...)` or `XxxState.from_context_auto_codec(ctx, store_name)` as in the tables above.
+
+---
+
+## 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.

docs/python-sdk-guide-zh.md → docs/Python-SDK/python-sdk-guide-zh.md

@@ -148,3 +148,11 @@ with FsClient(host="10.0.0.1", port=8080) as client:
 | BadRequestError (400) | YAML 配置不满足规范或 Kafka 参数错误   | 检查 WasmTaskBuilder 中的配置项。      |
 | ServerError (500)     | Server 侧运行时环境（如 RocksDB）异常 | 检查服务端 conf/config.yaml 存储路径权限。 |
 | NotFoundError (404)   | 操作了不存在的函数或无效的 Checkpoint   | 确认函数名是否输入正确。                   |
+
+---
+
+## 五、Advanced State API（高级状态 API）
+
+带类型状态（ValueState、ListState、MapState、Keyed\* 工厂等）及 `from_context` / `from_context_auto_codec` 用法请参见独立文档：
+
+- **[Python SDK — 高级状态 API](python-sdk-advanced-state-api-zh.md)**