> ## Documentation Index
> Fetch the complete documentation index at: https://docs.blockx.chaintable.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Plugin 系统

> Call Builder 与 Writer Plugin（ResultHandler）：Task 前后两端的可插拔阶段，如何注册、装配、执行和扩展

EVM 区块或 bundle 输入输出路径选择见 [Call Builder 与 Writer Plugin](/zh/concepts/builders-and-handlers)，协议配置集中在[任务配置](/zh/reference/task-configuration)。

Plugin 系统是 Worker 进程内的两组可插拔实现：**Call Builder** 在 Executor 阶段之前把 task 配置变成 `CallList`；**Writer Plugin**（task 侧叫 ResultHandler）在 Executor 阶段之后接收所有 call 的返回值并落库或透传。它们对应 [架构总览](/zh/architecture/overview) 中 Task 主链路的 Builder 阶段和 Writer 阶段，由 [Worker](/zh/components/worker) 的 orchestrator 调度。

## 职责与边界

* Call Builder 只负责"读输入、产出 `CallList`"。它不派发 call、不拉取函数代码、不缓存结果。
* Writer Plugin 只负责"消费 `outputs`、写出或返回结果"。它不参与 call 的重试或去重。
* 一个 task 只有一个 `FunctionCallConfig`（`type/config`）和至多一个 `ResultHandler`（`type/config`）。`type` 是注册名，`config` 是 `json.RawMessage`，由对应实现自己解析。
* Builder / Plugin 都不在构造时持有 IO；`TaskIOReader` / `TaskIO` 在 `Build` / `Execute` 调用时由 Worker 以参数传入，IO 限流和重试在 IO 子系统里做（见 [IO 访问子系统](/zh/components/io-subsystem)）。
* Worker 进程注册哪些 Builder / Plugin 由入口 `app.Profile` 声明。不在 Profile 里的 `type` 在运行期 lookup miss，task 以 `BUILDER_NOT_FOUND` / `PLUGIN_NOT_FOUND` 干净失败。
* Writer Plugin 是 fail-fast：`Execute` 返回 error 即 task 失败；是否允许上游重调度由 `PluginError.Retryable` 决定。
* `outputs` 中每个非 nil 元素必须是 `json.RawMessage`，内容是单个对象（单行）或对象数组（多行）。写表类插件用 `types.VisitOutputRows` 统一校验和遍历。
* 流式 Builder（`StreamingCallBuilder`）把"静态解析"和"扫描产出"拆成两阶段：`PrepareStream` 失败仍是零 call 派发的原子失败；`Run` 中途失败则先前批次可能已执行。

## 代码位置

| 路径 | 用途 |
| - | - |
| `internal/plugin/callbuilder/types/` | `CallBuilder` / `StreamingCallBuilder` 接口、`Call` / `CallList`、`CallBuilderName` 枚举、inline code 归一化、Substrait 风格 `Plan`（`operator.go`） |
| `internal/plugin/callbuilder/registry.go` | `BuilderRegistry`：线程安全的具名注册表 |
| `internal/plugin/callbuilder/dbscan/` | `DBScanBuilder`（`BlockTableCallConfig`）：按区块从 BlockDB 读行 |
| `internal/plugin/callbuilder/call_list/` | `CallListCallConfigBuilder`（`InputsCallConfig`）：显式 call 列表，无 IO |
| `internal/plugin/callbuilder/bundlescan/` | `BundleScanBuilder`（`BlockBundleCallConfig`）：DuckDB 读 S3 parquet，唯一实现了流式契约的 builder。展开见 [Bundle 集群](/zh/components/bundle) |
| `internal/plugin/event/types/` | `WriterPlugin` 接口、`PluginName` 枚举、`PluginResult`、`PluginError`、`VisitOutputRows` / `PreviewOutputs` |
| `internal/plugin/event/registry.go` | `PluginRegistry` |
| `internal/plugin/event/blocktablewrite/` | `BlockTableWriteHandler` 的部署感知入口（`plugin.go`）：按 `Deployment` 装配 `block/`（按区块写 BlockDB block 表）或 `bundle/`（经 BatchWrite 按 bundle 覆盖写，展开见 [Bundle 集群](/zh/components/bundle)） |
| `internal/plugin/event/returnvalue/` | `ReturnValueHandler`：把 outputs 拼成一份 JSON 数组透传给调用方 |
| `internal/plugin/event/tableupserts/` | `NormalTableWriteHandler`：写普通 L1 表，block/bundle 两种写路径 |
| `internal/plugin/event/batchwrite/` | BlockDB BatchWrite job 生命周期（Init / presigned PUT / Commit），被 tableupserts 与 blocktablewrite/bundle 共用 |
| `internal/plugin/schema/` | `ReadColumns` / `ReadTable`：经 `TaskIOReader` 读 logical-types 表结构 |
| `internal/worker/devstub/` | dev / e2e 用的 stub builder（`static` / `payload`）和 plugin（`log` / `test`） |
| `internal/worker/app/app.go` | `Profile.Builders` / `Profile.Plugins`；`newBuilderRegistry` / `newPluginRegistry` 按名字构造并注册 |
| `internal/worker/adapters/orchestrator_phases.go` | Worker 调用入口：`runBuilderPhase`、`runWriterPhase` |
| `cmd/worker/main.go`、`cmd/bundle_worker/main.go` | 两个入口各自的 Profile 清单 |
| `internal/plugin/**/*_test.go`、`internal/plugin/pipeline_integration_test.go` | 单元与集成测试 |
| `docs/specs/plugin-system.md` 等 | 设计 spec，见文末 |

