最近三个月，不是最近三条记录：用一段断月工资看懂SQL窗口

最近三个月不是最近三条记录用一段断月工资看懂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月为12026年12月为122027年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得到5006001100。所以员工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所以按行算会得到2005006001300。按月份算2月早就离开窗口了。PARTITION BY employee_id也不能省每个人的工资要分别算。员工2的2月和员工1的2月虽然编号相同不应该进入对方的总额。这里RANGE的偏移按那个数值排序表达式计算不能把2无条件理解成“两个月”。正因为本文排序列是连续月份编号差2才表示两个月。若排序列换成天数2就成了两天如果换成工资也不是月份了。最新月要按人找员工1最新有6月员工2最新只有4月。报表要去掉的是员工1的6月、员工2的4月。员工11 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、4ROWS和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.jsonPython保存为同目录的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_PATHstr(package / usr/lib/x86_64-linux-gnu)) run Path(tempfile.mkdtemp(prefixsalary-, dirbase)) (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-privNULL, --innodb-buffer-pool-size67108864, --innodb-redo-log-capacity16777216, --log-error str(run / server.log)] init subprocess.run(args [--initialize-insecure], envenv, capture_outputTrue, textTrue, timeout90) if init.returncode: raise RuntimeError(init.stderr) client [str(package / usr/bin/mysql), --no-defaults, --protocolSOCKET, --socket str(sock), -u, root, --batch, --skip-column-names, --connect-timeout2] server subprocess.Popen(args [--skip-networking, --mysqlx0, --socket str(sock), --pid-file str(run / mysql.pid), --tmpdir str(run)], envenv, stdoutsubprocess.DEVNULL, stderrsubprocess.DEVNULL) record {passed: False, run: str(run)} def sql(text): result subprocess.run(client, inputtext, envenv, capture_outputTrue, textTrue, timeout120) 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([fSELECT __step_{i}__;, step[sql]]) outputs chunks(sql(\n.join(commands)), __step_) assert outputs [s[expected] for s in steps], outputs record[articleSteps] [dict(s, actuala) 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), repeat8): cases.append([(e,m,0 if v1 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), fSELECT __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(timeout30) except subprocess.TimeoutExpired: server.kill() server.wait(timeout5) record[serverStopped] server.poll() is not None record[socketRemoved] not sock.exists() (run / results.json).write_text(json.dumps(record, indent2)) print(RESULTS str(run / results.json)) assert record[passed] and record[serverStopped] and record[socketRemoved] print(SALARY_CHECKS_PASSED)参考MySQL窗口范围说明。本文是自定义报表练习的本地验证不是LeetCode提交结果也不凭小数据测试宣称大表性能。