# 8 - 电商问数：表与字段信息同步到元数据库

---

**本章课程目标：**

- 理解为什么元数据知识库的第一步，一定是先把表信息和字段信息沉淀到 `Meta MySQL`。
- 看懂“配置描述 -> 业务实体 -> DW 补齐真实信息 -> Meta MySQL 落库”这条完整链路。
- 建立清晰的分层认识：`Service` 负责组织流程，`DW Repository` 负责查询真实结构，`Meta Repository` 负责落库，`Mapper` 负责对象转换。

**学习建议：** 这一章先抓住一条落库链路：配置文件描述表字段，系统去数仓补齐真实结构，再转换成内部实体，最后写入 Meta MySQL。ORM、Repository、事务这些细节都服务于这条链路。读代码时重点看哪些数据来自配置，哪些数据来自 DW，哪些数据最终沉淀到元数据库。

**对应代码分支：** `08-metadata-table-column-sync`

---

上一章我们已经把“元数据知识库构建”的入口说明白了：脚本如何启动、配置如何加载、服务层如何接过 `config_path` 并进入真正的业务流程。

从这一章开始，我们正式进入第一条真正落地的业务链路：**把配置文件里描述的表和字段，转换成系统内部统一的数据对象，并写入 `Meta MySQL`。**

这一步很基础，但也非常关键。因为后面的字段向量索引、字段取值全文索引、SQL 生成上下文组织，全部都建立在这批结构化元数据之上。也就是说，**只有表信息和字段信息先稳定落下来，后面的检索和问数链路才有可靠的基础。**

---

## 1、本章在整条构建流程中的位置

回到 `MetaKnowledgeService.build(config_path)`，表链路对应的是下面这一段：

项目对应文件路径：`shopkeeper-agent/app/services/meta_knowledge_service.py`

```python
async def build(self, config_path: Path):
    context = OmegaConf.load(config_path)
    schema = OmegaConf.structured(MetaConfig)
    meta_config: MetaConfig = OmegaConf.to_object(OmegaConf.merge(schema, context))

    if meta_config.tables:
        column_infos = await self._save_tables_to_meta_db(meta_config)
        await self._save_column_info_to_qdrant(column_infos)
        await self._save_value_info_to_es(meta_config, column_infos)
```

只要配置里存在 `tables`，系统就会依次做 3 件事：

1. 先把表信息和字段信息写入 `Meta MySQL`
2. 再根据 `column_infos` 为字段建立向量索引
3. 最后根据配置中的 `sync` 标记，为部分字段建立字段值全文索引

本章只讲第一步，也就是：

```python
column_infos = await self._save_tables_to_meta_db(meta_config)
```

可以把这一章看作：**先把“字段是什么”这件事讲清楚，后面再讲“字段怎么被检索”。**

---

## 2、先抓主线：\_save_tables_to_meta_db(...) 到底做了什么

进入细节之前，先看本章最核心的方法。

项目对应文件路径：`shopkeeper-agent/app/services/meta_knowledge_service.py`

```python
async def _save_tables_to_meta_db(self, meta_config: MetaConfig) -> list[ColumnInfo]:
    table_infos: list[TableInfo] = []
    column_infos: list[ColumnInfo] = []

    for table in meta_config.tables:
        table_info = TableInfo(
            id=table.name,
            name=table.name,
            role=table.role,
            description=table.description,
        )
        table_infos.append(table_info)

        column_types: dict[str, str] = await self.dw_mysql_repository.get_column_types(
            table.name
        )
        for column in table.columns:
            column_values: list = await self.dw_mysql_repository.get_column_values(
                table.name, column.name, 10
            )
            column_info = ColumnInfo(
                id=f"{table.name}.{column.name}",
                name=column.name,
                type=column_types[column.name],
                role=column.role,
                examples=column_values,
                description=column.description,
                alias=column.alias,
                table_id=table.name,
            )
            column_infos.append(column_info)

    async with self.meta_mysql_repository.session.begin():
        await self.meta_mysql_repository.save_table_infos(table_infos)
        await self.meta_mysql_repository.save_column_infos(column_infos)

    return column_infos
```

它做的事情其实非常清楚：

1. 遍历配置中的每张表，先构造 `TableInfo`
2. 去 `DW` 查询这张表所有字段的真实类型
3. 再遍历这张表下的每个字段，查询一部分真实取值作为示例
4. 构造 `ColumnInfo`
5. 开启事务，把表信息和字段信息统一写入 `Meta MySQL`