## 核心类型与接口

以下签名从代码复制。

```go theme={null}
// internal/plugin/callbuilder/types/types.go
type CallBuilder interface {
	Name() CallBuilderName
	Build(ctx context.Context, taskCtx *commontypes.TaskCtx, io TaskIOReader) (CallList, error)
}

type StreamingCallBuilder interface {
	CallBuilder
	PrepareStream(ctx context.Context, taskCtx *commontypes.TaskCtx, io TaskIOReader) (CallStreamProducer, error)
}

type CallStreamProducer interface {
	Run(ctx context.Context, emit BatchEmit) error
}
type BatchEmit func(ctx context.Context, batch CallList) error

// internal/plugin/event/types/types.go
type WriterPlugin interface {
	Name() PluginName
	Execute(ctx context.Context, taskCtx *commontypes.TaskCtx, outputs []any, io TaskIO) (any, error)
}
```

| 类型 / 函数 | 文件 | 说明 |
| - | - | - |
| `CallBuilderName` | `callbuilder/types/types.go` | 注册名。内置值：`BuilderDBScan = "BlockTableCallConfig"`、`BuilderCallList = "InputsCallConfig"`、`BuilderBundleScan = "BlockBundleCallConfig"` |
| `Call` | 同上 | 一个工作单元：`FunctionID`、`Code`（inline）、`Args` 或 `ArgsJSON`（紧凑预序列化，优先）、`CallID`、`NoResultCache`、`ArgsPooled`。后三个字段 `json:"-"`，只在进程内使用 |
| `CompactCall` / `CompactArgs` | 同上 | 扫描类 builder 用它把每行 args 一次 marshal 成 `ArgsJSON`，避免百万行 map 对象图常驻 |
| `NormalizeInlineCode` / `InlineCodeFunctionID` | 同上 | inline code 的可信 ingress：算一次 digest，`functionId` 与非空 `code.sourceCode` 互斥；inline 的 FunctionID 为 `inline-<first 8 chars of the digest>` |
| `ErrProductionStopped` / `CallStreamMarker` | 同上 | 流式契约的 sentinel：前者表示 Worker 受控停止产出，后者是 Builder 阶段告诉 core "进入流式 Calls 阶段"的 `PhaseOutcome.Result` |
| `Plan` 及 `ApplyInMemory` / `Validate` / `ToFilterSQLArgs` | `callbuilder/types/operator.go` | Substrait 风格 `relations` 数组（read / filter / deduplicate），DBScan 用内存解释器执行，BundleScan 用 SQL 翻译下推到 DuckDB |
| `TaskCtx` / `FunctionCallConfig` / `ResultHandler` | `internal/common/types/types.go` | `TaskCtx` 内嵌 `TaskInput`（`ID`、`FunctionCallConfig{Type, Config}`、`ResultHandler *{Type, Config}`、`TaskTimeoutMs`），加 `Stage`、`DebugMode`、`Deadline`、`Ctx`、`Cancel` |
| `BuilderRegistry` / `PluginRegistry` | `callbuilder/registry.go`、`event/registry.go` | `Register` 重名报错，`Lookup` 未命中返回 nil |
| `PluginName` | `event/types/types.go` | `PluginBlockDBWrite = "BlockTableWriteHandler"`、`PluginReturnValue = "ReturnValueHandler"`、`PluginTableUpserts = "NormalTableWriteHandler"` |
| `PluginError` / `IsRetryableError` | `event/types/errors.go` | 统一错误模型；`Retryable` 字段决定 orchestrator 是否标记可重调度。未使用 `PluginError` 的裸 error 默认视为可重试 |
| `VisitOutputRows` / `ConsumeOutputRows` / `PreviewOutputs` | `event/types/output_rows.go`、`output_preview.go` | 校验并逐行访问 outputs；`Preview` 供 `debug:true` 模式只统计行数、最多物化 3 行 |
| `PluginResult` | `internal/common/types/types.go` | Writer 阶段返回给调用方的标准化结果（`Success`、`FailureCode`、`Retryable`、`Result`） |
| `schema.ReadColumns` | `internal/plugin/schema/columns.go` | 写表类插件与 bundlescan 用它拿目标表列定义，只投影 output 中出现且属于目标表的列 |

