请教多标签查询怎么做效率高?

你这样完全没用上索引，你这个场景，最好就使用 postgresql ，使用 array 或者 json 存 tags ，然后建立 GIN 索引，然后每次查询把标签和子标签组成目标 tags ，然后用 select * from xxx where tags @> ARRAY[tag1,tag2,tag3]，就可以用上 GIN 索引，一个亿数据也没啥

@yjhatfdu2 #1 谢谢, 我去找找这方面的资料

@freewind #2 我们就是用的这个

100 万条数据，感觉直接弄个内存级的搜索引擎就行了，就是内存和数据库的一致性要写点逻辑维护下。

预算够的话，直接上 es 省事，弄个 nested array 字段就能解决

提示词：我正在做一个关系数据库的表设计。要保存的数据分为主表记录和标签两种，其中一条主表记录可关联多个标签，而标签存在层级结构。表设计需要满足下面的查询要求：以标签 ID 为条件，查询其本身和所有下级标签关联的主表记录。

下面是豆包的生成内容：
https://www.doubao.com/thread/w38665315320066

你也可以用其他 AI 尝试。

@wanniwa #3 是用了#1 的方法吗

@orriaty #4 小公司，预算不足

@yidinghe #5 多了一个关系表，不知道查询起来速度怎么样

如果打算用 pg ，也可以试试 [recursive cte]( https://www.postgresql.org/docs/current/queries-with.html#QUERIES-WITH-RECURSIVE) + [lateral]( https://www.postgresql.org/docs/current/queries-table-expressions.html#QUERIES-LATERAL) 一起使用，可以和上面的建议配合

实测来了，
创建测试表，并插入 1 亿条测试数据，每一条包含 50 个随机标签，每个标签有 4000 种可能
create table tags_test(id serial primary key,tags array[int]);
create function random_array() returns int[] as $$ select array_agg((4000*random())::int) from generate_series(1,50)) $$;
insert into tags_test(tags) select random_array() from generate_series(1,100000000);
创建 GIN 索引
//最好临时增加 maintenance_work_mem ，会加快索引构建
//set maintenance_work_mem to 1GB;
create index ON tags_test using gin(tags);
简单查个包含几个 tags 的数据
postgres=# select count(*) from tags_test where tags @>array[1,2,3,4];
count
-------
2
(1 row)

Time: 135.185 ms
postgres=# select count(*) from tags_test where tags @>array[1,2,3];
count
-------
190
(1 row)

Time: 123.327 ms
当然，实际情况下标签肯定不是随机分布的，可能会更快或者更慢，也可以试试用 jsonb 来存和查，结果应该差不多

@hnliuzesen #7 没用过，学习一下

@yjhatfdu2 #8 按这种方法，是不是要把标签平铺，先把所有下级标签存进去

@freewind 可以这样，也可以像#7 说的一样，每次用递归 CTE 把所有子层级标签查出来作为条件，这样标签层级可以更新，存储容量也可以小点，你们标签少可以考虑用更小的 int 做标签，也能省

@yjhatfdu2 #10 要是递归速度还行就打算先用这个, 数据的标签经常改, 标签数量也在不断增加

@yjhatfdu2 #8 这个索引，实际有多大呢？

我怀疑它是按照 (array_value, id) 来存的，即存了 50 亿条。

每条 10 字节的话，也要快 50 GB 了。。

自己写倒排序, 维护索引. 关键字: 广告系统布尔表达式, 倒排序

最简单的方案：标签字段合并、全文索引标签集合字段。
只是信息有限不知道符不符合你们的场景。

@wxf666 这个索引占用 9G 左右，pg 的 GIN 是通用反向索引，并不是直接按照这样展开存的，原理类似 es 的反向索引

@freewind 你这几千条标签递归查估计也就 1ms 之内的事情

还有一种 bitmap 存储，用位运算查

@yjhatfdu2 #15 是 (tag_id, main_ids_array) 这样存的吗？

即，有 4000 条记录，总计 50 亿个主表 ID （平均每条记录有 125W 个主表 ID ）？

每个 ID 有 4 字节，算下来也要 (5e9 * 4) / (1 << 20) = 18.6 GB 呀。。

@wxf666 你算的很对，表大概 20G 多一点，但是索引不是这么存的，你可以理解为反向索引存了 tag 和这个 tag 在哪些行中出现，这里面我怀疑是用 bitmap+压缩来实现的，有兴趣可以看下 pg 的 gin 相关的代码，所以实际上索引体积只有 9G 左右

好奇请教一下，“标签列做了索引, 用 a,b,c,c01 这样保存”

如果更新了标签的，或删除了标签，是全部标签列都匹配查询修改吗？

@jeesk #13
@GeekGao #14 我去研究一下

@yjhatfdu2 #16 这速度可以

@guochenglong #17 有了解过,感觉不太适合这个场景, 标签有些多

@yjhatfdu2 #19 我用 SQLite 试了下，感觉用普通 SQL ，也勉强能满足这个需求。

## 结果

