VictoriaLogs 设计与实现

// app/vlstorage/netinsert/netinsert.go
func (s *Storage) AddRow(streamHash uint64, r *logstorage.InsertRow) {
	idx := s.srt.getNodeIdx(streamHash)
	// sns 保存 vlstorage 节点列表
	sn := s.sns[idx]
	// sn.addRow 最终通过网络写入该 vlstorage 节点
	sn.addRow(r) 
}

// app/vlstorage/netinsert/netinsert.go
func (srt *streamRowsTracker) getNodeIdx(streamHash uint64) uint64 {
	if srt.nodesCount == 1 {
		// 如果只有一个 vlstorage 节点，那么就直接写入到该节点
		return 0
	}

    // streamHash := sid.id.lo ^ sid.id.hi，sid 即 streamID
    // sid.id = hash128(bb.B) 就是 stream 的标签和值 hash 得来的
    // 可以参见 LogRows.MustAdd 中的过程
	streamRows := srt.rowsPerStream[streamHash] + 1
	srt.rowsPerStream[streamHash] = streamRows

	// Stream 的前 1000 条日志写入同一个 vlstorage 组件
    // 对于只包含少量日志的 Stream，可提高局部性；当某个 Stream 量较大时则
    // 分散到不同的 vlstorage 节点，以提高查询性能。
	if streamRows <= 1000 {
		return streamHash % uint64(srt.nodesCount)
	}

	return uint64(fastrand.Uint32n(uint32(srt.nodesCount)))
}

// app/vlstorage/netinsert/netinsert.go
func (sn *storageNode) mustSendInsertRequest(pendingData *bytesutil.ByteBuffer) {
    // sn 代表当前选中的写入节点
	if err := sn.sendInsertRequest(pendingData); err == nil {
		return
	}

	// 初始节点写入失败后，从随机位置开始遍历节点列表重试
	for !sn.s.sendInsertRequestToAnyNode(pendingData) {
		// 省略
	}
}

// app/vlstorage/netinsert/netinsert.go
func (sn *storageNode) sendInsertRequest(pendingData *bytesutil.ByteBuffer) error {
	// 如果没有禁用压缩，就使用 zstd 压缩数据
	var body io.Reader
	if !sn.s.disableCompression {
		bb := zstdBufPool.Get()
		defer zstdBufPool.Put(bb)

		bb.B = zstd.CompressLevel(bb.B[:0], pendingData.B, 1)
		body = bb.NewReader()
	} else {
		body = pendingData.NewReader()
	}

	// 对 vlstorage 节点请求 internal/insert 接口
	reqURL := sn.getRequestURL("/internal/insert")
	req, err := http.NewRequestWithContext(ctx, "POST", reqURL, body)
	// ...
	resp, err := sn.c.Do(req)
	// ...
}

// app/vlselect/logsql/logsql.go
// ProcessQueryRequest handles /select/logsql/query request.
//
// See https://docs.victoriametrics.com/victorialogs/querying/#querying-logs
func ProcessQueryRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) {
	// 省略参数解析，metrics 采集等等

	// Execute the query
	if err := vlstorage.RunQuery(qctx, writeBlock); err != nil {
		// 省略错误处理
		return
	}
}

// app/vlstorage/main.go
func RunQuery(qctx *logstorage.QueryContext, writeBlock logstorage.WriteDataBlockFunc) error {
	// 这部分的目的是用来判断查询是否可以优化为：直接返回最后的 N 条日志，而不用先执行过滤再排序
	// 比如：
	// - 'sort by (_time desc) offset <offset> limit <limit>'
	// - 'first <limit> by (_time desc)'
	// - 'last <limit> by (_time)'
	qOpt, offset, limit := qctx.Query.GetLastNResultsQuery()
	if qOpt != nil {
		qctxOpt := qctx.WithQuery(qOpt)
		return runOptimizedLastNResultsQuery(qctxOpt, offset, limit, writeBlock)
	}

	// localStorage 在 vlstorage 节点中才会被赋值
	if localStorage != nil {
		return localStorage.RunQuery(qctx, writeBlock)
	}

	return netstorageSelect.RunQuery(qctx, writeBlock)
}

