# ODS→DWD 层确认修改清单

> 整理日期：2026-02-17（最后更新：2026-02-17）
> 来源：ODS 去重标准化 Spec 执行 + ODS→DWD 流程报告讨论 + 架构分析
> 状态标记：✅ 已完成 | 📋 待实施（用户已确认）

---

## 一、已完成的修改（✅）

### 1.1 ODS 去重标准化（Spec 全部 9 个任务）

| 任务 | 内容 | 涉及文件 |
|------|------|----------|
| Task 1 | 新增 `SnapshotMode` 枚举（`FULL_SET` / `TIME_WINDOW` / `NONE`） | `tasks/ods/ods_tasks.py` |
| Task 2 | content_hash 算法从 MD5 改为 SHA-256 | `tasks/ods/ods_tasks.py` |
| Task 3 | ODS DDL 补齐 `content_hash` 列 + `is_delete` 列 + `(pk, fetched_at DESC)` 索引 | `docs/database/ddl/etl_feiqiu__ods.sql`（原 `db/etl_feiqiu/schemas/ods.sql`，已归档） |
| Task 4 | 迁移脚本：为已有 ODS 表添加 `latest_version` 索引 | `db/_archived/ddl_baseline_2026-02-22/db/etl_feiqiu/migrations/2026-02-17__add_ods_latest_version_indexes.sql`（已归档） |
| Task 5 | 软删除重构：`_mark_missing_as_deleted` 改为 INSERT 删除版本行 | `tasks/ods/ods_tasks.py` |
| Task 6 | 属性测试覆盖去重和软删除逻辑 | `tests/unit/test_ods_dedup_properties.py` |
| Task 7-9 | 文档同步、集成验证 | 多文件 |

### 1.2 ODS 参数清理

| 修改 | 内容 |
|------|------|
| `run.snapshot_missing_delete` 默认值改为 `True` | 不再需要显式配置 |
| 删除 `run.ods_backfill_null_columns` 兼容代码 | 清理历史遗留 |

### 1.3 ODS→DWD 流程报告

- 生成完整报告：`docs/etl_tasks/ods_to_dwd_flow_report.md`（10 章 + 重构路线图）
- 报告经多轮 Q&A 修正，已确认的事实：
  - 所有 17 张维度表都有 SCD2 列 → Type1 分支不会触发
  - 所有 23 张事实表都没有 SCD2 列
  - 所有 ODS 表都有 `content_hash` → `snapshot_mode` 始终为 True
  - 水位线仅在自动水位模式下起作用，手动窗口模式不查水位线
  - 严格时间窗口下，回补步骤（`_insert_missing_by_pk`）基本不产生写入
  - `DwdLoadTask.extract()` 是空壳 — 只返回 `{"now": datetime.now()}`
  - E/T/L 模板在各层均未按设计使用（ODS 重写 execute、DWS 不继承 BaseTask）
  - `BaseOdsTask` 重复了 ~60% 的 `BaseTask` 窗口拆分代码

---

## 二、待实施的修改（📋 用户已确认方向）

### 2.1 统一窗口模式，去掉水位线（优先级：高）

**用户原话**："自动水位模式也确定 end 时间，去掉水位线机制"

现状：事实表增量有两条路径 — 配置了 `window_override` 时用 `WHERE fetched_at >= start AND fetched_at < end`，否则走自动水位 `WHERE fetched_at > watermark`（只有下界，无上界）。`load()` 中通过 `use_window = bool(config.get("run.window_override.start") and ...)` 判断走哪条路径。

改动：
- `DwdLoadTask.load()` 中去掉 `use_window` 判断，始终传 `context.window_start / window_end`
- 删除 `_get_fact_watermark()` 方法（约 30 行）
- `_merge_fact_increment` 中删除 watermark 分支，统一用 `WHERE fetched_at >= start AND fetched_at < end`
- `_merge_fact_increment` 签名中删除 `watermark` 参数

涉及文件：`tasks/dwd/dwd_load_task.py`

### 2.2 删除回补机制（优先级：高）

**用户原话**："类型转换失败需要直接报错的！不然除非对本次处理做数据上下游校验，不然开不开回补机制都会丢数据，需要报错提醒。"

确认逻辑：主增量写入是一条 SQL（`INSERT ... SELECT ... FROM ods表`），类型转换失败会导致整条 SQL 报错、整张表事务 rollback，不存在"部分行静默丢失"。回补无法兜底这种场景，应该直接报错。统一窗口后回补的 LEFT JOIN 结果集几乎一定为空。

