T25 · Blockchain Data Infrastructure

物理分割キー → 日付（ブロック時刻からUTCで導出）
区画内の順序 → ブロック高、次にlog_index

warehouse/
  raw_logs/
    dt=2024-03-01/part-0000.parquet
    dt=2024-03-02/part-0000.parquet
  decoded_transfers/
    dt=2024-03-01/part-0000.parquet
  curated_daily_metrics/
    dt=2024-03-01/part-0000.parquet

1. 区画全体を一時ディレクトリへ書く
   warehouse/decoded_transfers/_tmp_2024-03-01_run42/
2. 書き込みと検証後、正式名へ原子的に変更
   → warehouse/decoded_transfers/dt=2024-03-01/
3. manifestへ高さ区間、行数、checksum、decoder版を記録

create table warehouse_manifest (
  dataset        text    not null,      -- raw_logs / decoded_transfers / ...
  chain_id       integer not null,
  dt             date    not null,      -- 分割キー
  from_block     bigint  not null,      -- 区画が実際に含む高さ区間
  to_block       bigint  not null,
  row_count      bigint  not null,
  checksum       text    not null,      -- 再実行比較用の内容指紋
  decoder_version text   not null,      -- 解読コード版。由来の鍵
  status         text    not null,      -- pending / running / done / failed
  attempts       integer not null default 0,
  built_at       timestamptz not null default now(),
  primary key (dataset, chain_id, dt)
);

-- 1. 欠落はあるか？
select dt from generate_series($from::date, $to::date, '1 day') g(dt)
 where not exists (
   select 1 from warehouse_manifest m
    where m.dataset = 'decoded_transfers' and m.chain_id = $chain
      and m.dt = g.dt and m.status = 'done'
 );

-- 2. 隣接日の高さ区間は連続しているか？
select a.dt, a.to_block, b.dt, b.from_block
  from warehouse_manifest a
  join warehouse_manifest b
    on b.dataset = a.dataset and b.chain_id = a.chain_id
   and b.dt = a.dt + 1
 where b.from_block <> a.to_block + 1;

-- 3. 古いdecoderで作った再実行対象の区画は？
select dt from warehouse_manifest
 where dataset = 'decoded_transfers' and decoder_version <> $current;

-- 4. 再実行で内容が変わったか？区画ごとにchecksumを比較

業務DB（OLTP） ← 先端を追い、リアルタイムと再編を担当
        │
        │ block_numberがsafeHeight以下のものだけ移す
        ↓
倉庫（OLAP）   ← 数ブロック遅れ、保存後は不変

- 送信者だけか、送受信の両方か？
- 失敗取引を含むか？
- コントラクトを含むか？
- aggregator経由では中継者か最終利用者か？
- 一人の五アドレスは一人か五つか？
- UTCか現地時間か？

-- daily_active_addresses
-- 定義（2024-03確定）：
--   1. 成功取引のログだけを集計
--   2. fromとtoの両方を一度ずつ重複排除
--   3. mint/burnの相手であるzero addressを除外
--   4. 既知コントラクトは除外せず別途印を付ける
--   5. UTC暦日
create table curated_daily_active (
  chain_id   integer not null,
  dt         date    not null,
  contract   bytea   not null,
  addresses  bigint  not null,
  tx_count   bigint  not null,
  primary key (chain_id, dt, contract)
);

-- 12か月の月間アクティブアドレスと手数料
-- dtのpruningで対象12か月だけを読む
with e as (
  select dt, from_addr, to_addr, amount, gas_used, gas_price
    from decoded_transfers
   where chain_id = 1
     and contract = $contract
     and dt >= date '2024-01-01'
     and dt <  date '2025-01-01'
),
addrs as (
  select date_trunc('month', dt) as m, from_addr as addr from e
  union all
  select date_trunc('month', dt) as m, to_addr   as addr from e
)
select m,
       count(distinct addr)                      as active_addresses,
       sum(gas_used * gas_price) / 1e18          as fees_native
  from addrs
  join e on date_trunc('month', e.dt) = addrs.m
 group by m
 order by m;