最近三个月，不是最近三条记录：用一段断月工资看懂SQL窗口
有个人的工资记录是这样：
月份 1月 2月 3月 4月 5月 6月 金额 100 200 - - 500 600现在要查他5月的“最近三个月工资之和”，你会加哪些数？
如果从5月往前找两条记录，会找到2月、1月，得到800。但5月往前两个自然月明明是4月、3月。按月份看，这次应该只加到5月的500。
两种算法都能写成一个窗口，都能跑出结果，甚至在没有断月的样例里完全一样。得先确定窗口里装的到底是什么。
旧SQL学习笔记提过579“累计工资”的选题。官方题页目前需要订阅才能查看完整题面，本文另定一套规则来练习，不把下面的数据条件冒充原题。
这次到底要算什么？
每位员工每个月最多一条工资，金额是非负整数，不能为NULL。对已有记录的每个月，计算该员工当月及前两个月的工资总额；缺月按没有贡献处理，但不额外生成缺月的报表行。最后，报表不显示每位员工自己最新的那个月，按员工编号、月份升序输出。
为什么最后还有一个“不显示最新月份”？就当它是一次报表要求，我们也借它检查：到底要删谁的最新月，而不是看全表最大的月份就一起删。
为了先把月份范围看清楚，表里用连续月份编号month_no。以2026年1月为1，2026年12月为12，2027年1月继续用13。**不是每到一月又从1开始，也不是202612、202701那种年月拼接整数。**后者相减不代表相隔几个月。
在实际日期表里，可以用YEAR(d) * 12 + MONTH(d)得到另一套连续月份编号；起点不同不影响相邻月份差1。本文的简化表直接保存编号。
CREATE TABLE Salary ( employee_id INT NOT NULL, month_no INT NOT NULL CHECK (month_no >= 1), amount INT NOT NULL CHECK (amount >= 0), PRIMARY KEY (employee_id, month_no) ); INSERT INTO Salary VALUES (1, 1, 100), (1, 2, 200), (1, 5, 500), (1, 6, 600), (2, 2, 50), (2, 3, 80), (2, 4, 100);使用MySQL 8.0.46，在空的学习库里运行；如果已有同名表，换库，不要覆盖业务数据。本例已经按员工和月汇总。如果原表一月有多笔发薪，先决定哪些发薪计入，并汇总到这个粒度，不能直接把每笔记录当一个月。
拿着月份往前走，不要先拿窗口语法套
还是只看员工1。
1月往前取三个月，范围从编号-1到1。表里只有1月，所以结果100。窗口越过数据起点没关系，范围内有什么就算什么，不会造出负月份的数据。
2月看0、1、2，有1月和2月，结果300。
到了5月，我们把纸上的范围也向右挪到3、4、5：
记录月份：1 2 5 6 看2月时：[1 2] 看5月时： [3 4 5] 无 无 500这时1、2月都已经在范围外了。缺3、4月，不是让我们把窗口往更早的月份拉长，直到凑够三条。
再走到6月，范围是4、5、6，得到500+600=1100。
所以员工1的结果应该依次是100、300、500、1100。先保留这个手算结果，最后一步才处理“不显示最新月份”。
ROWS取前面的行，RANGE看排序值的范围
如果需求真是“当行和前两条记录”，可以写ROWS BETWEEN 2 PRECEDING AND CURRENT ROW。它没写错，只是回答了另一个问题。
这次想按月份编号取范围，可以写RANGE BETWEEN 2 PRECEDING AND CURRENT ROW。在升序月份编号下，当前是5，它就取排序值在3到5之间的记录。
把两者同时运行，就能看见差别：
SELECT employee_id, month_no, SUM(amount) OVER ( PARTITION BY employee_id ORDER BY month_no ROWS BETWEEN 2 PRECEDING AND CURRENT ROW ) AS last_three_rows, SUM(amount) OVER ( PARTITION BY employee_id ORDER BY month_no RANGE BETWEEN 2 PRECEDING AND CURRENT ROW ) AS last_three_months FROM Salary ORDER BY employee_id, month_no;员工1的四行：
月份 前三条记录 三个月范围 1 100 100 2 300 300 5 800 500 6 1300 1100到6月时，前三条是2、5、6，所以按行算会得到200+500+600=1300。按月份算，2月早就离开窗口了。
PARTITION BY employee_id也不能省：每个人的工资要分别算。员工2的2月和员工1的2月虽然编号相同，不应该进入对方的总额。
这里RANGE的偏移按那个数值排序表达式计算，不能把2无条件理解成“两个月”。正因为本文排序列是连续月份编号，差2才表示两个月。若排序列换成天数，2就成了两天；如果换成工资，也不是月份了。
最新月要按人找
员工1最新有6月，员工2最新只有4月。报表要去掉的是员工1的6月、员工2的4月。
员工1：1 2 5 [6] <- 去掉自己的6月 员工2： 2 3 [4] <- 去掉自己的4月如果只查全表MAX(month_no)，得到6。拿它过滤所有人，员工2的4月就漏删了。
可以在刚才算总额的那层，同时用MAX(month_no) OVER (PARTITION BY employee_id)给每一行添上所属员工的最新月，再到外层筛选。
WITH calculated AS ( SELECT employee_id, month_no, SUM(amount) OVER ( PARTITION BY employee_id ORDER BY month_no RANGE BETWEEN 2 PRECEDING AND CURRENT ROW ) AS salary_sum, MAX(month_no) OVER ( PARTITION BY employee_id ) AS latest_month FROM Salary ) SELECT employee_id, month_no, salary_sum FROM calculated WHERE month_no < latest_month ORDER BY employee_id, month_no;最后得到：
员工 月份 总额 1 1 100 1 2 300 1 5 500 2 2 50 2 3 130外层负责筛已经计算好的窗口结果。在定义latest_month的同一层，不能直接在WHERE里拿这个别名来过滤。
还有一个边界：员工如果只来过一个月，这个月也是他自己的最新月，最后就没有他的记录。我们不需要强行为他留一行。
这套查询不会生成缺少的3、4月，也不会把工资为0的已有行删掉。如果业务要展示所有员工每个月，包括从未有工资记录的月份，就得先补齐月份和员工的组合；那是另一种报表，不能把这两种输出要求混在一起。
验证别只用连续月份
如果测试数据永远是1、2、3、4，ROWS和RANGE很可能一起通过。这里最有用的反例，反而是只留1、2、5、6，让“凑三行”和“取三个月”不得不分开。
验证程序在Python里直接枚举每个员工已有月份：先找该员工最大月份，把它排除；对剩下的每条，遍历同一员工所有记录，只累加月份编号落在当月减2到当月之间的金额。它不使用SQL窗口函数，也不写另一版行号算法。
再把同一输入放到独立MySQL实例，执行从本文提取的SQL，比较完整的员工、月份、总额和顺序。还加入了不同员工最新月份不同、零工资、空表、仅一个月、跨年编号11到16，以及打乱插入顺序的情况。
错误查询也会真的运行：把RANGE换成ROWS、只向前取一个月、累计全部历史、漏掉员工分区、把全表最新月当个人最新月，以及忘记排除最新月。它们必须产生错误结果，不把SQL语法错误当成逻辑反例。
下面保留完整程序。使用Python 3及MySQL 8服务端、客户端和运行库；默认路径来自本文的WSL解包环境，可用BLOG_MYSQL_ROOT更改，并按安装布局调整运行库路径。它只启动自己临时数据目录中的MySQL，经Unix socket连接、不监听网络，结束时关闭子进程，不接触已有数据库。
JSON保存为article-steps.json，Python保存为同目录的verify_salary.py，执行python3 verify_salary.py。
[ { "id": "setup", "sql": "CREATE TABLE Salary (\n employee_id INT NOT NULL,\n month_no INT NOT NULL CHECK (month_no >= 1),\n amount INT NOT NULL CHECK (amount >= 0),\n PRIMARY KEY (employee_id, month_no)\n);\nINSERT INTO Salary VALUES\n (1, 1, 100), (1, 2, 200),\n (1, 5, 500), (1, 6, 600),\n (2, 2, 50), (2, 3, 80), (2, 4, 100);", "expected": [] }, { "id": "compare", "sql": "SELECT employee_id, month_no,\n SUM(amount) OVER (\n PARTITION BY employee_id ORDER BY month_no\n ROWS BETWEEN 2 PRECEDING AND CURRENT ROW\n ) AS last_three_rows,\n SUM(amount) OVER (\n PARTITION BY employee_id ORDER BY month_no\n RANGE BETWEEN 2 PRECEDING AND CURRENT ROW\n ) AS last_three_months\nFROM Salary\nORDER BY employee_id, month_no;", "expected": [ "1\t1\t100\t100", "1\t2\t300\t300", "1\t5\t800\t500", "1\t6\t1300\t1100", "2\t2\t50\t50", "2\t3\t130\t130", "2\t4\t230\t230" ] }, { "id": "final", "sql": "WITH calculated AS (\n SELECT employee_id, month_no,\n SUM(amount) OVER (\n PARTITION BY employee_id ORDER BY month_no\n RANGE BETWEEN 2 PRECEDING AND CURRENT ROW\n ) AS salary_sum,\n MAX(month_no) OVER (\n PARTITION BY employee_id\n ) AS latest_month\n FROM Salary\n)\nSELECT employee_id, month_no, salary_sum\nFROM calculated\nWHERE month_no < latest_month\nORDER BY employee_id, month_no;", "expected": [ "1\t1\t100", "1\t2\t300", "1\t5\t500", "2\t2\t50", "2\t3\t130" ] } ]import getpass import itertools import json import os import random import subprocess import tempfile import time from datetime import date, timedelta from pathlib import Path base = Path(__file__).resolve().parent steps = json.loads((base / 'article-steps.json').read_text()) package = Path(os.environ.get('BLOG_MYSQL_ROOT', '/mnt/d/blog-ops/experiments/2026-10-03-candidates/mysql-packages/root')) env = dict(os.environ, LD_LIBRARY_PATH=str(package / 'usr/lib/x86_64-linux-gnu')) run = Path(tempfile.mkdtemp(prefix='salary-', dir=base)) (run / 'data').mkdir() sock = Path('/dev/shm') / (run.name + '.sock') args = [str(package / 'usr/sbin/mysqld'), '--no-defaults', '--basedir=' + str(package / 'usr'), '--datadir=' + str(run / 'data'), '--lc-messages-dir=' + str(package / 'usr/share/mysql'), '--user=' + getpass.getuser(), '--secure-file-priv=NULL', '--innodb-buffer-pool-size=67108864', '--innodb-redo-log-capacity=16777216', '--log-error=' + str(run / 'server.log')] init = subprocess.run(args + ['--initialize-insecure'], env=env, capture_output=True, text=True, timeout=90) if init.returncode: raise RuntimeError(init.stderr) client = [str(package / 'usr/bin/mysql'), '--no-defaults', '--protocol=SOCKET', '--socket=' + str(sock), '-u', 'root', '--batch', '--skip-column-names', '--connect-timeout=2'] server = subprocess.Popen(args + ['--skip-networking', '--mysqlx=0', '--socket=' + str(sock), '--pid-file=' + str(run / 'mysql.pid'), '--tmpdir=' + str(run)], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) record = {'passed': False, 'run': str(run)} def sql(text): result = subprocess.run(client, input=text, env=env, capture_output=True, text=True, timeout=120) if result.returncode or result.stderr.strip(): raise RuntimeError(result.stdout + result.stderr) return result.stdout.splitlines() def oracle(rows): answer = [] for employee, month, amount in sorted(rows): latest = max(m for e, m, _ in rows if e == employee) if month == latest: continue total = sum(v for e, m, v in rows if e == employee and month - 2 <= m <= month) answer.append(f'{employee}\t{month}\t{total}') return answer def insert(rows): if not rows: return '' return 'INSERT INTO Salary VALUES ' + ','.join( f'({e},{m},{v})' for e, m, v in rows) + ';' def chunks(lines, prefix): groups = [] for line in lines: if line.startswith(prefix): groups.append([]) else: assert groups, line groups[-1].append(line) return groups try: for _ in range(200): if server.poll() is not None: raise RuntimeError((run / 'server.log').read_text()) if sock.exists(): try: sql('SELECT 1;') break except RuntimeError: pass time.sleep(0.1) else: raise TimeoutError('MySQL startup timed out') record['version'] = sql('SELECT VERSION();')[0] commands = ['CREATE DATABASE blog_salary; USE blog_salary;'] for i, step in enumerate(steps): commands.extend([f"SELECT '__step_{i}__';", step['sql']]) outputs = chunks(sql('\n'.join(commands)), '__step_') assert outputs == [s['expected'] for s in steps], outputs record['articleSteps'] = [dict(s, actual=a) for s, a in zip(steps, outputs)] final = next(s['sql'] for s in steps if s['id'] == 'final') main = [(1,1,100),(1,2,200),(1,5,500),(1,6,600),(2,2,50),(2,3,80),(2,4,100)] cases = [main, [], [(1,1,0)], [(1,11,10),(1,12,20),(1,13,30),(1,16,80),(2,12,20),(2,14,40)]] slots = [(e,m) for e in [1,2] for m in [1,2,5,6]] for states in itertools.product(range(3), repeat=8): cases.append([(e,m,0 if v==1 else 100) for (e,m),v in zip(slots,states) if v]) rng = random.Random(579) for _ in range(100): rows = [(e,m,rng.choice([0,1,100,500])) for e in [1,2,3] for m in range(10,19) if rng.random()<.55] rng.shuffle(rows) cases.append(rows) commands = ['USE blog_salary;'] for i, rows in enumerate(cases): commands.extend(['DELETE FROM Salary;', insert(rows), f"SELECT '__case_{i}__';", final]) outputs = chunks(sql('\n'.join(commands)), '__case_') assert len(outputs) == len(cases) for i, (got, rows) in enumerate(zip(outputs, cases)): assert got == oracle(rows), {'case':i,'input':rows,'actual':got,'expected':oracle(rows)} record['cases'] = len(cases) record['exhaustiveCases'] = 3**8 record['randomSeed'] = 579 latest = 'MAX(month_no) OVER (\n PARTITION BY employee_id\n )' variants = { 'rows-not-months': final.replace('RANGE BETWEEN', 'ROWS BETWEEN'), 'only-two-months': final.replace('2 PRECEDING', '1 PRECEDING'), 'all-history': final.replace('RANGE BETWEEN 2 PRECEDING', 'ROWS BETWEEN UNBOUNDED PRECEDING'), 'mix-employees': final.replace('PARTITION BY employee_id ', '', 1), 'global-latest': final.replace(latest, '(SELECT MAX(month_no) FROM Salary)'), 'include-latest': final.replace('month_no < latest_month', 'month_no <= latest_month'), 'reverse-output': final.replace('ORDER BY employee_id, month_no;', 'ORDER BY employee_id, month_no DESC;') } record['mutants'] = [] for name, query in variants.items(): assert query != final, name witness = cases[3] if name == 'only-two-months' else main got = sql('USE blog_salary; DELETE FROM Salary;' + insert(witness) + query) assert got != oracle(witness), name record['mutants'].append({'name':name,'input':witness,'actual':got,'expected':oracle(witness),'rejected':True}) record['passed'] = True finally: server.terminate() try: server.wait(timeout=30) except subprocess.TimeoutExpired: server.kill() server.wait(timeout=5) record['serverStopped'] = server.poll() is not None record['socketRemoved'] = not sock.exists() (run / 'results.json').write_text(json.dumps(record, indent=2)) print('RESULTS=' + str(run / 'results.json')) assert record['passed'] and record['serverStopped'] and record['socketRemoved'] print('SALARY_CHECKS_PASSED')参考：MySQL窗口范围说明。本文是自定义报表练习的本地验证，不是LeetCode提交结果，也不凭小数据测试宣称大表性能。