<Note>
  `event/types` 里还声明了 `SchemaProvider` 接口，但当前没有生产调用方；实际列定义都走 `schema.ReadColumns`。
</Note>

## 数据流 / 执行流程

```mermaid theme={null}
sequenceDiagram
    participant O as Orchestrator
    participant BR as BuilderRegistry
    participant B as CallBuilder
    participant D as Dispatcher / Executor
    participant PR as PluginRegistry
    participant P as WriterPlugin

    O->>BR: Lookup(FunctionCallConfig.Type)
    alt 非流式或 StreamBuild.Enabled=false
        O->>B: Build(ctx, taskCtx, ioScope)
        B-->>O: CallList
    else StreamingCallBuilder 且开关打开
        O->>B: PrepareStream(ctx, taskCtx, ioScope)
        B-->>O: CallStreamProducer
        O->>B: producer.Run(ctx, emit) 在 Calls 阶段 scanSem 下
    end
    O->>D: 派发 calls, 收集 outputs []json.RawMessage
    O->>PR: Lookup(ResultHandler.Type)
    O->>P: Execute(ctx, taskCtx, outputs, ioScope)
    P-->>O: result any, error 封装为 PluginResult
```

<Steps>
  <Step title="激活：构造 TaskCtx">
    `Orchestrator.activateAndRun`（`orchestrator_phases.go`）从 core 取回解析好的 `TaskCtx`，注入 `Ctx` / `Cancel`，并把 `ResultHandler.Config` 里的 `debug:true` 提升为 `TaskCtx.DebugMode`。
  </Step>

  <Step title="Builder 阶段">
    `runBuilderPhase` 先取 `builderSem`（`Admission.BuilderSlots`），按 `FunctionCallConfig.Type` 在 `builderRegistry` 里 `Lookup`。命中 `StreamingCallBuilder` 且 `StreamBuild.Enabled` 时走 `PrepareStream`，否则走 `Build`。成功后 `releaseBuilderConfigPayload` 把 `FunctionCallConfig.Config` 置 nil 释放请求 JSON。失败码：`BUILDER_NOT_FOUND` / `BUILDER_FAILED`，`Retryable` 取决于是否是可重试 `IOError`。
  </Step>

  <Step title="Calls 阶段">
    非流式：`runCallPhase` 把整份 `CallList` 交给 dispatcher。流式：`runCallStreamPhase` 在 `scanSem` 下跑 `producer.Run`，`makeBatchEmit` 提供带信用（`MaxOutstandingBatches`）的背压 emit。详见 [Call 执行子系统](/zh/components/call-execution)。
  </Step>

  <Step title="Writer 阶段">
    `runWriterPhase` 取 `pluginSem`（`Admission.PluginSlots`），若 `ResultHandler` 非 nil 则 `Lookup` 并 `plugin.Execute(resultCtx, ioTaskCtx, cmd.Outputs, ioScope)`；返回值封装为 `commontypes.PluginResult` 进入 `TaskResult.ExecuteResult.PluginResults`。`ResultHandler` 为 nil 时阶段直接成功。
  </Step>