// app/vlstorage/netselect/netselect.go
func (s *Storage) RunQuery(qctx *logstorage.QueryContext, writeBlock logstorage.WriteDataBlockFunc) error {
	// nqr 代表 NetQueryRunner，用于执行分布式查询
	nqr, err := logstorage.NewNetQueryRunner(qctx, s.RunQuery, writeBlock)
	if err != nil {
		return err
	}

	search := func(stopCh <-chan struct{}, q *logstorage.Query, writeBlock logstorage.WriteDataBlockFunc) error {
		qctxLocal := qctx.WithQuery(q)
		return s.runQuery(stopCh, qctxLocal, writeBlock)
	}

	// nqr.Run 实际执行的还是 Storage.runQuery 方法
	// nqr.Run 其中会将 pipe 分成 remote pipe 和 local pipe 
	// 分别执行分布式查询和本地查询
	concurrency := qctx.Query.GetConcurrency()
	return nqr.Run(qctx.Context, concurrency, search)
}

// app/vlstorage/netselect/netselect.go
func (s *Storage) runQuery(stopCh <-chan struct{}, qctx *logstorage.QueryContext, writeBlock logstorage.WriteDataBlockFunc) error {

	// ...

	// runQuery 并发请求所有 vlstorage 节点
	for i := range s.sns {
		go func(nodeIdx int) {
			err := sn.runQuery(qctxLocal, func(db *logstorage.DataBlock) {
				writeBlock(uint(nodeIdx), db)
			})
		}(i)
	}

	// ...
}

func (sn *storageNode) runQuery(qctx *logstorage.QueryContext, processBlock func(db *logstorage.DataBlock)) error {
	path := "/internal/select/query"
	responseBody, reqURL, err := sn.getResponseBodyForPathAndArgs(qctx.Context, path, args)
	if err != nil {
		return err
	}
	defer responseBody.Close()

	// 解析响应，省略
}

// app/vlselect/internalselect/internalselect.go
func processQueryRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
	// ...

	// 调用与 vlselect 查询链相同的 RunQuery
	if err := vlstorage.RunQuery(qctx, writeBlock); err != nil {
		return err
	}
}

// app/vlstorage/main.go
func RunQuery(qctx *logstorage.QueryContext, writeBlock logstorage.WriteDataBlockFunc) error {
	// ...

	// vlstorage 使用本地存储执行查询
	if localStorage != nil {
		return localStorage.RunQuery(qctx, writeBlock)
	}
	return netstorageSelect.RunQuery(qctx, writeBlock)
}

// app/vlinsert/internalinsert/internalinsert.go
func RequestHandler(w http.ResponseWriter, r *http.Request) {
	// ...

	// CommonParams 包含了写入时的一些公共参数, 如：
	// 	TenantID         logstorage.TenantID
	// TimeFields       []string
	// MsgFields        []string
	// StreamFields     []string
	// IgnoreFields     []string
	// 等等
	cp, err := insertutil.GetCommonParams(r)
	
	// 根据压缩类型解析 body 中的数据, 再通过 parseData 转为 InsertRow 类型
	err = protoparserutil.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
		lmp := cp.NewLogMessageProcessor("internalinsert", false)
		irp := lmp.(insertutil.InsertRowProcessor)
		err := parseData(irp, data)
		lmp.MustClose()
		return err
	})
}

// app/vlinsert/internalinsert/internalinsert.go
func parseData(irp insertutil.InsertRowProcessor, data []byte) error {
	// 从对象池中获取 InsertRow 对象
	r := logstorage.GetInsertRow()

	src := data
	i := 0
	for len(src) > 0 {
		tail, err := r.UnmarshalInplace(src)
		
		src = tail
		i++

		// 将解析后的 InsertRow 添加到 InsertRowProcessor 中
		irp.AddInsertRow(r)
	}
}

// app/vlinsert/insertutil/common_params.go
func (lmp *logMessageProcessor) AddInsertRow(r *logstorage.InsertRow) {
	// 超过 MaxFieldsPerLine（默认 1000） 个字段的日志行会被丢弃
	if len(r.Fields) > *MaxFieldsPerLine {
		return
	}

	// 调用 logstorage.LogRows 将 InsertRow 添加到 LogRows 中
	lmp.lr.MustAddInsertRow(r)

	// 如果需要 flush，则调用 flushLocked 方法
	if lmp.lr.NeedFlush() {
		lmp.flushLocked()
	}
}

// app/vlinsert/insertutil/common_params.go
func (lmp *logMessageProcessor) flushLocked() {
	logRowsStorage.MustAddRows(lmp.lr)
}

// app/vlstorage/main.go
// 这个方法在 vlinsert 中也被调用，只不过就是使用的 netstorageInsert.AddRow 方法
func (*Storage) MustAddRows(lr *logstorage.LogRows) {
	if localStorage != nil {
		localStorage.MustAddRows(lr)
	} else {
		// Store lr across the remote storage nodes.
		lr.ForEachRow(netstorageInsert.AddRow)
	}
}

