> ## Documentation Index
> Fetch the complete documentation index at: https://docs.chaintable.com/llms.txt
> Use this file to discover all available pages before exploring further.
# 业务抽象表
> 了解 BlockDB 普通表、业务抽象表、影子表机制和区块维度读取语义。
***
## blockdb的业务抽象表介绍
| 层 | 别名 | SDK 类 | 职责 |
| ------------- | ----------- | -------------------------- | ------------------------------------------ |
| **普通表**
| 存储层
| `Table` | 普通在线表，高吞吐读写。既存非链上业务数据，也作为业务抽象表的物理存储底座。 |
| **业务抽象表** | 业务抽象层
| `BlockTable` / `TimeTable` | 提供业务数据生产场景的高级语义API，提供共识高度等高级功能。**本身不存数据**。 |
```Python theme={null}
from blockdb import Table, BlockTable
# L1：普通在线表 —— 自己建一张简单小表 demo.rules，先写后读
rules = Table("demo.rules") # 建表走页面（SDK 不开放 Meta），这里直接用
# 写：upsert_rows，sync=True 等服务端确认；返回 job_id
rules.upsert_rows([
{"id": "rule_1", "name": "min_value", "threshold": 100},
{"id": "rule_2", "name": "max_value", "threshold": 999},
], sync=True)
# 读：按主键 / 批量 / 全表扫描
print(rules.get_row("rule_1")) # 单行
print(rules.batch_get_rows(["rule_1", "rule_2"])) # 批量
rows = rules.scan("SELECT * FROM demo.rules") # 小表全表扫描，再本地 for 循环计算
for r in rows:
... # 本地复杂计算逻辑
# 业务抽象：链上业务表 —— 按区块语义读
token = BlockTable("token.token.eth")
```
## **1.3 影子表机制（业务抽象表 → 普通表 自动映射）**
建一张业务抽象表时，BlockDB 自动在存储层创建多张协作的普通表：
| 业务抽象 类型 | 普通表 映射 | 例子 |
| ------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Block State 表**（state，随高度更新、保留历史快照） | - 主表（最新值）
- `_archive`（历史高度值，主键(id)）
- `_height`（处理进度/共识高度）
- `_bundle`（数据区间的hash表） | 
|
| **Block Event 表**（event，发生即不可变、只追加）
\*\*不维护 \*\***`_archive`** | - 主表（原始事件）、
- `_height`。
- `_bundle`（数据区间的hash表） | 
|
| **TimeTable** （自动按时间分桶） | - 主表（latest）
- `_archive`（历史，复合主键 `(id,time_at)`） | |
> 子表名是**下划线**前缀：`<业务抽象_name>._archive` / `<业务抽象_name>._height`。
* **\_height表**：id, block\_height\_start, block\_height\_end。每行是一个连续区块区间。可以看这个表排查处理过的区块范围，包括最新高度。排查问题使用，写代码几乎用不到
* **\_archive表**：和源表一样，只是源表id字段变original\_id，新增新的id字段
* \*\*\_bundle表：\*\*维护了每1000个block的数据hash和，主要给S3同步系统（未来可能有别的数据对比功能基于它）
## **1.4 BlockTable —— 区块维度**
按**区块高度**变化。读取自动注入 `height <= 目标高度`，保证不穿越共识。
* `get_event` / `get_state`：以 block 为锚点（None=最新）查事件 / 状态快照。\*\*必须传 \*\***`current_block`**
* `Block` = `{id, height, timestamp}`，是 业务抽象 对外数据单位。
* `get_block_rows(block)`：取某高度整块所有行。
* `upsert_block(block, rows)`：按区块**原子**写入。
* `upsert_block_bundle(bundle, rows)`：按 `bundle`（`int64` 高度区间标识，即1000个block）**原子**写入一批行，不携带单个 `Block` 元信息。如果rows里只有一个block，也认为1000个block全部推进
```Python theme={null}
from blockdb import BlockTable, Block
# 演示表（已建好，可直接跑）；block 三元组按表里真实数据 hardcode
EV = BlockTable("chaintabledev.zichao_业务抽象evt_test") # event：evt_h 在高度 h，共识=4
ST = BlockTable("chaintabledev.zichao_业务抽象state_test") # state：acct_A 在 h0=0、h3=300
# ── get_event：事件 height ≤ current_block.height 才可见 ──
blk2 = Block(id="0xaa4f6cff6a8181ef2e76113989a75c312f305db34539cd75ddb3093d8eb6aa3a",
height=2, timestamp=1700000002)
print(EV.get_event("evt_2", blk2)) # -> {'id':'evt_2','value':2.0,...} （h2 ≤ 2，命中）
# ── get_state：取 height ≤ 该高度 的最新快照（换高度 = 查历史快照）──
blk3 = Block(id="0x0c76b9c84c58c1d77cd4d6bd70c3cf27d7b7ae151a26c00403191f3be18114a0",
height=3, timestamp=1700000003)
print(ST.get_state("acct_A", blk3)) # -> value=300（h3 已更新）
blk2s = Block(id="0x7f865288162b3e5e49b2eefb5af1282e6e2d85fdff50049d2891295626bc27fe",
height=2, timestamp=1700000002)
print(ST.get_state("acct_A", blk2s)) # -> value=0 （≤2 最新还是 h0）
```
注意：`get_event` / `get_state` 必须传 `current_block`
* **关键：高度要达到共识。** `current_block.height` 必须 ≤ 表的**共识高度**（= `_height` 表从 **0 起连续**覆盖到的最高高度）
## **1.5 TimeTable —— 物理时间维度**
和 BlockTable 并列的另一种 业务抽象 模型：
* BlockTable 描述**区块维度**（某高度的状态 / 事件）
* TimeTable 描述某对象**随物理时间变化的状态序列**。表结构固定 `{id, time_at, value}`。
| | BlockTable | TimeTable |
| --- | ---------------- | ------------------------ |
| 维度 | 区块高度 | 物理时间（60s 桶，时间久的自动压缩可能更长） |
| 锚点 | `current_block` | `time_at` |
| 读语义 | 取 height ≤ 锚点的最新 | 取 time\_at ≤ 锚点的最近已记录桶 |
关键特性：
* **分桶**：时间桶 `time_at` 固定 60s，写「向后取整、秒归零」对齐（`17:04:35`→`17:05:00`），同一对象 60s 内只存 1 个点。
* **自动压缩、控制数据分布**：越久远的数据记录越稀疏，相邻两点数值变化不大时不重复记录——所以**写入前不用自己去重，也不用担心数据膨胀**。
* **读取的点可能是稀疏的**：读某时刻返回的是「≤ 该时刻、最近一个已记录桶」的值，越久远越稀疏（具体稀疏 / 压缩规则见 TimeTable 手册）。
以自建的 `token_time_demo` 表为例（存某 token 每分钟的价格）：
```Python theme={null}
from blockdb import TimeTable, Table
NAME = "token_time_demo"
ID = "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48" # 某 token
# ① 写 TimeTable（业务抽象 接口）：按时间桶 upsert，写入时间自动向后取整到整分钟
tt = TimeTable(NAME)
tt.upsert_time_rows(
"2026-05-18 17:04:35", # 自动归到 17:05:00 桶
[{"id": ID, "value": 2543.12}],
sync=True, # 等服务端确认写入完成
)
# ② 点查某表(如token表)最新 time 的值（L1 接口直接读 L1 latest 主表），例如token最新价格
latest = Table(NAME).get_row(ID) # 主表存的就是最新桶的值，batch_get_rows批量
print(latest) # -> {"id": ID, "time_at": "2026-05-18 17:05:00", "value": 2543.12}
# ③ 拿某个时刻的价格切面值（业务抽象 读接口，读也按桶对齐）
row = tt.get_value(ID, "2026-05-18 17:04:40") # 对齐到 17:04:00 桶
print(row)
# ④ 取一个时间区间内的值（读 L1 _archive 历史表）；filters 用 SQL-like 字符串
# 查询历史价格 （sdk未来会包一层）
archive = Table(f"{NAME}._archive")
series = archive.filter_rows(
filters=f"id = '{ID}' AND time_at >= '2026-05-18 00:00:00' AND time_at <= '2026-05-18 23:59:00'",
order_by="time_at asc", limit=0,
)
print(len(series), "buckets")
```