</Steps>

### 内置 Call Builder

* **DBScanBuilder**（`dbscan/dbscan.go`）：解析 `DBScanBuilderDecl{Block, TriggerSources}`；每个 `TriggerSource{Table, Operator, FunctionID, Code, Params}` 一个 goroutine（`errgroup.SetLimit(3)`）。对每个 source：校验表名标识符 → `NormalizeInlineCode` → `Operator.Validate` → `io.Read` 发 `blockdb.ReadOp[*GetBlockRowsRequest]`（`Op: "GetBlockRows"`，逻辑表名原样传给 BlockDB）→ `Operator.ApplyInMemory(rows)`（先 filter 后 deduplicate，按 relations 链顺序）→ 按 `Params` 模板拼 args。`Params` 中等于 `"${<table>}"` 的位置替换成行 map；`Params` 为空则整行作为唯一参数；`Params` 非空但没有占位符则所有 call 共享同一份静态 `ArgsJSON`。`CallID` 格式为 `{taskID}-{table}-{height}-{i}`。
* **CallListCallConfigBuilder**（`call_list/call_list.go`）：解析 `{function, code, callList: [][]any}`，不做 IO，第 i 项生成 `Call{FunctionID, Args, CallID: "{taskID}-{i}"}`。
* **BundleScanBuilder**（`bundlescan/`）：从 `blockBundle.number` + `triggerSources[].tableId`（或兼容字段 `bundleBucket`）解析 parquet 位置，经 DuckDB 流式读行并把 `Operator` 翻译成 SQL 下推。同时实现 `StreamingCallBuilder`（`stream.go`，批边界 `defaultBatchMaxRows = 8192` / `defaultBatchMaxArgsBytes = 32 MiB`），并在可证明行 key 唯一时置位 `Call.NoResultCache`。细节见 [Bundle 集群](/zh/components/bundle)。

### Operator（Plan）

`TriggerSource.Operator` 是 `*types.Plan`，JSON 形如 `{"relations": [{"read": ...}, {"filter": ...}, {"deduplicate": ...}]}`，relation 之间用 `input.relation_id`（数组下标）串成线性链。`Plan.Validate` 要求恰好一个 `read`、链无分叉、所有 `selection.field` 落在 `read.base_schema.names` 内。filter 用 SQL 三值逻辑（NULL 比较得 `UNKNOWN`，只保留 `TRUE`），`equal(col, null)` 按 `IS NULL` 解释；deduplicate key 用带类型标签和长度前缀的确定性编码（`appendDedupValue`）。支持的 `function_reference`：`and`、`or`、`not`、`equal`、`not_equal`、`gt`、`gte`、`lt`、`lte`、`is_null`、`is_not_null`。

`Plan.UnmarshalJSON` 还接受一种 legacy 形态：顶层带 `type` 字段的表达式树（`Attribute` / `Literal` / `EqualTo` / `In` / `And` / `Or` / `Not` 等），会被 `legacyConditionToPlan` 转成单个 filter relation。

完整格式见 blockx 仓库 `docs/specs/substrait_ast.md`。

### 内置 Writer Plugin

| 名字 | 包 | 配置（`ResultHandler.Config`） | 行为 |
| - | - | - | - |
| `BlockTableWriteHandler` | `blocktablewrite`（block 部署 → `block/`，bundle 部署 → `bundle/`） | block：`{targetTable, block{id,height,timestamp}}`；bundle：`{targetTable, blockBundle{number}}` | block：`schema.ReadColumns` 取列 → `VisitOutputRows` 投影成 `WriteRow` → 一次 `blockdb.BlockWriteReq`（`Sync:false`），返回 `{written_rows, finish_time}`。bundle：经 `batchwrite` 做 Init / presigned PUT / Commit 的覆盖写，见 [Bundle 集群](/zh/components/bundle)。形状不匹配时以 `block deployment requires config.block` / `bundle deployment requires config.blockBundle` 失败 |
| `ReturnValueHandler` | `returnvalue` | 无 | 把 outputs 原样拼成一个 JSON 数组（`json.RawMessage`）放进 `PluginResult.Result`；nil 元素写 `null` |
| `NormalTableWriteHandler` | `tableupserts` | `{targetTable, condition?{column, policy}}` | block profile 用 `TableWriteService.UpsertRows`；bundle profile 用 `batchwrite.Writer.Submit`（`NormalTableUpsertStrategy`）。`policy` 支持 `UpdateIfSmaller` / `UpdateIfLarger` |