改动：
- 删除 `_insert_missing_by_pk()` 方法（约 100 行）
- 删除 `_merge_fact_increment()` 末尾的回补调用及 `missing_inserted` 相关代码
- 删除 `FACT_MISSING_FILL_TABLES` 常量

注意：`dwd.fact_upsert_batch_size` 等批量参数不受影响 — 这些参数是 `dwd_verifier.py`（验证模块）在用，不是回补在用。

前置依赖：2.1（统一窗口模式）
涉及文件：`tasks/dwd/dwd_load_task.py`

### 2.3 去掉冗余范围机制（优先级：高）

**用户原话**："去掉冗余范围机制"

与 2.1 合并实施。统一窗口后，`overlap_seconds` 仍保留（用于自动模式下 start 的回退），但不再有水位线 vs 窗口两套并行的范围过滤。

### 2.4 清理死代码和未使用常量（优先级：高）

**确认删除的项目：**

| 清理项 | 操作 | 说明 |
|--------|------|------|
| `BaseDwdTask`（`base_dwd_task.py`） | 删除整个文件 | 死代码，`DwdLoadTask` 直接继承 `BaseTask`，从未使用 `BaseDwdTask` |
| `_pick_order_column()` 方法 | 删除 | 从未被调用，`_merge_fact_increment` 硬编码 `order_col = "fetched_at"` |
| `FACT_ORDER_CANDIDATES` 常量 | 删除 | 配合 `_pick_order_column` 使用，一起删除 |
| `FACT_MISSING_FILL_TABLES` 常量 | 删除 | 回补机制的一部分，随 2.2 删除 `_insert_missing_by_pk` 时一并清理 |
| `_upsert_scd2_row()` 方法 | 删除 | 早期逐行 SCD2 处理，已被批量方法 `_close_current_dim_bulk` / `_insert_dim_rows_bulk` 替代 |
| `_close_current_dim()` 方法 | 删除 | 同上，逐行关闭旧版本 |
| `_insert_dim_row()` 方法 | 删除 | 同上，逐行插入新版本 |
| `_merge_dim_type1_upsert()` 方法 | 删除 | 所有 17 张维度表都有 SCD2 列，Type1 分支永远不触发；未来新维度表也统一走 SCD2 |
| `_merge_dim()` 中 Type1 分支判断 | 删除 | 删除 `_merge_dim_type1_upsert` 后，`_merge_dim()` 中 `if scd2_cols:` 的分支判断也不再需要，直接走 SCD2 |

**确认修复的 bug：**

| 修复项 | 操作 | 说明 |
|--------|------|------|
| `_build_column_mapping()` 中引用未定义变量 `ods_table` / `cur` | 将 `ods_table` 和 `cur` 加入方法参数签名 | 当前因 `if "fetched_at" not in ods_set` 永远为 False 而未触发，但未来可能导致 `NameError` |

**确认保留的项目：**

| 保留项 | 原因 |
|--------|------|
| `dwd.fact_upsert` 及相关批量参数 | 验证模块 `dwd_verifier.py` 在使用，与回补无关 |
| `FACT_MAPPINGS` 字典名称 | 虽然维度表也用它（名字有误导），但改名影响面大（多个脚本和测试引用），暂不改 |

涉及文件：`tasks/dwd/dwd_load_task.py`、`tasks/dwd/base_dwd_task.py`

### 2.5 DWD content_hash 优化 + 参数化设计（优先级：中）

**用户原话**："DWD 可不可以也增加一个 HASH"、"做一个参数，使用 HASH 还是使用展平的数据，以便应对 DWD 表结构改变"

方案：
- 所有 40 张 DWD 表增加 `content_hash VARCHAR(64)` 列
- 新增配置 `dwd.change_detection_mode`：`"hash"`（默认）/ `"column"`
- hash 模式：复用 ODS 的 `content_hash`（基于 payload + is_delete 的 SHA-256），一列对比
- column 模式：保留当前逐列对比逻辑（DWD 表结构变更后的首次全量刷新用）
- 两种模式都依赖 `FACT_MAPPINGS` 做列映射（column 模式额外用映射后的值做对比）

hash 等价性说明：ODS 的 `content_hash` 基于原始 payload 计算，DWD 列是 payload 的确定性派生。日常运行时"payload 不变 = DWD 列不变"成立。但 DWD 表结构变更（新增列、修改映射规则）时 hash 等价关系被打破，需临时切 column 模式全量刷新。