// lib/logstorage/storage.go
type Storage struct {
	path string 			// 存储目录的路径
	retention time.Duration // 数据保留时间
	flockF *os.File 		// 用于确保 Storage 仅被单个进程打开的文件锁

	partitions []*partitionWrapper // 分区列表，按时间排序，例如 partitions[0] 是最早的分区
	ptwHot *partitionWrapper       // 最新的分区，用于写入新的日志行

	// ... 省略其他字段
}

func (s *Storage) MustAddRows(lr *LogRows) {
	// Fast path: 
	// 尝试将 LogRows 全部写入到 ptwHot 中
	ptwHot := s.ptwHot
	if ptwHot != nil {
		if ptwHot.canAddAllRows(lr) {
			ptwHot.pt.mustAddRows(lr)
			return
		}
	}

	// Slow path:
	// 如果 LogRows 不能被 ptwHot 全部写入，那么就需要把 LogRows 拆分成多个 LogRows，写入到不同的分区中
	// PS: 历史的分区可能需要从磁盘加载，会增加写入延迟
	now := time.Now().UnixNano()
	minAllowedDay := s.getMinAllowedDay(now) // 保留策略：过去的时间
	maxAllowedDay := s.getMaxAllowedDay(now) // 保留策略：未来的时间
	minAllowedTimestamp := now - s.maxBackfillAge.Nanoseconds() // 最大能接受的历史日志时间

	// 遍历 LogRows 中的每个日志行，过滤掉不符合要求的日志行，
	// 并根据日志行的时间戳，将其添加到同一分区（日期）的 LogRows 中
	m := make(map[int64]*LogRows)
	for i, ts := range lr.timestamps {
		day := ts / nsecsPerDay		
		// 根据保留策略 和 最大能接受的历史日志时间 过滤掉不符合要求的日志行
		// ...

		lrPart := m[day]
		lrPart.mustAddInternal(lr.streamIDs[i], ts, lr.rows[i], lr.streamTagsCanonicals[i])
	}
	for day, lrPart := range m {
		ptw := s.getPartitionForWriting(day)
		if ptw != nil {
			ptw.pt.mustAddRows(lrPart)
		}
	}
}

// lib/logstorage/partition.go
type partition struct {
	s *Storage

	// path 是分区的完整目录路径。例如 /data/logstorage/partitions/20230801
	// /data/logstorage 可配置，partitions/20230801 是 VictoriaLogs 的目录结构
	path string 
	name string // name 是分区的名称，是目录名。如：20230801

	idb *indexdb // idb 是索引数据库，用于存储日志行的索引信息
	ddb *datadb  // ddb 是数据数据库，用于存储日志行的原始数据

	// ....
}

// lib/logstorage/partition.go
// mustAddRows 将数据写入 indexdb 和 datadb，并跳过已注册 Stream 的重复索引写入
func (pt *partition) mustAddRows(lr *LogRows) {
	// 将 新增 的 Stream 注册到 indexdb 中
	if !pt.idb.hasStreamID(streamID) {
		streamTagsCanonical := streamTagsCanonicals[rowIdx]
		pt.idb.mustRegisterStream(streamID, streamTagsCanonical)
	}

	// 最后把 LogRows 写入到 datadb 中
	pt.ddb.mustAddRows(lr)
}

// lib/logstorage/indexdb.go
type indexdb struct {
	path string          // path 是索引数据库的目录路径：如 path/to/partition/indexdb 目录
	partitionName string // partitionName 是索引数据库所属的分区名称(天)
	tb *mergeset.Table   // tb 是索引数据库的存储，用于存储索引信息
	s *Storage 	         // s 是 indexdb 所属的 Storage
}

// lib/logstorage/indexdb.go
func (idb *indexdb) mustRegisterStream(streamID *streamID, streamTagsCanonical string) {
	// Register tenantID:streamID entry.
	bufLen := len(buf)
	buf = marshalCommonPrefix(buf, nsPrefixStreamID, tenantID)
	buf = streamID.id.marshal(buf)
	items = append(items, buf[bufLen:])

	// Register tenantID:streamID -> streamTagsCanonical entry.
	bufLen = len(buf)
	buf = marshalCommonPrefix(buf, nsPrefixStreamIDToStreamTags, tenantID)
	buf = streamID.id.marshal(buf)
	buf = append(buf, streamTagsCanonical...)
	items = append(items, buf[bufLen:])

	// Register tenantID:name:value -> streamIDs entries.
	tags := st.tags
	for i := range tags {
		bufLen = len(buf)
		buf = marshalCommonPrefix(buf, nsPrefixTagToStreamIDs, tenantID)
		buf = tags[i].indexdbMarshal(buf)
		buf = streamID.id.marshal(buf)
		items = append(items, buf[bufLen:])
	}
	
	// Add items to the storage
	idb.tb.AddItems(items)
}