所有写表插件在 `TaskCtx.DebugMode` 为 true 时只 `PreviewOutputs` 统计行数并打 `table_write_debug` 日志，不读 schema、不发写请求。

### devstub

`internal/worker/devstub/` 提供 `StaticCallBuilder`（名字 `static`，返回固定 call）、`PayloadCallBuilder`（`payload`，从 `config.calls` 取 call，`config.builderFail` 模拟失败）、`LogWriterPlugin`（`log`）、`TestWriterPlugin`（`test`，`config.pluginFail` 模拟失败）。它们只被 `cmd/worker` 的 Profile 列入，`cmd/bundle_worker` 不注册；用于 e2e / perf（如 `e2e/perf/s2_call_count_payload_test.go`）和本地联调，不要在生产 task 里使用。

## 状态与生命周期

Builder 与 Writer Plugin 的注册集合在进程启动时由 `app.Profile` 确定，启动完成后不会按 task 动态增删。普通 Builder 在 Builder 阶段执行一次；流式 Builder 的 producer 延续到 Calls 阶段并在完成、失败或停止派发时结束；Writer Plugin 在 Calls 成功汇聚后执行一次。阶段状态、超时和终态都由 Worker 管理，不保存在 plugin registry 中。

Builder 和 Writer Plugin 每次调用都接收当前 task 的 `TaskCtx` 与 IO 接口。它们不拥有 Worker slot、task 终态或跨 task 的 IO scope。

## 配置

Plugin 本身没有独立配置文件；相关项都在 `internal/worker/app/config.go` 的 `WorkerFullConfig`：

| 字段 | 默认 | 作用 |
| - | - | - |
| `PhaseIOTimeoutMs` | `10000` | 传给 `NewDBScanBuilder` / `NewBlockDBWritePlugin` 等的 `ioTimeout`，约束 Builder / Writer 阶段单次 IO。环境变量 `PHASE_IO_TIMEOUT_MS` |
| `Admission.BuilderSlots` | `16` | `builderSem`，并发 Builder 阶段数（`internal/worker/core/config.go`） |
| `Admission.PluginSlots` | `16` | `pluginSem`，并发 Writer 阶段数 |
| `Admission.ScanSlots` | `8` | `scanSem`，并发流式产出数（每个产出内 bundlescan 最多 3 个 source 并发） |
| `StreamBuild.Enabled` | `false`（`cmd/bundle_worker` 的 `TuneDefaults` 打开） | 是否对 `StreamingCallBuilder` 走两阶段流式路径 |
| `StreamBuild.BatchMaxRows` / `BatchMaxArgsBytes` | `8192` / `32 << 20` | 单批上限，先到先切；单 call `ArgsJSON` 超字节上限是 production error |
| `StreamBuild.MaxOutstandingBatches` | `2` | 生产者信用窗口，背压来源 |
| `BlockDB.BatchWriteAddr` | 空 | bundle profile 含 `PluginBlockDBWrite` 或 `PluginTableUpserts` 时必填（block profile 不要求），否则 `validateProfileRuntimeConfig` 在启动期失败。环境变量 `BLOCKDB_BATCH_WRITE_ADDR` |

## 扩展点

### 新增一个 Call Builder

