> ## 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.

# Call Builder 与 Writer Plugin

> task 前后两端的可插拔阶段：三类 Call Builder 决定输入从哪来、展开成哪些 call；三类 Writer Plugin 决定 outputs 写到哪。本页给出每一类的语义、配置形状与组合方式。

task 通过两个 `(type, config)` 对声明前后两端：`functionCallConfig` 选一个 Call Builder，`resultHandler` 选一个 Writer Plugin（blockx-py 里叫 Handler）。`type` 必须是当前 Worker 进程注册过的名字；`config` 由对应实现自己解析，Worker 在提交时只校验它是合法 JSON。

## Call Builder

| `functionCallConfig.type` | 实现         | 输入                                                           | 每个 call 的参数                           |
| ------------------------- | ---------- | ------------------------------------------------------------ | ------------------------------------- |
| `BlockTableCallConfig`    | dbScan     | 指定区块上、触发表的所有行（BlockDB `GetBlockRows`），可先经 `operator` 过滤、去重   | 按 `params` 模板拼装，`"${表名}"` 的位置换成行 dict |
| `InputsCallConfig`        | callList   | task 自带的 `callList`                                          | 每行一个 call，行本身就是参数数组                   |
| `BlockBundleCallConfig`   | bundleScan | 触发表在某个 bundle 分区里的 parquet 行（DuckDB 流式读取，`operator` 下推为 SQL） | 按 `param` 模板拼装，规则同上                   |

### BlockTableCallConfig

```json theme={null}
{
  "block": { "id": "0x…", "height": 18000000, "timestamp": "2024-06-08T00:00:00Z" },
  "triggerSources": [
    {
      "table": "chain.trace.eth",
      "functionId": "",
      "code": { "sourceCode": "def _(chain_id, record):\n    ...\n" },
      "params": ["eth", "${chain.trace.eth}"],
      "operator": { "relations": [] }
    }
  ]
}
```

* 一个 task 可以有多个 `triggerSources`，各自独立扫描后合并成一份 call 列表。
* `functionId` 与 `code.sourceCode` 二选一：前者是注册函数，后者是 inline 源码。
* `params` 为空时整行作为唯一参数；`params` 非空但没有 `"${表名}"` 占位符时，所有 call 共享同一份静态参数，因此只会执行一次。
* `operator` 可选，省略时触发表在该区块上的每一行都生成一个 call。

### InputsCallConfig

```json theme={null}
{
  "function": "",
  "code": { "sourceCode": "def _(name):\n    ...\n" },
  "callList": [["ct2"], ["blockx"]]
}
```

* 不做任何 IO。`callList[i]` 直接成为第 i 个 call 的参数数组。
* 字段名是 `function`，不是 `functionId`；blockx-py 会自动做这层重命名。
* 结果按完成序汇聚，不保证与 `callList` 的行序一致。需要对应关系时，让返回值带上入参。

### BlockBundleCallConfig

```json theme={null}
{
  "blockBundle": { "number": 25001 },
  "triggerSources": [
    {
      "tableId": "chain.trace.op",
      "function": "",
      "code": { "sourceCode": "def _(chain_id, record):\n    ...\n" },
      "param": ["op", "${chain.trace.op}"],
      "operator": { "relations": [] }
    }
  ]
}
```

* bundle 是按 1000 个区块为一段切分的数据分区，由 `blockBundle.number` 标识。
* `tableId` 只报源表身份。Worker 按 `(tableId, number)` 从该表的 Iceberg 快照解析这个 bundle 的数据文件；state 表经 `<tableId>._archive` 解析。
* 字段名是 `function` / `param`（单数），与 `BlockTableCallConfig` 的 `functionId` / `params` 不同。
* bundle worker 默认开启 stream build：扫描与执行流水线化，百万行也不必先物化整份 call 列表。

### Operator：过滤与去重

`operator` 是一个 Substrait 风格的 `relations` 链：恰好一个 `read`，后接任意个 `filter` 和 `deduplicate`，用 `input.relation_id` 串成线性链。

* `filter` 支持 `and`、`or`、`not`、`equal`、`not_equal`、`gt`、`gte`、`lt`、`lte`、`is_null`、`is_not_null`，按 SQL 三值逻辑求值，只保留结果为 `TRUE` 的行。
* `deduplicate` 按指定列去重，key 用带类型标签的确定性编码。
* dbScan 在 Worker 内存里执行这条链；bundleScan 把它翻译成 SQL 下推给 DuckDB。
* blockdb-py 的 `filter()` / `Operator` 直接产出这个结构，可以原样放进 `operator`。

格式细节见 blockx 仓库 `docs/specs/substrait_ast.md`。

## Writer Plugin

