# SQL 执行结果一致性判断逻辑详细说明

## 一、概述

`compare_results` 函数位于 `shared/utils/result_compare.py`，用于多层次比较原始 SQL 和优化后 SQL 的执行结果，判断两者是否一致。该函数采用**四层递进式判断机制**，确保结果比较的准确性和全面性。

## 二、数据准备阶段

### 2.1 结果数据格式

执行 SQL 时，结果会被保存为 JSON 格式（`sql_exe.py` 中的 `execute_sql_and_save` 函数生成）：

```json
{
    "rows": [...],              // 结果集样本（前10行）
    "row_count": 100,           // 总行数
    "columns": ["col1", "col2"], // 列名列表
    "numeric_stats": {...},     // 数值列统计信息
    "text_stats": {...},        // 文本列统计信息
    "sample_hashes": {...},     // 样本哈希值
    "exe_time": 8.498           // 执行时间
}
```

### 2.2 数据提取

`compare_results` 函数首先提取结果集中的行数据：

#### 2.2.1 pymysql 引擎的原始返回格式

**pymysql 使用 `DictCursor` 游标**（`client.py` 第65行）：

```python
cursorclass=DictCursor  # 使用字典游标
```

**`cursor.fetchall()` 返回格式**：

- 返回类型：`List[Dict[str, Any]]`
- 每个元素是一个字典，键为列名，值为列值
- 示例：
  ```python
  [
      {"ledger_date": "20210728", "captl_acct_bal": "17313000.00", "currency": "CNY"},
      {"ledger_date": "20210729", "captl_acct_bal": "17313000.00", "currency": "CNY"},
      ...
  ]
  ```

**`TDSQLClient.execute_query()` 返回格式**（`client.py` 第165-168行）：

```python
{
    'query_results': List[Dict[str, Any]],  # pymysql 返回的字典列表
    'execution_time': float                  # 执行时间（秒）
}
```

#### 2.2.2 保存到文件后的格式转换

**`sql_exe.py` 中的 `execute_sql_and_save` 函数**（第192-218行）会将 pymysql 的原始结果转换为结构化格式：

**转换过程**：

1. 从 `results['query_results']` 获取原始行数据（字典列表）
   - 数据顺序：按照 pymysql `cursor.fetchall()` 返回的顺序（即数据库返回的顺序）
   - **注意**：`sample_rows` 提取时**没有排序**，只是简单取前10行
2. 提取前10行作为样本：`sample_rows = query_rows[:10]`
   - **没有排序**：直接按数据库返回的顺序取前10行
   - 这些样本主要用于调试和查看，**不用于结果比较**
   - 结果比较主要使用 `sample_hashes`（见层级4说明）
3. 计算统计信息：
   - `numeric_stats`：数值列统计（count, sum, min, max）
   - `text_stats`：文本列统计（distinct_count）
   - `sample_hashes`：样本哈希值（前5后5的 MD5）
4. 保存为 JSON 文件：
   ```json
   {
       "rows": [...],              // 前10行样本
       "row_count": 100,           // 总行数
       "columns": ["col1", "col2"], // 列名列表
       "numeric_stats": {...},     // 数值列统计
       "text_stats": {...},        // 文本列统计
       "sample_hashes": {...},     // 样本哈希值
       "exe_time": 8.498           // 执行时间
   }
   ```

#### 2.2.3 `_extract_rows` 的数据提取逻辑

**`_extract_rows(result)`** 函数从保存的 JSON 文件中提取行数据：

**当前使用的格式**：

- **新格式**：`result.get("rows")` - **这是现在主要使用的格式**
  - 直接返回列表（前10行样本）
  - 格式：`{"rows": [{"col1": "val1"}, ...], "row_count": 100, "columns": [...], ...}`
  - 所有新生成的结果文件都使用此格式

**兼容旧格式**（仅用于读取历史数据文件）：

- `result.get("query_results")` - 仅用于兼容旧数据文件
  - 如果是列表类型，直接返回
  - 如果是字符串类型，尝试 JSON 解析：
    - 解析后是列表 → 返回列表
    - 解析后是字典 → 转换为单行列表 `[parsed]`（单行结果的情况）
  - 格式示例：`{"query_results": "{\"col1\": \"val1\", ...}", "exe_time": 8.498}`

**容错处理**：

- 如果所有尝试都失败，返回空列表 `[]`
- 如果 `result` 为空或 `None`，返回空列表 `[]`

**数据流转过程**：

```
pymysql (DictCursor)
  ↓ fetchall()
List[Dict[str, Any]]  # [{"col1": "val1"}, {"col2": "val2"}, ...]
  ↓ execute_query()
{"query_results": List[Dict], "execution_time": float}
  ↓ execute_sql_and_save()
{"rows": List[Dict], "row_count": int, "numeric_stats": {...}, ...}
  ↓ 保存为 JSON 文件
JSON 文件
  ↓ _extract_rows()
List[Dict[str, Any]]  # 提取的行数据，用于比较
```

## 三、四层判断机制

### 层级 1：行数比较（`counts_equal`）

**判断逻辑**：