所以后面整章的内容，本质上就是把这 5 步一层层拆开来看。

---

## 3、第一步：配置里的表和字段，如何变成业务实体

这一层最重要的认知是：**配置文件不是直接入库的，配置文件会先被转换成业务实体。**

如果你对“配置文件、业务实体、ORM 模型、mappers”这几类角色的区别还不够稳，这里建议先回看第 7 章前面的概念说明。第 8 章从这里开始，默认你已经先接受一个前提：

- 配置文件负责描述“这次要处理什么”
- 业务实体负责表示“系统内部怎么统一表达这些数据”
- `Mapper` 负责在业务实体和 ORM 模型之间做转换
- ORM 模型负责把数据映射到数据库表

也就是说，这一章不是把 `YAML` 逐行塞进数据库，而是先把配置中的描述组织成系统内部统一的数据对象。

### 3.1 先看配置长什么样

项目对应文件路径：`shopkeeper-agent/conf/meta_config.yaml`

```yaml
tables:
  - name: dim_region
    role: dim
    description: 地区维度表，用于描述订单发生的地理区域信息。
    columns:
      - name: region_id
        role: primary_key
        description: 地区唯一标识。
        alias: [地区ID, 区域ID]
        sync: false

      - name: region_name
        role: dimension
        description: 订单所属的大区名称，如华东、华南等。
        alias: [地区, 区域, 大区]
        sync: true
```

这份配置主要负责告诉系统两件事：哪些表、哪些字段要进入元数据知识库；这些表和字段的业务语义是什么。

这里能看到的内容，大多是业务描述信息，比如：表角色 `role`，字段角色 `role`，字段说明

### 3.2 再看系统内部统一的数据对象

项目对应文件路径：

- `shopkeeper-agent/app/entities/table_info.py`
- `shopkeeper-agent/app/entities/column_info.py`

```python
@dataclass
class TableInfo:
    id: str
    name: str
    role: str
    description: str


@dataclass
class ColumnInfo:
    id: str
    name: str
    type: str
    role: str
    examples: list[Any]
    description: str
    alias: list[str]
    table_id: str
```

这里的 `TableInfo` 是系统内部流转的表元数据对象，`ColumnInfo` 则是系统内部流转的字段元数据对象。

它们还不是 ORM 模型，而是更轻量、更偏业务语义的数据载体。

### 3.3 为什么 ColumnInfo.id 不是直接用字段名

对应代码：

```python
id=f"{table.name}.{column.name}"
```

原因很简单：真实数仓里会有大量同名字段。例如很多表里都会出现：`id`、`customer_id`、`product_id`、`date_id`等同名字段。

如果只用字段名做唯一标识，不同表中的同名字段很快就会冲突。所以当前项目把字段 ID 设计成：**表名 + 点号 + 字段名**

例如：

- `fact_order.order_amount`
- `dim_customer.customer_id`
- `dim_product.category`

这样字段在系统内部才具备稳定、明确的唯一性。

### 3.4 为什么 type 和 examples 不是直接从配置里来

看 `ColumnInfo` 的构造代码：

```python
column_info = ColumnInfo(
    id=f"{table.name}.{column.name}",
    name=column.name,
    type=column_types[column.name],
    role=column.role,
    examples=column_values,
    description=column.description,
    alias=column.alias,
    table_id=table.name,
)
```

这里很值得注意：

- `role`、`description`、`alias` 来自配置
- `type`、`examples` 来自运行时查询

这背后对应的是一个很重要的工程原则：**能从系统自动获取的真实信息，就尽量不要让用户手工重复维护。**

字段类型本来就是数据库表结构的一部分，示例值本来就存在于真实数据里。如果这些内容也要求用户在配置里手填，会带来两个明显问题：维护成本高，很容易和真实数仓脱节。所以配置文件主要负责描述**业务语义**，而程序运行时负责补齐**真实结构和真实样本**。

---

## 4、第二步：为什么要专门有一个 DW Repository

前面已经看到，字段类型和字段示例值都不是配置里直接给出的，而是运行时从数仓中查出来的。
这部分职责，就落在 `DWMySQLRepository` 上。

项目对应文件路径：`shopkeeper-agent/app/repositories/mysql/dw/dw_mysql_repository.py`

```python
async def get_column_types(self, table_name: str) -> dict[str, str]:
    sql = f"show columns from {table_name}"
    result = await self.session.execute(text(sql))
    return {row.Field: row.Type for row in result.fetchall()}


async def get_column_values(self, table_name: str, column_name: str, limit: int):
    sql = f"select distinct {column_name} from {table_name} limit {limit}"
    result = await self.session.execute(text(sql))
    return result.scalars().fetchall()
```