- 数据：4000 标签，1 亿数据，50 标签/数据（即，倒排索引中，约 50 亿数据 ID ）
- 性能：0.8s 查询一个标签，拥有的 625W 数据 ID ，升序输出
- 体积：倒排索引表，约 11 GB

## 环境

- SYS：Win11
- CPU：i7-4790
- MEM：16 GB ，1600 MHz
- HDD：顺序读 150 MB/s ，4K 随机读 0.65 MB/s

## 多标签呢？

SQLite 不支持 Sort Merge Join ，多个标签一起查时，很慢。。
需要物化每个标签的结果，再由 SQLite 进行布隆匹配、B 树二分查找。。

（测前，都用 RamMap 清除了，系统对文件的缓存）
2 个 625W 数据 ID 的标签，结果 250W ，耗时 4.9 秒，
3 个 625W 数据 ID 的标签，结果 205W ，耗时 8.5 秒，
4 个 625W 数据 ID 的标签，结果 120W ，耗时 11.9 秒。。

不知道能不能用 Virtual Table ，写个表值函数（知道有，没写过），自己实现：

- Sort Merge Join 的效果，流式匹配所有标签，免除物化表、上千万次 B 树匹配的开销
- 提速读取数据 ID （ SQL 中，若写为 `SELECT COUNT(*)`，不计算具体数据 ID ，会提速到 0.1s ）

## 倒排索引结构

- 表结构：`（＜标签 ID ，高位数据 ID ＞，低 11 位数据 ID 集合 BLOB ）`
- 若 > 146 个低位数据 ID ，第 3 列视为 293 字节的 Bitmap ，但每字节最高位弃用（ SQLite 中没有便捷转换 128~255 为单字节 BLOB 的方法）
- 否则，每个低位数据 ID ，转为二字节 UTF-8 字符串（ 0~127 需后补 `x'00'`），并串联起来。

## 测试数据

- 标签 ID：`[0, 3999]`
- 数据 ID：`[1, 1e8]`
- 每个数据拥有的标签 ID：`[(数据 ID * 1) % 4000, (数据 ID * 2) % 4000, ..., (数据 ID * 50) % 4000]`
- 标签 ID 为 400 、1600 、2800 、3600 时，拥有的数据 ID 最多，都为 625W 个

## 查询 SQL

```sql
-- V 站吞空格，缩进改为全角空格了
WITH
　　 t1 AS MATERIALIZED (
　　　　 SELECT (data_hi_id << 11) |
　　　　　　 CASE WHEN i.stop = 7 THEN -- 低位数据 ID 量 <= 146 ，逐 2 字节翻译
　　　　　　　　 IFNULL(unicode(substr(data_lo_ids, j.value, 2)), 0)
　　　　　　 WHEN j.stop & (1 << (6 + j.value)) THEN -- Bitmap ，逐 bit 翻译
　　　　　　　　 i.value + j.value - 1
　　　　　　 END AS data_id
　　　　 FROM tag_data
　　　　 JOIN generate_series(7, IIF(length(data_lo_ids) < 293, 1, 293) * 7, 7) i
　　　　 JOIN generate_series(
　　　　　　 IIF(i.stop > 7, -6, 1),
　　　　　　 IIF(i.stop > 7, unicode(substr(data_lo_ids, i.value / 7, 1)), length(data_lo_ids)),
　　　　　　 II(i.stop > 7, 1, 2)) j
　　　　 WHERE tag_id = '《第一个标签 ID 》'
　　　　-- 这句加了好慢。。怀疑是反复计算 data_id 导致
　　　　-- AND data_id IS NOT NULL
　　),
　　-- 以下结构类似，省略了
　　 t2 AS (...),
　　 t3 AS (...),
　　 t4 AS (...)
-- 写成 COUNT(*) 且仅 t1 时，飞快
-- 估计是不用算具体 data_id 了
SELECT COUNT(data_id)
FROM t1
JOIN t2 USING(data_id)
JOIN t3 USING(data_id)
JOIN t4 USING(data_id)
-- 匹配到后面，预计数据量很少时，可采用这种方式：
/*
WHERE EXISTS (
　　 SELECT 1
　　 FROM tag_data
　　 WHERE tag_id = '《第四个标签 ID 》'
　　 AND data_hi_id = data_id >> 11
　　 AND CASE WHEN octet_length(data_lo_ids) <= 292 THEN
　　　　　　 instr(data_lo_ids, IIF(data_id & 0x780, char(data_id & 0x7FF), char(data_id & 0x7FF, 0)))
　　　　 ELSE
　　　　　　 unicode(substr(data_lo_ids, (data_id & 0x7FF) / 7 + 1, 1)) & (1 << (data_id & 0x7FF) % 7)
　　　　 END)
*/
```

## 增删改

用触发器实现。但很慢。。

C++ 新人，正好练练手，生成倒排索引库（这都要花十几分钟。。）：

