SQL 语义复核修正：同一个口径不再因为子查询里的别名、HAVING 里的别名被报成不同；取哪几行逐段比，连同 OFFSET 和取前几行时的排序

条件里套着的子查询（EXISTS、IN (SELECT …)）原来按外层的来源整个照抄进外层那个条件，别名原样留着：带别名和不带别名的同一条 NOT EXISTS 被报成两处筛选不同，双盲复算就会说盲算换了口径。现在每段 SELECT 在语法树上记段号（复制出来的子查询靠它找回原树上那一段算好的来源，相关子查询照样到外层找），改写时每一列按它所在那一段认，子查询里的表去掉别名；比较两边排序时也按那一段的来源排。HAVING 引用 SELECT 别名（HAVING s > 100）换成别名指的表达式，和 GROUP BY 别名一个读法，聚合函数里的同名列不换。取哪几行原来只看最外层 LIMIT 的行数：CTE 里 LIMIT 10 和 LIMIT 100、OFFSET 0 和 OFFSET 10、取前 10 行时降序和升序都报成没有不同，取到的却是不同的行。Facets 加 limits，逐段收集（最外层、CTE、FROM 里的子查询；条件里的子查询已经整个写在条件里，不单算，免得报两遍）行数、跳过的行数，以及这一段的排序（ORDER BY 1、别名换成 SELECT 里的表达式）；OFFSET 0 算没写，排序只在限了行的段里比。原来的 limit（最外层行数）照旧保留。
2026-11-15 01:41:57 +00:00 · 2026-10-05 01:46:55 +08:00
parent f1c068d6a9
commit a427b53874
2 changed files with 346 additions and 53 deletions
@@ -7,13 +7,15 @@
 - normalize_sql：字面量换成占位符之后的规范文本，加上字面量本身（按规范文本里的顺序）。两条 SQL 只有字面量不同，
  规范文本就相同；关键字大小写、空白、注释、不加引号的名字的大小写都不算不同。
 - facets：把一条查询拆成几个方面——用到的表、连接（类型、关联进来的表、ON 条件）、WHERE 的各个合取式、其中的
-  时间条件（列、运算符、取值）、分组、聚合（函数、参数、是否去重）、HAVING、LIMIT。别名一律换成真实表名，
-  等式、比较的两边排好顺序：o.paid_at >= '2026-09-01' 和 '2026-09-01' <= orders.paid_at 是同一个条件。
+  时间条件（列、运算符、取值）、分组、聚合（函数、参数、是否去重）、HAVING、取哪几行（LIMIT、OFFSET、取前几行
+  时的排序）。别名一律换成真实表名（条件里套着的子查询也是），等式、比较的两边排好顺序：o.paid_at >= '2026-09-01'
+  和 '2026-09-01' <= orders.paid_at 是同一个条件。
 - diff_facets：逐个方面比两份 facets，每处不同给一句中文说明。

-**逐段收集。** CTE、子查询里的条件和聚合也算进来：盲算可能把有效记录条件写在 CTE 里，原流程写在外层，按
+**逐段收集。** CTE、子查询里的条件、聚合和取行也算进来：盲算可能把有效记录条件写在 CTE 里，原流程写在外层，按
 「整条语句里有哪些条件」比才不会把同一个口径报成不同。逗号连接（以及 INNER JOIN 之外又写在 WHERE 里的等值关联）
-写在 WHERE 里的列等式算作关联条件，不算筛选。
+写在 WHERE 里的列等式算作关联条件，不算筛选。条件里套着的子查询（EXISTS、IN (SELECT …)）既单算一段，也按它自己
+那一段的来源整个写进外层那个条件：别名在子查询里定义，拿外层的来源去认，同一个条件只因别名不同就会被报成不同。

 **和 sqlcheck 一样宁可漏报。** 解析不了（语法错、方言特性、多条语句、不是查询）就带 parse_error 返回，从不抛
 异常。没加表前缀、又有好几张表可能有这一列的，照写列名（给了表结构才去对）。只读语法树，不执行 SQL。
@@ -24,6 +26,7 @@ import logging
 import re
 from collections.abc import Iterable, Iterator, Mapping
 from dataclasses import dataclass, fields, is_dataclass
+from itertools import zip_longest
 from typing import Any

 from app.data import catalog
@@ -36,6 +39,9 @@ logger = logging.getLogger(__name__)
 PLACEHOLDER = "?"
 #: 太长的 SQL 不做语法分析（normalize_sql 退回逐词规范化，facets 带 parse_error）：解析耗时随长度增长
 MAX_SQL_CHARS = 100_000