典型工作流：
```
日常运行：change_detection_mode = "hash"（快速）
  → 修改了 DWD 表结构或 FACT_MAPPINGS
  → 临时切换：change_detection_mode = "column" + window_override 覆盖全量
  → 全量刷新完成后切回 "hash"
```

涉及文件：DDL（40 张 DWD 表）、迁移脚本、`tasks/dwd/dwd_load_task.py`、`config/defaults.py`

### 2.6 跨层参数提取与层级控制（优先级：中）

**用户原话**："很多任务和参数都是 API-ODS-DWD 通用的，甚至 API-ODS-DWD-DWS 与未来 API-ODS-DWD-CORE 都通用，把这部分提取出来，做层级控制，是否合理？"（确认合理）

现状问题：
- 各层对配置的使用方式不统一：ODS 读 `run.*`，DWD 读 `dwd.*`，DWS 读 `dws.*`
- 共用参数（如 `overlap_seconds`、`window_minutes`）散落在 `run.*` 下
- 如果未来 CORE 层也需要 `overlap_seconds` 但值不同，没法区分

方案：
- 配置增加 `layer.<层名>.*` 结构，支持层级覆盖
- `AppConfig` 增加 `get_for_layer(layer, key)` 方法：先查 `layer.<层>.<key>`，fallback 到 `run.<key>`
- 向后兼容：旧的 `run.*` 键仍可用，`layer.*` 优先
- `BaseDwsTask` 重构为继承 `BaseTask`，复用时间窗口和窗口拆分逻辑

BaseDwsTask 重构说明：当前 `BaseDwsTask`（1243 行）不继承 `BaseTask`，有完全独立的时间范围计算逻辑（`get_time_layer_range` / `get_time_window_range`）。重构后继承 `BaseTask`，复用窗口计算和段循环，但保留 DWS 特有的业务方法（配置缓存、业绩等级、奖金计算等）。这是一个大改动，需要仔细设计。

涉及文件：`config/defaults.py`、`config/settings.py`、`tasks/base_task.py`、`tasks/dws/base_dws_task.py`、各层基类

### 2.7 连接器与多门店配置（优先级：中）

**用户原话**："store_id 之上有个连接器 ID，也就是说现在 Feiqiu 的平台可能会有若干个 store 进行接入。每个 store 都有自己的 Token（其他都一样）"

方案：

**数据库设计：**
```sql
-- 连接器注册表
CREATE TABLE meta.connector (
    connector_id    SERIAL PRIMARY KEY,
    connector_code  VARCHAR(50) NOT NULL UNIQUE,  -- 如 'feiqiu'
    display_name    VARCHAR(100),
    api_base_url    VARCHAR(255),
    is_active       BOOLEAN DEFAULT TRUE,
    created_at      TIMESTAMPTZ DEFAULT now()
);

-- 门店注册表
CREATE TABLE meta.connector_store (
    store_id         BIGINT PRIMARY KEY,
    connector_id     INT NOT NULL REFERENCES meta.connector(connector_id),
    store_name       VARCHAR(100),
    api_token_enc    TEXT NOT NULL,                 -- AES 加密后的 Token
    token_updated_at TIMESTAMPTZ,                   -- Token 最后更新时间
    is_active        BOOLEAN DEFAULT TRUE,
    created_at       TIMESTAMPTZ DEFAULT now()
);
```

**Token 安全：**
- 应用层 AES 加密存储，加密密钥通过环境变量 `TOKEN_ENCRYPT_KEY` 传入
- 管理后台不显示 Token 明文，只显示：更新时间（`token_updated_at`）+ 有效性状态
- 后端提供 Token 有效性检查接口（调上游 API 验证），管理后台展示检查结果
- 日志中已有脱敏机制（`security.redact_keys` 包含 `token`）

**配置体系变更：**
- .env 中用 `CONNECTOR_CODE=feiqiu` 替代 `STORE_ID` + `API_TOKEN`
- 运行时从 DB 加载门店列表，遍历执行
- 向后兼容：.env 中仍有 `STORE_ID` + `API_TOKEN` 时走旧的单门店模式
- 两种模式互斥，启动时检测并提示

**维护方式：**
- 管理后台（主）：连接器/门店 CRUD、Token 更新（只能设置不能查看）、有效性检查
- CLI 命令（辅）：`python -m cli.main connector add/list/update`
- 共用后端 API

涉及文件：DDL（2 张新表）、迁移脚本、`config/`、`orchestration/`、后端 API、管理后台、`.env.template`、相关文档

### 2.8 架构简化 A3：E/T/L 模板替换为 `process_segment()` 钩子（优先级：中）