```c++
// 同上，缩进是全角空格
#include <string>
#include <vector>
#include <iostream>
#include <string_view>
#include "sqlite_modern_cpp.h"

constexpr size_t MAX_DATA_ID = 1e8;
constexpr size_t MAX_TAG_COUNT = 50;
constexpr size_t MAX_TAG_ID = 4000 - 1;
constexpr size_t MAX_MEMORY = 8ull << 30;
constexpr auto DB_PATH = "D:/out.db";

int main() {

　　 struct TagData {
　　　　 struct {
　　　　　　 uint8_t data_lo_ids[(2048 + 6) / 7]{};
　　　　} data_hi_ids[(MAX_DATA_ID >> 11) + 1]{};
　　};

　　 constexpr int tag_steps = MAX_MEMORY / sizeof(TagData);

　　 sqlite::database db(DB_PATH);
　　 db << "CREATE TABLE IF NOT EXISTS tag_data ("
　　　　　"　　 tag_id INT,"
　　　　　"　　 data_hi_id INT,"
　　　　　"　　 count INT,"
　　　　　"　　 data_lo_ids BLOB,"
　　　　　"　　 PRIMARY KEY (tag_id, data_hi_id)"
　　　　　") WITHOUT ROWID";

　　 std::string blob_buf;
　　 std::vector<uint16_t> data_lo_ids;
　　 auto insert_stmt = db << "INSERT INTO tag_data VALUES (?, ?, ?, CAST(? AS BLOB))";

　　 for (size_t tag_id_beg = 0; tag_id_beg <= MAX_TAG_ID; tag_id_beg += tag_steps) {

　　　　 auto tag_id_end = (std::min)(tag_id_beg + tag_steps - 1, MAX_TAG_ID);
　　　　 auto tags = std::make_unique<TagData[]>(tag_id_end - tag_id_beg + 1);

　　　　 db << "BEGIN";
　　　　 std::cout
　　　　　　<< "正在计算 1~" << MAX_DATA_ID << "，每个数的 1~" << MAX_TAG_COUNT << " 倍，÷" << MAX_TAG_ID + 1
　　　　　　<< " 后的余数，是否在 [" << tag_id_beg << ", " << tag_id_end << "] 范围内……" << std::endl;

　　　　 for (size_t data_id = 1; data_id <= MAX_DATA_ID; data_id++) {
　　　　　　 for (size_t times = 1; times <= MAX_TAG_COUNT; times++) {
　　　　　　　　 size_t tag_id = (data_id * times) % (MAX_TAG_ID + 1);
　　　　　　　　 if (tag_id >= tag_id_beg && tag_id <= tag_id_end)
　　　　　　　　　　 tags[tag_id - tag_id_beg]
　　　　　　　　　　　　.data_hi_ids[data_id >> 11]
　　　　　　　　　　　　.data_lo_ids[(data_id & 0x7FF) / 7] |= 1 << ((data_id & 0x7FF) % 7);
　　　　　　}
　　　　}

　　　　 for (size_t tag_id = tag_id_beg; tag_id <= tag_id_end; tag_id++) {

　　　　　　 auto& tag = tags[tag_id - tag_id_beg];
　　　　　　 std::cout << "正在写入 tag_id = " << tag_id << " 至数据库……\r";

　　　　　　 for (auto& data_hi: tag.data_hi_ids) {
　　　　　　　　 data_lo_ids.clear();
　　　　　　　　 for (size_t i = 0; i < sizeof data_hi.data_lo_ids; i++)
　　　　　　　　　　 if (data_hi.data_lo_ids[i])
　　　　　　　　　　　　 for (size_t j = 0; j < 7; j++)
　　　　　　　　　　　　　　 if (data_hi.data_lo_ids[i] & (1 << j))
　　　　　　　　　　　　　　　　 data_lo_ids.push_back(i * 7 + j);

　　　　　　　　 if (data_lo_ids.size() > 292 / 2)
　　　　　　　　　　 insert_stmt
　　　　　　　　　　　　<< tag_id << (&data_hi - tag.data_hi_ids) << data_lo_ids.size()
　　　　　　　　　　　　<< std::string_view((char *) data_hi.data_lo_ids, sizeof data_hi.data_lo_ids)
　　　　　　　　　　　　>> []{};

　　　　　　　　 else if (!data_lo_ids.empty()) {
　　　　　　　　　　 blob_buf.clear();
　　　　　　　　　　 for (auto lo_id: data_lo_ids)
　　　　　　　　　　　　 if (lo_id > 127)
　　　　　　　　　　　　　　 blob_buf.append({char(0xC0 | (lo_id >> 6)), char(0x80 | (lo_id & 0x3F))});
　　　　　　　　　　　　 else
　　　　　　　　　　　　　　 blob_buf.append({char(lo_id), 0});
　　　　　　　　　　 insert_stmt
　　　　　　　　　　　　<< tag_id << (&data_hi - tag.data_hi_ids) << data_lo_ids.size() << blob_buf
　　　　　　　　　　　　>> []{};
　　　　　　　　}
　　　　　　}
　　　　}

　　　　 db << "COMMIT";
　　}
}
```

@tedzhou1221 #20 修改标签不动,删除会把标签都清除

@wxf666 #23 学习了