<Steps>
  <Step title="定义名字">
    在 `internal/plugin/callbuilder/types/types.go` 的 `const` 块加一个 `CallBuilderName`，值就是 task 里 `functionCallConfig.type` 的字符串。
  </Step>

  <Step title="实现接口">
    新建 `internal/plugin/callbuilder/<name>/`，实现 `types.CallBuilder`。在 `Build` 里用 `sonic.Config{UseNumber: true}` 解析 `taskCtx.FunctionCallConfig.Config`，用 `types.NormalizeInlineCode` 处理 inline code，通过传入的 `io.Read` 做所有读 IO。逐行 args 用 `types.CompactCall` 产出。若要支持流式，再实现 `types.StreamingCallBuilder`（`PrepareStream` 里做完全部静态校验，`Run` 里按批 `emit`，每批换新的 backing array，遇到 `types.ErrProductionStopped` 原样返回）。
  </Step>

  <Step title="装配">
    在 `internal/worker/app/app.go` 的 `newBuilderRegistry` 的 `switch name` 加一个 case 构造它；再把名字加进需要它的 Profile（`cmd/worker/main.go` 或 `cmd/bundle_worker/main.go` 的 `Builders` 列表）。不加 case 会在启动时报 `unknown call builder`。
  </Step>

  <Step title="测试">
    包内单测（参考 `dbscan/dbscan_test.go`、`call_list/call_list_test.go`：用 fake `TaskIOReader` 断言请求类型、CallID、args、错误分类）；`internal/worker/app/profile_test.go` 的 `TestNewBuilderRegistry_ProfileMembership` 系列补名字；如需端到端，参考 `e2e/system/logical_type/dbscan_operator_test.go`。
  </Step>
</Steps>

```go theme={null}
package mybuilder

import (
	"context"
	"fmt"

	"github.com/bytedance/sonic"

	commontypes "github.com/Chaintable/blockx/internal/common/types"
	"github.com/Chaintable/blockx/internal/plugin/callbuilder/types"
)

var jsonAPI = sonic.Config{UseNumber: true}.Froze()

type decl struct {
	Function string `json:"function"`
	Rows     []any  `json:"rows"`
}

// BuilderMyScan 需先加到 internal/plugin/callbuilder/types/types.go 的 CallBuilderName 常量块。
type Builder struct{}

func (b *Builder) Name() types.CallBuilderName { return types.BuilderMyScan }

func (b *Builder) Build(ctx context.Context, taskCtx *commontypes.TaskCtx, io types.TaskIOReader) (types.CallList, error) {
	var d decl
	if err := jsonAPI.Unmarshal(taskCtx.FunctionCallConfig.Config, &d); err != nil {
		return nil, fmt.Errorf("myScan: unmarshal config: %w", err)
	}
	calls := make(types.CallList, 0, len(d.Rows))
	for i, row := range d.Rows {
		c, err := types.CompactCall(d.Function, nil, []any{row}, fmt.Sprintf("%s-%d", taskCtx.ID, i))
		if err != nil {
			return nil, err
		}
		calls = append(calls, c)
	}
	return calls, nil
}
```

### 新增一个 Writer Plugin

<Steps>
  <Step title="定义名字">
    在 `internal/plugin/event/types/types.go` 的 `const` 块加一个 `PluginName`，值即 `resultHandler.type`。
  </Step>

  <Step title="实现接口">
    新建 `internal/plugin/event/<name>/`，实现 `types.WriterPlugin`。`Execute` 里：`taskCtx.ResultHandler` 为 nil 或 config 解析失败时返回 `*types.PluginError{Kind: types.PluginErrExecution, ...}`；用 `types.VisitOutputRows`（或 `ConsumeOutputRows`）遍历 outputs；写 IO 走传入的 `io.Write`；下游错误用 `types.IsRetryableError(err)` 决定 `Retryable`。尊重 `taskCtx.DebugMode`：只 `PreviewOutputs`，不发外部 IO。成功时返回要透传的数据（无则 `nil`）。
  </Step>

  <Step title="装配">
    在 `internal/worker/app/app.go` 的 `newPluginRegistry` 加 case；把名字加进 `cmd/*/main.go` 的 `Plugins` 列表。若依赖新的 endpoint，在 `validateProfileRuntimeConfig` 里加启动期校验。
  </Step>

  <Step title="测试">
    包内单测（参考 `blocktablewrite/plugin_test.go`、`blocktablewrite/bundle/write_test.go` 和 `tableupserts/table_upserts_test.go`：单行 / 多行 / 缺列 / 显式 nil / 写失败可重试性 / debug 模式）；`event/types/errors_test.go` 的 `TestIsRetryableError` 覆盖新错误分类；`internal/worker/app/profile_test.go` 的 `TestNewPluginRegistry_ProfileMembership`；orchestrator 侧 `TestOrchestrator_EventPhase_*` 已覆盖通用路径，一般不用改。
  </Step>