type rawItemsShard struct {
	ibs []*inmemoryBlock
}

type inmemoryBlock struct {
    commonPrefix []byte // 公共前缀，减少重复存储
    data         []byte
    items        []Item
}

// github.com/VictoriaMetrics/VictoriaMetrics/lib/mergeset/encoding.go
func (ib *inmemoryBlock) Add(x []byte) bool {
	data := ib.data
	// 如果写入当前索引条目后，内存块大小超过了 maxInmemoryBlockSize(64KB)，
	// 那么就不能写入当前索引条目，返回 false
	if len(x)+len(data) > maxInmemoryBlockSize {
		return false
	}

	dataLen := len(data)
	data = append(data, x...)
	ib.items = append(ib.items, Item{
		Start: uint32(dataLen),
		End:   uint32(len(data)),
	})
	ib.data = data
	return true
}

// github.com/VictoriaMetrics/VictoriaMetrics/lib/mergeset/table.go
func (riss *rawItemsShards) addIbsToFlush(tb *Table, ibsToFlush []*inmemoryBlock) {
	if len(ibsToFlush) == 0 {
		return
	}

	var ibsToMerge []*inmemoryBlock

	riss.ibsToFlushLock.Lock()
	if len(riss.ibsToFlush) == 0 {
		riss.updateFlushDeadline()
	}

	// 将当前分片的内存块加入到 ibsToFlush 中
	riss.ibsToFlush = append(riss.ibsToFlush, ibsToFlush...)

	// 如果待合并的内存块数量超过了 maxBlocksPerShard(256) * cpus 这一阈值，
	// 那么就需要进行合并操作，将这些内存块合并成一个更大的内存块（inmemoryPart）
	if len(riss.ibsToFlush) >= maxBlocksPerShard*cgroup.AvailableCPUs() {
		ibsToMerge = riss.ibsToFlush
		riss.ibsToFlush = nil
	}
	riss.ibsToFlushLock.Unlock()

	// 将内存块合并成一个更大的内存块（inmemoryPart）
	tb.flushBlocksToInmemoryParts(ibsToMerge, false)
}

func (tb *Table) flushBlocksToInmemoryParts(ibs []*inmemoryBlock, isFinal bool) {
	pws := make([]*partWrapper, 0, (len(ibs)+defaultPartsToMerge-1)/defaultPartsToMerge)

	// 按最多 defaultPartsToMerge(16) 大小分成多个 chunk，
	// 每个 chunk 合并成一个更大的内存块（inmemoryPart）
	for len(ibs) > 0 {
		n := defaultPartsToMerge
		if n > len(ibs) {
			n = len(ibs)
		}
		
		go func(ibsChunk []*inmemoryBlock) {
			if pw := tb.createInmemoryPart(ibsChunk); pw != nil {
				pws = append(pws, pw)
			}
		}(ibs[:n])
		ibs = ibs[n:]
	}

	// 处理所有的 chunk (partWrapper), 将其合并成一个更大的内存块（inmemoryPart）
	// 除非超过大小限制
	maxPartSize := getMaxInmemoryPartSize()
	for len(pws) > 1 {
		// 合并 inmemoryPart
		pws = tb.mustMergeInmemoryParts(pws)
		pwsRemaining := pws[:0]
		for _, pw := range pws {
			// 如果大小超过 maxPartSize，那么就直接加入到 inmemoryParts 中
			if pw.p.size >= maxPartSize {
				tb.addToInmemoryParts(pw, isFinal)
			} else {
				pwsRemaining = append(pwsRemaining, pw)
			}
		}
		pws = pwsRemaining
	}

	if len(pws) == 1 {
		tb.addToInmemoryParts(pws[0], isFinal)
	}
}

// github.com/VictoriaMetrics/VictoriaMetrics/lib/mergeset/table.go
func (tb *Table) nextMergeIdx() uint64 {
	return tb.mergeIdx.Add(1)
}

// github.com/VictoriaMetrics/VictoriaMetrics/lib/mergeset/table.go
func (tb *Table) mergeParts(pws []*partWrapper, stopCh <-chan struct{}, isFinal bool) error {
	mergeIdx := tb.nextMergeIdx()
	dstPartPath := ""
	if dstPartType == partFile {
		// 输出是磁盘 filePart，目录名格式为 %016X，如 18804EBAD6A80650
		dstPartPath = filepath.Join(tb.path, fmt.Sprintf("%016X", mergeIdx))
	}

	// 只有一个 inmemoryPart 时，直接将其写入到磁盘中
	if isFinal && len(pws) == 1 && pws[0].mp != nil {
		mp := pws[0].mp
		mp.MustStoreToDisk(dstPartPath)
		// 打开新创建的 filePart
		pwNew := tb.openCreatedPart(pws, nil, dstPartPath)
		// 用新建的 filePart 替换参与合并的源 part
		tb.swapSrcWithDstParts(pws, pwNew, dstPartType)
		return nil
	}
}

