如何在 Sqlite 中使用时间序列，以及快速的时间范围查询？

如何解决如何在 Sqlite 中使用时间序列，以及快速的时间范围查询？

假设我们使用 Unix 时间戳列 ts 在 Sqlite 数据库中记录事件：

CREATE TABLE data(ts INTEGER,text TEXT);   -- more columns in reality

并且我们想要快速查找日期时间范围，例如：

SELECT text FROM data WHERE ts BETWEEN 1608710000 and 1608718654;

像这样，EXPLAIN QUERY PLAN 给出了 SCAN TABLE data，这是不好的，所以一个明显的解决方案是使用 CREATE INDEX dt_idx ON data(ts)创建索引。

那么问题就解决了，但是对于一个已经增加的序列/已经排序的列 ts 维护一个索引是一个糟糕的解决方案，我们可以使用直接在 O(log n) 中进行 B 树搜索。在内部，这将是索引：

ts           rowid
1608000001   1
1608000002   2
1608000012   3
1608000077   4

这会浪费数据库空间（当查询必须首先查看索引时会浪费 CPU）。

为了避免这种情况：

• (1) 我们可以使用 ts 作为 INTEGER PRIMARY KEY，所以 ts 将是 rowid 本身。但这失败了，因为 ts 不是唯一的：2 个事件可以在同一秒（甚至在同一毫秒）发生。

  例如参见SQLite Autoincrement中给出的信息。

• (2) 我们可以使用 rowid 作为时间戳 ts 与递增的数字连接。示例：

   16087186540001      
   16087186540002
   [--------][--]
       ts     increasing number 
  

  那么 rowid 是唯一的并且严格递增（假设每秒事件数少于 10k），并且不需要索引。查询 WHERE ts BETWEEN a AND b 将简单地变为 WHERE rowid BETWEEN a*10000 AND b*10000+9999。

  但是有没有一种简单的方法可以让 Sqlite INSERT 一个 rowid 大于或等于给定值的项目？假设当前时间戳是 1608718654 并且出现两个事件：

    CREATE TABLE data(ts_and_incr INTEGER PRIMARY KEY AUTOINCREMENT,text TEXT);
    INSERT INTO data VALUES (NEXT_UNUSED(1608718654),"hello")  #16087186540001 
    INSERT INTO data VALUES (NEXT_UNUSED(1608718654),"hello")  #16087186540002
  

更一般地说，如何使用 Sqlite 以最佳方式创建时间序列，以进行快速查询 WHERE timestamp BETWEEN a AND b？

解决方法

第一个解决方案

问题中详述的方法（2）似乎效果很好。在基准测试中，我获得了：

• 简单的方法，没有索引：18 MB 的数据库，86 毫秒的查询时间
• 简单方法，带索引：32 MB 数据库，12 毫秒查询时间
• 方法 (2)：18 MB 数据库，12 毫秒查询时间

这里的关键是使用 dt 作为 INTEGER PRIMARY KEY，所以 它将是行 ID 本身（另见 Is an index needed for a primary key in SQLite?），使用B 树，并且不会另一个隐藏的 rowid 列。因此，我们避免了一个额外的索引，它会产生对应关系 dt => rowid：这里 dt 是 行 ID。

我们还使用了 AUTOINCREMENT，它在内部创建了一个 sqlite_sequence 表，用于跟踪最后添加的 ID。这在插入时很有用：因为两个事件可能具有相同的以秒为单位的时间戳（即使使用毫秒或微秒时间戳也是可能的，操作系统可能会截断精度），我们使用 timestamp*10000 和last_added_ID + 1 以确保它是独一无二的：

 MAX(?,(SELECT seq FROM sqlite_sequence) + 1)

代码：

import sqlite3,random,time
db = sqlite3.connect('test.db')
db.execute("CREATE TABLE data(dt INTEGER PRIMARY KEY AUTOINCREMENT,label TEXT);")