</Steps>

```go theme={null}
package myplugin

import (
	"context"

	commontypes "github.com/Chaintable/blockx/internal/common/types"
	"github.com/Chaintable/blockx/internal/plugin/event/types"
)

// PluginMyWrite 需先加到 internal/plugin/event/types/types.go 的 PluginName 常量块。
type Plugin struct{}

func (p *Plugin) Name() types.PluginName { return types.PluginMyWrite }

func (p *Plugin) Execute(ctx context.Context, taskCtx *commontypes.TaskCtx, outputs []any, io types.TaskIO) (any, error) {
	if taskCtx.ResultHandler == nil {
		return nil, &types.PluginError{Kind: types.PluginErrExecution, PluginName: p.Name(), Message: "result handler config is nil"}
	}
	rowCount, err := types.VisitOutputRows(outputs, func(row map[string]any) error {
		// 把 row 投影成写请求；不要保留 row map 本身
		return nil
	})
	if err != nil {
		return nil, &types.PluginError{Kind: types.PluginErrExecution, PluginName: p.Name(), Message: err.Error()}
	}
	// 写 IO 走传入的 io.Write；下游错误保留可重试分类：
	// if _, err := io.Write(ctx, req); err != nil {
	//     return nil, &types.PluginError{Kind: types.PluginErrExecution, PluginName: p.Name(),
	//         Message: "write failed", Cause: err, Retryable: types.IsRetryableError(err)}
	// }
	return map[string]any{"written_rows": rowCount}, nil
}
```

<Warning>
  `Call.ArgsJSON`、`NoResultCache`、`ArgsPooled` 都是进程内字段：`Call` 不能被 JSON 往返当作 args 的载体，`ArgsPooled` 只在恰好一个 call 引用该缓冲时才可置位。共享静态 args 的 builder 一律不设 `NoResultCache`。
</Warning>

## 测试

```bash theme={null}
go test ./internal/plugin/... -short              # builder / plugin / operator 单测与包内集成
go test ./internal/worker/app/ -run 'Registry|Profile'   # Profile 装配
go test ./internal/worker/adapters/ -run 'Orchestrator_EventPhase|OrchestratorStream|BuilderFails'
go test ./internal/worker/devstub/
```

* `internal/plugin/callbuilder/callbuilder_test.go`、`internal/plugin/event/event_test.go`：注册表和"runner"式集成用例；`internal/plugin/pipeline_integration_test.go`：起 gRPC mock BlockDB 跑 DBScan → BlockDBWrite 全链路。
* `internal/plugin/callbuilder/types/operator_test.go`：三值逻辑、大整数精确比较、legacy operator 转换、dedup key。
* `internal/plugin/callbuilder/bundlescan/stream_test.go`：流式批边界、单行超限、`PrepareStream` 静态错误零 emit。
* e2e：`e2e/system/logical_type/dbscan_operator_test.go`（真实 Worker 进程 + gRPC mock BlockDB），`e2e/perf/s1_builder_dbscan_test.go`、`s5_plugin_return_value_test.go`。总览见 [测试](/zh/development/testing)。

## 相关文档

* blockx 仓库 `docs/specs/plugin-system.md`：接口、数据结构、装配、DBScan / BlockDBWrite / BlockBundle / TableUpserts 详细设计（重点）。
* blockx 仓库 `docs/specs/architecture.md` §4.2.3：Plugin 模块在整体架构中的位置。
* blockx 仓库 `docs/specs/substrait_ast.md`：Operator（Plan）JSON 格式与 Substrait 差异。
* blockx 仓库 `docs/specs/2026-07-14-dbscan-in-memory-operator.md`：DBScan 改用 `GetBlockRows` + 内存解释器的决策记录。
* 站内：[Worker](/zh/components/worker)（orchestrator 与阶段准入）、[Call 执行子系统](/zh/components/call-execution)（Calls 阶段与流式派发）、[IO 访问子系统](/zh/components/io-subsystem)（`TaskIOReader` / `TaskIO` 的实现与限流）、[Bundle 集群](/zh/components/bundle)（bundlescan / blocktablewrite/bundle 展开）、[Task 生命周期](/zh/architecture/task-lifecycle)。


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.