refactor(foundation): 采用 Daft-native step 架构，将注册与编排迁入 pipeline 包 (!3) · Merge requests · public_collaboration / Platform Core Public

Merged yujiwen requested to merge feature/daft-operator-adapter into master Jul 06, 2026

摘要

本 MR 将 backend/foundation 从 BaseOperator + mapper（item/arrow） 主线，重构为 Daft-native step function 架构，并收紧 pipeline ↔ operators 模块边界，为后续扩展 Excel source、tabular step、更多 pipeline 配置打下基础。

核心变化：

删除 BaseOperator、operators/mapper/、旧 eval/zh_t2s/ 等 legacy 代码
新增 operators/text/ 示范 step（text.uppercase、text.zh_t2s）
新增 pipeline/ 包：config、registry、graph、runner
STEP_REGISTRY 与 build_plan_from_config 归属 pipeline/registry.py（operators 不再 import pipeline）
配置解析增强：source/sink 结构化解析、未知顶层 key 拒绝、错误 step 参数统一报 INVALID_PIPELINE_CONFIG
测试分层：tests/（CI）+ eval/golden/（正确性）+ eval/bench/（性能）
文档全面同步（含 Legacy 平台栈与 foundation 边界区分）

背景与动机

此前 foundation 同时存在 BaseOperator/mapper 与 Daft 方向的设计文档，模块边界不清晰：

问题	风险
`STEP_REGISTRY` 放在 `operators/core/registry.py` 并 import `pipeline.graph`	违反 `pipeline → operators` 单向依赖
mapper 以 item/arrow 为主路径	与 Daft lazy DataFrame 范式重复，后续扩算子需再重构
配置 typo 抛出裸 `TypeError`	配置驱动系统难以给用户稳定错误信息
`source`/`sink` 配置被静默丢弃	与设计文档不一致，误导后续实现
文档仍引用 `BaseOperator`、`register_operator`	容易让 AI/新同学继续往错误路径加代码

本 MR 目标是提交一个 干净、可扩展的 Daft-native 基线，避免下一轮加算子时再次搬目录。

架构决策（SSOT）

Source → daft.DataFrame → step(df, spec) → compose_plan → collect_with_metrics(df)

决策	选择
Step 签名	`(df, frozen_spec) -> df`，不传 `OperatorContext`
追踪/审计	仅在 `collect_with_metrics(..., ctx=...)`
依赖方向	`pipeline → operators → contracts`
Registry key	`{domain}.{name}@{version}`，如 `text.zh_t2s@1.0.0`
列契约	`SchemaSpec` + `validate_schema`（含 `allow_overwrite`）
物化边界	唯一物化点：`pipeline/runner.collect_with_metrics`
领域模型	`contracts/` Pydantic SSOT（不 import daft）

设计文档 SSOT：docs/design/08_daft_native_operators.md

主要代码变更

新增

路径	说明
`operators/text/`	`uppercase.py`、`zh_t2s.py`、`specs.py`、开发模板 README
`operators/core/`	`schema.py`（`SchemaSpec`）、`errors.py`
`pipeline/config.py`	`PipelineConfig`、`SourceConfig`、`SinkConfig`、`PipelineStepConfig`
`pipeline/registry.py`	`STEP_REGISTRY`、`resolve_step`、`build_plan_from_config`
`pipeline/graph.py`	`compose_plan`
`pipeline/runner.py`	`collect_with_metrics`、`CollectHooks`
`eval/golden/`	JSONL golden + `pipeline_text_v1.json` 配置回归
`eval/bench/`	`zh_t2s` batch UDF 性能基线
`eval/README.md`	Golden vs Bench 分层说明

删除

路径	说明
`operators/base.py`	旧 `BaseOperator` 基类
`operators/mapper/`	item/arrow mapper 实现
`operators/registry.py`	旧算子注册
`eval/zh_t2s/`	旧 mapper 端到端 eval
`tests/test_*_mapper.py`	mapper 单测

已实现 Step