t = 1600000000
for i in range(1000*1000):
    if random.randint(0,100) == 0:  # timestamp increases of 1 second with probability 1%
        t += 1
    db.execute("INSERT INTO data(dt,label) VALUES (MAX(?,(SELECT seq FROM sqlite_sequence) + 1),'hello');",(t*10000,))
db.commit()

# t will range in a ~ 10 000 seconds window
t1,t2 = 1600005000*10000,1600005100*10000  # time range of width 100 seconds (i.e. 1%)
start = time.time()
for _ in db.execute("SELECT 1 FROM data WHERE dt BETWEEN ? AND ?",(t1,t2)): 
    pass
print(time.time()-start)

---

使用 WITHOUT ROWID 表

这是另一种使用 WITHOUT ROWID 的方法，它提供 8 毫秒 查询时间。我们必须自己实现一个自动递增的 id，因为使用 WITHOUT ROWID 时 AUTOINCREMENT 不可用。
WITHOUT ROWID 当我们想要使用 PRIMARY KEY(dt,another_column1,another_column2,id) 并避免有额外的 rowid 列时很有用。我们将只有一个，而不是为 rowid 提供一棵 B 树和为 (dt,...) 提供一棵 B 树。

db.executescript("""
    CREATE TABLE autoinc(num INTEGER); INSERT INTO autoinc(num) VALUES(0);

    CREATE TABLE data(dt INTEGER,id INTEGER,label TEXT,PRIMARY KEY(dt,id)) WITHOUT ROWID;
    
    CREATE TRIGGER insert_trigger BEFORE INSERT ON data BEGIN UPDATE autoinc SET num=num+1; END;
    """)

t = 1600000000
for i in range(1000*1000):
    if random.randint(0,100) == 0: # timestamp increases of 1 second with probabibly 1%
        t += 1
    db.execute("INSERT INTO data(dt,id,label) VALUES (?,(SELECT num FROM autoinc),?);",(t,'hello'))
db.commit()

# t will range in a ~ 10 000 seconds window
t1,t2 = 1600005000,1600005100  # time range of width 100 seconds (i.e. 1%)
start = time.time()
for _ in db.execute("SELECT 1 FROM data WHERE dt BETWEEN ? AND ?",t2)): 
    pass
print(time.time()-start)

---

粗略排序的 UUID

更一般地说，问题与按日期时间“粗略排序”的 ID 有关。更多相关信息：

• ULID（通用唯一的字典序可排序标识符）
• Snowflake
• MongoDB ObjectId

所有这些方法都使用一个 ID，即：

[---- timestamp ----][---- random and/or incremental ----]

,

我不是 SqlLite 方面的专家，但使用过数据库和时间序列。我以前也遇到过类似的情况，我来分享一下我的概念解决方案。

你的问题中有一些答案的部分内容，但没有答案。

我这样做的方式，创建了 2 个表，一个表 (main_logs) 将记录时间以秒为增量作为日期作为主键的整数，其他表日志包含在该特定时间创建的所有日志 (main_sub_logs)您的案例每秒最多可以处理 10000 条日志。 main_sub_logs 引用了 main_logs，它包含每个日志秒和 X 个日志，属于该秒并具有自己的计数器 id，然后重新开始。

通过这种方式，您可以将时间序列的查找时间限制为事件窗口的秒数，而不是将所有日志集中在一处。

通过这种方式，您可以连接这两个表，当您在 2 个特定时间之间从第一个表中查找时，您将获得所有日志。

那么这里是我如何创建我的 2 个表：

CREATE TABLE IF NOT EXISTS main_logs (
  id INTEGER PRIMARY KEY
);

CREATE TABLE IF NOT EXISTS main_sub_logs (
   id INTEGER,ref INTEGER,log_counter INTEGER,log_text text,PRIMARY KEY (id),FOREIGN KEY (ref) REFERENCES main_logs(id)
)

我插入了一些虚拟数据：

现在让我们查询 1608718655 和 1608718656 之间的所有日志

SELECT * FROM main_logs AS A
JOIN main_sub_logs AS B ON A.id == B.Ref
WHERE A.id >= 1608718655 AND A.id <= 1608718656

会得到这样的结果：