**来源**：架构分析发现 E/T/L 模板在各层均未按设计使用

现状问题：
- `DwdLoadTask.extract()` 是空壳 — 只返回 `{"now": datetime.now()}`
- `DwdLoadTask.transform()` 直接透传（继承自 BaseTask 默认实现）
- ODS 层完全重写了 `execute()`，不走 E/T/L 模板
- DWS 层不继承 `BaseTask`，有自己的时间窗口逻辑
- E/T/L 三步拆分在实际业务中没有意义，每层都是"读 → 处理 → 写"一体化

改动：
- `BaseTask.execute()` 改为：窗口计算 → 窗口拆分 → 循环调用 `process_segment(context)`
- 删除 `extract()` / `transform()` / `load()` 三个钩子方法
- 子类只需实现 `process_segment(context) -> dict`（返回统计计数）
- `DwdLoadTask` 将当前 `load()` 逻辑移入 `process_segment()`
- `DwdLoadTask.extract()` 中的 `{"now": datetime.now()}` 移入 `process_segment()` 内部
- ODS 层不再需要重写 `execute()`

涉及文件：`tasks/base_task.py`、`tasks/dwd/dwd_load_task.py`、`tasks/ods/ods_tasks.py`

### 2.9 架构简化 A4：统一 BaseOdsTask 和 BaseTask 窗口拆分逻辑（优先级：中）

**来源**：架构分析发现 BaseOdsTask 重复了 ~60% 的 BaseTask 窗口拆分代码

现状问题：
- `BaseOdsTask` 当前不继承任何基类（独立实现）
- `BaseOdsTask._resolve_window()` 和 `BaseTask._get_time_window()` 逻辑高度重复（手动窗口覆盖、闲忙时判断、overlap 回退）
- `BaseOdsTask.execute()` 自己做了窗口拆分循环（调用 `build_window_segments`），与 `BaseTask.execute()` 重复
- 两套代码独立维护，容易不一致

改动：
- `BaseOdsTask` 改为继承 `BaseTask`
- 删除 `BaseOdsTask._resolve_window()`，复用 `BaseTask._get_time_window()`
- 删除 `BaseOdsTask.execute()` 中的窗口拆分循环，复用 `BaseTask.execute()` 的段循环
- ODS 层实现 `process_segment(context)` 钩子（与 2.8 配合）
- ODS 特有逻辑（API 调用、快照去重、软删除等）保留在 `process_segment()` 中

前置依赖：2.8（process_segment 钩子）
涉及文件：`tasks/base_task.py`、`tasks/ods/ods_tasks.py`

---

## 三、实施顺序

```
第一阶段（低风险，连续做）：
  2.1 统一窗口 → 2.2 删除回补 → 2.3 去冗余 → 2.4 清理死代码 + bug 修复
  改动集中在 dwd_load_task.py，一次性完成

第二阶段（架构重构）：
  2.8 E/T/L → process_segment() 钩子（改 BaseTask + DwdLoadTask）
  2.9 统一 ODS/BaseTask 窗口拆分（改 BaseOdsTask 继承 BaseTask）

第三阶段（中风险，需设计）：
  2.6 跨层参数提取 — 改 defaults.py + AppConfig + 各层基类
      含 BaseDwsTask 继承重构（大改动）

第四阶段（可并行）：
  2.5 DWD content_hash — DDL + 对比逻辑改造
  2.7 连接器与多门店 — meta 表 + 加载逻辑 + 管理后台页面
```

---

## 四、已决定不做的项目

| 项目 | 决定 | 原因 |
|------|------|------|
| ODS `is_delete` 列索引 | 不做 | 低选择性（绝大多数行 is_delete=0），等实际出现慢查询时再 `EXPLAIN ANALYZE` 评估 |
| ODS 单列 `(fetched_at)` 索引 | 不做 | 数据库已有该索引（经查询确认），无需重复创建 |
| `FACT_MAPPINGS` 改名 | 暂不做 | 虽然维度表也用它（名字有误导），但改名影响面大（`gen_dataflow_doc.py`、`gen_full_dataflow_doc.py`、`dataflow_analyzer.py`、`test_dataflow_analyzer.py` 等多处引用），收益不大 |

---

## 五、保留但不改动的项目

| 项目 | 说明 |
|------|------|
| `dwd.fact_upsert` 及批量参数 | `fact_upsert_batch_size` / `min_batch_size` / `max_retries` / `retry_backoff_sec` / `lock_timeout_ms` 是验证模块 `dwd_verifier.py` 在使用，与回补机制无关，保留 |
