turbo-transforms

Turbo Transforms

Turbo パイプライン設定向けの SQL、TypeScript、動的テーブル変換を作成、理解、デバッグします。

ユーザーのニーズ（デコード、フィルタリング、再構成、結合、独自ロジック、ルックアップ結合）を特定し、以下の関連セクションを使用してください。変換スニペットだけでなく、パイプライン全体の YAML を生成する場合は、ユーザーに提示する前に必ず goldsky turbo validate で検証してください。

専門トピックのリファレンスファイル：

• references/evm-patterns.md — 高度な EVM パターン（1 回デコードして複数条件でフィルタリング、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 シンクを使用します
• ウィンドウ関数（OVER、PARTITION BY、ROW_NUMBER）
• サブクエリ — 代わりに変換を連鎖させます
• CTE（WITH...AS）はサポートされています

_gs_op 列

すべてのレコードには、操作の種類を示す _gs_op 列が含まれます：'i'（挿入）、'u'（更新）、'd'（削除）。データベースシンクで正しいアップサートのセマンティクスを維持するため、この列を変換後も保持してください。

---

主な SQL 関数

よく使われる関数のクイックリファレンスです。関数の全リファレンスについては、ドキュメントのSQL transformsを参照してください。

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 を 1 つの配列に含めてください。効率化のため、ソースレベルでコントラクトアドレスによる事前フィルタリングを行ってください。

その他の主な関数

関数	用途
fetch_abi(url, format)	URL から ABI を取得（'raw' または 'etherscan' 形式）
_gs_keccak256(string)	Keccak256 ハッシュ（0x 接頭辞付きの 16 進数を返します）
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...')

変換の連結

各変換が前の変換から読み取る、複数ステップのパイプラインを構築します。

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 — 複数のイベントタイプの結合

スキーマが同一の変換を結合し、単一の出力にまとめます。

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

すべての分岐で、同じ順序の同じ列を指定する必要があります。プレースホルダーとして '' または 0 を使用してください。

高度な EVM パターン（1 回のデコードで複数条件のフィルタリング、異なるイベントの正規化、複数イベントのアクティビティフィード）については、references/evm-patterns.md をお読みください。

---

ソースレベルのフィルタリング

効率化のため、データが変換に到達する前にソースでフィルタリングします。

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: → 粗い事前フィルタリング（コントラクトアドレス、ブロック範囲）
• 変換の WHERE → 細かいフィルタリング（イベントタイプ、パラメーター値、除外条件）

---

シンクのバッチ処理の設定

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

---

変換のデバッグ

「Unknown source reference」 — FROM 句の名前が、どのソースまたは変換のキーとも一致していません。入力ミスがないか確認してください。

「Missing primary_key」 — すべての変換に 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 スクリプト変換

カスタム解析、BigInt 演算、状態を保持する処理など、SQL では表現できないロジックに使用します。スキーマ型：string、uint64、int64、float64、boolean、bytes。

主なルール：function invoke(data) を定義し、フィルタリングする場合は null を返し、スキーマに一致するオブジェクトを返します。async/await や外部インポートは使用できません。

ドキュメント、例、SQL とスクリプトの選択表については、references/typescript-transforms.md を参照してください。HTTP エンリッチメント用の Handler 変換も含まれています。

---

動的テーブル変換

再デプロイせずに、許可リスト、ブロックリスト、結合形式のエンリッチメントに使用できる、更新可能なルックアップテーブルです。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／16 進数関数の使用方法については、references/solana-patterns.md を参照してください。

---

関連項目

• /turbo-builder — これらの変換を使って、対話形式でパイプラインを構築・デプロイします
• /turbo-doctor — パイプラインの問題（変換エラーを含む）を診断・修正します
• /turbo-pipelines — パイプラインの YAML 設定とアーキテクチャのリファレンス
• /turbo-operations — ライフサイクルコマンドとモニタリングのリファレンス
• /datasets — ブロックチェーンデータセットとチェーンプレフィックスのリファレンス