- 比较原始 SQL 和优化 SQL 的结果行数
- 优先使用 `result.get("row_count")`，如果不存在则使用 `len(rows)`

**判断标准**：

```python
counts_equal = org_row_count == opt_row_count
```

**失败原因**：

- 如果行数不同，记录：`"row_count org={org_row_count} opt={opt_row_count}"`

**意义**：最基础的检查，如果行数不同，结果肯定不一致。

---

### 层级 2：Schema 比较（`schema_equal`）

**判断逻辑**：

- 比较原始 SQL 和优化 SQL 的结果列结构（列名集合）
- 使用 `_extract_schema(result, rows)` 提取列名：
  - 优先使用 `result.get("columns")`
  - 如果不存在，从第一行数据的 keys 中提取

**判断标准**：

```python
schema_equal = org_schema == opt_schema
```

**失败原因**：

- 如果列结构不同，记录：`"schema_diff org_only={...} opt_only={...}"`
- 显示原始结果独有的列和优化结果独有的列

**意义**：确保结果集的列结构一致，这是后续统计比较的基础。

---

### 层级 3：聚合统计比较（`aggregates_equal`）

**判断逻辑**：

- 对**共享列**（`shared_cols = org_schema & opt_schema`）进行聚合统计比较
- 分为两类统计：
  1. **数值列统计**（`numeric_stats`）
  2. **文本列统计**（`text_stats`）

#### 3.1 数值列统计（`build_numeric_stats`）

**统计指标**（对每个数值列）：

- `count`：非空数值的数量
- `sum`：所有数值的总和（保留6位小数）
- `min`：最小值
- `max`：最大值

**数值识别规则**（`_to_number`）：

- 支持 `int`、`float`、`Decimal` 类型
- 字符串尝试转换为数字（去除逗号后转换）
- `None`、`bool` 类型不视为数值

**判断标准**：

```python
org_num_stats == opt_num_stats
```

#### 3.2 文本列统计（`build_text_stats`）

**统计指标**（对每个文本列）：

- `distinct_count`：去重后的唯一值数量

**文本规范化规则**（`_normalize_value`）：

- `None` → 空字符串 `""`
- 字符串 → `strip().lower()`（去除首尾空格并转小写）
- 其他类型 → `str(value)`

**判断标准**：

```python
org_text_stats == opt_text_stats
```

**失败原因**：

- 如果数值统计不同：`"numeric_stats_diff {org_stats} vs {opt_stats}"`
- 如果文本统计不同：`"text_stats_diff {org_stats} vs {opt_stats}"`

**意义**：通过聚合统计快速判断结果集的数据分布是否一致，避免逐行比较（性能优化）。

---

### 层级 4：样本比较（`sample_equal`）

**判断逻辑**：

- 比较结果集的样本数据，验证实际数据内容是否一致
- 使用两种方式：
  1. **哈希比较**（优先）：如果结果中包含 `sample_hashes`，直接比较哈希值
  2. **序列化比较**（备选）：如果没有哈希，则序列化后比较

#### 4.1 哈希比较（`sample_hashes`）

**哈希生成逻辑**（`build_sample_hashes`）：

1. 将每行数据序列化为 JSON 字符串（`json.dumps(row, sort_keys=True)`）
   - `sort_keys=True` 确保字典键按字母顺序排序，使序列化结果稳定
2. **对所有序列化后的字符串进行排序**（`serialized.sort()`）
   - 这一步很重要：对行进行排序，确保即使数据库返回顺序不同，也能正确比较
   - 排序后的字符串列表用于提取样本
3. 取前 `sample_size` 个（默认5个）作为 `prefix`
4. 取后 `sample_size` 个（默认5个）作为 `suffix`
5. 对每个字符串计算 MD5 哈希值

**⚠️ 重要区别**：

- **`rows` 字段**（保存的前10行样本）：**没有排序**，按数据库返回顺序
- **`sample_hashes` 字段**（用于比较）：**有排序**，对序列化后的字符串进行排序后再提取样本
- 因此，即使原始 SQL 和优化 SQL 返回的行顺序不同，只要数据内容相同，`sample_hashes` 也能正确匹配

**判断标准**：

```python
org_hashes == opt_hashes  # 比较 prefix 和 suffix 的哈希值列表
```

**⚠️ 列名差异的影响**：

哈希比较确实会受到列名差异的影响：

1. **列名集合不同**：

   - 如果原始 SQL 和优化 SQL 的列名集合不同（例如：`{"col1": 1, "col2": 2}` vs `{"col1": 1, "col3": 2}`）
   - 即使数据值相同，序列化后的 JSON 字符串也会不同（因为键不同）
   - 哈希值也会不同，导致 `sample_equal = False`
   - **但这种情况在层级2（Schema比较）中已经被检测出来**，`schema_equal = False`，所以 `all_equal` 也会是 `False`
   - 因此，列名集合不同时，哈希比较失败是**预期的行为**
   
2. **列名集合相同，但顺序不同**：

   - 如果列名集合相同，但顺序不同（例如：`{"col1": 1, "col2": 2}` vs `{"col2": 2, "col1": 1}`）
   - 由于使用了 `sort_keys=True`，`json.dumps()` 会按字母顺序排序键
   - 序列化后的字符串会相同：`{"col1": 1, "col2": 2}`
   - 哈希值也会相同，不会因为列名顺序不同而失败
   