这两个方法分别做两件事：

- `get_column_types(table_name)`：查一张表所有字段的真实类型
- `get_column_values(table_name, column_name, limit)`：查某个字段的一部分真实取值

### 4.1 为什么字段类型是一张表查一次

先看这行代码：

```python
column_types = await self.dw_mysql_repository.get_column_types(table.name)
```

它放在外层表循环里，是因为字段类型的查询粒度是“按表”。

例如查询 `fact_order`，返回结果可能类似：

```python
{
    "order_id": "varchar(30)",
    "customer_id": "varchar(20)",
    "order_amount": "float",
}
```

这样后面遍历字段时，就可以直接通过：

```python
column_types[column.name]
```

拿到当前字段的真实类型。

### 4.2 为什么字段示例值要一列一查

再看这行代码：

```python
column_values = await self.dw_mysql_repository.get_column_values(
    table.name, column.name, 10
)
```

它放在内层字段循环里，是因为示例值的查询粒度是“按字段”。

例如：

```sql
select distinct category from dim_product limit 10
```

可能返回：

```python
["饮料", "休闲食品", "粮油调味", "家居清洁"]
```

这些值最终会写入：

```python
examples=column_values
```

写进 `ColumnInfo.examples`。

这里要特别注意一点：**这一章拿到的是“示例值”，不是该字段的全部取值。**

所以这一层的目标不是为了直接构建全文索引，而是为了让字段元数据先带上一小部分有代表性的真实样本。真正的大规模字段值同步，会在后面的 `Elasticsearch` 章节展开。

---

## 5、第三步：为什么 Service 先处理业务实体，而不是直接操作 ORM

这是本章非常值得建立起来的分层意识。当前项目里，`Service` 层并没有直接构造 `SQLAlchemy` 的 ORM 模型，而是先处理：`TableInfo`、`ColumnInfo`这样的业务实体。这样做的关键目的，不是“多套一层”，而是为了**解耦**。

如果 `Service` 层直接依赖 ORM 模型，会带来一个明显问题：业务逻辑会直接绑定到底层存储实现，一旦 ORM 模型变化，业务层代码也要跟着改。

而现在这样分层之后：

- `Service` 层只关心业务对象怎么组织
- `Repository` 层才关心底层如何写入数据库

这也是为什么后面同一批 `ColumnInfo` 还能继续被拿去构建 `Qdrant` 向量索引、参与 `Elasticsearch` 取值索引，而不必让业务层到处带着 ORM 对象跑。

---

## 6、第四步：Meta Repository 和 Mapper 到底负责什么

现在我们已经有了两类业务实体：`TableInfo`、`ColumnInfo`。接下来要做的，就是把它们写入 `Meta MySQL`。这部分职责在 `MetaMySQLRepository` 中。

项目对应文件路径：`shopkeeper-agent/app/repositories/mysql/meta/meta_mysql_repository.py`

```python
async def save_table_infos(self, table_infos: list[TableInfo]):
    models = [TableInfoMapper.to_model(table_info) for table_info in table_infos]
    self.session.add_all(models)


async def save_column_infos(self, columns_info: list[ColumnInfo]):
    models = [ColumnInfoMapper.to_model(column_info) for column_info in columns_info]
    self.session.add_all(models)
```

这里最关键的点，不是 `add_all()`，而是：`Repository` 对外暴露的输入仍然是业务实体，而不是 ORM 模型。

也就是说，`Service` 层只需要把“业务上已经整理好的数据对象”交给仓储层，至于怎么转换、怎么落库，由仓储层继续往下吸收。

### 6.1 Mapper 的作用是翻译

项目对应文件路径：

- `shopkeeper-agent/app/repositories/mysql/meta/mappers/table_info_mapper.py`
- `shopkeeper-agent/app/repositories/mysql/meta/mappers/column_info_mapper.py`

```python
class TableInfoMapper:
    @staticmethod
    def to_model(table_info: TableInfo) -> TableInfoMySQL:
        return TableInfoMySQL(**asdict(table_info))


class ColumnInfoMapper:
    @staticmethod
    def to_model(column_info: ColumnInfo) -> ColumnInfoMySQL:
        return ColumnInfoMySQL(**asdict(column_info))
```

这里的角色分工非常清楚：

- `TableInfo`、`ColumnInfo` 是业务实体
- `TableInfoMySQL`、`ColumnInfoMySQL` 是 ORM 模型
- `Mapper` 负责把前者转换成后者