| `resultHandler.type`      | 目标                  | `config`                                                                                              | 注册在                                                |
| ------------------------- | ------------------- | ----------------------------------------------------------------------------------------------------- | -------------------------------------------------- |
| `BlockTableWriteHandler`  | BlockDB block 表（L2） | block 集群：`{targetTable, block{id, height, timestamp}}`；bundle 集群：`{targetTable, blockBundle{number}}` | 两套集群，实现按部署形态装配                                     |
| `NormalTableWriteHandler` | BlockDB 普通表（L1）     | `{targetTable, condition?: {column, policy}}`                                                         | 两套集群；block 集群直接 `UpsertRows`，bundle 集群走 BatchWrite |
| `ReturnValueHandler`      | 回传给 Client          | `{}`                                                                                                  | 仅 block 集群                                         |

共同行为：

* 写表插件先读目标表的列定义，只投影返回 dict 中属于目标表的列；`outputs` 里的 `null` 被跳过。
* 任意 handler 的 `config` 里带 `"debug": true` 时，写表插件只统计行数并打 `table_write_debug` 日志，不读 schema、不发写请求。blockx-py 用环境变量 `BLOCKDB_DEBUG=1` 打开它。
* 不带 `resultHandler` 也是合法的：Calls 阶段成功后 task 直接进入终态，`outputs` 被丢弃。
* Writer Plugin 是 fail-fast：写入失败即 task `PLUGIN_FAILED`，`retryable` 由插件的错误分类决定。

### BlockTableWriteHandler

* 一个名字对应两套实现，由 Worker 进程的部署形态（`block` / `bundle`）在启动时装配，调用方不能选。
* block 集群的实现要求 `config.block`，一次 task 一次写入，返回 `{written_rows, finish_time}`。
* bundle 集群的实现要求 `config.blockBundle`，经 BlockDB BatchWrite job（Init → presigned PUT → Commit）覆盖写目标表的同一个 bundle。
* 形状不匹配时在 Writer 阶段以 `PLUGIN_FAILED` 失败（`block deployment requires config.block` / `bundle deployment requires config.blockBundle`），此时 call 已经全部执行过。blockx-py 在提交前先校验形状与目标集群，避免白算一遍。

### NormalTableWriteHandler

* 按主键 `id` upsert。
* `condition` 可选：`{"column": "ts", "policy": "UpdateIfSmaller"}` 表示只有新值在该列上更小才覆盖；`policy` 取 `UpdateIfSmaller` 或 `UpdateIfLarger`。乱序或重放写入下天然幂等。
* 不带区块上下文，同一份配置可以在实时和回填两条路径里复用。

### ReturnValueHandler

* 把 `outputs` 原样拼成一个 JSON 数组放进 `pluginResults[i].result`。blockx-py 里读 `result.handler_result`。
* bundle worker 没有注册它，带它的 task 提交到 bundle 集群会在 `SubmitTask` 时被 `InvalidArgument` 拒绝；blockx-py 因此把这类 task 一律路由到 block 集群。

## 组合速查

| 场景                      | Call Builder            | Writer Plugin                           | 集群                   |
| ----------------------- | ----------------------- | --------------------------------------- | -------------------- |
| 实时：每个新区块算一次，写 block 表   | `BlockTableCallConfig`  | `BlockTableWriteHandler`（`block`）       | block                |
| 实时：每个新区块算一次，写普通表        | `BlockTableCallConfig`  | `NormalTableWriteHandler`               | block                |
| 历史回填：按 bundle 写 block 表 | `BlockBundleCallConfig` | `BlockTableWriteHandler`（`blockBundle`） | bundle               |
| 历史回填：按 bundle 写普通表      | `BlockBundleCallConfig` | `NormalTableWriteHandler`               | bundle               |
| 一次性计算，把结果拿回来            | `InputsCallConfig`      | `ReturnValueHandler`                    | block                |
| 一批参数算完写普通表              | `InputsCallConfig`      | `NormalTableWriteHandler`               | bundle（blockx-py 默认） |

## 小结

* Call Builder 决定输入：`BlockTableCallConfig` 读区块、`InputsCallConfig` 用自带参数、`BlockBundleCallConfig` 读 bundle。
* Writer Plugin 决定结果去向：`BlockTableWriteHandler` 写 L2、`NormalTableWriteHandler` 写 L1、`ReturnValueHandler` 回传。
* `BlockTableWriteHandler` 的 `config` 形状必须跟集群走：block 集群带 `block`，bundle 集群带 `blockBundle`。
* `operator` 在 Builder 阶段过滤、去重触发行，减少 call 数量。

继续阅读：

* [Plugin 系统](/components/plugins)：接口、注册表、装配与扩展一个新的 Builder / Plugin。
* [Bundle 集群](/components/bundle)：bundleScan 流式扫描与 BatchWrite 写路径。
* [提交 task 示例](/development/submit-task-example)：每一种组合的 blockx-py 写法。