适用场景

订单日报按自然日统计,数据库把时间保存为 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 只是把问题绑定到当前精度,后续从毫秒切到微秒还可能出错。

排查顺序

  1. 先记录查询的日期、业务时区和实际绑定的起止值,不要只看前端传来的日期。
  2. 从前一天末尾、当天零点、当天最后一秒、次日零点各取一条脱敏记录,核对归属。
  3. 检查写入格式是否一致:空格与 T 分隔、是否带偏移量、秒的小数位、是否混有整数时间戳。
  4. 检查过滤条件是否对时间列施加了 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 时区数据、并发一致性或大表性能。

总结

日报边界应定义成 [当天零点, 次日零点)。闭区间会重复计算次日零点,手写最后一秒又可能漏掉小数秒。把时间格式、业务时区和区间规则放进同一份约定,再用边界记录做回归测试,才能让相邻日报稳定衔接。