// github.com/VictoriaMetrics/VictoriaMetrics/lib/mergeset/inmemory_part.go
func (mp *inmemoryPart) MustStoreToDisk(path string) {
	fs.MustMkdirFailIfExist(path)

	metaindexPath := filepath.Join(path, metaindexFilename)  // metaindex.bin
	indexPath := filepath.Join(path, indexFilename)          // index.bin
	itemsPath := filepath.Join(path, itemsFilename)          // items.bin
	lensPath := filepath.Join(path, lensFilename)            // lens.bin

	var psw filestream.ParallelStreamWriter
	psw.Add(metaindexPath, &mp.metaindexData)
	psw.Add(indexPath, &mp.indexData)
	psw.Add(itemsPath, &mp.itemsData)
	psw.Add(lensPath, &mp.lensData)

	// 并发执行前面添加的刷写任务
	psw.Run()

	mp.ph.MustWriteMetadata(path) // 写入当前 part 目录下的 metadata.json

	fs.MustSyncPathAndParentDir(path)
}

// lib/logstorage/datadb.go
func (ddb *datadb) mustFlushLogRows(lr *logRows) {
	mp.mustInitFromRows(lr)
	p := mustOpenInmemoryPart(ddb.pt, mp)

	// 将 inmemoryPart 加入到 datadb.inmemoryParts 中
	ddb.inmemoryParts = append(ddb.inmemoryParts, pw)
	// 触发 inmemoryParts 合并
	ddb.startInmemoryPartsMergerLocked()
}

// lib/logstorage/datadb.go
func (mp *inmemoryPart) mustInitFromRows(lr *logRows) {
	sort.Sort(lr)         // 根据 streamID 排序，再按 timestamp 排序
	lr.sortFieldsInRows() // 根据 field 的 Name 进行排序
	
	// 将 inmemoryPart 的各种 buffer 赋值给 blockStreamWriter，
	// 后续 bsw.Finalize 实际上也是将 bsw 的数据写入到 inmemoryPart 中
	bsw.MustInitForInmemoryPart(mp)

	var sidPrev *streamID

	uncompressedBlockSizeBytes := uint64(0)
	timestamps := lr.timestamps
	rows := lr.rows
	streamIDs := lr.streamIDs

	for i := range timestamps {
		streamID := &streamIDs[i]
		if sidPrev == nil {
			sidPrev = streamID
		}
		
		// 注意：将相同流的日志写入同一个 block 中，如果超过 maxUncompressedBlockSize (2MB) 则会写入下一个 block
		// 看 blockStreamWriter.MustWriteRows 方法
		if uncompressedBlockSizeBytes >= maxUncompressedBlockSize || !streamID.equal(sidPrev) {
			bsw.MustWriteRows(sidPrev, trs.timestamps, trs.rows)
			trs.reset()
			sidPrev = streamID
			uncompressedBlockSizeBytes = 0
		}

		fields := rows[i]
		trs.timestamps = append(trs.timestamps, timestamps[i])
		trs.rows = append(trs.rows, fields)
		uncompressedBlockSizeBytes += uint64(EstimatedJSONRowLen(fields))
	}
	bsw.MustWriteRows(sidPrev, trs.timestamps, trs.rows)

	// 将 bsw 中的数据更新到 inmemoryPart
	bsw.Finalize(&mp.ph)	
}

// lib/logstorage/block_stream_writer.go
func (bsw *blockStreamWriter) MustWriteRows(sid *streamID, timestamps []int64, rows [][]Field) {
	if len(timestamps) == 0 {
		return
	}

	b := getBlock()

	// 将原始日志字段写入 block
	b.MustInitFromRows(timestamps, rows)
	bsw.MustWriteBlock(sid, b)
	putBlock(b)
}


type block struct {
	timestamps []int64
	columns []column     // 所有 row 会根据按列进行存储
	constColumns []Field // 所有 row 这一列的值相同就可以存储在 constColumns 中，但值不能超过 256 字节
}

