feat: add MetadataHolder embedding for operator metadata reuse

Introduce MetadataHolder struct that stores DSL-declared field-name
slices and provides a default SetMetadata implementation. Operators
embed it via Go struct embedding instead of writing per-operator
SetMetadata methods, reducing boilerplate.

Migrated all engine operators (filter_condition, merge_dedup,
reorder_sort, transform_normalize, transform_dispatch) to use
MetadataHolder embedding. Updated tests, design docs, and codegen
output. Bump to v0.2.5.

Author: Liam0205

apple/_version.py

@@ -1 +1 @@
-__version__ = "0.2.4"
+__version__ = "0.2.5"

apple_generated/operators.py

@@ -8,14 +8,12 @@ class FilterConditionOp(BaseOp):
     """Operator: filter_condition"""
     _name = "filter_condition"
     _params_schema = {
-        "field": {"type": "string", "required": True},
         "value": {"type": "any", "required": True},
     }
 
     def __call__(
         self,
         *,
-        field: str = ...,
         value: Any = ...,
         common_input: list[str] | None = None,
         common_output: list[str] | None = None,
@@ -28,7 +26,6 @@ def __call__(
     ) -> "FilterConditionOp":
         return self._apply(
             params={
-                "field": field,
                 "value": value,
             },
             common_input=common_input,
@@ -79,14 +76,12 @@ class MergeDedupOp(BaseOp):
     """Operator: merge_dedup"""
     _name = "merge_dedup"
     _params_schema = {
-        "dedup_by": {"type": "string", "required": True},
         "strategy": {"type": "string", "required": False, "default": "first"},
     }
 
     def __call__(
         self,
         *,
-        dedup_by: str = ...,
         strategy: str = "",
         common_input: list[str] | None = None,
         common_output: list[str] | None = None,
@@ -99,7 +94,6 @@ def __call__(
     ) -> "MergeDedupOp":
         return self._apply(
             params={
-                "dedup_by": dedup_by,
                 "strategy": strategy,
             },
             common_input=common_input,
@@ -185,14 +179,12 @@ class ReorderSortOp(BaseOp):
     """Operator: reorder_sort"""
     _name = "reorder_sort"
     _params_schema = {
-        "field": {"type": "string", "required": True},
         "order": {"type": "string", "required": False, "default": "desc"},
     }
 
     def __call__(
         self,
         *,
-        field: str = ...,
         order: str = "",
         common_input: list[str] | None = None,
         common_output: list[str] | None = None,
@@ -205,7 +197,6 @@ def __call__(
     ) -> "ReorderSortOp":
         return self._apply(
             params={
-                "field": field,
                 "order": order,
             },
             common_input=common_input,
@@ -262,15 +253,11 @@ class TransformDispatchOp(BaseOp):
     """Operator: transform_dispatch"""
     _name = "transform_dispatch"
     _params_schema = {
-        "common_field": {"type": "string", "required": True},
-        "item_field": {"type": "string", "required": True},
     }
 
     def __call__(
         self,
         *,
-        common_field: str = ...,
-        item_field: str = ...,
         common_input: list[str] | None = None,
         common_output: list[str] | None = None,
         item_input: list[str] | None = None,
@@ -282,8 +269,6 @@ def __call__(
     ) -> "TransformDispatchOp":
         return self._apply(
             params={
-                "common_field": common_field,
-                "item_field": item_field,
             },
             common_input=common_input,
             common_output=common_output,
@@ -299,17 +284,13 @@ class TransformNormalizeOp(BaseOp):
     """Operator: transform_normalize"""
     _name = "transform_normalize"
     _params_schema = {
-        "field": {"type": "string", "required": True},
         "method": {"type": "string", "required": False, "default": "min_max"},
-        "output_field": {"type": "string", "required": False},
     }
 
     def __call__(
         self,
         *,
-        field: str = ...,
         method: str = "",
-        output_field: str = "",
         common_input: list[str] | None = None,
         common_output: list[str] | None = None,
         item_input: list[str] | None = None,
@@ -321,9 +302,7 @@ def __call__(
     ) -> "TransformNormalizeOp":
         return self._apply(
             params={
-                "field": field,
                 "method": method,
-                "output_field": output_field,
             },
             common_input=common_input,
             common_output=common_output,

design_doc/03_data_abstraction.md

@@ -270,3 +270,81 @@ func (out *OperatorOutput) SetWarning(err error)
 - **统一接口，按需使用**：所有算子共享同一个 `Operator` 接口和 `OperatorOutput`。不同类型的算子使用不同的方法子集，不使用的方法不调用即可。引擎通过 JSON 元数据（`recall: true`、`sources` 等）或 Go 接口断言识别算子类别，决定写回策略。
 - **无状态可重入**：算子在 `Init` 后不持有可变状态，`Execute` 可被多个 goroutine 并发调用。算子可持有只读配置和线程安全资源（如连接池），不可持有请求级状态。
 - **错误约定**：`return nil` 表示正常执行；`output.SetWarning(err)` 表示可恢复错误（DAG 继续）；`return err` 表示不可恢复错误（DAG 终止）。
+
+#### MetadataAware — 字段名自省
+
+算子操作的字段名已在 DSL 的 `common_input` / `common_output` / `item_input` / `item_output` 中声明。通过实现可选接口 `MetadataAware`，算子可以在初始化阶段获取这些声明的字段名，而无需通过 `Params` 重复指定。
+
+```go
+// MetadataAware is an optional interface. The engine calls SetMetadata
+// after Init for operators that implement it.
+type MetadataAware interface {
+    SetMetadata(commonInput, commonOutput, itemInput, itemOutput []string)
+}
+```
+
+引擎在 `Init(params)` 之后自动检测算子是否实现 `MetadataAware`，若实现则调用 `SetMetadata` 注入 `$metadata` 中声明的字段名。
+
+##### MetadataHolder — 嵌入式默认实现
+
+引擎提供 `MetadataHolder` 结构体，存储四个字段名切片并提供默认的 `SetMetadata` 实现。算子通过 Go embedding 嵌入即可自动满足 `MetadataAware` 接口，无需每个算子手写 `SetMetadata`：
+
+```go
+// MetadataHolder 存储 DSL 声明的字段名，提供默认 SetMetadata。
+type MetadataHolder struct {
+    CommonInput  []string
+    CommonOutput []string
+    ItemInput    []string
+    ItemOutput   []string
+}
+
+func (m *MetadataHolder) SetMetadata(commonInput, commonOutput, itemInput, itemOutput []string) {
+    m.CommonInput = commonInput
+    m.CommonOutput = commonOutput
+    m.ItemInput = itemInput
+    m.ItemOutput = itemOutput
+}
+```
+
+**使用方式**——算子嵌入 `pine.MetadataHolder`，在 `Execute` 中直接访问字段名：
+
+```go
+type SortOp struct {
+    pine.MetadataHolder   // 自动实现 MetadataAware
+    ascending bool
+}
+
+func (o *SortOp) Execute(ctx context.Context, in *pine.OperatorInput, out *pine.OperatorOutput) error {
+    field := o.ItemInput[0]   // 直接从嵌入的 MetadataHolder 获取
+    // ...
+}
+```
+
+如果算子需要自定义 metadata 处理逻辑（极少见），可以覆写 `SetMetadata` 方法，此时嵌入的默认实现不再生效。
+
+**何时用 `MetadataAware` vs `Params`**：
+
+| 场景 | 推荐方式 | 说明 |
+|------|---------|------|
+| 算子操作的字段名由调用方决定 | 嵌入 `MetadataHolder` | 字段名已在 DSL 声明中给出，不应重复 |
+| 算子有固定的业务语义字段 | 硬编码 | 如 `filter_paginate` 固定读 `page`/`size` |
+| 算子有与字段名无关的配置 | `Params` | 如 `reorder_sort` 的 `order="desc"` |
+
+**反模式**：通过 `Params` 传入字段名，导致 DSL 调用时必须同时声明 `common_input=["x"]` 和 `field="x"`——信息重复，容易不一致。
+
+**正确模式示例**（以 `reorder_sort` 为例）：
+
+```python
+# 反模式: field 名在 item_input 和 params 中重复
+flow.reorder_sort(
+    item_input=["score"],
+    field="score",       # 冗余！
+    ascending=False,
+)
+
+# 正确模式: 算子通过 MetadataHolder 获取 item_input=["score"]
+flow.reorder_sort(
+    item_input=["score"],
+    order="desc",
+)
+```

doc/operators/filter_condition.md

@@ -8,7 +8,6 @@ Removes items where a specified field equals a given value.
 
 | Name | Type | Required | Default | Description |
 |------|------|----------|---------|-------------|
-| field | string | Yes | - | Item field to check. |
 | value | any | Yes | - | Items where field == value are removed. |
 
 ## Metadata Contract
@@ -24,7 +23,6 @@ Removes items where a specified field equals a given value.
 
 ```python
 flow.filter_condition(
-    field=...,
     value=...,
     common_input=[...],
     item_input=[...],

doc/operators/merge_dedup.md

@@ -8,7 +8,6 @@ Deduplicates items by a key field, keeping the first occurrence.
 
 | Name | Type | Required | Default | Description |
 |------|------|----------|---------|-------------|
-| dedup_by | string | Yes | - | Field name to deduplicate on. |
 | strategy | string | No | `"first"` | Dedup strategy — "first" keeps first occurrence. |
 
 ## Metadata Contract
@@ -24,7 +23,6 @@ Deduplicates items by a key field, keeping the first occurrence.
 
 ```python
 flow.merge_dedup(
-    dedup_by=...,
     strategy=...,
     common_input=[...],
     item_input=[...],

doc/operators/reorder_sort.md

@@ -8,7 +8,6 @@ Sorts items by a numeric field in ascending or descending order.
 
 | Name | Type | Required | Default | Description |
 |------|------|----------|---------|-------------|
-| field | string | Yes | - | Item field to sort by. |
 | order | string | No | `"desc"` | Sort direction — "asc" or "desc". |
 
 ## Metadata Contract
@@ -24,7 +23,6 @@ Sorts items by a numeric field in ascending or descending order.
 
 ```python
 flow.reorder_sort(
-    field=...,
     order=...,
     common_input=[...],
     item_input=[...],

doc/operators/transform_dispatch.md

@@ -8,8 +8,6 @@ Copies a common-side field value to every item as an item-side field.
 
 | Name | Type | Required | Default | Description |
 |------|------|----------|---------|-------------|
-| common_field | string | Yes | - | Source common field to read. |
-| item_field | string | Yes | - | Target item field to write. |
 
 ## Metadata Contract
 
@@ -24,8 +22,6 @@ Copies a common-side field value to every item as an item-side field.
 
 ```python
 flow.transform_dispatch(
-    common_field=...,
-    item_field=...,
     common_input=[...],
     item_input=[...],
     item_output=[...],

doc/operators/transform_normalize.md

@@ -8,9 +8,7 @@ Normalizes a numeric item field using min-max scaling to [0, 1].
 
 | Name | Type | Required | Default | Description |
 |------|------|----------|---------|-------------|
-| field | string | Yes | - | Item field to normalize. |
 | method | string | No | `"min_max"` | Normalization method. |
-| output_field | string | No | `""` | Target field for normalized values. |
 
 ## Metadata Contract
 
@@ -25,9 +23,7 @@ Normalizes a numeric item field using min-max scaling to [0, 1].
 
 ```python
 flow.transform_normalize(
-    field=...,
     method=...,
-    output_field=...,
     common_input=[...],
     item_input=[...],
     item_output=[...],

internal/types/operator.go

@@ -148,6 +148,31 @@ type MetadataAware interface {
 	SetMetadata(commonInput, commonOutput, itemInput, itemOutput []string)
 }
 
+// MetadataHolder stores the four DSL-declared field-name slices and provides
+// a default SetMetadata implementation. Embed it in an operator struct to
+// satisfy MetadataAware automatically:
+//
+//	type SortOp struct {
+//	    pine.MetadataHolder
+//	    ascending bool
+//	}
+//
+// The operator can then access o.CommonInput, o.ItemInput, etc. directly.
+type MetadataHolder struct {
+	CommonInput  []string
+	CommonOutput []string
+	ItemInput    []string
+	ItemOutput   []string
+}
+
+// SetMetadata implements MetadataAware.
+func (m *MetadataHolder) SetMetadata(commonInput, commonOutput, itemInput, itemOutput []string) {
+	m.CommonInput = commonInput
+	m.CommonOutput = commonOutput
+	m.ItemInput = itemInput
+	m.ItemOutput = itemOutput
+}
+
 // ParamSpec describes a single operator parameter for schema validation.
 type ParamSpec struct {
 	Type        string // "string", "int64", "float64", "bool", "any"

operator.go

@@ -33,3 +33,7 @@ type OperatorSchema = types.OperatorSchema
 // MetadataAware is an optional interface for operators that need access to
 // their declared input/output field names from $metadata.
 type MetadataAware = types.MetadataAware
+
+// MetadataHolder stores DSL-declared field-name slices and provides a default
+// SetMetadata. Embed it in operator structs for automatic MetadataAware compliance.
+type MetadataHolder = types.MetadataHolder