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

[freewind]

主表大概 100 万条数据，标签有 4000 个左右，有层级划分，每条数据有几十个标签。

按标签查询的时候要把打了当前标签和它下级标签的数据全部查出来

一次要查询多个标签, 还会和其它多个条件组合查询

标签列做了索引, 用 a,b,c,c01 这样保存

现在是用 find_in_set 和 like 查询有些慢

请问要怎么做才能提高查询速度?

数据库是 mysql 5.7

[yjhatfdu2]

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

[freewind]

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

[wanniwa]

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

[Morriaty]

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

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

[yidinghe]

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

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

你也可以用其他 AI 尝试。

[freewind]

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

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

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

[hnliuzesen]

如果打算用 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) 一起使用，可以和上面的建议配合

[yjhatfdu2]

实测来了，
创建测试表，并插入 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 来存和查，结果应该差不多

[freewind]

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

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

[yjhatfdu2]

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

[freewind]

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

[wxf666]

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

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

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

[jeesk]

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

[GeekGao]

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

[yjhatfdu2]

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

[yjhatfdu2]

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

[guochenglong]

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

[wxf666]

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

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

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

[yjhatfdu2]

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

[tedzhou1221]

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

[tedzhou1221]

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

[freewind]

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

@yjhatfdu2 #16 这速度可以

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

[wxf666]

@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)),
　　　　　　 IIF(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";
　　}
}
```

[freewind]

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

@wxf666 #23 👍 学习了