所以整条写库链路可以压缩成下面这样：

```text
MetaConfig
  -> TableInfo / ColumnInfo
  -> TableInfoMapper / ColumnInfoMapper
  -> TableInfoMySQL / ColumnInfoMySQL
  -> Meta MySQL
```

### 6.2 add_all() 不等于真正提交数据库

很多初学者第一次接触 ORM 时，看到：

```python
self.session.add_all(models)
```

很容易以为“已经写进数据库了”。其实不是。

`add_all()` 做的事情，本质上只是：

- 把这些 ORM 对象加入当前 `Session`
- 告诉 ORM：这些对象后续需要被持久化

它本身还不是最终的提交动作。真正的数据落库，要发生在事务提交阶段。

---

## 7、第五步：为什么事务要放在 Service 层统一管理

回到本章主方法：

```python
async with self.meta_mysql_repository.session.begin():
    await self.meta_mysql_repository.save_table_infos(table_infos)
    await self.meta_mysql_repository.save_column_infos(column_infos)
```

这里使用的是：

```python
async with session.begin():
```

这种写法在当前场景下非常合适，因为它把事务边界说得很清楚：“保存表信息”和“保存字段信息”属于同一批业务操作，它们要么一起成功，要么一起失败。

这背后体现的是一个很重要的分层原则：

- `Repository` 负责“怎么写”
- `Service` 负责“这些写操作要以什么业务事务组织起来”

所以事务边界放在 `Service` 层，是合理的。它还有两个很实用的好处：1. 如果上下文中的操作都成功，事务会自动提交。2. 如果中途抛出异常，事务会自动回滚。

这样代码会比手写 `try/except + commit + rollback` 更简洁，也更不容易出错。

---

## 8、整条链路回顾

到这里，我们就可以把“一次表信息与字段信息同步到元数据库”的完整顺序串起来了：

1. 脚本读取 `meta_config.yaml`
2. `MetaKnowledgeService.build(config_path)` 进入表链路
3. 遍历配置中的每张表，构造 `TableInfo`
4. 通过 `DWMySQLRepository.get_column_types(...)` 查询该表所有字段的真实类型
5. 遍历该表中的每个字段，再通过 `DWMySQLRepository.get_column_values(...)` 查询示例值
6. 构造 `ColumnInfo`
7. 打开一笔事务
8. `MetaMySQLRepository` 接收业务实体，并通过 `Mapper` 转成 ORM 模型
9. `session.add_all()` 把这些 ORM 对象加入当前 `Session`
10. 事务结束后统一提交，数据真正写入 `Meta MySQL`

如果把这段流程再压缩成一句话，可以记成：**配置描述先变成业务实体，业务实体再补齐真实结构与样本，最后以事务方式落到元数据库。**

---

## 9、验证本章效果

这一章完成后，最合适的验证方式不是先去看 `Qdrant` 或 `Elasticsearch`，而是先确认 `Meta MySQL` 里的结构化元数据是否已经正确落下来。

### 9.1 运行同步脚本

```bash
python -m app.scripts.build_meta_knowledge -c conf/meta_config.yaml
```

### 9.2 重点检查 table_info 和 column_info

这一章最值得重点看的，不是有没有“插入成功”这 4 个字，而是下面这些字段是否真的带上了应有的信息。

先看 `table_info`：

- 是否已经写入每张配置中的表
- `id`、`name`、`role`、`description` 是否正确

再看 `column_info`：

- `id` 是否是 `表名.字段名`
- `type` 是否已经带上真实字段类型
- `examples` 是否已经带上一部分真实示例值
- `description`、`alias`、`table_id` 是否和配置保持一致

其中最关键的是 `type` 和 `examples`。
只要这两个字段已经有值，就说明本章最核心的“配置 + 动态补齐 + 落库”链路已经跑通了。

![验证本章效果](images/8/8-9-2-1.png)

---

**本章小结：**

- 元数据知识库的第一步，不是建向量索引，而是先把表和字段沉淀成结构化元数据。
- 配置文件负责描述业务语义，真实字段类型和字段示例值由程序运行时从 `DW` 自动补齐。
- `Service` 负责编排流程，`DW Repository` 负责查真实结构，`Meta Repository` 负责落库，`Mapper` 负责对象转换，事务边界由 `Service` 统一管理。

下一章开始，我们继续沿着这条表链路往后走，进入第二步：**为字段信息建立向量索引，并写入 `Qdrant`。**