+#: 每段 SELECT 在语法树上记段号的 meta 键。条件、表达式要复制一份再改写，复制出来的子查询靠段号找回原来那一段
+#: 算好的来源（别名、外层）：复制品没有父节点，自己找不到外层
+_SEGMENT = "sql_semantics_segment"


 # ==========================================================================
@@ -249,9 +255,25 @@ class Aggregate:
    text: str


+@dataclass(frozen=True)
+class LimitFacet:
+    """一段查询取哪几行。count、offset 是行数和跳过行数的写法（没写为 None，OFFSET 0 也算没写）；order 是这一段的
+    排序键 (Part, 是否降序)：只取前几行时，按什么排决定了取到的是哪几行。scope 是哪一段：None 是最外层，CTE 写它的
+    名字，FROM 里的子查询写它的别名（没有别名写「子查询」）。条件里套着的子查询不单算：它整个写在那个条件里比。"""
+
+    count: str | None
+    offset: str | None = None
+    order: tuple[tuple[Part, bool], ...] = ()
+    scope: str | None = None
+
+
@dataclass(frozen=True)
 class Facets:
-    """一条查询的各个方面。每一项都按在 SQL 里出现的先后排、按 key 去重；parse_error 不为空时其余都是空的。"""
+    """一条查询的各个方面。每一项都按在 SQL 里出现的先后排、按 key 去重；parse_error 不为空时其余都是空的。
+
+    limit 是最外层 LIMIT 的行数（写法）；limits 是各段取哪几行（最外层在前，见 LimitFacet），只有写了 LIMIT 或
+    OFFSET 的段才有。
+    """

    dialect: str | None = None
    tables: tuple[Part, ...] = ()
@@ -262,6 +284,7 @@ class Facets:
    aggregates: tuple[Aggregate, ...] = ()
    having: tuple[Part, ...] = ()
    limit: str | None = None
+    limits: tuple[LimitFacet, ...] = ()
    parse_error: str | None = None

    def as_dict(self) -> dict[str, Any]:
@@ -417,6 +440,11 @@ class _Statement:

        self.root = root
        self.read = read
+        #: 各段 SELECT（原树上的节点），下标就是记在 meta 里的段号
+        self.segments: list[Any] = []
+        for select in root.find_all(exp.Select, bfs=False):
+            select.meta[_SEGMENT] = len(self.segments)
+            self.segments.append(select)
        self.ctes: dict[str, Any] = {}
        for cte in root.find_all(exp.CTE):
            if cte.alias:
@@ -552,36 +580,82 @@ class _Statement:

    # ---- 写成 key / text

