turbo-transforms

Turbo 转换

为 Turbo 流水线配置编写、理解和调试 SQL、TypeScript 及动态表转换。

先确定用户的需求（解码、筛选、重塑、合并、自定义逻辑或查找联接），然后使用下方相关章节。如果要生成完整的流水线 YAML（而不仅是转换代码片段），务必先运行 goldsky turbo validate 进行验证，再呈现给用户。

专业主题参考文件：

• references/evm-patterns.md — 高级 EVM 模式（一次解码、多次筛选、UNION ALL、架构规范化）
• references/typescript-transforms.md — TypeScript/WASM 脚本转换和处理器转换
• references/dynamic-tables.md — 动态表转换（允许列表、查找联接）
• references/solana-patterns.md — Solana 指令/日志解码和函数示例

---

转换基础

YAML 结构

transforms:
  my_transform:
    type: sql
    primary_key: id
    sql: |
      SELECT id, block_number, address
      FROM my_source
      WHERE address = '0xabc...'

必需字段

字段	必需	描述
type	是	sql、script、handler、dynamic_table 或 throttle
primary_key	是*	用于保证唯一性和排序的列（*不适用于 throttle）
sql	是	SQL 查询（适用于 sql 类型的转换）

引用数据

• 通过 YAML 键名引用源：FROM my_source
• 通过 YAML 键名引用其他转换：FROM my_transform
• SQL 转换无需 from 字段 — SQL 中的 FROM 子句负责处理

SQL 流式处理限制

Turbo SQL 由 Apache DataFusion 驱动，采用流式模式。以下功能不受支持：

• 联接 — 查找式联接请改用 dynamic_table 转换
• 聚合（GROUP BY、COUNT、SUM、AVG）— 请改用 postgres_aggregate sink
• 窗口函数（OVER、PARTITION BY、ROW_NUMBER）
• 子查询 — 请改用转换链
• 支持 CTE（WITH...AS）

_gs_op 列

每条记录都包含一个 _gs_op 列，用于跟踪操作类型：'i'（插入）、'u'（更新）、'd'（删除）。请在转换过程中保留此列，以确保数据库 sink 的 upsert 语义正确。

---

关键 SQL 函数

常用函数速查。完整函数参考请参阅文档中的 SQL 转换。

evm_log_decode() — 解码原始 EVM 日志

evm_log_decode(abi_json, topics, data) -> STRUCT<name, event_params>
-- Aliases: _gs_log_decode, evm_decode_log

返回 decoded.event_signature（事件名称）和 decoded.event_params[N]（从 1 开始索引、以字符串形式表示的参数）。

示例：

SELECT
  _gs_log_decode('[{"anonymous":false,"inputs":[{"indexed":true,"name":"from","type":"address"},{"indexed":true,"name":"to","type":"address"},{"indexed":false,"name":"value","type":"uint256"}],"name":"Transfer","type":"event"}]',
    topics, data) AS decoded,
  id, block_number, transaction_hash, address, block_timestamp
FROM my_raw_logs

提示： 为保留字添加反引号转义（`data`、`decoded`）。将所有事件 ABI 都包含在同一个数组中。为提高效率，请在源级别按合约地址进行预筛选。

其他关键函数

函数	用途
fetch_abi(url, format)	从 URL 获取 ABI（格式为 'raw' 或 'etherscan'）
_gs_keccak256(string)	Keccak256 哈希（返回带 0x 前缀的十六进制值）
xxhash(string)	用于复合键的快速非加密哈希
to_u256(value) / u256_to_string(u256)	256 位整数运算（避免精度损失）
dynamic_table_check(table, value)	检查值是否存在于动态查找表中
to_timestamp(seconds)	将 Unix 秒数转换为时间戳

---

常见转换模式

表别名

SELECT balance.token_type, balance.owner_address,
  CAST(`balance` AS STRING) AS balance_amount
FROM my_balances_source balance
WHERE balance.token_type = 'ERC_20' AND balance.balance IS NOT NULL

简单筛选

transforms:
  usdc_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT * FROM base_transfers
      WHERE address = lower('0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913')

列投影和别名

transforms:
  clean_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT id, block_number, to_timestamp(block_timestamp) AS block_time,
        address AS token_address, sender, recipient, amount
      FROM my_source

类型转换和数值缩放

-- Token amount conversions
(CAST(decoded.event_params[3] AS DOUBLE) / 1e6) AS amount_usdc    -- 6 decimals
(CAST(decoded.event_params[3] AS DOUBLE) / 1e18) AS amount_eth    -- 18 decimals
CAST(`balance` AS STRING) AS balance_amount                        -- preserve precision

代币	小数位数	除数
USDC、USDT	6	1e6
WBTC	8	1e8
ETH、大多数 ERC-20	18	1e18

使用 CASE WHEN 的条件逻辑

CASE WHEN decoded.event_params[4] = '0' THEN 'BUY' ELSE 'SELL' END AS side

排除筛选

WHERE decoded.event_params[1] NOT IN ('0x4bfb...', '0xc5d5...', '0x4d97...')

Transform 链式处理

构建多步管道，让每个 transform 从上一步读取数据：

transforms:
  decoded:
    type: sql
    primary_key: id
    sql: |
      SELECT _gs_log_decode('[...]', topics, data) AS decoded,
        id, block_number, transaction_hash, address, block_timestamp
      FROM raw_logs_source
  transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT id, block_number, decoded.event_params[1] AS from_address,
        decoded.event_params[2] AS to_address,
        (CAST(decoded.event_params[3] AS DOUBLE) / 1e18) AS amount
      FROM decoded
      WHERE decoded.event_signature = 'Transfer'