// lib/logstorage/storage_search.go
func (s *Storage) runQuery(qctx *QueryContext, writeBlock writeBlockResultFunc) error {
	// 从查询条件中解析出查询需要的参数：
	// type storageSearchOptions struct {
	// 	tenantIDs          []TenantID
	// 	streamIDs          []streamID
	// 	minTimestamp       int64
	// 	maxTimestamp       int64
	// 	streamFilter       *StreamFilter
	// 	filter             filter
	// 	fieldsFilter       *prefixfilter.Filter
	// 	hiddenFieldsFilter *prefixfilter.Filter
	// 	timeOffset         int64
	// }
	sso := s.getSearchOptions(qctx.TenantIDs, q, qctx.HiddenFieldsFilters)
	s.searchParallel(workersCount, sso, qctx.QueryStats, stopCh, writeBlockToPipes)
}

// lib/logstorage/storage_search.go
func (s *Storage) searchParallel(workersCount int, sso *storageSearchOptions, qs *QueryStats, stopCh <-chan struct{}, writeBlock writeBlockResultFunc) {
	// ...

	// 启动多个 blockSearch 协程，每个协程从 workCh 中取出一个 blockSearchWorkBatch 进行搜索
	workCh := make(chan *blockSearchWorkBatch, workersCount)
	for workerID := 0; workerID < workersCount; workerID++ {
		go func(workerID uint) {
			for bswb := range workCh {
				for i := range bsws {
					// bs blockSearch 代表对一个 block 进行搜索
					bs.search(qsLocal, bsw, bm)
					if bs.br.rowsLen > 0 {
						writeBlock(workerID, &bs.br)
					}	
				}
			}
		}(uint(workerID))
	}

	// 根据时间范围确定需要查询的 Partition
	ptws, ptwsDecRef := s.getPartitionsForTimeRange(sso.minTimestamp, sso.maxTimestamp)
	defer ptwsDecRef()

	// 并发查询每个 Partition
	for i, ptw := range ptws {
		go func(idx int, pt *partition) {
			psfs[idx] = pt.search(sso, qsLocal, workCh, stopCh)
		}(i, ptw.pt)
	}
}

// lib/logstorage/storage_search.go
func (pt *partition) search(sso *storageSearchOptions, qs *QueryStats, workCh chan<- *blockSearchWorkBatch, stopCh <-chan struct{}) partitionSearchFinalizer {
	// 如果查询条件中包含了 _streamFilter 那么从 indexdb 中获取到对应的 streamID
	pso := pt.getSearchOptions(sso)
	// 在 datadb 中执行查询
	return pt.ddb.search(pso, qs, workCh, stopCh)
}

// lib/logstorage/storage_search.go
func (ddb *datadb) search(pso *partitionSearchOptions, qs *QueryStats, workCh chan<- *blockSearchWorkBatch, stopCh <-chan struct{}) partitionSearchFinalizer {
	// 按照时间范围确定需要查询的 Part
	pws, pwsDecRef := ddb.getPartsForTimeRange(pso.minTimestamp, pso.maxTimestamp)

	// Apply search to matching parts
	for _, pw := range pws {
		pw.p.search(pso, qs, workCh, stopCh)
	}

	return pwsDecRef
}

// lib/logstorage/storage_search.go
func (p *part) search(pso *partitionSearchOptions, qs *QueryStats, workCh chan<- *blockSearchWorkBatch, stopCh <-chan struct{}) {
	bhss := getBlockHeaders()
	if len(pso.tenantIDs) > 0 {
		p.searchByTenantIDs(pso, qs, bhss, workCh, stopCh)
	} else {
		// 这个方法是对 part 中的 index
		p.searchByStreamIDs(pso, qs, bhss, workCh, stopCh)
	}
	putBlockHeaders(bhss)
}

func (is *indexSearch) getStreamIDsForNonEmptyTagValue(tenantID TenantID, tagName, tagValue string) map[u128]struct{} {
	ids := make(map[u128]struct{})
	
	ts := &is.ts
	kb := &is.kb

	// 构建查询前缀：nsPrefixTagToStreamIDs + tenantID + tagName + tagValue
	kb.B = marshalCommonPrefix(kb.B[:0], nsPrefixTagToStreamIDs, tenantID)
	kb.B = marshalTagValue(kb.B, bytesutil.ToUnsafeBytes(tagName))
	kb.B = marshalTagValue(kb.B, bytesutil.ToUnsafeBytes(tagValue))
	prefix := kb.B

	// 找到第一条以 prefix 开头的记录
	ts.Seek(prefix)
	for ts.NextItem() {
		if !bytes.HasPrefix(item, prefix) {
			break
		}

		// 解析 streamID 并更新结果
		tail := item[len(prefix):]
		sp.UpdateStreamIDs(ids, tail)
	}

	return ids
}

