> ## 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 前后两端的可插拔阶段，如何注册、装配、执行和扩展 Plugin 系统是 Worker 进程内的两组可插拔实现：**Call Builder** 在 Executor 阶段之前把 task 配置变成 `CallList`；**Writer Plugin**（task 侧叫 ResultHandler）在 Executor 阶段之后接收所有 call 的返回值并落库或透传。它们对应 [架构总览](/architecture/overview) 中 Task 主链路的 Builder 阶段和 Writer 阶段，由 [Worker](/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 访问子系统](/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`（`CallListCallConfig`）：显式 call 列表，无 IO | | `internal/plugin/callbuilder/bundlescan/` | `BundleScanBuilder`（`BlockBundleCallConfig`）：DuckDB 读 S3 parquet，唯一实现了流式契约的 builder。展开见 [Bundle 集群](/components/bundle) | | `internal/plugin/event/types/` | `WriterPlugin` 接口、`PluginName` 枚举、`PluginResult`、`PluginError`、`VisitOutputRows` / `PreviewOutputs` | | `internal/plugin/event/registry.go` | `PluginRegistry` | | `internal/plugin/event/blockdbwrite/` | `BlockWriteResultHandler`：按区块写 BlockDB 表 | | `internal/plugin/event/returnvalue/` | `ReturnValueResultHandler`：把 outputs 拼成一份 JSON 数组透传给调用方 | | `internal/plugin/event/tableupserts/` | `TableUpsertsResultHandler`：写普通 L1 表，block/bundle 两种写路径 | | `internal/plugin/event/batchwrite/` | BlockDB BatchWrite job 生命周期（Init / presigned PUT / Commit），被 tableupserts 与 bundlewrite 共用 | | `internal/plugin/event/bundlewrite/` | `BlockBundleWriteResultHandler`：按 bundle 覆盖写。展开见 [Bundle 集群](/components/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 = "CallListCallConfig"`、`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-` | | `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 = "BlockWriteResultHandler"`、`PluginReturnValue = "ReturnValueResultHandler"`、`PluginBundleWrite = "BlockBundleWriteResultHandler"`、`PluginTableUpserts = "TableUpsertsResultHandler"` | | `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 中出现且属于目标表的列 | `event/types` 里还声明了 `SchemaProvider` 接口，但当前没有生产调用方；实际列定义都走 `schema.ReadColumns`。 ## 数据流 / 执行流程 ```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 ``` `Orchestrator.activateAndRun`（`orchestrator_phases.go`）从 core 取回解析好的 `TaskCtx`，注入 `Ctx` / `Cancel`，并把 `ResultHandler.Config` 里的 `debug:true` 提升为 `TaskCtx.DebugMode`。 `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`。 非流式：`runCallPhase` 把整份 `CallList` 交给 dispatcher。流式：`runCallStreamPhase` 在 `scanSem` 下跑 `producer.Run`，`makeBatchEmit` 提供带信用（`MaxOutstandingBatches`）的背压 emit。详见 [Call 执行子系统](/components/call-execution)。 `runWriterPhase` 取 `pluginSem`（`Admission.PluginSlots`），若 `ResultHandler` 非 nil 则 `Lookup` 并 `plugin.Execute(resultCtx, ioTaskCtx, cmd.Outputs, ioScope)`；返回值封装为 `commontypes.PluginResult` 进入 `TaskResult.ExecuteResult.PluginResults`。`ResultHandler` 为 nil 时阶段直接成功。 ### 内置 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` 中等于 `"${}"` 的位置替换成行 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 集群](/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`） | 行为 | | ------------------------------- | -------------- | ------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `BlockWriteResultHandler` | `blockdbwrite` | `{targetTable, block{id,height,timestamp}}` | `schema.ReadColumns` 取列 → `VisitOutputRows` 投影成 `WriteRow` → 一次 `blockdb.BlockWriteReq`（`Sync:false`）。返回 `{written_rows, finish_time}` | | `ReturnValueResultHandler` | `returnvalue` | 无 | 把 outputs 原样拼成一个 JSON 数组（`json.RawMessage`）放进 `PluginResult.Result`；nil 元素写 `null` | | `TableUpsertsResultHandler` | `tableupserts` | `{targetTable, condition?{column, policy}}` | block profile 用 `TableWriteService.UpsertRows`；bundle profile 用 `batchwrite.Writer.Submit`（`NormalTableUpsertStrategy`）。`policy` 支持 `UpdateIfSmaller` / `UpdateIfLarger` | | `BlockBundleWriteResultHandler` | `bundlewrite` | `{targetTable, blockBundle{number}}` | 经 `batchwrite` 做 Init / presigned PUT / Commit 的覆盖写。见 [Bundle 集群](/components/bundle) | 所有写表插件在 `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 里使用。 ## 配置 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` | 空 | Profile 含 `PluginBundleWrite`，或 bundle profile 含 `PluginTableUpserts` 时必填，否则 `validateProfileRuntimeConfig` 在启动期失败。环境变量 `BLOCKDB_BATCH_WRITE_ADDR` | ## 扩展点 ### 新增一个 Call Builder 在 `internal/plugin/callbuilder/types/types.go` 的 `const` 块加一个 `CallBuilderName`，值就是 task 里 `functionCallConfig.type` 的字符串。 新建 `internal/plugin/callbuilder//`，实现 `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` 原样返回）。 在 `internal/worker/app/app.go` 的 `newBuilderRegistry` 的 `switch name` 加一个 case 构造它；再把名字加进需要它的 Profile（`cmd/worker/main.go` 或 `cmd/bundle_worker/main.go` 的 `Builders` 列表）。不加 case 会在启动时报 `unknown call builder`。 包内单测（参考 `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`。 ```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 在 `internal/plugin/event/types/types.go` 的 `const` 块加一个 `PluginName`，值即 `resultHandler.type`。 新建 `internal/plugin/event//`，实现 `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`）。 在 `internal/worker/app/app.go` 的 `newPluginRegistry` 加 case；把名字加进 `cmd/*/main.go` 的 `Plugins` 列表。若依赖新的 endpoint，在 `validateProfileRuntimeConfig` 里加启动期校验。 包内单测（参考 `blockdbwrite` 的测试和 `tableupserts/table_upserts_test.go`：单行 / 多行 / 缺列 / 显式 nil / 写失败可重试性 / debug 模式）；`event/types/errors_test.go` 的 `TestIsRetryableError` 覆盖新错误分类；`internal/worker/app/profile_test.go` 的 `TestNewPluginRegistry_ProfileMembership`；orchestrator 侧 `TestOrchestrator_EventPhase_*` 已覆盖通用路径，一般不用改。 ```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 } ``` `Call.ArgsJSON`、`NoResultCache`、`ArgsPooled` 都是进程内字段：`Call` 不能被 JSON 往返当作 args 的载体，`ArgsPooled` 只在恰好一个 call 引用该缓冲时才可置位。共享静态 args 的 builder 一律不设 `NoResultCache`。 ## 测试 ```bash theme={null} cd ~/Work/blockx 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`。总览见 [测试](/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](/components/worker)（orchestrator 与阶段准入）、[Call 执行子系统](/components/call-execution)（Calls 阶段与流式派发）、[IO 访问子系统](/components/io-subsystem)（`TaskIOReader` / `TaskIO` 的实现与限流）、[Bundle 集群](/components/bundle)（bundlescan / bundlewrite 展开）、[Task 生命周期](/architecture/task-lifecycle)。