+    @staticmethod
+    def _owner(node: Any) -> Any:
+        """node 往上最近的一段 SELECT（带段号的，node 本身是也算）；没有就是 None：node 在调用方那一段里。"""
+        from sqlglot import exp
+
+        while node is not None:
+            if isinstance(node, exp.Select) and node.meta.get(_SEGMENT) is not None:
+                return node
+            node = node.parent
+        return None
+
+    def scope(self, node: Any, default: dict[str, _Source]) -> dict[str, _Source]:
+        """node 所在那一段的来源。条件里套着的子查询（EXISTS、IN (SELECT …)）是另一段：别名在那一段里定义，相关
+        子查询还要到外层找。复制出来的子查询带着段号，按段号找回原树上那一段算好的来源；不在子查询里就是 default。"""
+        owner = self._owner(node)
+        return default if owner is None else self.sources(self.segments[owner.meta[_SEGMENT]])
+
    def render(self, node: Any, sources: dict[str, _Source], mode: str = "text") -> str:
        """四种写法（见 Part）：key、loose 比较用，名字按 name_key，key 写真实表名、loose 不写；text、full 给人看，
-        text 只在好几张表都有这一列时写真实表名，full 一律写（关联条件说的就是两张表各出哪一列，用 full）。"""
+        text 只在好几张表都有这一列时写真实表名，full 一律写（关联条件说的就是两张表各出哪一列，用 full）。
+
+        node 里套着的子查询逐段改写：每一列按它所在那一段的来源认，子查询里的表去掉别名（列已经写成了真实表名）。
+        子查询里拿两张表的列互相比的（相关子查询），给人看时和关联条件一样写表名。
+        """
        from sqlglot import exp

        copy = node.copy()
-        for col in _own_columns(copy):
-            source, name = self.resolve(col, sources)
-            quoted = isinstance(col.this, exp.Identifier) and bool(col.this.args.get("quoted"))
-            if mode in ("key", "loose"):
-                col.set("this", exp.to_identifier(name_key(name)))
-                col.set("table", exp.to_identifier(source.key) if source is not None and mode == "key" else None)
-                col.set("db", None)
-                col.set("catalog", None)
-                continue
-            col.set("this", exp.Identifier(this=name, quoted=quoted or not _PLAIN_NAME.match(name)))
-            if source is None or (source.kind != "other" and mode != "full"
-                                  and (self.single or name_key(name) not in self.shared())):
-                col.set("table", None)
-                col.set("db", None)
-                col.set("catalog", None)
-            else:
-                *heads, last = source.idents or (exp.to_identifier(source.name),)
-                col.set("table", last.copy())
-                col.set("db", heads[-1].copy() if heads else None)
-                col.set("catalog", heads[-2].copy() if len(heads) > 1 else None)
+        groups: dict[int, tuple[Any, list[Any]]] = {}
+        for col in copy.find_all(exp.Column):
+            if col.name and not col.is_star:
+                owner = self._owner(col)
+                groups.setdefault(id(owner), (owner, []))[1].append(col)
+        for owner, cols in groups.values():
+            scope = sources if owner is None else self.sources(self.segments[owner.meta[_SEGMENT]])
+            found = [(col, *self.resolve(col, scope)) for col in cols]
+            here = mode
+            if owner is not None and mode == "text" and len({s.key for _, s, _ in found if s is not None}) > 1:
+                here = "full"
+            for col, source, name in found:
+                self._write(col, source, name, here)
+        for table in copy.find_all(exp.Table):
+            if isinstance(table.this, exp.Identifier) and self._owner(table) is not None:
+                table.set("alias", None)
+                if mode in ("key", "loose"):
+                    for arg in ("catalog", "db", "this"):
+                        ident = table.args.get(arg)
+                        if isinstance(ident, exp.Identifier):
+                            table.set(arg, exp.to_identifier(name_key(ident.name)))
        return copy.sql(dialect=self.read, comments=False)

+    def _write(self, col: Any, source: _Source | None, name: str, mode: str) -> None:
+        """按 render 的写法改写一列（就地改）。"""
+        from sqlglot import exp
+
+        quoted = isinstance(col.this, exp.Identifier) and bool(col.this.args.get("quoted"))
+        if mode in ("key", "loose"):
+            col.set("this", exp.to_identifier(name_key(name)))
+            col.set("table", exp.to_identifier(source.key) if source is not None and mode == "key" else None)
+            col.set("db", None)
+            col.set("catalog", None)
+            return
+        col.set("this", exp.Identifier(this=name, quoted=quoted or not _PLAIN_NAME.match(name)))
+        if source is None or (source.kind != "other" and mode != "full"
+                              and (self.single or name_key(name) not in self.shared())):
+            col.set("table", None)
+            col.set("db", None)
+            col.set("catalog", None)
+        else:
+            *heads, last = source.idents or (exp.to_identifier(source.name),)
+            col.set("table", last.copy())
+            col.set("db", heads[-1].copy() if heads else None)
+            col.set("catalog", heads[-2].copy() if len(heads) > 1 else None)
+
    def canonical(self, node: Any, sources: dict[str, _Source]) -> Any:
-        """比较的两边排好顺序（列在左、字面量在右；两边都有列的按 key 排），IN 的取值排好序。返回新的树。"""
+        """比较的两边排好顺序（列在左、字面量在右；两边都有列的按 key 排），IN 的取值排好序。返回新的树。
+        套着的子查询里的比较按它那一段的来源排：拿外层的来源去认子查询的别名，排出来的先后就取决于别名怎么起。"""
        from sqlglot import exp

        def fix(n: Any) -> Any:
@@ -589,8 +663,8 @@ class _Statement:
            if found is not None and isinstance(n, exp.Binary):
                left, right = n.this, n.expression
                lc, rc = bool(_own_columns(left)), bool(_own_columns(right))
-                if (rc and not lc) or (lc and rc and self.render(left, sources, "key")
-                                       > self.render(right, sources, "key")):
+                here = self.scope(n, sources) if lc and rc else sources
+                if (rc and not lc) or (lc and rc and self.render(left, here, "key") > self.render(right, here, "key")):
                    return getattr(exp, found[1])(this=right, expression=left)
            elif isinstance(n, exp.In) and n.expressions and not n.args.get("query") \
                    and not any(_own_columns(e) for e in n.expressions):
@@ -656,9 +730,7 @@ class _Statement:
            joins += [self._join(d, sources) for d in drafts]
            group += self._group(select, sources)
            aggregates += self._aggregates(select, sources)
-            clause = select.args.get("having")
-            having += [self.part(self.canonical(c, sources), sources)
-                       for c in (_conjuncts(clause.this) if clause is not None else [])]
+            having += self._having(select, sources)
        return Facets(
            dialect=self.read,
            tables=tuple(Part.plain(s.key, s.name) for s in self.tables),
@@ -669,6 +741,7 @@ class _Statement:
            aggregates=_dedupe(aggregates, lambda a: (a.func, a.arg.key, a.distinct)),
            having=_dedupe(having, lambda p: p.key),
            limit=self._limit(),
+            limits=tuple(self._limits()),
        )

    def _join_drafts(self, select: Any, sources: dict[str, _Source]) -> list[dict[str, Any]]:
@@ -757,25 +830,59 @@ class _Statement:
        return TimePredicate(column=self.part(cols[0], sources), op=op, value=value,
                             condition=self.part(cond, sources, excludes=True))

-    def _group(self, select: Any, sources: dict[str, _Source]) -> list[Part]:
-        """分组键。GROUP BY 1 换成 SELECT 里对应的表达式，GROUP BY 别名换成别名指的表达式（MySQL、SQLite 的读法）。"""
+    @staticmethod
+    def _aliases(query: Any) -> dict[str, Any]:
+        """SELECT 里的别名 {name_key: 别名指的表达式}。"""
        from sqlglot import exp

+        return {name_key(s.alias): s.this for s in query.selects if isinstance(s, exp.Alias) and s.alias}
+
+    @staticmethod
+    def _target(item: Any, selects: list[Any], aliases: dict[str, Any]) -> Any:
+        """GROUP BY、ORDER BY 里的一项：写序号（1）的换成 SELECT 里对应的表达式，写别名的换成别名指的表达式
+        （MySQL、SQLite 的读法），其余原样。"""
+        from sqlglot import exp
+
+        target = _unparen(item)
+        if isinstance(target, exp.Literal) and not target.is_string and str(target.this).isdigit():
+            index = int(target.this) - 1
+            if 0 <= index < len(selects):
+                return selects[index].this if isinstance(selects[index], exp.Alias) else selects[index]
+        elif isinstance(target, exp.Column) and not target.table and name_key(target.name) in aliases:
+            return aliases[name_key(target.name)]
+        return target
+
+    def _group(self, select: Any, sources: dict[str, _Source]) -> list[Part]:
+        """分组键。GROUP BY 1 换成 SELECT 里对应的表达式，GROUP BY 别名换成别名指的表达式（MySQL、SQLite 的读法）。"""
        group = select.args.get("group")
        if group is None:
            return []
-        selects = select.selects
-        by_alias = {name_key(s.alias): s.this for s in selects if isinstance(s, exp.Alias) and s.alias}
+        selects, aliases = select.selects, self._aliases(select)
+        return [self.part(self.canonical(self._target(item, selects, aliases), sources), sources)
+                for item in group.expressions]
+
+    def _having(self, select: Any, sources: dict[str, _Source]) -> list[Part]:
+        """HAVING 的各个合取式。引用 SELECT 别名的列（HAVING s > 100，MySQL、SQLite 的写法）换成别名指的表达式，
+        和 GROUP BY 别名一个读法：不换的话，写别名和写表达式的同一个条件会被报成两个不同的条件。"""
+        from sqlglot import exp
+
+        clause = select.args.get("having")
+        if clause is None:
+            return []
+        aliases = self._aliases(select)
        out = []
-        for item in group.expressions:
-            target = _unparen(item)
-            if isinstance(target, exp.Literal) and not target.is_string and str(target.this).isdigit():
-                index = int(target.this) - 1
-                if 0 <= index < len(selects):
-                    target = selects[index].this if isinstance(selects[index], exp.Alias) else selects[index]
-            elif isinstance(target, exp.Column) and not target.table and name_key(target.name) in by_alias:
-                target = by_alias[name_key(target.name)]
-            out.append(self.part(self.canonical(target, sources), sources))
+        for cond in _conjuncts(clause.this):
+            copy = cond.copy()
+            for col in _own_columns(copy):
+                target = aliases.get(name_key(col.name)) if not col.table else None
+                # 聚合函数里的名字指的是表里的列，不是别名：SUM(amount) AS amount 之后的 SUM(amount) 不能再换
+                if target is None or any(isinstance(p, exp.AggFunc) for p in _parents(col, stop=None)):
+                    continue
+                if col is copy:
+                    copy = target.copy()
+                    break
+                col.replace(target.copy())
+            out.append(self.part(self.canonical(copy, sources), sources))
        return out

    def _aggregates(self, select: Any, sources: dict[str, _Source]) -> list[Aggregate]:
@@ -811,6 +918,76 @@ class _Statement:
        value = node.expression if isinstance(node, exp.Limit) else node.args.get("count")
        return value.sql(dialect=self.read, comments=False) if value is not None else None

+    def _limits(self) -> list[LimitFacet]:
+        """各段取哪几行：最外层在前，其余按出现的先后。条件里套着的子查询（WHERE、HAVING、ON 里的）不单算：
+        它整个写在那个条件的 key 里，单算的话同一处不同会报两遍。"""
+        from sqlglot import exp
+
+        out = []
+        for query in self.root.find_all(exp.Select, exp.SetOperation, bfs=False):
+            if query is not self.root and self._in_condition(query):
+                continue
+            if (found := self._limit_of(query)) is not None:
+                out.append(found)
+        return out
+
+    def _in_condition(self, node: Any) -> bool:
+        from sqlglot import exp
+
+        while node is not None and node is not self.root:
+            if isinstance(node, (exp.Where, exp.Having)) or \
+                    (isinstance(node.parent, exp.Join) and node.arg_key == "on"):
+                return True
+            node = node.parent
+        return False
+
+    def _limit_of(self, query: Any) -> LimitFacet | None:
+        """一段的 LIMIT / OFFSET，连同这一段的排序；两样都没写返回 None。"""
+        from sqlglot import exp
+
+        def written(node: Any) -> str | None:
+            return node.sql(dialect=self.read, comments=False) if node is not None else None
+
+        limit, offset = query.args.get("limit"), query.args.get("offset")
+        count = written(limit.expression if isinstance(limit, exp.Limit) else limit.args.get("count")) \
+            if limit is not None else None
+        if offset is None and isinstance(limit, exp.Limit):
+            offset = limit.args.get("offset")        # 有的方言把 OFFSET 挂在 LIMIT 底下
+        skip = written(offset.expression if isinstance(offset, exp.Offset) else offset)
+        if skip is not None and skip.strip() == "0":
+            skip = None                              # OFFSET 0 等于没写
+        if limit is None and skip is None:
+            return None
+        return LimitFacet(count=count, offset=skip, order=self._order(query), scope=self._scope_name(query))
+
+    def _order(self, query: Any) -> tuple[tuple[Part, bool], ...]:
+        """这一段的排序键 (Part, 是否降序)。ORDER BY 1、ORDER BY 别名换成 SELECT 里的表达式（UNION 按第一支的）。"""
+        from sqlglot import exp
+
+        order = query.args.get("order")
+        if order is None:
+            return ()
+        select = query if isinstance(query, exp.Select) else None
+        sources = self.sources(select) if select is not None else _Sources()
+        selects = list(query.selects)
+        aliases = self._aliases(query)
+        out = []
+        for item in order.expressions:
+            target = self._target(item.this if isinstance(item, exp.Ordered) else item, selects, aliases)
+            desc = bool(item.args.get("desc")) if isinstance(item, exp.Ordered) else False
+            out.append((self.part(self.canonical(target, sources), sources), desc))
+        return tuple(out)
+
+    def _scope_name(self, query: Any) -> str | None:
+        from sqlglot import exp
+
+        if query is self.root:
+            return None
+        parent = query.parent
+        if isinstance(parent, (exp.CTE, exp.Subquery)) and parent.alias:
+            return parent.alias
+        return "子查询"
+

 # ==========================================================================
 # 比较
@@ -820,7 +997,8 @@ class _Statement:
@dataclass(frozen=True)
 class FacetDiff:
    """一处不同。aspect：time（时间列）/ time_range（同一时间列、范围不同）/ filter / aggregate / join / tables /