UNION ALL — 合并多种事件类型

将架构相同的 transform 合并为单个输出：

  all_events:
    type: sql
    primary_key: id
    sql: |
      SELECT * FROM transfers
      UNION ALL
      SELECT * FROM approvals

所有分支的列必须相同且顺序一致。使用 '' 或 0 作为占位符。

有关高级 EVM 模式（解码一次、筛选多次；规范化不同事件；多事件活动流），请阅读 references/evm-patterns.md。

---

源级筛选

在数据到达 transform 之前 在源端 筛选数据，以提高效率：

sources:
  my_logs:
    type: dataset
    dataset_name: matic.raw_logs
    version: 1.0.0
    start_at: earliest
    filter: >-
      address IN ('0xabc...', '0xdef...')
      AND block_number >= 82422949

• 源 filter: → 粗粒度预筛选（合约地址、区块范围）
• Transform WHERE → 精细筛选（事件类型、参数值、排除项）

---

Sink 批处理配置

sinks:
  my_sink:
    type: clickhouse
    from: my_transform
    table: my_table
    secret_name: MY_SECRET
    primary_key: id
    batch_size: 1000
    batch_flush_interval: 300ms

用例	batch_flush_interval	batch_size
实时仪表板	300ms	1000
中等吞吐量	1000ms	1000
高吞吐量流式处理	10s	100000

---

调试 Transforms

“Unknown source reference” — FROM 子句名称与任何源或 transform 键都不匹配。检查是否有拼写错误。

“Missing primary_key” — 每个 transform 都需要 primary_key。几乎总是使用 id。

“Column not found” — 使用 goldsky turbo inspect <pipeline> -n <source_node> -p 查看实际列。

解码事件的结果为空：

• 确认 ABI JSON 与实际合约事件匹配
• 检查 decoded.event_signature 是否与事件名称完全一致（区分大小写）
• 确保 address 筛选条件匹配正确的合约
• 参数从 1 开始索引：decoded.event_params[1]，而不是 [0]

UNION ALL 中的类型不匹配 — 所有分支的列数必须相同，且类型兼容。

goldsky turbo inspect <pipeline-name> -n <transform_name> -p

---

TypeScript / WASM 脚本 Transforms

适用于 SQL 无法表达的逻辑：自定义解析、BigInt 运算、有状态处理。架构类型：string、uint64、int64、float64、boolean、bytes。

关键规则：定义 function invoke(data)，返回 null 以筛选数据，返回与架构匹配的对象，不使用 async/await 或外部导入。

完整文档、示例以及 SQL 与脚本的选择表，请参阅 references/typescript-transforms.md。其中还包含用于 HTTP 数据增强的 Handler transforms。

---

动态表 Transforms

用于允许列表、阻止列表和类似联接的数据增强的可更新查找表，无需重新部署。底层使用 PostgreSQL（持久化）或内存（快速）。在 SQL 中使用 dynamic_table_check('table_name', column)。

完整配置、后端选项、REST API 更新和钱包跟踪示例，请参阅 references/dynamic-tables.md。

---

节流转换（流量控制）

通过将记录缓冲成批次，并按固定的最小间隔发送每个批次，限制流的吞吐量。节流不会修改数据——每条输入记录都会原样传递。可用于：

• 避免超过下游接收器或外部 API 的速率限制
• 将突发式数据源平滑为稳定、可预测的速率
• 向 HTTP handler 按节奏发送记录，避免接收服务不堪重负
• 在开发或测试期间减慢处理速度

transforms:
  throttled:
    type: throttle
    from: my_source       # source or transform
    max_batch_size: 100
    min_batch_interval: 10s

字段	必填	说明
type	是	throttle
from	是	要读取的源或转换
max_batch_size	否	每批次的最大记录数
min_batch_interval	否	批次之间的最小间隔（例如 10s、500ms、1m）

只有在累积了 max_batch_size 条记录且自上一批次以来已过 min_batch_interval 后，批次才会向下游发送。有效最大吞吐量约为每秒 max_batch_size / min_batch_interval 条记录。

最佳实践：

• 将节流放在靠近瓶颈的位置（就在受速率限制的接收器/处理器之前），这样上游转换仍可全速处理。
• 调整 max_batch_size，使其符合下游接收器偏好的批次大小。
• 节流是一种 transform——像使用其他转换一样，将其他转换串接在它的下游（例如，在接收器中使用 from: throttled）。
• 节流没有 primary_key 字段（记录会原样传递）。
• 尽可能在生产环境中移除节流；它会按设计限制吞吐量。

Solana 转换模式

基于 IDL 的指令解码、特定程序解码器（Token、System、Stake、Vote）、SPL 代币跟踪。

有关所有内置解码器、完整管道示例以及数组/JSON/十六进制函数的用法，请参阅 references/solana-patterns.md。

---

相关内容

• /turbo-builder — 使用这些转换以交互方式构建和部署管道
• /turbo-doctor — 诊断并修复管道问题（包括转换错误）
• /turbo-pipelines — 管道 YAML 配置和架构参考
• /turbo-operations — 生命周期命令和监控参考
• /datasets — 区块链数据集和链前缀参考