适用场景
订单日报按自然日统计,数据库把时间保存为 TEXT。开发把查询区间设成“当天零点到次日零点”,却发现每天零点的订单同时进入两天报表。查询执行成功、金额也能加总,错误只在日期交界处出现。
本文用内存 SQLite 复现,不涉及线上订单。示例约定:所有时间均为业务所在地固定 UTC+08:00 时间,格式统一为 YYYY-MM-DD HH:MM:SS.ffffff,使用默认 BINARY 排序,字段非空。这个约定必须由写入边界保证,不能仅靠 TEXT 类型声明。
先区分重复与遗漏
下面的条件包含两个端点:
SELECT id FROM orders
WHERE created_at BETWEEN :start_at AND :end_at;
若绑定值分别为 2026-11-11 00:00:00.000000 和 2026-11-12 00:00:00.000000,次日零点也会入选。SQLite 表达式文档明确了 BETWEEN 与大于等于、 小于等于比较的等价关系。
另一种补丁是把结束值改成 23:59:59.000000,这又会漏掉最后一秒内的订单,例如 23:59:59.500000。继续堆叠 9 只是把问题绑定到当前精度,后续从毫秒切到微秒还可能出错。
排查顺序
- 先记录查询的日期、业务时区和实际绑定的起止值,不要只看前端传来的日期。
- 从前一天末尾、当天零点、当天最后一秒、次日零点各取一条脱敏记录,核对归属。
- 检查写入格式是否一致:空格与 T 分隔、是否带偏移量、秒的小数位、是否混有整数时间戳。
- 检查过滤条件是否对时间列施加了
date()等函数;这可能影响普通时间索引的利用,需要查看实际执行计划。
本例时间采用同一时区、同一精度和固定宽度,才可以直接依赖文本比较。SQLite 没有专门的日期存储类型,可使用 TEXT、REAL 或 INTEGER;这不代表任意日期字符串都能正确排序,参见官方类型说明。
修复:统一使用左闭右开区间
SELECT id FROM orders
WHERE created_at >= :start_at
AND created_at < :end_at
ORDER BY created_at, id;
start_at 是当天零点,end_at 是下一天零点。当天零点属于今天,次日零点只属于明天。两个相邻日报区间在边界处没有交集,也不需要猜测一天最后一个可表示时刻。
时间列可以建立普通索引:
CREATE INDEX idx_orders_created_at ON orders(created_at);
是否实际使用索引取决于表规模、统计信息与查询条件。上线前用真实规模的脱敏样本查看 EXPLAIN QUERY PLAN,不要把本地小表的执行耗时当作性能结论。
可直接运行的复现与回归测试
将下面完整代码保存为 daily_range.py,使用 Python 3.10 或以上执行 python daily_range.py。只使用标准库,不读写业务文件,不访问网络。
import sqlite3
import unittest
from contextlib import closing
from datetime import date, datetime, time, timedelta
SCHEMA = """
CREATE TABLE orders (
id INTEGER PRIMARY KEY,
created_at TEXT NOT NULL
);
CREATE INDEX idx_orders_created_at ON orders(created_at);
"""
def day_bounds(day_text: str) -> tuple[str, str]:
"""严格接受日期,在固定业务时区的本地表示中生成日界。"""
if not isinstance(day_text, str):
raise TypeError("日期必须是字符串")
if len(day_text) != 10:
raise ValueError("日期必须使用 YYYY-MM-DD 格式")
day = date.fromisoformat(day_text)
if day.isoformat() != day_text:
raise ValueError("日期必须使用 YYYY-MM-DD 格式")
if day == date.max:
raise ValueError("日期超出可生成次日日界的范围")
start = datetime.combine(day, time.min)
end = start + timedelta(days=1)
return (
start.isoformat(sep=" ", timespec="microseconds"),
end.isoformat(sep=" ", timespec="microseconds"),
)
def find_ids(connection: sqlite3.Connection, day_text: str) -> list[int]:
"""按左闭右开区间查询,日期参数不参与 SQL 拼接。"""
start, end = day_bounds(day_text)
rows = connection.execute(
"SELECT id FROM orders WHERE created_at >= ? "
"AND created_at < ? ORDER BY created_at, id",
(start, end),
).fetchall()
return [row[0] for row in rows]
class DailyRangeTests(unittest.TestCase):
"""覆盖日界、最后一秒、相邻区间及输入失败路径。"""
def setUp(self) -> None:
self.connection = sqlite3.connect(":memory:")
self.addCleanup(self.connection.close)
self.connection.executescript(SCHEMA)
self.connection.executemany(
"INSERT INTO orders VALUES (?, ?)",
[
(1, "2026-11-10 23:59:59.999999"),
(2, "2026-11-11 00:00:00.000000"),
(3, "2026-11-11 23:59:59.500000"),
(4, "2026-11-12 00:00:00.000000"),
],
)
def test_today(self) -> None:
self.assertEqual(find_ids(self.connection, "2026-11-11"), [2, 3])
def test_adjacent_days(self) -> None:
before = find_ids(self.connection, "2026-11-10")
today = find_ids(self.connection, "2026-11-11")
after = find_ids(self.connection, "2026-11-12")
self.assertEqual(before, [1])
self.assertEqual(after, [4])
self.assertEqual(before + today + after, [1, 2, 3, 4])
def test_empty_day(self) -> None:
self.assertEqual(find_ids(self.connection, "2026-11-13"), [])
def test_calendar_boundaries(self) -> None:
for day_text, next_day in [
("2024-02-29", "2024-03-01"),
("2026-12-31", "2027-01-01"),
]:
with self.subTest(day=day_text):
start, end = day_bounds(day_text)
self.assertEqual(start, day_text + " 00:00:00.000000")
self.assertEqual(end, next_day + " 00:00:00.000000")
def test_invalid_dates(self) -> None:
for value in ["2026-02-29", "20261111", "2026-11-11' OR 1=1", "9999-12-31"]:
with self.subTest(value=value), self.assertRaises(ValueError):
day_bounds(value)
def test_wrong_type(self) -> None:
with self.assertRaises(TypeError):
day_bounds(None)
if __name__ == "__main__":
with closing(sqlite3.connect(":memory:")) as connection:
connection.executescript(SCHEMA)
connection.executemany(
"INSERT INTO orders VALUES (?, ?)",
[(1, "2026-11-11 23:59:59.500000"),
(2, "2026-11-12 00:00:00.000000")],
)
print(connection.execute(
"SELECT id FROM orders WHERE created_at BETWEEN ? AND ? ORDER BY id",
day_bounds("2026-11-11"),
).fetchall())
print(find_ids(connection, "2026-11-11"))
unittest.main()
前两行输出应分别为 [(1,), (2,)] 和 [1]:旧查询混入了次日零点,修复后只保留当天订单。随后六项测试应通过。
test_today 同时覆盖当天起点和最后一秒的小数部分;test_adjacent_days 验证连续三天拼接后没有重复或遗漏;日期输入测试拒绝非法日期和不符合约定的字符串。SQL 中的两个问号是绑定占位符,不能改成字符串插值。实际多租户系统还必须从服务端鉴权结果增加租户过滤,不能把传入的租户号视为授权。
上线前需要确认的边界
- 本文按固定 UTC+08:00 业务日解释无偏移时间,代码不负责时区转换。若数据库保存 UTC,应先在业务地区构造当天和次日的本地零点,再分别转换到 UTC;有夏令时的地区不能直接把 UTC 起点加 24 小时当作下一天。
- 所有历史行和新写入值都应遵守统一格式。直接比较 TEXT 不会替你校验日期,也不会统一时区。先审计再迁移,不能在生产库盲目批量替换分隔符。
- 若使用整数时间戳,统一秒、毫秒或微秒单位,并将两个边界转换到同一单位,再使用同样的
>=和<条件。 - 此修复解决日期归属,不保证多次查询看到同一业务快照。迟到订单、订单状态变化和补录应另设报表截止策略及重算规则。
- 本地验证只覆盖内存 SQLite 的查询结果及日期参数生成,没有验证生产 ORM、IANA 时区数据、并发一致性或大表性能。
总结
日报边界应定义成 [当天零点, 次日零点)。闭区间会重复计算次日零点,手写最后一秒又可能漏掉小数秒。把时间格式、业务时区和区间规则放进同一份约定,再用边界记录做回归测试,才能让相邻日报稳定衔接。
Discussion
评论