-    group_by / having / limit / parse。a、b 是两边各自的写法（给人看，没有就是 None），text 是一句中文说明。
+    group_by / having / limit（取哪几行：行数、跳过的行数、取前几行时的排序）/ parse。a、b 是两边各自的写法（给人看，
+    没有就是 None），text 是一句中文说明。
    detail 只在 time 上有：两边的时间列 [{table, column}]，调用方据此对照数据目录（业务日期）。"""

    aspect: str
@@ -910,6 +1088,12 @@ def _rows(limit: str | None) -> str:
    return f"只取前 {limit} 行" if limit.isdigit() else f"按 LIMIT {limit} 取行"


+def _skip(offset: str | None) -> str:
+    if offset is None:
+        return "不跳过行"
+    return f"跳过前 {offset} 行" if offset.isdigit() else f"按 OFFSET {offset} 跳过行"
+
+
 class _Compare:
    """一次比较：两边的叫法、列的中文名，以及按哪种写法比。

@@ -1099,14 +1283,55 @@ class _Compare:
             for p in self.b.having if self.k(p) not in ka]

    def limit(self) -> list[FacetDiff]:
-        a, b = self.a.limit, self.b.limit
-        return [FacetDiff("limit", a, b, _say(self.la, _rows(a), "，", self.lb, _rows(b)))] if a != b else []
+        """取哪几行：最外层两边比；CTE、子查询里的按出现的先后一一配对（两边给它起的名字可以不同），整组一样就不报。"""
+        def outer(f: Facets) -> LimitFacet | None:
+            return next((x for x in f.limits if x.scope is None), None)
+
+        out = self.rows(outer(self.a), outer(self.b))
+        ia = [x for x in self.a.limits if x.scope is not None]
+        ib = [x for x in self.b.limits if x.scope is not None]
+        if [self.lk(x) for x in ia] != [self.lk(x) for x in ib]:
+            for x, y in zip_longest(ia, ib):
+                out += self.rows(x, y)
+        return out
+
+    def lk(self, x: LimitFacet | None) -> tuple[Any, ...] | None:
+        if x is None:
+            return None
+        return x.count, x.offset, tuple((self.k(p), desc) for p, desc in x.order)
+
+    def sorting(self, x: LimitFacet, *, full: bool = False) -> str:
+        if not x.order:
+            return "没有排序"
+        keys = "、".join(_say(p.full if full else p.text, "降序" if desc else "升序") for p, desc in x.order)
+        return _say("按", keys, "排")
+
+    def rows(self, x: LimitFacet | None, y: LimitFacet | None) -> list[FacetDiff]:
+        """一对段（一边可以没有）取的行有什么不同：行数、跳过的行数，两边都限了行时再比排序。"""
+        if self.lk(x) == self.lk(y):
+            return []
+        sa, sb = (x.scope if x else None), (y.scope if y else None)
+        la = _say(self.la, "在", sa, "里") if sa else self.la
+        lb = _say(self.lb, "在", sb, "里") if sb and sb != sa else self.lb
+        out = []
+        ca, cb = (x.count if x else None), (y.count if y else None)
+        if ca != cb:
+            out.append(FacetDiff("limit", ca, cb, _say(la, _rows(ca), "，", lb, _rows(cb))))
+        oa, ob = (x.offset if x else None), (y.offset if y else None)
+        if oa != ob:
+            out.append(FacetDiff("limit", oa, ob, _say(la, _skip(oa), "，", lb, _skip(ob))))
+        if x is not None and y is not None and self.lk(x)[2] != self.lk(y)[2]:
+            wa, wb = self.sorting(x), self.sorting(y)
+            if wa == wb:
+                wa, wb = self.sorting(x, full=True), self.sorting(y, full=True)
+            out.append(FacetDiff("limit", wa, wb, _say(la, "取行前", wa, "，", lb, wb)))
+        return out


 def diff_facets(a: Facets, b: Facets, labels: tuple[str, str] = ("A", "B"), *,
                names: Mapping[str, str] | None = None) -> list[FacetDiff]:
    """逐个方面比两份 facets，每处不同一条，按对结果影响大致从大到小排：时间、筛选、聚合、连接和表、分组、HAVING、
-    LIMIT。完全一致返回空列表。
+    取哪几行。完全一致返回空列表。

    labels 是两边的叫法（缺省 A / B，双盲复算可以传「原流程」「盲算」）；names 是列的中文名（{「表.列」或「列」:
    中文名}，可以用 catalog_labels 从数据目录取），给了的话时间列写成「支付时间 paid_at」。任一边解析不了就只返回
@@ -1124,5 +1349,5 @@ def diff_facets(a: Facets, b: Facets, labels: tuple[str, str] = ("A", "B"), *,
    return cmp.time() + cmp.filters() + cmp.aggregates() + cmp.joins() + cmp.group() + cmp.having() + cmp.limit()


-__all__ = ["MAX_SQL_CHARS", "PLACEHOLDER", "Aggregate", "ColumnRef", "FacetDiff", "Facets", "JoinFacet",
+__all__ = ["MAX_SQL_CHARS", "PLACEHOLDER", "Aggregate", "ColumnRef", "FacetDiff", "Facets", "JoinFacet", "LimitFacet",
           "NormalizedSql", "Part", "TimePredicate", "catalog_labels", "diff_facets", "facets", "normalize_sql"]
@@ -3,7 +3,8 @@
 - normalize_sql：字面量换成占位符后的规范文本 + 字面量。只有字面量不同的两条 SQL 规范文本相同（定时运行每期的日期
  不同，不能算模板变了）；关键字大小写、空白、注释也不算不同；解析不了的退回逐词规范化，照样认得出只改了字面量
 - facets：表、连接（类型 / 关联进来的表 / ON 条件）、WHERE 合取式、时间条件（列 / 运算符 / 取值）、分组、聚合
-  （函数 / 参数 / 去重）、HAVING、LIMIT。别名换成真实表名、比较的两边排好顺序；解析不了带 parse_error，不抛
+  （函数 / 参数 / 去重）、HAVING、取哪几行（各段的 LIMIT / OFFSET / 排序）。别名换成真实表名（条件里套着的子查询、
+  HAVING 里的别名也是）、比较的两边排好顺序；解析不了带 parse_error，不抛
 - diff_facets：每处不同一句中文：「A 按支付时间 paid_at 筛时间，B 按下单时间 created_at」「A 排除了
  is_deleted = 1 的行，B 没有」「A 对 order_id 去重计数，B 没去重」……写法不同、口径相同的不报

@@ -186,6 +187,27 @@ def test_having_and_limit():
    assert facets("SELECT a FROM orders LIMIT 5, 20", "mysql").limit == "20"


+def test_having_reads_select_aliases():
+    """HAVING 里引用 SELECT 的别名（MySQL、SQLite 的写法）换成别名指的表达式，和 GROUP BY 别名一样。"""
+    by_alias = facets("SELECT region AS r, SUM(amount) AS s FROM orders GROUP BY r HAVING s > 100")
+    explicit = facets("SELECT o.region, SUM(o.amount) FROM orders o GROUP BY o.region HAVING SUM(o.amount) > 100")
+    assert [p.key for p in by_alias.having] == [p.key for p in explicit.having] == ["SUM(orders.amount) > 100"]
+
+
+def test_limits_are_collected_per_segment_with_offset_and_order():
+    """LIMIT 逐段收集（CTE、FROM 里的子查询各算一段），连同 OFFSET 和这一段的排序：只取前几行时，按什么排、
+    跳过几行决定了取到的是哪几行。OFFSET 0 等于没写。"""
+    f = facets("WITH top AS (SELECT amount FROM orders ORDER BY amount DESC LIMIT 10) "
+               "SELECT SUM(amount) FROM top LIMIT 1")
+    assert [(x.scope, x.count, x.offset) for x in f.limits] == [(None, "1", None), ("top", "10", None)]
+    [(key, desc)] = f.limits[1].order
+    assert key.text == "amount" and desc is True
+    mysql = facets("SELECT a FROM orders ORDER BY 1 LIMIT 5, 20", "mysql")
+    assert (mysql.limits[0].count, mysql.limits[0].offset) == ("20", "5")
+    assert [(k.text, d) for k, d in mysql.limits[0].order] == [("a", False)], "ORDER BY 1 换成 SELECT 里的第一项"
+    assert facets("SELECT a FROM orders LIMIT 10 OFFSET 0").limits[0].offset is None
+
+
@pytest.mark.parametrize("dialect,sql", [
    ("sqlite", 'SELECT SUM("金额") AS "销售额" FROM "订单" WHERE "是否作废" = 0'),
    ("mysql", "SELECT SUM(`金额`) AS `销售额` FROM `订单` WHERE `是否作废` = 0"),
@@ -211,8 +233,18 @@ def test_a_correlated_subquery_reads_the_outer_alias():
    """子查询里引用外层的别名（o.id）到外层去找：写成真实表名，两张表的列互相比的条件也写出表名。"""
    f = facets("SELECT o.id FROM orders o WHERE NOT EXISTS (SELECT 1 FROM refunds r WHERE r.order_id = o.id)")
    assert "orders.id = refunds.order_id" in [p.text for p in f.where]
-    assert facets("SELECT orders.id FROM orders WHERE NOT EXISTS "
-                  "(SELECT 1 FROM refunds WHERE orders.id = refunds.order_id)").where[1].key == f.where[1].key
+    plain = facets("SELECT orders.id FROM orders WHERE NOT EXISTS "
+                   "(SELECT 1 FROM refunds WHERE orders.id = refunds.order_id)")
+    assert plain.where[1].key == f.where[1].key
+    # 外层那个条件（NOT EXISTS 整句）也一样：子查询按它自己那一段的来源写，别名换成真实表名
+    assert plain.where[0].key == f.where[0].key
+    assert f.where[0].excludes == "EXISTS(SELECT 1 FROM refunds WHERE orders.id = refunds.order_id)"
+
+
+def test_a_subquery_in_a_condition_is_written_with_its_own_tables():
+    f = facets("SELECT SUM(o.amount) FROM orders o WHERE o.user_id IN (SELECT u.id FROM users u WHERE u.vip = 1)")
+    assert f.where[0].text == "user_id IN (SELECT id FROM users WHERE vip = 1)"
+    assert f.where[0].key == "orders.user_id IN (SELECT users.id FROM users WHERE users.vip = 1)"


 def test_the_schema_tells_which_table_an_unqualified_column_belongs_to():
@@ -380,11 +412,47 @@ def test_an_unparsable_side_is_reported_once_without_item_by_item_comparison():
    ("SELECT SUM(o.amount) FROM orders o, stores s WHERE o.store_id = s.id AND s.city = '虚构市'",
     "SELECT SUM(orders.amount) FROM orders INNER JOIN stores ON stores.id = orders.store_id "
     "WHERE stores.city = '虚构市'"),
+    # 条件里套着的子查询用了别名：NOT EXISTS、IN (SELECT …)
+    ("SELECT o.id FROM orders o WHERE NOT EXISTS (SELECT 1 FROM refunds r WHERE r.order_id = o.id)",
+     "SELECT orders.id FROM orders WHERE NOT EXISTS (SELECT 1 FROM refunds WHERE refunds.order_id = orders.id)"),
+    ("SELECT SUM(o.amount) FROM orders o WHERE o.user_id IN (SELECT u.id FROM users u WHERE u.vip = 1)",
+     "SELECT SUM(amount) FROM orders WHERE user_id IN (SELECT id FROM users WHERE vip = 1)"),
+    # HAVING 里用 SELECT 的别名
+    ("SELECT o.region, SUM(o.amount) FROM orders o GROUP BY o.region HAVING SUM(o.amount) > 100",
+     "SELECT region AS r, SUM(amount) AS s FROM orders GROUP BY r HAVING s > 100"),
+    # OFFSET 0 等于没写；取前几行时 ORDER BY 1 和写出列名是同一个排序
+    ("SELECT amount FROM orders ORDER BY 1 DESC LIMIT 10 OFFSET 0",
+     "SELECT o.amount FROM orders o ORDER BY o.amount DESC LIMIT 10"),
 ])
 def test_the_same_meaning_written_differently_is_no_difference(a, b):
    assert texts(a, b) == []


+def test_a_real_difference_inside_a_subquery_is_still_reported():
+    a = "SELECT SUM(o.amount) FROM orders o WHERE o.user_id IN (SELECT u.id FROM users u WHERE u.vip = 1)"
+    b = "SELECT SUM(o.amount) FROM orders o WHERE o.user_id IN (SELECT u.referrer_id FROM users u WHERE u.vip = 1)"
+    assert texts(a, b) == ["A 按 user_id IN (SELECT id FROM users WHERE vip = 1) 筛选，"
+                           "B 按 user_id IN (SELECT referrer_id FROM users WHERE vip = 1) 筛选"]
+
+
+@pytest.mark.parametrize("a,b,text", [
+    ("WITH top AS (SELECT amount FROM orders ORDER BY amount DESC LIMIT 10) SELECT SUM(amount) FROM top",
+     "WITH top AS (SELECT amount FROM orders ORDER BY amount DESC LIMIT 100) SELECT SUM(amount) FROM top",
+     "A 在 top 里只取前 10 行，B 只取前 100 行"),
+    ("SELECT amount FROM orders LIMIT 10 OFFSET 0", "SELECT amount FROM orders LIMIT 10 OFFSET 10",
+     "A 不跳过行，B 跳过前 10 行"),
+    ("SELECT amount FROM orders ORDER BY 1 DESC LIMIT 10", "SELECT amount FROM orders ORDER BY 1 ASC LIMIT 10",
+     "A 取行前按 amount 降序排，B 按 amount 升序排"),
+    ("SELECT amount FROM orders ORDER BY amount DESC LIMIT 10", "SELECT amount FROM orders LIMIT 10",
+     "A 取行前按 amount 降序排，B 没有排序"),
+])
+def test_which_rows_are_taken_is_compared(a, b, text):
+    """取哪几行：每一段的行数、跳过的行数，以及取前几行时的排序。排序只在有 LIMIT / OFFSET 的段里比：
+    没有取前几行，排序不影响算出来的数。"""
+    assert texts(a, b) == [text]
+    assert texts("SELECT SUM(amount) FROM orders ORDER BY 1 DESC", "SELECT SUM(amount) FROM orders") == []
+
+
@pytest.mark.parametrize("dialect", DIALECTS)
 def test_each_dialect_gives_the_same_three_findings(dialect):
    q = {"sqlite": '"', "postgres": '"', "mysql": "`"}[dialect]