3. **列名相同，但数据值不同**：

   - 如果列名相同，但数据值不同（例如：`{"col1": 1, "col2": 2}` vs `{"col1": 1, "col2": 3}`）
   
   - 序列化后的 JSON 字符串会不同
   
   - 哈希值也会不同，导致 `sample_equal = False`
   
     

**总结**：

- 哈希比较会正确检测列名差异（列名集合不同时）
- 但列名集合不同的情况已经在层级2中被检测出来
- 列名顺序不同不会影响哈希比较（因为 `sort_keys=True`）
- 哈希比较主要用于验证**数据内容**是否一致，而不是列结构（列结构由层级2负责）

#### 4.2 序列化比较（`_compare_samples`）

**当没有哈希值时**：

1. 序列化所有行：`_serialize_rows(rows)`
   - 每行转为 JSON 字符串（`sort_keys=True` 确保字段顺序一致）
   - 对所有行进行排序
2. 比较前缀：前 `sample_size` 个元素
3. 比较后缀：后 `sample_size` 个元素

**判断标准**：

```python
# 前缀比较
for idx in range(min(sample_size, len(org), len(opt))):
    if org_serialized[idx] != opt_serialized[idx]:
        return False

# 后缀比较
for offset in range(1, min(sample_size, len(org), len(opt)) + 1):
    if org_serialized[-offset] != opt_serialized[-offset]:
        return False
```

**失败原因**：

- 如果样本不同，记录：`"sample_rows_diff"`

**意义**：验证实际数据内容是否一致，这是最严格的检查。

---

## 四、最终判断

### 4.1 一致性判断

**判断标准**：

```python
all_equal = all([
    counts_equal,      # 行数相同
    schema_equal,      # 列结构相同
    aggregates_equal,  # 聚合统计相同
    sample_equal       # 样本数据相同
])
```

**只有四个层级全部通过，才认为结果一致**。

### 4.2 差异原因记录

函数会记录所有不一致的原因到 `diff_reason` 列表：

- 如果所有检查都通过，记录：`"matched"`
- 如果有任何不一致，记录具体原因（如 `row_count`、`schema_diff`、`numeric_stats_diff` 等）

### 4.3 返回值

```python
{
    "counts_equal": bool,        # 行数是否相同
    "schema_equal": bool,        # 列结构是否相同
    "aggregates_equal": bool,    # 聚合统计是否相同
    "sample_equal": bool,        # 样本数据是否相同
    "all_equal": bool,           # 是否完全一致（所有层级都通过）
    "diff_reason": List[str]     # 差异原因列表
}
```

## 五、项目中的调用场景

### 5.1 `ana_multi_version.py`（多版本分析）

**调用位置**：`analyze_single_version` 函数的层级4

**使用方式**：

```python
comparison = compare_results(org_result_data, opt_result_data)
result['result_identical'] = comparison.get("all_equal", False)

if not result['result_identical']:
    result['status'] = '结果不一致'
    # 即使结果不一致，也继续计算性能指标
```

**作用**：判断优化后的 SQL 结果是否与原始 SQL 一致，用于状态判断。



## 六、判断流程总结

```
输入：org_result, opt_result
  ↓
提取行数据（_extract_rows）
  ↓
【层级1】行数比较
  ├─ 相同 → 继续
  └─ 不同 → 记录原因，继续（但 all_equal=False）
  ↓
【层级2】Schema 比较
  ├─ 相同 → 继续
  └─ 不同 → 记录原因，继续（但 all_equal=False）
  ↓
【层级3】聚合统计比较
  ├─ 数值统计比较（count, sum, min, max）
  ├─ 文本统计比较（distinct_count）
  ├─ 相同 → 继续
  └─ 不同 → 记录原因，继续（但 all_equal=False）
  ↓
【层级4】样本比较
  ├─ 优先：哈希比较（sample_hashes）
  ├─ 备选：序列化比较（前5后5样本）
  ├─ 相同 → all_equal=True
  └─ 不同 → 记录原因，all_equal=False
  ↓
返回：{counts_equal, schema_equal, aggregates_equal, sample_equal, all_equal, diff_reason}
```



## 七、待讨论问题

问题总结： 

1. 结果上返回大概率不一致， 包括：

   1. count(0)， count(*) 及大小写 写法问题
   2. 起列别名，如count(0)  ->> table_count, 字段查询也存在这种问题

   【改不了强行改，不去该 

   提示词限制 -> 辅助强行矫正

   结果纠正】

   --> 这里： 现在的哈希比较(字符比较) 是否合理？

2. 行列增删问题， 具体包括：

   1. 返回结果 字段缺失/ 增加无效字段
   2. 增加筛选条件（limit/distinct/日期）， -> 行数不一致

3. 聚合操作问题：

   1. 丢失count

4. 逻辑不一致但是结果一致？ -> 是正确优化还恰巧碰上

    ----> 风险提示/开发确认