Registry Key	模块	策略
`text.uppercase@1.0.0`	`operators/text/uppercase.py`	native_expr（`with_column` + `upper`）
`text.zh_t2s@1.0.0`	`operators/text/zh_t2s.py`	`@daft.func.batch` UDF

配置解析行为

source:   # 已解析，执行层待后续 PR
  type: excel
  params: { ... }
pipeline:
  - op: text.zh_t2s
    version: "1.0.0"
    params: { content_col: content }
sink:     # 已解析，执行层待后续 PR
  type: memory

未知顶层 key（如 metadata）→ INVALID_PIPELINE_CONFIG
step 参数缺字段/多余字段 → INVALID_PIPELINE_CONFIG（含 op、version、params details）
build_plan_from_config 当前仅执行 steps

Breaking Changes

本 MR 不做向后兼容，有意删除 mapper 主线。

变更	迁移方式
`BaseOperator` / mapper API 删除	改用 `step(df, spec) -> df`
Registry 位置变更	`from pipeline import STEP_REGISTRY, build_plan_from_config`
Registry key 命名	`text_uppercase@1.0.0` → `text.uppercase@1.0.0`
`operators.__init__` 不再导出 pipeline 能力	从 `pipeline` 包导入
旧 `eval/zh_t2s/` 删除	改用 `eval/golden/` + `eval/bench/`

文档变更

文档	变更
`docs/OPERATORS.md`	Daft-native 框架总览、目录与开发三步
`docs/PDF_PIPELINE_SCHEMAS.md`	改为 `SchemaSpec` + `contracts/` SSOT
`docs/design/08_daft_native_operators.md`	架构 SSOT（新增）
`docs/design/10_excel_text_quality_pipeline.md`	Excel/tabular 扩展路线（新增）
`docs/requirement/modularization_vision.md`	`STEP_REGISTRY` 归 pipeline；去掉 mapper 主线表述
`docs/requirement/design/05_platform_pipeline_ray_operators.md`	foundation 代码地图 + Legacy 平台栈分节
`AGENTS.md` / `CLAUDE.md`	标明本仓库仅含 foundation；`backend/app` 标 Legacy

测试计划

CI（默认，每次 PR）

cd backend/foundation
uv run pytest tests -v          # 32 passed
uv run ruff check .

Golden 正确性回归（合并前建议跑）

uv run pytest eval/golden -m eval -v   # 3 passed

覆盖：

text.uppercase smoke（含 null）
text.zh_t2s smoke（含 null）
pipeline_text_v1.json 配置驱动串联

Bench 性能基线（可选）

uv run pytest eval/bench -m bench -v -s   # 1 passed

手动验证清单

PipelineConfig.from_dict 含 source/sink 时不丢字段
错误 step 参数返回 INVALID_PIPELINE_CONFIG 而非 TypeError
from pipeline import STEP_REGISTRY 可解析 text.zh_t2s@1.0.0
operators 包内无 import pipeline

依赖变更

新增并 pin：daft==0.7.16
pyproject.toml setuptools include：operators*、contracts*、pipeline*

后续工作（本 MR 范围外）

按 docs/design/10_excel_text_quality_pipeline.md 建议顺序：

pipeline/sources/excel.py — source 执行层
operators/tabular/dedupe.py — tabular.dedupe_by_key
operators/text/typo_check.py — text.typo_check
sink 物化（memory → parquet / DB）

Review 关注点

模块边界：确认 pipeline → operators → contracts 无反向 import
配置错误语义：bad params / unknown keys 是否足够用于上层 UI 展示
Registry 命名空间：text.* / tabular.* 是否符合团队约定
文档一致性：Legacy 平台栈（backend/app）与 foundation 边界是否清晰
Breaking 可接受性：mapper 删除是否影响其他并行分支（需协调合并）

关联信息

项	值
Source branch	`feature/daft-operator-adapter`
Target branch	`master`
Commit	`630886b`
变更规模	66 files, +3143 / -3073

可直接将以上内容粘贴到 GitLab MR 的 Description。若需要我代为创建 MR，告诉我即可。