// indexBlockHeader 覆盖多个 block
type indexBlockHeader struct {
	streamID streamID       // 覆盖范围内最小的 streamID
	minTimestamp int64      // 覆盖范围内最小的 timestamp
	maxTimestamp int64      // 覆盖范围内最大的 timestamp
	indexBlockOffset uint64 // 在 index.bin 中的偏移量
	indexBlockSize uint64   // 在 index.bin 中的数据大小
}

// blockHeader 保存单个 block 的元数据
type blockHeader struct {
	streamID streamID       // block 所属的 streamID
	uncompressedSizeBytes uint64 // 未压缩数据大小
	rowsCount uint64 // 日志行数
	timestampsHeader timestampsHeader // 时间戳元数据
	columnsHeaderIndexOffset uint64 // columnsHeader 在 columns_header_index.bin 中的偏移量
	columnsHeaderIndexSize uint64 // columnsHeader 索引大小
	columnsHeaderOffset uint64 // columnsHeader 在 columns_header.bin 中的偏移量
	columnsHeaderSize uint64 // columnsHeader 数据大小
}

type timestampsHeader struct {
	blockOffset uint64 // block 在 timestamps.bin 中的偏移量
	blockSize uint64 // block 在 timestamps.bin 中的大小

	minTimestamp int64 // block 内最小时间戳
	maxTimestamp int64 // block 内最大时间戳

	marshalType encoding.MarshalType // 时间戳编码类型
}

// lib/logstorage/storage_search.go
func (bs *blockSearch) search(qs *QueryStats, bsw *blockSearchWork, bm *bitmap) {
	bs.reset()

	bs.qs = qs
	bs.bsw = bsw

	// bitmap 用来存储 block 中命中的 log entry 的‘索引’
	bm.init(int(bsw.bh.rowsCount))
	// 将所有的位标记为 1（初始状态，代表所有行都需要检测）
	bm.setBits()

	// filter 是一个接口，用来实现 log entries 的过滤
	// 除 =（filterExact）外，还支持 AND（filterAnd）等组合 filter
	bs.bsw.pso.filter.applyToBlockSearch(bs, bm)

	// 如果没有命中任何 log entry，直接返回
	if bm.isZero() {
		return
	}

	// 将 block 中的
	bs.br.mustInit(bs, bm)

	// 获取需要的列，通过 “ | fields level, streamID, timestamp, message” 这种方式来指定
	bs.br.initColumns(bsw.pso.fieldsFilter)
}

type filter interface {
	String() string // 返回 filter 的字符串表示
	updateNeededFields(pf *prefixfilter.Filter)
	matchRow(fields []Field) bool

	// 即根据 filter 过滤出 bs 中命中的 log entry 的‘索引’，并更新到 bm 中 (标记或者取消标记)
	applyToBlockSearch(bs *blockSearch, bm *bitmap)

	// 即根据 filter 过滤出 br 中命中的 log entry 的‘索引’，并更新到 bm 中(标记或者取消标记)
	applyToBlockResult(br *blockResult, bm *bitmap)
}

// filterExact 匹配指定字段的精确值
//
// Example LogsQL: `fieldName:exact("foo bar")` of `fieldName:="foo bar"`
type filterExact struct {
	fieldName string // 待匹配字段名
	value     string // 待匹配字段值

	tokens       []string
	tokensHashes []uint64
}

// lib/logstorage/filter_exact.go
func (fe *filterExact) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
	fieldName := fe.fieldName
	value := fe.value

	// 先判断是否是 const 列 (所有行这一列的值都是相同的)。
	// 如果是，可以直接比较 value 是否相等，且只需要比较一次就能确定是否命中
	v := bs.getConstColumnValue(fieldName)
	if v != "" {
		if value != v {
			bm.resetBits()
		}
		return
	}

	// 如果不是 const 列，检查列是否存在，并获取到 column header：
	// 1. `column_names.bin` 提供 column_name 和 column_id 的映射
	// 2. `columns_header_index.bin` 提供 column_id 到 column header 的偏移量映射
	// 3. `column_idxs.bin` 提供 column_name 到 bloom/values shardId 的映射
	ch := bs.getColumnHeader(fieldName)
	if ch == nil {
		// 列不存在的情况下，如果查询值不是空，那么需要重置 bitmap
		if value != "" {
			bm.resetBits()
		}
		return
	}

	tokens := fe.getTokensHashes()

	switch ch.valueType {
	case valueTypeString:
		matchStringByExactValue(bs, ch, bm, value, tokens)
	case valueTypeDict:
		matchValuesDictByExactValue(bs, ch, bm, value)
	case valueTypeUint8:
		matchUint8ByExactValue(bs, ch, bm, value, tokens)
	// ... 省略其他数值类型
	case valueTypeIPv4:
		matchIPv4ByExactValue(bs, ch, bm, value, tokens)
	case valueTypeTimestampISO8601:
		matchTimestampISO8601ByExactValue(bs, ch, bm, value, tokens)
	default:
		logger.Panicf("FATAL: %s: unknown valueType=%d", bs.partPath(), ch.valueType)
	}
}

