This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 28cd2a7a2 Support displaying a measure's indexed tags in the dump tool
(#1143)
28cd2a7a2 is described below
commit 28cd2a7a214f9526e7ae58fe21a3b124c4d617e1
Author: mrproliu <[email protected]>
AuthorDate: Thu May 28 20:46:24 2026 +0800
Support displaying a measure's indexed tags in the dump tool (#1143)
* Support displaying a measure's indexed tags in the dump tool
---
CHANGES.md | 1 +
banyand/cmd/dump/measure.go | 84 +++++--
banyand/internal/dump/index_resolver.go | 202 ++++++++++++++++
banyand/internal/dump/measure/iterator.go | 28 +++
banyand/internal/dump/measure/measure.go | 7 +
banyand/internal/dump/measure/suite_test.go | 357 +++++++++++++++++++++++++++-
pkg/index/index.go | 5 +
pkg/index/inverted/inverted_series.go | 53 +++++
pkg/index/inverted/inverted_series_test.go | 66 +++++
9 files changed, 780 insertions(+), 23 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index b86953289..abcd47d66 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -57,6 +57,7 @@ Release Notes.
- Fix the `modRevision` contract on no-op Update RPCs
(`MeasureRegistryService.Update`, etc.). Previously `updateResource` detected
unchanged content via `CheckerMap` and short-circuited without writing to the
property store, but the caller had already fabricated `modRevision =
time.Now().UnixNano()` and returned it. The returned revision never appeared in
the property watch stream, so `AwaitRevisionApplied(R)` would hang.
`updateResource` now returns `(int64, error)` — the existing pr [...]
- Add end-to-end observability for liaison internal queue pipelines with
per-topic metrics for queue_sub and queue_pub, along with Grafana panels and
troubleshooting docs.
- Introduce measure migration tool.
+- Support displaying a measure's indexed tags in the dump tool, resolved per
part so peak memory is bounded by the part rather than a segment-wide series
map.
### Bug Fixes
diff --git a/banyand/cmd/dump/measure.go b/banyand/cmd/dump/measure.go
index 252299123..1bcd430c6 100644
--- a/banyand/cmd/dump/measure.go
+++ b/banyand/cmd/dump/measure.go
@@ -27,7 +27,6 @@ import (
"github.com/spf13/cobra"
"google.golang.org/protobuf/encoding/protojson"
- "github.com/apache/skywalking-banyandb/api/common"
databasev1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
modelv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/internal/dump"
@@ -136,7 +135,7 @@ func dumpMeasureShard(opts measureDumpOptions) error {
type measureDumpContext struct {
tagFilter logical.TagFilter
fileSystem fs.FileSystem
- seriesMap map[common.SeriesID]string
+ indexResolver *dump.IndexResolver
writer *csv.Writer
opts measureDumpOptions
partIDs []uint64
@@ -164,12 +163,13 @@ func newMeasureDumpContext(opts measureDumpOptions)
(*measureDumpContext, error)
ctx.partIDs = partIDs
fmt.Fprintf(os.Stderr, "Found %d parts in shard\n", len(partIDs))
- ctx.seriesMap, err = dump.LoadSegmentSeriesMap(opts.segmentPath)
+ // The offline CLI has no schema, so it resolves indexed tags by
IndexRuleID
+ // (ruleToTag is nil). EntityValues come from each part on demand, so no
+ // segment-wide series map is loaded.
+ ctx.indexResolver, err = dump.NewIndexResolver(opts.segmentPath,
dump.DefaultIndexCacheSize, nil)
if err != nil {
- fmt.Fprintf(os.Stderr, "Warning: Failed to load series
information: %v\n", err)
- ctx.seriesMap = nil
- } else {
- fmt.Fprintf(os.Stderr, "Loaded %d series from segment\n",
len(ctx.seriesMap))
+ fmt.Fprintf(os.Stderr, "Warning: Failed to open index resolver
(indexed tags unavailable): %v\n", err)
+ ctx.indexResolver = nil
}
if opts.criteriaJSON != "" {
@@ -237,7 +237,7 @@ func (ctx *measureDumpContext) initOutput() error {
}
ctx.writer = csv.NewWriter(os.Stdout)
- header := []string{"PartID", "Timestamp", "Version", "SeriesID",
"Series"}
+ header := []string{"PartID", "Timestamp", "Version", "SeriesID",
"Series", "IndexedTags"}
header = append(header, ctx.fieldColumns...)
header = append(header, ctx.tagColumns...)
if err := ctx.writer.Write(header); err != nil {
@@ -250,6 +250,9 @@ func (ctx *measureDumpContext) close() {
if ctx.writer != nil {
ctx.writer.Flush()
}
+ if ctx.indexResolver != nil {
+ _ = ctx.indexResolver.Close()
+ }
}
func (ctx *measureDumpContext) processParts() error {
@@ -261,6 +264,9 @@ func (ctx *measureDumpContext) processParts() error {
fmt.Fprintf(os.Stderr, "Warning: failed to open part
%016x: %v\n", partID, err)
continue
}
+ if ctx.indexResolver != nil {
+ reader.SetIndexResolver(ctx.indexResolver)
+ }
partRowCount := 0
it := reader.Iterator()
@@ -317,11 +323,11 @@ func (ctx *measureDumpContext) shouldSkip(tags
map[string][]byte) bool {
func (ctx *measureDumpContext) writeRow(partID uint64, row dumpmeasure.Row)
error {
if ctx.opts.csvOutput {
- if err := writeMeasureRowAsCSV(ctx.writer, partID, row,
ctx.fieldColumns, ctx.tagColumns, ctx.seriesMap); err != nil {
+ if err := writeMeasureRowAsCSV(ctx.writer, partID, row,
ctx.fieldColumns, ctx.tagColumns); err != nil {
return err
}
} else {
- writeMeasureRowAsText(partID, row, ctx.rowNum+1,
ctx.projectionTags, ctx.projectionFields, ctx.seriesMap)
+ writeMeasureRowAsText(partID, row, ctx.rowNum+1,
ctx.projectionTags, ctx.projectionFields)
}
ctx.rowNum++
return nil
@@ -360,16 +366,46 @@ func parseMeasureProjectionTags(projectionStr string)
[]string {
return result
}
-func measureSeriesText(row dumpmeasure.Row, seriesMap
map[common.SeriesID]string) string {
- if len(row.EntityValues) > 0 {
- return string(row.EntityValues)
+func measureSeriesText(row dumpmeasure.Row) string {
+ return string(row.EntityValues)
+}
+
+func sortedRuleIDs(m map[uint32][][]byte) []uint32 {
+ ids := make([]uint32, 0, len(m))
+ for id := range m {
+ ids = append(ids, id)
}
- if seriesMap != nil {
- if text, ok := seriesMap[row.SeriesID]; ok {
- return text
+ sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
+ return ids
+}
+
+// formatIndexedTagValues renders one IndexRuleID's values (several for an
array
+// tag). The index does not store the value type, so values are shown as text
+// when printable and as a byte size otherwise.
+func formatIndexedTagValues(values [][]byte) string {
+ parts := make([]string, 0, len(values))
+ for _, v := range values {
+ if isPrintable(v) {
+ parts = append(parts, string(v))
+ } else {
+ parts = append(parts, fmt.Sprintf("(binary:%d bytes)",
len(v)))
}
}
- return ""
+ return strings.Join(parts, ",")
+}
+
+// formatIndexedTagsCSV serializes all indexed tags into one CSV cell as
+// "ruleID=v1,v2;ruleID=...", sorted by IndexRuleID.
+func formatIndexedTagsCSV(m map[uint32][][]byte) string {
+ if len(m) == 0 {
+ return ""
+ }
+ ids := sortedRuleIDs(m)
+ parts := make([]string, 0, len(ids))
+ for _, id := range ids {
+ parts = append(parts, fmt.Sprintf("%d=%s", id,
formatIndexedTagValues(m[id])))
+ }
+ return strings.Join(parts, ";")
}
func writeMeasureRowAsText(
@@ -378,7 +414,6 @@ func writeMeasureRowAsText(
rowNum int,
projectionTags []string,
projectionFields []string,
- seriesMap map[common.SeriesID]string,
) {
fmt.Printf("Row %d:\n", rowNum)
fmt.Printf(" PartID: %d (0x%016x)\n", partID, partID)
@@ -386,7 +421,7 @@ func writeMeasureRowAsText(
fmt.Printf(" Version: %d\n", row.Version)
fmt.Printf(" SeriesID: %d\n", row.SeriesID)
- if seriesText := measureSeriesText(row, seriesMap); seriesText != "" {
+ if seriesText := measureSeriesText(row); seriesText != "" {
fmt.Printf(" Series: %s\n", seriesText)
}
@@ -441,6 +476,13 @@ func writeMeasureRowAsText(
}
}
}
+
+ if len(row.IndexedTags) > 0 {
+ fmt.Printf(" IndexedTags (by IndexRuleID):\n")
+ for _, ruleID := range sortedRuleIDs(row.IndexedTags) {
+ fmt.Printf(" %d: %s\n", ruleID,
formatIndexedTagValues(row.IndexedTags[ruleID]))
+ }
+ }
fmt.Printf("\n")
}
@@ -450,14 +492,14 @@ func writeMeasureRowAsCSV(
row dumpmeasure.Row,
fieldColumns []string,
tagColumns []string,
- seriesMap map[common.SeriesID]string,
) error {
csvRow := []string{
fmt.Sprintf("%d", partID),
formatTimestamp(row.Timestamp),
fmt.Sprintf("%d", row.Version),
fmt.Sprintf("%d", row.SeriesID),
- measureSeriesText(row, seriesMap),
+ measureSeriesText(row),
+ formatIndexedTagsCSV(row.IndexedTags),
}
// Add field values
diff --git a/banyand/internal/dump/index_resolver.go
b/banyand/internal/dump/index_resolver.go
new file mode 100644
index 000000000..c4d97629c
--- /dev/null
+++ b/banyand/internal/dump/index_resolver.go
@@ -0,0 +1,202 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package dump
+
+import (
+ "bytes"
+ "context"
+ "path/filepath"
+
+ lru "github.com/hashicorp/golang-lru"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ modelv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/index/inverted"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+)
+
+// DefaultIndexCacheSize is the default number of series whose resolved indexed
+// tags an IndexResolver keeps in its frequency-aware cache.
+const DefaultIndexCacheSize = 4096
+
+// IndexedTagSpec describes an indexed tag so the resolver can decode its raw
+// stored value into a typed TagValue keyed by name. Callers that know the
schema
+// pass a IndexRuleID -> IndexedTagSpec map to NewIndexResolver.
+type IndexedTagSpec struct {
+ Family string
+ Name string
+ Type pbv1.ValueType
+}
+
+// IndexResolver recovers a measure's indexed (non-entity) tag values, which
the
+// write path stores only in the segment-level series index (sidx) and never as
+// data columns. Lookups are on-demand per series and held in a fixed-size,
+// frequency-aware (2Q) cache, so the potentially large indexed values are
never
+// all held in memory at once.
+//
+// Resolve returns values keyed by IndexRuleID (no schema needed). When the
+// caller supplies a IndexRuleID -> IndexedTagSpec mapping, DecodeTagValues
+// additionally decodes a resolved map into typed TagValues keyed by
"family.tag".
+type IndexResolver struct {
+ store index.SeriesStore
+ cache *lru.TwoQueueCache
+ ruleToTag map[uint32]IndexedTagSpec
+}
+
+// NewIndexResolver opens the segment's series index under segmentPath/sidx for
+// read-only lookups. ruleToTag is optional: when the schema is known it lets
+// DecodeTagValues decode values by tag name, otherwise pass nil and use
Resolve.
+// cacheSize bounds the frequency-aware cache; values <= 0 fall back to
+// DefaultIndexCacheSize. Call Close when done.
+func NewIndexResolver(segmentPath string, cacheSize int,
+ ruleToTag map[uint32]IndexedTagSpec,
+) (*IndexResolver, error) {
+ if cacheSize <= 0 {
+ cacheSize = DefaultIndexCacheSize
+ }
+ cache, err := lru.New2Q(cacheSize)
+ if err != nil {
+ return nil, err
+ }
+ store, err := inverted.NewStore(inverted.StoreOpts{
+ Path: filepath.Join(segmentPath, dirNameSidx),
+ Logger: logger.GetLogger(logName),
+ })
+ if err != nil {
+ return nil, err
+ }
+ return &IndexResolver{
+ store: store,
+ cache: cache,
+ ruleToTag: ruleToTag,
+ }, nil
+}
+
+// Resolve returns the indexed tag values stored for the series identified by
+// seriesID, whose EntityValues the caller already knows (the index document is
+// keyed by EntityValues). Each rule id maps to one or more raw value byte
slices
+// (one for a scalar tag, several for an array tag). It returns nil when
+// entityValues is empty or the series has no indexed tags. Results are cached
+// per seriesID, so a series spanning several blocks or parts resolves once.
+func (r *IndexResolver) Resolve(seriesID common.SeriesID, entityValues []byte)
(map[uint32][][]byte, error) {
+ if v, ok := r.cache.Get(seriesID); ok {
+ return v.(map[uint32][][]byte), nil
+ }
+ if len(entityValues) == 0 {
+ return nil, nil
+ }
+ raw, err := r.store.StoredFields(context.Background(), entityValues)
+ if err != nil {
+ return nil, err
+ }
+ result := parseIndexedFields(raw)
+ r.cache.Add(seriesID, result)
+ return result, nil
+}
+
+// PartSeriesMap builds a SeriesID -> EntityValues map scoped to the given set
of
+// series by scanning the series index once and keeping only the entries whose
+// SeriesID (Hash of EntityValues) is in seriesIDs. It is the fallback that
+// recovers EntityValues for a part with no part-level series metadata, while
+// keeping peak memory bounded by the part's distinct series count. It stops
+// early once every requested series has been found.
+func (r *IndexResolver) PartSeriesMap(seriesIDs map[common.SeriesID]struct{})
(map[common.SeriesID][]byte, error) {
+ if len(seriesIDs) == 0 {
+ return nil, nil
+ }
+ iter, err := r.store.SeriesIterator(context.Background())
+ if err != nil {
+ return nil, err
+ }
+ result := make(map[common.SeriesID][]byte, len(seriesIDs))
+ for iter.Next() {
+ entityValues := iter.Val().EntityValues
+ if len(entityValues) == 0 {
+ continue
+ }
+ seriesID := common.SeriesID(convert.Hash(entityValues))
+ if _, ok := seriesIDs[seriesID]; !ok {
+ continue
+ }
+ result[seriesID] = bytes.Clone(entityValues)
+ if len(result) == len(seriesIDs) {
+ break
+ }
+ }
+ if closeErr := iter.Close(); closeErr != nil {
+ return nil, closeErr
+ }
+ return result, nil
+}
+
+// DecodeTagValues decodes already-resolved indexed tags (a Row's IndexedTags)
+// into typed TagValues keyed by "family.tag" (or just "tag" when the family is
+// empty), using the ruleToTag mapping given to NewIndexResolver. It does no
I/O.
+// It returns nil when that mapping is absent (schema unknown) or there are no
+// indexed tags.
+func (r *IndexResolver) DecodeTagValues(indexed map[uint32][][]byte)
map[string]*modelv1.TagValue {
+ if r.ruleToTag == nil || len(indexed) == 0 {
+ return nil
+ }
+ out := make(map[string]*modelv1.TagValue, len(indexed))
+ for ruleID, values := range indexed {
+ spec, ok := r.ruleToTag[ruleID]
+ if !ok {
+ continue
+ }
+ key := spec.Name
+ if spec.Family != "" {
+ key = spec.Family + "." + spec.Name
+ }
+ var scalar []byte
+ if len(values) > 0 {
+ scalar = values[0]
+ }
+ // DecodeTagValue uses value for scalar types and valueArr for
array
+ // types; passing both lets the value type pick the right one.
+ out[key] = DecodeTagValue(spec.Type, scalar, values)
+ }
+ return out
+}
+
+// Close releases the underlying series index store.
+func (r *IndexResolver) Close() error {
+ return r.store.Close()
+}
+
+// parseIndexedFields keys a document's stored fields by IndexRuleID. An
+// index-rule field name is a marshaled uint32 IndexRuleID (exactly 4 bytes);
+// any other-width field is skipped, so a non-rule name (e.g. an index-mode
+// tag name) can neither panic BytesToUint32 nor map to a garbage rule id. A
+// rule id maps to one value for a scalar tag, several for an array tag.
+func parseIndexedFields(raw map[string][][]byte) map[uint32][][]byte {
+ if len(raw) == 0 {
+ return nil
+ }
+ result := make(map[uint32][][]byte, len(raw))
+ for name, values := range raw {
+ if len(name) != 4 {
+ continue
+ }
+ result[convert.BytesToUint32([]byte(name))] = values
+ }
+ return result
+}
diff --git a/banyand/internal/dump/measure/iterator.go
b/banyand/internal/dump/measure/iterator.go
index 0c75c54f4..9802a303a 100644
--- a/banyand/internal/dump/measure/iterator.go
+++ b/banyand/internal/dump/measure/iterator.go
@@ -20,6 +20,7 @@ package measure
import (
"fmt"
+ "github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/banyand/internal/dump"
"github.com/apache/skywalking-banyandb/pkg/compress/zstd"
"github.com/apache/skywalking-banyandb/pkg/encoding"
@@ -46,6 +47,21 @@ func (p *PartReader) Iterator() *dump.Iterator[Row] {
}
blocks = append(blocks, bms...)
}
+ // When the part has no part-level series metadata (the standalone
write path
+ // writes none), recover EntityValues for just this part's series by
scanning
+ // the series index scoped to the part — keeping peak memory bounded by
the
+ // part rather than loading a segment-wide map.
+ if p.seriesMap == nil && p.indexResolver != nil {
+ seriesIDs := make(map[common.SeriesID]struct{})
+ for _, bm := range blocks {
+ seriesIDs[bm.seriesID] = struct{}{}
+ }
+ seriesMap, err := p.indexResolver.PartSeriesMap(seriesIDs)
+ if err != nil {
+ return dump.NewErrIterator[Row](fmt.Errorf("cannot
build part series map: %w", err))
+ }
+ p.seriesMap = seriesMap
+ }
var decoder encoding.BytesBlockDecoder
return dump.NewIterator(len(blocks), func(blockIdx int) ([]Row, error) {
return p.decodeBlock(&decoder, blocks[blockIdx])
@@ -104,6 +120,17 @@ func (p *PartReader) decodeBlock(decoder
*encoding.BytesBlockDecoder, bm *blockM
entityValues = p.seriesMap[bm.seriesID]
}
+ // Indexed (non-column) tags live only in the series index; resolve
them once
+ // per block (one series) and share across the block's rows.
+ var indexedTags map[uint32][][]byte
+ if p.indexResolver != nil {
+ var resolveErr error
+ indexedTags, resolveErr = p.indexResolver.Resolve(bm.seriesID,
entityValues)
+ if resolveErr != nil {
+ return nil, fmt.Errorf("cannot resolve indexed tags for
series %d: %w", bm.seriesID, resolveErr)
+ }
+ }
+
rows := make([]Row, 0, len(timestamps))
for i := range timestamps {
tags := make(map[string][]byte, len(tagsByDataPoint))
@@ -131,6 +158,7 @@ func (p *PartReader) decodeBlock(decoder
*encoding.BytesBlockDecoder, bm *blockM
Fields: fields,
TagTypes: tagTypesForRow,
FieldTypes: fieldTypesForRow,
+ IndexedTags: indexedTags,
})
}
return rows, nil
diff --git a/banyand/internal/dump/measure/measure.go
b/banyand/internal/dump/measure/measure.go
index b992d5735..c6cb40581 100644
--- a/banyand/internal/dump/measure/measure.go
+++ b/banyand/internal/dump/measure/measure.go
@@ -104,6 +104,7 @@ type PartReader struct {
tagFamilyMetadata map[string]fs.Reader
tagFamilies map[string]fs.Reader
seriesMap map[common.SeriesID][]byte // part-level
smeta.bin, nil if absent
+ indexResolver *dump.IndexResolver // optional; resolves
indexed (non-column) tags
path string
primaryBlockMetadata []primaryBlockMetadata
partMetadata PartMetadata
@@ -117,6 +118,7 @@ type Row struct {
Fields map[string][]byte
TagTypes map[string]pbv1.ValueType
FieldTypes map[string]pbv1.ValueType
+ IndexedTags map[uint32][][]byte
EntityValues []byte
Timestamp int64
Version int64
@@ -240,6 +242,11 @@ func (p *PartReader) PartID() uint64 { return
p.partMetadata.ID }
// smeta.bin, or nil if smeta.bin is absent.
func (p *PartReader) SeriesMap() map[common.SeriesID][]byte { return
p.seriesMap }
+// SetIndexResolver attaches an index resolver so the iterator fills each Row's
+// IndexedTags with the series' indexed (non-column) tag values. Optional; the
+// resolver is owned by the caller (the PartReader does not close it).
+func (p *PartReader) SetIndexResolver(r *dump.IndexResolver) { p.indexResolver
= r }
+
// Close releases all file handles. Safe to call multiple times.
func (p *PartReader) Close() error {
closePart(p)
diff --git a/banyand/internal/dump/measure/suite_test.go
b/banyand/internal/dump/measure/suite_test.go
index 634e4d72a..c5f77a046 100644
--- a/banyand/internal/dump/measure/suite_test.go
+++ b/banyand/internal/dump/measure/suite_test.go
@@ -21,6 +21,7 @@ import (
"context"
"fmt"
"io/fs"
+ "os"
"path/filepath"
"strconv"
"testing"
@@ -30,6 +31,7 @@ import (
"github.com/stretchr/testify/require"
"google.golang.org/protobuf/types/known/timestamppb"
+ "github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/api/data"
commonv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
@@ -42,14 +44,19 @@ import (
"github.com/apache/skywalking-banyandb/banyand/protector"
"github.com/apache/skywalking-banyandb/banyand/queue"
"github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
localfs "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/index"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/test"
)
const (
- rtGroup = "dump_rt_group"
- rtMeasure = "dump_rt_measure"
+ rtGroup = "dump_rt_group"
+ rtMeasure = "dump_rt_measure"
+ idxrGroup = "dump_idxr_group"
+ idxrMeasure = "dump_idxr_measure"
)
// TestMeasureWriteRoundTrip drives the top-level measure write path (the same
@@ -245,3 +252,349 @@ func findPartDirs(root string) []string {
})
return dirs
}
+
+// TestMeasureIndexedTagResolvedFromIndex verifies the dump recovers a
measure's
+// indexed (non-entity) tag values from the series index via an IndexResolver,
+// even though they are absent from the data columns. It covers a scalar
string,
+// a scalar int and a string-array indexed tag.
+func TestMeasureIndexedTagResolvedFromIndex(t *testing.T) {
+ req := require.New(t)
+ require.NoError(t, logger.Init(logger.Logging{Env: "dev", Level:
"warn"}))
+ gomega.RegisterFailHandler(func(message string, _ ...int) {
panic(message) })
+
+ pipeline := queue.Local()
+ metaSvc, err := metadataservice.NewService()
+ req.NoError(err)
+ metricSvc := obsservice.NewMetricService(metaSvc, pipeline, "test", nil)
+ pm := protector.NewMemory(metricSvc)
+ measureSvc, err := measure.NewStandalone(metaSvc, pipeline, nil,
metricSvc, pm)
+ req.NoError(err)
+
+ metaPath, metaDefer, err := test.NewSpace()
+ req.NoError(err)
+ defer metaDefer()
+ ports, err := test.AllocateFreePorts(1)
+ req.NoError(err)
+ rootPath, rootDefer, err := test.NewSpace()
+ req.NoError(err)
+ defer rootDefer()
+
+ flags := []string{
+ "--schema-server-root-path=" + metaPath,
+ fmt.Sprintf("--schema-server-grpc-port=%d", ports[0]),
+ "--schema-server-grpc-host=127.0.0.1",
+ "--measure-root-path=" + rootPath,
+ "--measure-flush-timeout=200ms",
+ }
+ moduleDefer := test.SetupModules(flags, pipeline, metaSvc, measureSvc)
+ stopped := false
+ defer func() {
+ if !stopped {
+ moduleDefer()
+ }
+ }()
+
+ registerIndexedResolveMeasure(t, metaSvc)
+ require.Eventually(t, func() bool {
+ _, ok := measureSvc.LoadGroup(idxrGroup)
+ return ok
+ }, 30*time.Second, 200*time.Millisecond, "measure group not loaded")
+ time.Sleep(time.Second)
+
+ const total = 2
+ baseTS := time.Now().Truncate(time.Minute)
+ bp := pipeline.NewBatchPublisher(5 * time.Second)
+ for i := 0; i < total; i++ {
+ iStr := strconv.Itoa(i)
+ ts := baseTS.Add(time.Duration(i) * time.Minute)
+ writeReq := &measurev1.WriteRequest{
+ Metadata: &commonv1.Metadata{Name: idxrMeasure, Group:
idxrGroup},
+ DataPoint: &measurev1.DataPointValue{
+ Timestamp: timestamppb.New(ts),
+ TagFamilies: []*modelv1.TagFamilyForWrite{{
+ Tags: []*modelv1.TagValue{
+ strTagValue("series" + iStr),
// series (entity)
+ strTagValue("plain" + iStr),
// strTag (plain column)
+ strTagValue("indexed" + iStr),
// idxStr (indexed)
+ intTagValue(int64(70 + i)),
// idxInt (indexed)
+ strArrTagValue("arr"+iStr+"a",
"arr"+iStr+"b"), // idxArr (indexed)
+ },
+ }},
+ Fields: []*modelv1.FieldValue{
+ {Value: &modelv1.FieldValue_Int{Int:
&modelv1.Int{Value: int64(100 + i)}}},
+ },
+ },
+ }
+ _, errPub := bp.Publish(context.TODO(), data.TopicMeasureWrite,
bus.NewMessage(bus.MessageID(i), &measurev1.InternalWriteRequest{
+ EntityValues: []*modelv1.TagValue{strTagValue("series"
+ iStr)},
+ Request: writeReq,
+ }))
+ req.NoError(errPub)
+ }
+ req.Empty(bp.Close())
+
+ var partDirs []string
+ require.Eventually(t, func() bool {
+ partDirs = findPartDirs(rootPath)
+ return len(partDirs) > 0
+ }, 30*time.Second, 200*time.Millisecond, "no on-disk part was flushed")
+
+ // Resolve each rule's assigned IndexRuleID while the schema server is
still
+ // up, so we can later assert per-rule that multiple rules on one row
stay
+ // separate (each under its own id).
+ reg := metaSvc.SchemaRegistry()
+ ruleID := func(name string) uint32 {
+ ir, gErr := reg.GetIndexRule(context.TODO(),
&commonv1.Metadata{Name: name, Group: idxrGroup})
+ req.NoError(gErr)
+ return ir.GetMetadata().GetId()
+ }
+ strRuleID, intRuleID, arrRuleID := ruleID("idxr_str_rule"),
ruleID("idxr_int_rule"), ruleID("idxr_arr_rule")
+
+ // Stop the live service so it releases bluge's exclusive lock on the
series
+ // index; the dump (like the offline CLI) reads the index from a
quiesced
+ // database. Deferred cleanup is suppressed via the stopped flag.
+ moduleDefer()
+ stopped = true
+
+ segmentPath := findSidxSegmentPath(t, rootPath)
+
+ ruleToTag := map[uint32]dump.IndexedTagSpec{
+ strRuleID: {Family: "default", Name: "idxStr", Type:
pbv1.ValueTypeStr},
+ intRuleID: {Family: "default", Name: "idxInt", Type:
pbv1.ValueTypeInt64},
+ arrRuleID: {Family: "default", Name: "idxArr", Type:
pbv1.ValueTypeStrArr},
+ }
+ // The standalone write path persists no part-level smeta.bin, so the
resolver
+ // recovers EntityValues for each part by scanning the series index
scoped to
+ // that part (PartSeriesMap) — the per-part bounded fallback, no
segment-wide
+ // map.
+ resolver, err := dump.NewIndexResolver(segmentPath, 0, ruleToTag)
+ req.NoError(err)
+ // Closed explicitly before the smeta-path phase reopens the same index
+ // (bluge holds an exclusive lock, so only one store may be open at a
time).
+
+ fileSystem := localfs.NewLocalFileSystem()
+ seen := 0
+ entityBySeries := map[common.SeriesID]string{}
+ partSeriesByDir := map[string]map[common.SeriesID][]byte{}
+ for _, partDir := range partDirs {
+ partID, parseErr := strconv.ParseUint(filepath.Base(partDir),
16, 64)
+ req.NoError(parseErr)
+ p, openErr := OpenPart(partID, filepath.Dir(partDir),
fileSystem)
+ req.NoError(openErr)
+ // Standalone parts carry no smeta.bin, so the iterator must
take the
+ // per-part fallback scan to recover EntityValues.
+ req.Nil(p.SeriesMap(), "standalone part has no smeta.bin")
+ p.SetIndexResolver(resolver)
+ curSeries := map[common.SeriesID][]byte{}
+ it := p.Iterator()
+ for it.Next() {
+ r := it.Row()
+ seen++
+ // Identify the row by its plain (column) tag, e.g.
"plain1" -> n=1.
+ plain :=
dump.DecodeTagValue(r.TagTypes["default.strTag"], r.Tags["default.strTag"],
nil).GetStr().GetValue()
+ n, atoiErr := strconv.Atoi(plain[len("plain"):])
+ req.NoError(atoiErr)
+ iStr := strconv.Itoa(n)
+ // Indexed and entity tags are NOT stored as data
columns; the plain
+ // tag and the field are.
+ req.NotContains(r.Tags, "default.idxStr")
+ req.NotContains(r.Tags, "default.idxInt")
+ req.NotContains(r.Tags, "default.idxArr")
+ req.NotContains(r.Tags, "default.series")
+ req.Contains(r.Fields, "intField")
+ // No part-level smeta exists (standalone write path),
so EntityValues
+ // here proves the per-part fallback scan recovered it;
the entity tag
+ // value is part of the encoded EntityValues.
+ req.NotEmpty(r.EntityValues, "fallback scan must
recover EntityValues")
+ req.Contains(string(r.EntityValues), "series"+iStr)
+ entityBySeries[r.SeriesID] = string(r.EntityValues)
+ curSeries[r.SeriesID] = append([]byte(nil),
r.EntityValues...)
+ // All three rules' values are recovered, each under
its own
+ // IndexRuleID (multiple rules on one row stay
separate, not merged).
+ req.Len(r.IndexedTags, 3, "all three index rules
resolved")
+ strVals := r.IndexedTags[strRuleID]
+ req.Len(strVals, 1, "scalar string indexed tag -> one
value")
+ req.Equal("indexed"+iStr, string(strVals[0]))
+ intVals := r.IndexedTags[intRuleID]
+ req.Len(intVals, 1, "scalar int indexed tag -> one
value")
+ req.Equal(convert.Int64ToBytes(int64(70+n)), intVals[0])
+ arrVals := r.IndexedTags[arrRuleID]
+ req.Len(arrVals, 2, "array indexed tag -> two values")
+ arrSet := map[string]bool{string(arrVals[0]): true,
string(arrVals[1]): true}
+ req.True(arrSet["arr"+iStr+"a"] &&
arrSet["arr"+iStr+"b"], "both array elements recovered")
+ // With the rule->tag mapping, the row's
already-resolved IndexedTags
+ // decode to typed TagValues keyed by "family.tag".
+ named := resolver.DecodeTagValues(r.IndexedTags)
+ req.Equal("indexed"+iStr,
named["default.idxStr"].GetStr().GetValue())
+ req.Equal(int64(70+n),
named["default.idxInt"].GetInt().GetValue())
+ req.ElementsMatch([]string{"arr" + iStr + "a", "arr" +
iStr + "b"}, named["default.idxArr"].GetStrArray().GetValue())
+ }
+ req.NoError(it.Err())
+ it.Close()
+ p.Close()
+ partSeriesByDir[partDir] = curSeries
+ }
+ req.Equal(total, seen, "every data point should be read back")
+ req.Len(entityBySeries, total, "each entity maps to a distinct series")
+
+ // Direct coverage of the bounded per-part scan: it returns
EntityValues only
+ // for the requested series, and nothing for unknown series.
+ allIDs := make(map[common.SeriesID]struct{}, len(entityBySeries))
+ for sid := range entityBySeries {
+ allIDs[sid] = struct{}{}
+ }
+ full, err := resolver.PartSeriesMap(allIDs)
+ req.NoError(err)
+ req.Len(full, len(entityBySeries), "scan returns exactly the requested
series")
+ for sid, ev := range entityBySeries {
+ req.Equal(ev, string(full[sid]))
+ }
+
+ var oneID common.SeriesID
+ for sid := range allIDs {
+ oneID = sid
+ break
+ }
+ subset, err :=
resolver.PartSeriesMap(map[common.SeriesID]struct{}{oneID: {}})
+ req.NoError(err)
+ req.Len(subset, 1, "a subset request stays scoped to that subset")
+ req.Equal(entityBySeries[oneID], string(subset[oneID]))
+
+ bogus, err :=
resolver.PartSeriesMap(map[common.SeriesID]struct{}{common.SeriesID(0xdeadbeef):
{}})
+ req.NoError(err)
+ req.Empty(bogus, "an unknown series resolves to nothing")
+
+ // Distributed write-path coverage: a part produced by the data-node
path
+ // carries a part-level smeta.bin, so OpenPart loads its series map
directly
+ // and the iterator skips the index scan. Simulate that by writing a
real
+ // smeta.bin per part, then reopen with a fresh resolver (empty cache)
so the
+ // indexed tags are genuinely re-resolved from the smeta-provided
EntityValues.
+ req.NoError(resolver.Close())
+ smetaResolver, err := dump.NewIndexResolver(segmentPath, 0, ruleToTag)
+ req.NoError(err)
+ defer smetaResolver.Close()
+ smetaSeen := 0
+ for partDir, series := range partSeriesByDir {
+ partID, parseErr := strconv.ParseUint(filepath.Base(partDir),
16, 64)
+ req.NoError(parseErr)
+ docs := make(index.Documents, 0, len(series))
+ for _, entityValues := range series {
+ docs = append(docs, index.Document{EntityValues:
entityValues})
+ }
+ smeta, marshalErr := docs.Marshal()
+ req.NoError(marshalErr)
+ req.NoError(os.WriteFile(filepath.Join(partDir, "smeta.bin"),
smeta, 0o600))
+
+ p, openErr := OpenPart(partID, filepath.Dir(partDir),
fileSystem)
+ req.NoError(openErr)
+ req.NotNil(p.SeriesMap(), "smeta.bin present -> series map
loaded, index scan skipped")
+ p.SetIndexResolver(smetaResolver)
+ it := p.Iterator()
+ for it.Next() {
+ r := it.Row()
+ smetaSeen++
+ req.Equal(series[r.SeriesID], r.EntityValues,
"EntityValues sourced from smeta.bin")
+ plain :=
dump.DecodeTagValue(r.TagTypes["default.strTag"], r.Tags["default.strTag"],
nil).GetStr().GetValue()
+ n, atoiErr := strconv.Atoi(plain[len("plain"):])
+ req.NoError(atoiErr)
+ iStr := strconv.Itoa(n)
+ req.Len(r.IndexedTags, 3, "indexed tags still resolve
on the smeta fast path")
+ req.Equal("indexed"+iStr,
string(r.IndexedTags[strRuleID][0]))
+ req.Equal(convert.Int64ToBytes(int64(70+n)),
r.IndexedTags[intRuleID][0])
+ arrVals := r.IndexedTags[arrRuleID]
+ req.Len(arrVals, 2)
+ arrSet := map[string]bool{string(arrVals[0]): true,
string(arrVals[1]): true}
+ req.True(arrSet["arr"+iStr+"a"] &&
arrSet["arr"+iStr+"b"], "both array elements recovered")
+ }
+ req.NoError(it.Err())
+ it.Close()
+ p.Close()
+ }
+ req.Equal(total, smetaSeen, "every data point read back on the smeta
path too")
+}
+
+func registerIndexedResolveMeasure(t *testing.T, metaSvc
metadataservice.Service) {
+ reg := metaSvc.SchemaRegistry()
+ ctx := context.TODO()
+ _, err := reg.CreateGroup(ctx, &commonv1.Group{
+ Metadata: &commonv1.Metadata{Name: idxrGroup},
+ Catalog: commonv1.Catalog_CATALOG_MEASURE,
+ ResourceOpts: &commonv1.ResourceOpts{
+ ShardNum: 1,
+ SegmentInterval: &commonv1.IntervalRule{Unit:
commonv1.IntervalRule_UNIT_DAY, Num: 1},
+ Ttl: &commonv1.IntervalRule{Unit:
commonv1.IntervalRule_UNIT_DAY, Num: 7},
+ },
+ })
+ require.NoError(t, err)
+ _, err = reg.CreateMeasure(ctx, &databasev1.Measure{
+ Metadata: &commonv1.Metadata{Name: idxrMeasure, Group:
idxrGroup},
+ TagFamilies: []*databasev1.TagFamilySpec{{
+ Name: "default",
+ Tags: []*databasev1.TagSpec{
+ {Name: "series", Type:
databasev1.TagType_TAG_TYPE_STRING},
+ {Name: "strTag", Type:
databasev1.TagType_TAG_TYPE_STRING},
+ {Name: "idxStr", Type:
databasev1.TagType_TAG_TYPE_STRING},
+ {Name: "idxInt", Type:
databasev1.TagType_TAG_TYPE_INT},
+ {Name: "idxArr", Type:
databasev1.TagType_TAG_TYPE_STRING_ARRAY},
+ },
+ }},
+ Fields: []*databasev1.FieldSpec{
+ {
+ Name: "intField",
+ FieldType:
databasev1.FieldType_FIELD_TYPE_INT,
+ EncodingMethod:
databasev1.EncodingMethod_ENCODING_METHOD_GORILLA,
+ CompressionMethod:
databasev1.CompressionMethod_COMPRESSION_METHOD_ZSTD,
+ },
+ },
+ Entity: &databasev1.Entity{TagNames: []string{"series"}},
+ })
+ require.NoError(t, err)
+ for _, ir := range []struct {
+ name string
+ tag string
+ }{
+ {"idxr_str_rule", "idxStr"},
+ {"idxr_int_rule", "idxInt"},
+ {"idxr_arr_rule", "idxArr"},
+ } {
+ _, err = reg.CreateIndexRule(ctx, &databasev1.IndexRule{
+ Metadata: &commonv1.Metadata{Name: ir.name, Group:
idxrGroup},
+ Tags: []string{ir.tag},
+ Type: databasev1.IndexRule_TYPE_INVERTED,
+ })
+ require.NoError(t, err)
+ }
+ _, err = reg.CreateIndexRuleBinding(ctx, &databasev1.IndexRuleBinding{
+ Metadata: &commonv1.Metadata{Name: "idxr_binding", Group:
idxrGroup},
+ Subject: &databasev1.Subject{Name: idxrMeasure, Catalog:
commonv1.Catalog_CATALOG_MEASURE},
+ Rules: []string{"idxr_str_rule", "idxr_int_rule",
"idxr_arr_rule"},
+ BeginAt: timestamppb.New(time.Now().Add(-time.Hour)),
+ ExpireAt: timestamppb.New(time.Now().Add(24 * time.Hour)),
+ })
+ require.NoError(t, err)
+}
+
+func strTagValue(s string) *modelv1.TagValue {
+ return &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str:
&modelv1.Str{Value: s}}}
+}
+
+func intTagValue(v int64) *modelv1.TagValue {
+ return &modelv1.TagValue{Value: &modelv1.TagValue_Int{Int:
&modelv1.Int{Value: v}}}
+}
+
+func strArrTagValue(vals ...string) *modelv1.TagValue {
+ return &modelv1.TagValue{Value: &modelv1.TagValue_StrArray{StrArray:
&modelv1.StrArray{Value: vals}}}
+}
+
+func findSidxSegmentPath(t *testing.T, root string) string {
+ t.Helper()
+ var seg string
+ _ = filepath.WalkDir(root, func(path string, d fs.DirEntry, err error)
error {
+ if err == nil && d.IsDir() && d.Name() == "sidx" {
+ seg = filepath.Dir(path)
+ }
+ return nil
+ })
+ require.NotEmpty(t, seg, "sidx directory not found under %s", root)
+ return seg
+}
diff --git a/pkg/index/index.go b/pkg/index/index.go
index 2c6372e99..2e5f11db6 100644
--- a/pkg/index/index.go
+++ b/pkg/index/index.go
@@ -880,6 +880,11 @@ type SeriesStore interface {
SeriesIterator(context.Context) (FieldIterator[Series], error)
SeriesSort(ctx context.Context, indexQuery Query, orderBy *OrderBy,
preLoadSize int, fieldKeys []FieldKey) (iter
FieldIterator[*DocumentResult], err error)
+ // StoredFields returns the document's stored fields keyed by name (a
field
+ // may have several values), always excluding internal bookkeeping
fields
+ // (_id, _timestamp, ...). A non-empty projection limits the result to
the
+ // listed fields. Returns nil when no document matches.
+ StoredFields(ctx context.Context, docID []byte, projection ...FieldKey)
(map[string][][]byte, error)
}
// SeriesMatcherType represents the type of series matcher.
diff --git a/pkg/index/inverted/inverted_series.go
b/pkg/index/inverted/inverted_series.go
index 5a64a4283..9c749ec00 100644
--- a/pkg/index/inverted/inverted_series.go
+++ b/pkg/index/inverted/inverted_series.go
@@ -226,6 +226,59 @@ func (s *store) Search(ctx context.Context,
return parseResult(dmi, projection, limit)
}
+// StoredFields implements index.SeriesStore.
+func (s *store) StoredFields(ctx context.Context, docID []byte, projection
...index.FieldKey) (map[string][][]byte, error) {
+ reader, err := s.writer.Reader()
+ if err != nil {
+ return nil, err
+ }
+ defer func() {
+ if r := recover(); r != nil {
+ _ = reader.Close()
+ panic(r)
+ }
+ _ = reader.Close()
+ }()
+
+ q := bluge.NewTermQuery(convert.BytesToString(docID))
+ q.SetField(docIDField)
+ dmi, err := reader.Search(ctx, bluge.NewAllMatches(q))
+ if err != nil {
+ return nil, err
+ }
+ match, err := dmi.Next()
+ if err != nil {
+ return nil, err
+ }
+ if match == nil {
+ return nil, nil
+ }
+ var want map[string]struct{}
+ if len(projection) > 0 {
+ want = make(map[string]struct{}, len(projection))
+ for i := range projection {
+ want[projection[i].Marshal()] = struct{}{}
+ }
+ }
+ fields := make(map[string][][]byte)
+ if visitErr := match.VisitStoredFields(func(field string, value []byte)
bool {
+ switch field {
+ case docIDField, seriesIDField, timestampField, versionField:
+ return true // always skip internal bookkeeping fields,
even if projected
+ }
+ if want != nil {
+ if _, ok := want[field]; !ok {
+ return true // not in the requested projection
+ }
+ }
+ fields[field] = append(fields[field], bytes.Clone(value))
+ return true
+ }); visitErr != nil {
+ return nil, visitErr
+ }
+ return fields, nil
+}
+
func parseResult(dmi search.DocumentMatchIterator, loadedFields
[]index.FieldKey, limit int) ([]index.SeriesDocument, error) {
result := make([]index.SeriesDocument, 0, 10)
next, err := dmi.Next()
diff --git a/pkg/index/inverted/inverted_series_test.go
b/pkg/index/inverted/inverted_series_test.go
index 3814915eb..c841dc697 100644
--- a/pkg/index/inverted/inverted_series_test.go
+++ b/pkg/index/inverted/inverted_series_test.go
@@ -1007,3 +1007,69 @@ func updateData(tester *require.Assertions, s
index.SeriesStore) {
tester.NoError(s.UpdateSeriesBatch(b1))
tester.NoError(s.UpdateSeriesBatch(b2))
}
+
+func TestStore_StoredFields(t *testing.T) {
+ tester := require.New(t)
+ path, fn := setUp(tester)
+ s, err := NewStore(StoreOpts{
+ Path: path,
+ Logger: logger.GetLogger("test"),
+ })
+ tester.NoError(err)
+ defer func() {
+ tester.NoError(s.Close())
+ fn()
+ }()
+ insertData(tester, s)
+ ctx := context.TODO()
+
+ // No projection: every stored field of test2 is returned, internal
+ // bookkeeping fields (_id, _timestamp, ...) excluded.
+ all, err := s.StoredFields(ctx, []byte("test2"))
+ tester.NoError(err)
+ tester.Equal([][]byte{convert.Int64ToBytes(100)},
all[fieldKeyDuration.Marshal()])
+ tester.Equal([][]byte{[]byte("svc2")},
all[fieldKeyServiceName.Marshal()])
+ tester.Equal([][]byte{convert.Int64ToBytes(100)},
all[fieldKeyStartTime.Marshal()])
+ tester.NotContains(all, docIDField)
+ tester.NotContains(all, timestampField)
+
+ // Projection: only the requested field is returned.
+ only, err := s.StoredFields(ctx, []byte("test2"), fieldKeyDuration)
+ tester.NoError(err)
+ tester.Len(only, 1)
+ tester.Equal([][]byte{convert.Int64ToBytes(100)},
only[fieldKeyDuration.Marshal()])
+
+ // A two-field projection returns exactly those two.
+ two, err := s.StoredFields(ctx, []byte("test2"), fieldKeyDuration,
fieldKeyServiceName)
+ tester.NoError(err)
+ tester.Len(two, 2)
+ tester.Contains(two, fieldKeyDuration.Marshal())
+ tester.Contains(two, fieldKeyServiceName.Marshal())
+
+ // A missing document yields nil.
+ none, err := s.StoredFields(ctx, []byte("does-not-exist"))
+ tester.NoError(err)
+ tester.Nil(none)
+
+ // An array indexed tag stores several values under one field name;
they all
+ // come back (the result is [][]byte per field, not a single value).
+ arrKey := index.FieldKey{IndexRuleID: 4099}
+ tester.NoError(s.InsertSeriesBatch(index.Batch{Documents:
[]index.Document{{
+ EntityValues: []byte("arr1"),
+ Fields: []index.Field{
+ field(arrKey, []byte("a"), true),
+ field(arrKey, []byte("b"), true),
+ field(fieldKeyServiceName, []byte("svcArr"), true),
+ },
+ }}}))
+ arr, err := s.StoredFields(ctx, []byte("arr1"))
+ tester.NoError(err)
+ tester.ElementsMatch([][]byte{[]byte("a"), []byte("b")},
arr[arrKey.Marshal()])
+ tester.Equal([][]byte{[]byte("svcArr")},
arr[fieldKeyServiceName.Marshal()])
+
+ // Internal bookkeeping fields are excluded even when explicitly
projected.
+ internalProj, err := s.StoredFields(ctx, []byte("test2"),
index.FieldKey{TagName: docIDField}, fieldKeyDuration)
+ tester.NoError(err)
+ tester.NotContains(internalProj, docIDField)
+ tester.Equal([][]byte{convert.Int64ToBytes(100)},
internalProj[fieldKeyDuration.Marshal()])
+}