// columnHeader 代表列的元数据信息, 它一定只对应一个 Block 中的一列
// 注意如果 name 为空，那么它代表的 message(_msg) 列
type columnHeader struct {
	name string

	valueType valueType // 列中存储的值的类型

	minValue uint64 // 列中存储的最小值, 用于快速判断是否在给定的范围内。适用于 uint*, ipv4, timestamp 和 float64 类型
	maxValue uint64 // 列中存储的最大值, 用于快速判断是否在给定的范围内。适用于 uint*, ipv4, timestamp 和 float64 类型
	valuesDict valuesDict // 列中存储的唯一值字典, 适用于 valueType = valueTypeDict 类型

	valuesOffset uint64 // 列中存储的 values 的偏移量, 用于快速定位到 values.bin 中的对应位置
	valuesSize uint64   // 列中存储的 values 的大小, 用于快速定位到 values.bin 中的对应位置

	bloomFilterOffset uint64 // 列中存储的 bloom filter 的偏移量, 用于快速定位到 bloom.bin 中的对应位置
	bloomFilterSize uint64   // 列中存储的 bloom filter 的大小, 用于快速定位到 bloom.bin 中的对应位置
}

// lib/logstorage/filter_exact.go
func matchStringByExactValue(bs *blockSearch, ch *columnHeader, bm *bitmap, value string, tokens []uint64) {
	// 先判断 bloom filter 是否命中, 如果 bloom filter 不命中, 那么直接返回
	if !matchBloomFilterAllTokens(bs, ch, tokens) {
		bm.resetBits()
		return
	}

	visitValues(bs, ch, bm, func(v string) bool {
		return v == value
	})
}

// lib/logstorage/filter_phrase.go
func visitValues(bs *blockSearch, ch *columnHeader, bm *bitmap, f func(value string) bool) {
	if bm.isZero() {
		// Fast path - nothing to visit
		return
	}

	// 根据 列头 来获取 values
	// 1. 根据列名确定 values.bin 所在的分片
	// 2. 根据 columnHeader 中的 valuesOffset 和 valuesSize 读取 values
	values := bs.getValuesForColumn(ch)
	bm.forEachSetBit(func(idx int) bool {
		return f(values[idx])
	})
}

/storage/
 ├── partitions/
 │   ├── 20251210/
 │   │   ├── indexdb
 │   │   │   ├── 18804EBAD6A6ECA1
 │   │   │   │    ├── index.bin
 │   │   │   │    ├── items.bin
 │   │   │   │    ├── lens.bin
 │   │   │   │    ├── metadata.json
 │   │   │   │    └── metaindex.bin
 │   │   │   ├── 18804EBAD6A6ECB9
 │   │   │   ├── ...
 │   │   │   ├── 18804EBAD6A6ECB8
 │   │   │   └── parts.json
 │   │   │
 │   │   └── datadb/
 │   │       ├── 18804EBAD6A80650
 │   │       │    ├── bloom.bin0
 │   │       │    ├── bloom.bin1
 │   │       │    ├── bloom.bin2
 │   │       │    ├── bloom.bin3
 │   │       │    ├── bloom.bin4
 │   │       │    ├── column_idxs.bin
 │   │       │    ├── column_names.bin
 │   │       │    ├── columns_header.bin
 │   │       │    ├── columns_header_index.bin
 │   │       │    ├── index.bin
 │   │       │    ├── message_bloom.bin
 │   │       │    ├── message_values.bin
 │   │       │    ├── metadata.json
 │   │       │    ├── metaindex.bin
 │   │       │    ├── timestamps.bin
 │   │       │    ├── values.bin0
 │   │       │    ├── values.bin1
 │   │       │    ├── values.bin2
 │   │       │    ├── values.bin3
 │   │       │    └── values.bin4
 │   │       │
 │   │       ├── 18804EBAD6A80FF6
 │   │       ├── ...
 │   │       ├── 18804EBAD6A81004
 │   │       └── parts.json
 │   │
 │   │── 20251211/
 │   │── ...
 │   └── 20251213/
│
└── flock.lock