You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/09/02 00:44:04 UTC
(skywalking-banyandb) branch main updated: feat(index): walk stored documents natively for migration source reads (#1318)
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 411a39414 feat(index): walk stored documents natively for migration source reads (#1318)
411a39414 is described below
commit 411a394142c364c5e437ca19d2bcb51b3f8153c7
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Sep 2 08:43:59 2026 +0800
feat(index): walk stored documents natively for migration source reads (#1318)
---
banyand/internal/migration/unionsidx.go | 163 +++---
.../migration/unionsidx_native_source_test.go | 328 +++++++++++
banyand/internal/migration/unionsidx_test.go | 44 ++
banyand/measure/migration_indexmode_copy.go | 133 +----
banyand/measure/migration_indexmode_copy_test.go | 9 +
.../migration_indexmode_native_source_test.go | 375 +++++++++++++
banyand/measure/migration_verify.go | 123 ++++-
pkg/index/inverted/internal/nativeice/nativeice.go | 614 +++++++++++++++++++--
pkg/index/inverted/inverted.go | 51 ++
pkg/index/inverted/native_dependency_test.go | 5 +
pkg/index/inverted/native_fixture_nidx01c_test.go | 331 +++++++++++
pkg/index/inverted/native_multichunk_walk_test.go | 97 ++++
pkg/index/inverted/native_nidx01b_boundary_test.go | 15 +-
pkg/index/inverted/native_nidx01c_walk_test.go | 463 ++++++++++++++++
.../inverted/testdata/nidx01c/provenance.json | 90 +++
.../testdata/nidx01c/sourceA/000000000002.seg | Bin 0 -> 1638 bytes
.../testdata/nidx01c/sourceA/000000000003.snp | Bin 0 -> 66 bytes
.../testdata/nidx01c/sourceB/000000000001.snp | Bin 0 -> 48 bytes
.../testdata/nidx01c/sourceB/000000000002.seg | Bin 0 -> 1241 bytes
19 files changed, 2593 insertions(+), 248 deletions(-)
diff --git a/banyand/internal/migration/unionsidx.go b/banyand/internal/migration/unionsidx.go
index 29d519db0..7eecd7525 100644
--- a/banyand/internal/migration/unionsidx.go
+++ b/banyand/internal/migration/unionsidx.go
@@ -19,6 +19,7 @@ package migration
import (
"context"
+ "errors"
"fmt"
"os"
"path/filepath"
@@ -28,11 +29,13 @@ import (
"sync/atomic"
"github.com/blugelabs/bluge"
- blugesearch "github.com/blugelabs/bluge/search"
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/index/inverted"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/run"
)
// The union sidx is rebuilt read-only via raw bluge — no banyandb store
@@ -78,37 +81,21 @@ func BuildGroupUnionSidx(ctx context.Context, srcGroupRoots []string, stagingPat
return "", fmt.Errorf("open union sidx writer at %q: %w", stagingPath, err)
}
closed := false
+ published := false
defer func() {
if !closed {
_ = writer.Close()
}
+ if !published {
+ _ = os.RemoveAll(stagingPath)
+ }
}()
- // Collect every source sidx directory across all node + segment
- // roots up front so the worker pool can fan out cleanly.
- var sidxPaths []string
- for _, srcGroupRoot := range srcGroupRoots {
- if ctx.Err() != nil {
- return "", ctx.Err()
- }
- segEntries, err := os.ReadDir(srcGroupRoot)
- if err != nil {
- if os.IsNotExist(err) {
- continue
- }
- return "", fmt.Errorf("read src group root %q: %w", srcGroupRoot, err)
- }
- for _, se := range segEntries {
- if !se.IsDir() || !strings.HasPrefix(se.Name(), segPrefix) {
- continue
- }
- srcSidxPath := filepath.Join(srcGroupRoot, se.Name(), sidxDirName)
- info, statErr := os.Stat(srcSidxPath)
- if statErr != nil || !info.IsDir() {
- continue
- }
- sidxPaths = append(sidxPaths, srcSidxPath)
- }
+ // Collect every source sidx directory across all node + segment roots up
+ // front so the worker pool can fan out cleanly.
+ sidxPaths, pathsErr := collectSourceSidxPaths(ctx, srcGroupRoots)
+ if pathsErr != nil {
+ return "", pathsErr
}
seen := make(map[common.SeriesID]struct{}, 1_000_000)
@@ -134,17 +121,31 @@ func BuildGroupUnionSidx(ctx context.Context, srcGroupRoots []string, stagingPat
var wg sync.WaitGroup
workerCtx, cancelWorkers := context.WithCancel(ctx)
defer cancelWorkers()
+ workerLogger := logger.GetLogger("migration")
+ workerTasks := make([]*run.Task, 0, workerCount)
+ recordWorkerPanic := func(panicValue any) {
+ panicErr := fmt.Errorf("union sidx source reader panicked: %v", panicValue)
+ if firstErr.CompareAndSwap(nil, &panicErr) {
+ cancelWorkers()
+ }
+ }
for i := 0; i < workerCount; i++ {
wg.Add(1)
- go func() {
+ workerTask := run.Go(workerCtx, "union sidx source reader", workerLogger, func(taskCtx context.Context) {
defer wg.Done()
+ defer func() {
+ if panicValue := recover(); panicValue != nil {
+ recordWorkerPanic(panicValue)
+ panic(panicValue)
+ }
+ }()
for srcSidxPath := range pathCh {
- if workerCtx.Err() != nil {
+ if taskCtx.Err() != nil {
return
}
logf("union sidx: start scanning %s", srcSidxPath)
- count, scanned, mergeErr := mergeOneSourceSidxInto(workerCtx, srcSidxPath, writer, seen, &seenMu, &writerMu)
+ count, scanned, mergeErr := mergeOneSourceSidxInto(taskCtx, srcSidxPath, writer, seen, &seenMu, &writerMu)
if mergeErr != nil {
e := fmt.Errorf("merge %s: %w", srcSidxPath, mergeErr)
if firstErr.CompareAndSwap(nil, &e) {
@@ -158,16 +159,25 @@ func BuildGroupUnionSidx(ctx context.Context, srcGroupRoots []string, stagingPat
logf("union sidx: scanned %d/%d sidx dir(s) (%.1f%%): %s",
done, len(sidxPaths), float64(done)*100/float64(len(sidxPaths)), srcSidxPath)
}
- }()
+ })
+ workerTasks = append(workerTasks, workerTask)
}
- for _, p := range sidxPaths {
+
+dispatch:
+ for _, srcSidxPath := range sidxPaths {
select {
- case pathCh <- p:
+ case pathCh <- srcSidxPath:
case <-workerCtx.Done():
+ break dispatch
}
}
close(pathCh)
wg.Wait()
+ for _, workerTask := range workerTasks {
+ if outcome := workerTask.Wait(); outcome != nil && outcome.Panicked {
+ recordWorkerPanic(outcome.PanicValue)
+ }
+ }
if errPtr := firstErr.Load(); errPtr != nil {
return "", *errPtr
}
@@ -181,12 +191,40 @@ func BuildGroupUnionSidx(ctx context.Context, srcGroupRoots []string, stagingPat
return "", fmt.Errorf("close union sidx writer: %w", closeErr)
}
if inserted == 0 {
- _ = os.RemoveAll(stagingPath)
return "", nil
}
+ published = true
return stagingPath, nil
}
+func collectSourceSidxPaths(ctx context.Context, srcGroupRoots []string) ([]string, error) {
+ var sidxPaths []string
+ for _, srcGroupRoot := range srcGroupRoots {
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ return nil, ctxErr
+ }
+ segmentEntries, readErr := os.ReadDir(srcGroupRoot)
+ if readErr != nil {
+ if os.IsNotExist(readErr) {
+ continue
+ }
+ return nil, fmt.Errorf("read src group root %q: %w", srcGroupRoot, readErr)
+ }
+ for _, segmentEntry := range segmentEntries {
+ if !segmentEntry.IsDir() || !strings.HasPrefix(segmentEntry.Name(), segPrefix) {
+ continue
+ }
+ sourceSidxPath := filepath.Join(srcGroupRoot, segmentEntry.Name(), sidxDirName)
+ info, statErr := os.Stat(sourceSidxPath)
+ if statErr != nil || !info.IsDir() {
+ continue
+ }
+ sidxPaths = append(sidxPaths, sourceSidxPath)
+ }
+ }
+ return sidxPaths, nil
+}
+
func mergeOneSourceSidxInto(
ctx context.Context,
srcPath string,
@@ -195,25 +233,6 @@ func mergeOneSourceSidxInto(
seenMu *sync.Mutex,
writerMu *sync.Mutex,
) (inserted, scanned int, err error) {
- reader, err := bluge.OpenReader(bluge.DefaultConfig(srcPath))
- if err != nil {
- // A sidx directory may exist on disk while carrying no committed
- // bluge snapshot — e.g. a fresh segment created by the runtime
- // whose sidx writer never received a doc. Treat that as "no
- // sidx to merge for this seg" instead of aborting the whole
- // per-group union build.
- if strings.Contains(err.Error(), "unable to find a usable snapshot") {
- return 0, 0, nil
- }
- return 0, 0, fmt.Errorf("open reader: %w", err)
- }
- defer func() { _ = reader.Close() }()
-
- dmi, err := reader.Search(ctx, bluge.NewAllMatches(bluge.NewMatchAllQuery()))
- if err != nil {
- return 0, 0, fmt.Errorf("search: %w", err)
- }
-
batch := bluge.NewBatch()
batched := 0
@@ -223,46 +242,46 @@ func mergeOneSourceSidxInto(
return dst.Batch(batch)
}
- for {
- next, nextErr := dmi.Next()
- if nextErr != nil {
- return inserted, scanned, fmt.Errorf("iterate docs: %w", nextErr)
- }
- if next == nil {
- break
- }
+ walkErr := inverted.ReadOnlyWalkDocuments(ctx, srcPath, func(source inverted.StoredDocument) error {
scanned++
- doc, dup, buildErr := buildDocFromMatchLocked(next, seen, seenMu)
+ doc, dup, buildErr := buildDocFromStoredDocumentLocked(source, seen, seenMu)
if buildErr != nil {
- return inserted, scanned, buildErr
+ return buildErr
}
if dup || doc == nil {
- continue
+ return nil
}
batch.Insert(doc)
batched++
inserted++
if batched >= unionSidxBatchSize {
- if err := flush(); err != nil {
- return inserted, scanned, fmt.Errorf("flush batch: %w", err)
+ if flushErr := flush(); flushErr != nil {
+ return fmt.Errorf("flush batch: %w", flushErr)
}
batch = bluge.NewBatch()
batched = 0
}
+ return nil
+ })
+ if walkErr != nil {
+ if errors.Is(walkErr, inverted.ErrNoCommittedIndex) {
+ return 0, 0, nil
+ }
+ return inserted, scanned, fmt.Errorf("walk source %s: %w", srcPath, walkErr)
}
if batched > 0 {
- if err := flush(); err != nil {
- return inserted, scanned, fmt.Errorf("flush tail batch: %w", err)
+ if flushErr := flush(); flushErr != nil {
+ return inserted, scanned, fmt.Errorf("flush tail batch: %w", flushErr)
}
}
return inserted, scanned, nil
}
-// buildDocFromMatchLocked rebuilds one series-index doc from its stored
-// fields, deduplicating by SeriesID under seenMu.
-func buildDocFromMatchLocked(
- match *blugesearch.DocumentMatch,
+// buildDocFromStoredDocumentLocked rebuilds one series-index doc from its
+// stored fields, deduplicating by SeriesID under seenMu.
+func buildDocFromStoredDocumentLocked(
+ source inverted.StoredDocument,
seen map[common.SeriesID]struct{},
seenMu *sync.Mutex,
) (*bluge.Document, bool, error) {
@@ -272,7 +291,7 @@ func buildDocFromMatchLocked(
value []byte
}
var fields []storedField
- visitErr := match.VisitStoredFields(func(field string, value []byte) bool {
+ visitErr := source.VisitStoredFields(func(field string, value []byte) bool {
switch field {
case sidxDocIDField:
entityValues = append([]byte(nil), value...)
diff --git a/banyand/internal/migration/unionsidx_native_source_test.go b/banyand/internal/migration/unionsidx_native_source_test.go
new file mode 100644
index 000000000..5d3bc6698
--- /dev/null
+++ b/banyand/internal/migration/unionsidx_native_source_test.go
@@ -0,0 +1,328 @@
+// 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 migration
+
+import (
+ "context"
+ "go/ast"
+ "go/parser"
+ "go/token"
+ "os"
+ "path/filepath"
+ "sort"
+ "strconv"
+ "strings"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/index/inverted"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+)
+
+const (
+ // nativeVisitorPackage and nativeVisitorFunc name the BanyanDB-owned
+ // read-only document visitor issue #14010 introduces.
+ nativeVisitorPackage = "github.com/apache/skywalking-banyandb/pkg/index/inverted"
+ nativeVisitorFunc = "ReadOnlyWalkDocuments"
+
+ // retiredReaderEntryPoint is the call that opens a retired third-party
+ // index reader directly on a source directory. This package reads sources
+ // and writes the union destination; only the read side changes, so the
+ // destination writer's own entry point is untouched by this assertion.
+ retiredReaderEntryPoint = "OpenReader"
+
+ // iceFooterLength is the fixed width of an ICE v3 segment footer. A segment
+ // file shorter than it cannot carry a valid one.
+ iceFooterLength = 60
+
+ // unionSourceTagName is the stored tag the seeded sources carry, and
+ // publishedSeriesLimit bounds a destination lookup so a stray duplicate is
+ // reported rather than silently truncated away.
+ unionSourceTagName = "service"
+ publishedSeriesLimit = 100
+)
+
+// TestSeriesUnionSourceReadReachesNativeVisitor is the structural half of the
+// series-union cutover. Publishing the right series is not enough on its own:
+// the point of the milestone is that the source read stops going through the
+// retired reader, and a union that is right for the wrong reason would hide
+// that.
+//
+// Requirement proved here:
+//
+// R5 -- series union's source read reaches the BanyanDB-owned native document
+// visitor, and no file in this package opens a retired index reader
+// directly any more.
+func TestSeriesUnionSourceReadReachesNativeVisitor(t *testing.T) {
+ tester := require.New(t)
+
+ entries, err := os.ReadDir(".")
+ tester.NoError(err)
+ reachesVisitor := false
+ sources := 0
+ for _, entry := range entries {
+ name := entry.Name()
+ if entry.IsDir() || !strings.HasSuffix(name, ".go") || strings.HasSuffix(name, "_test.go") {
+ continue
+ }
+ sources++
+ fileSet := token.NewFileSet()
+ file, parseErr := parser.ParseFile(fileSet, name, nil, parser.ParseComments)
+ tester.NoError(parseErr)
+ imports := importPathsByQualifier(t, file)
+ visitorQualifiers := qualifiersForImport(imports, nativeVisitorPackage)
+ ast.Inspect(file, func(node ast.Node) bool {
+ selector, isSelector := node.(*ast.SelectorExpr)
+ if !isSelector {
+ return true
+ }
+ tester.NotEqual(retiredReaderEntryPoint, selector.Sel.Name,
+ "%s must not open a retired index reader directly at %s", name, fileSet.Position(selector.Pos()))
+ qualifier, isIdent := selector.X.(*ast.Ident)
+ if !isIdent || selector.Sel.Name != nativeVisitorFunc {
+ return true
+ }
+ if _, viaVisitor := visitorQualifiers[qualifier.Name]; viaVisitor {
+ reachesVisitor = true
+ }
+ return true
+ })
+ }
+ tester.NotZero(sources, "the migration package must have source files")
+ tester.True(reachesVisitor, "series union must read its source documents through %s.%s",
+ nativeVisitorPackage, nativeVisitorFunc)
+}
+
+// importPathsByQualifier returns the imported package paths of one parsed file,
+// keyed by the local name each is referred to by.
+func importPathsByQualifier(t *testing.T, file *ast.File) map[string]string {
+ t.Helper()
+ imports := map[string]string{}
+ for _, spec := range file.Imports {
+ importPath, err := strconv.Unquote(spec.Path.Value)
+ require.NoError(t, err)
+ name := filepath.Base(importPath)
+ if spec.Name != nil {
+ name = spec.Name.Name
+ }
+ imports[name] = importPath
+ }
+ return imports
+}
+
+// qualifiersForImport returns the local names one import path is referred to by.
+func qualifiersForImport(imports map[string]string, wanted string) map[string]struct{} {
+ names := map[string]struct{}{}
+ for name, importPath := range imports {
+ if importPath == wanted {
+ names[name] = struct{}{}
+ }
+ }
+ return names
+}
+
+// unionSourceSeries is one series a legacy-created source segment index holds.
+// The label is issue #14010's name for the document; a series identifier is a
+// hash of the marshaled entity buffer, so the entity value, not the label, is
+// what the fixture chooses.
+type unionSourceSeries struct {
+ label string
+ entityValue string
+ tagValue string
+ timestamp int64
+ version int64
+ deleted bool
+}
+
+// unionNativeSourceSeries is the content issue #14010 declares for the union
+// case: two live series and one deleted series that must not be published.
+var unionNativeSourceSeries = []unionSourceSeries{
+ {label: "101", entityValue: "series-101", tagValue: "blue", timestamp: 100, version: 2},
+ {label: "202", entityValue: "series-202", tagValue: "red", timestamp: 200, version: 1},
+ {label: "303", entityValue: "series-303", tagValue: "gray", timestamp: 300, version: 3, deleted: true},
+}
+
+// unionSourceIdentity returns the marshaled entity buffer the store records as
+// one declared series' identity.
+func unionSourceIdentity(t *testing.T, declared unionSourceSeries) []byte {
+ t.Helper()
+ series := &pbv1.Series{Subject: "svc", EntityValues: []*modelv1.TagValue{{
+ Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: declared.entityValue}},
+ }}}
+ require.NoError(t, series.Marshal())
+ return append([]byte(nil), series.Buffer...)
+}
+
+// seedUnionSourceSidx writes the declared series into
+// "<groupRoot>/<segName>/sidx" with the compatibility writer and deletes the
+// ones declared deleted, so the source is a directory a released BanyanDB
+// produced rather than one this test hand-assembled.
+func seedUnionSourceSidx(t *testing.T, groupRoot, segName string, seriesSet []unionSourceSeries) string {
+ t.Helper()
+ sidxDir := filepath.Join(groupRoot, segName, sidxDirName)
+ require.NoError(t, os.MkdirAll(sidxDir, storage.DirPerm))
+
+ store, err := inverted.NewStore(inverted.StoreOpts{Path: sidxDir, BatchWaitSec: 0})
+ require.NoError(t, err)
+ documents := make(index.Documents, 0, len(seriesSet))
+ var deletedIdentities [][]byte
+ for _, declared := range seriesSet {
+ identity := unionSourceIdentity(t, declared)
+ tag := index.NewBytesField(index.FieldKey{TagName: unionSourceTagName}, convert.StringToBytes(declared.tagValue))
+ tag.Store = true
+ tag.Index = true
+ tag.NoSort = true
+ documents = append(documents, index.Document{
+ Fields: []index.Field{tag},
+ EntityValues: identity,
+ Timestamp: declared.timestamp,
+ DocID: convert.Hash(identity),
+ Version: declared.version,
+ })
+ if declared.deleted {
+ deletedIdentities = append(deletedIdentities, identity)
+ }
+ }
+ require.NoError(t, store.UpdateSeriesBatch(index.Batch{Documents: documents}))
+ if len(deletedIdentities) > 0 {
+ require.NoError(t, store.Delete(deletedIdentities))
+ }
+ require.NoError(t, store.Close())
+ return sidxDir
+}
+
+// openPublishedDestination opens a published union index through the retained
+// series-store surface, which this milestone does not touch, so what it reports
+// cannot agree with a faulty source read by construction.
+func openPublishedDestination(t *testing.T, sidxDir string) index.SeriesStore {
+ t.Helper()
+ store, err := inverted.NewStore(inverted.StoreOpts{Path: sidxDir, BatchWaitSec: 0})
+ require.NoError(t, err)
+ return store
+}
+
+// publishedUnionSeries looks one identity up in a published union index and
+// reports the stored tag values it holds, or reports that the index holds no
+// live document under that identity. The lookup is an exact one so that the
+// answer for one series can never be contaminated by another's.
+func publishedUnionSeries(t *testing.T, store index.SeriesStore, identity []byte) ([]string, bool) {
+ t.Helper()
+ query, err := store.BuildQuery([]index.SeriesMatcher{
+ {Type: index.SeriesMatcherTypeExact, Match: identity},
+ }, nil, nil)
+ require.NoError(t, err)
+ documents, err := store.Search(context.Background(),
+ []index.FieldKey{{TagName: unionSourceTagName}}, query, publishedSeriesLimit)
+ require.NoError(t, err)
+ if len(documents) == 0 {
+ return nil, false
+ }
+ require.Len(t, documents, 1, "identity %q must resolve to at most one live published document", identity)
+ var values []string
+ if value, stored := documents[0].Fields[unionSourceTagName]; stored && value != nil {
+ values = append(values, string(value))
+ }
+ sort.Strings(values)
+ return values, true
+}
+
+// truncateUnionSourceSegment cuts a source segment file short of the footer its
+// on-disk grammar requires, which is what a segment published by a process that
+// died mid-write looks like. The directory still holds a committed manifest
+// naming that segment, so the damage is only visible to a reader that validates
+// the segment it is about to read.
+func truncateUnionSourceSegment(t *testing.T, sidxDir string) {
+ t.Helper()
+ segments, err := filepath.Glob(filepath.Join(sidxDir, "*.seg"))
+ require.NoError(t, err)
+ require.NotEmpty(t, segments, "the seeded source must hold a committed segment")
+ sort.Strings(segments)
+ for _, segmentPath := range segments {
+ require.NoError(t, os.Truncate(segmentPath, iceFooterLength-8))
+ }
+}
+
+// TestE2ESeriesUnionNativeSource walks the production situation this milestone
+// exists for: a lifecycle migration unions the series indexes of several node
+// replicas into one staged index before broadcasting it. The union reads every
+// source through the native document visitor and re-emits the survivors through
+// its retained destination writer, which this milestone does not change.
+//
+// Requirements proved here:
+//
+// R3 -- the union reads both sources natively and publishes one copy of each
+// live series. The second source repeats the first's series 101, and the
+// existing first-seen dedup winner rule is unchanged, so the destination
+// holds 101 once and 202 once; the deleted series 303 is never
+// published.
+//
+// R4 -- a source whose committed segment is damaged fails the union with the
+// native typed corruption error and publishes no staged destination, so
+// a migration aborts on unreadable input instead of broadcasting a
+// partial series index. The byte-level stored-chunk failures R4 also
+// names are proved against the visitor itself in pkg/index/inverted,
+// where no retired reader stands in the way.
+func TestE2ESeriesUnionNativeSource(t *testing.T) {
+ tester := require.New(t)
+
+ firstRoot := t.TempDir()
+ secondRoot := t.TempDir()
+ seedUnionSourceSidx(t, firstRoot, "seg-20260621", unionNativeSourceSeries)
+ seedUnionSourceSidx(t, secondRoot, "seg-20260622", unionNativeSourceSeries[:1])
+
+ staging := filepath.Join(t.TempDir(), "union")
+ published, err := BuildGroupUnionSidx(context.Background(), []string{firstRoot, secondRoot}, staging, nil)
+ tester.NoError(err)
+ tester.Equal(staging, published, "a union that read live series must publish its staged index")
+
+ publishedCount, countErr := inverted.ReadOnlyDocCount(published)
+ tester.NoError(countErr)
+ tester.Equal(int64(2), publishedCount,
+ "the union must publish exactly the two live series, each once, even though the first is read twice")
+
+ destination := openPublishedDestination(t, published)
+ defer func() {
+ tester.NoError(destination.Close())
+ }()
+ for _, declared := range unionNativeSourceSeries {
+ identity := unionSourceIdentity(t, declared)
+ values, found := publishedUnionSeries(t, destination, identity)
+ if declared.deleted {
+ tester.False(found, "the deleted series labeled %s must not be published", declared.label)
+ continue
+ }
+ tester.True(found, "the series labeled %s must be published", declared.label)
+ tester.Equal([]string{declared.tagValue}, values,
+ "series %s must keep its declared stored tag value", declared.label)
+ }
+
+ corruptRoot := t.TempDir()
+ truncateUnionSourceSegment(t, seedUnionSourceSidx(t, corruptRoot, "seg-20260621", unionNativeSourceSeries))
+ corruptStaging := filepath.Join(t.TempDir(), "corrupt-union")
+
+ corruptPublished, corruptErr := BuildGroupUnionSidx(context.Background(),
+ []string{corruptRoot}, corruptStaging, nil)
+ tester.ErrorIs(corruptErr, inverted.ErrCorruptIndex,
+ "a damaged source must be reported as a corrupt index, not as an opaque read failure")
+ tester.Empty(corruptPublished, "a union that failed on a damaged source must publish no staged index")
+}
diff --git a/banyand/internal/migration/unionsidx_test.go b/banyand/internal/migration/unionsidx_test.go
index ac83819d3..33d3a5849 100644
--- a/banyand/internal/migration/unionsidx_test.go
+++ b/banyand/internal/migration/unionsidx_test.go
@@ -257,6 +257,50 @@ func TestBuildGroupUnionSidx_EmptyGroupYieldsEmptyPath(t *testing.T) {
}
}
+// TestBuildGroupUnionSidx_UncommittedSourceSidxIsEmpty verifies a discovered
+// source sidx with no committed generation is tolerated as an empty source.
+func TestBuildGroupUnionSidx_UncommittedSourceSidxIsEmpty(t *testing.T) {
+ srcGroupRoot := t.TempDir()
+ srcSidxPath := filepath.Join(srcGroupRoot, segPrefix+"20260101", sidxDirName)
+ if err := os.MkdirAll(srcSidxPath, storage.DirPerm); err != nil {
+ t.Fatalf("mkdir uncommitted source sidx: %v", err)
+ }
+ stagingPath := filepath.Join(t.TempDir(), "union", "sidx")
+
+ resultPath, err := BuildGroupUnionSidx(context.Background(), []string{srcGroupRoot}, stagingPath, nil)
+ if err != nil {
+ t.Fatalf("BuildGroupUnionSidx: %v", err)
+ }
+ if resultPath != "" {
+ t.Fatalf("expected empty result path for an uncommitted source sidx, got %q", resultPath)
+ }
+}
+
+// TestBuildGroupUnionSidx_RecoveredWorkerPanicReturnsError verifies a panic
+// recovered from a source worker fails the union rather than publishing it.
+func TestBuildGroupUnionSidx_RecoveredWorkerPanicReturnsError(t *testing.T) {
+ srcGroupRoot := t.TempDir()
+ segPath := filepath.Join(srcGroupRoot, segPrefix+"20260101")
+ writeSidxAt(t, segPath, nil)
+ stagingPath := filepath.Join(t.TempDir(), "union", "sidx")
+
+ resultPath, buildErr := BuildGroupUnionSidx(context.Background(), []string{srcGroupRoot}, stagingPath,
+ func(format string, _ ...any) {
+ if format == "union sidx: scanned %d/%d sidx dir(s) (%.1f%%): %s" {
+ panic("source worker panic")
+ }
+ })
+ if buildErr == nil {
+ t.Fatal("expected recovered source-worker panic to fail the union")
+ }
+ if resultPath != "" {
+ t.Fatalf("expected no published path after a recovered worker panic, got %q", resultPath)
+ }
+ if _, statErr := os.Stat(stagingPath); !os.IsNotExist(statErr) {
+ t.Fatalf("expected failed union staging path to be removed, stat error: %v", statErr)
+ }
+}
+
func TestBuildGroupUnionSidx_MissingSrcRootIsNoop(t *testing.T) {
stagingPath := filepath.Join(t.TempDir(), "union", "sidx")
resultPath, err := BuildGroupUnionSidx(context.Background(), []string{filepath.Join(t.TempDir(), "no-such-dir")}, stagingPath, nil)
diff --git a/banyand/measure/migration_indexmode_copy.go b/banyand/measure/migration_indexmode_copy.go
index d6f2c51c0..64aa17074 100644
--- a/banyand/measure/migration_indexmode_copy.go
+++ b/banyand/measure/migration_indexmode_copy.go
@@ -20,6 +20,7 @@ package measure
import (
"context"
"encoding/binary"
+ "errors"
"fmt"
"hash/fnv"
"io"
@@ -28,9 +29,6 @@ import (
"sort"
"strings"
- "github.com/blugelabs/bluge"
- blugesearch "github.com/blugelabs/bluge/search"
-
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/banyand/internal/migration"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
@@ -124,64 +122,31 @@ func classifyStoredField(name string, tagNames map[string]struct{}, ruleByID map
return index.FieldKey{TagName: name}, false, 0
}
-// readIndexModeDocs scans every committed bluge doc under sidxDir and rebuilds
-// each as an index.Document from TWO sources:
-//
-// A) stored fields -> regular tag fields (+ timestamp/version), and
-// B) the doc _id unmarshaled to a pbv1.Series -> regenerate the index-only
-// _im_name / _im_entity_tag_* fields, which the write path never stores.
-//
-// ruleByID restores Analyzer/NoSort on indexed fields; schemasBySubject maps
-// each series' Subject to its measure schema so the entity-derived fields are
-// regenerated with the right tag names/types and skip conditions.
-func readIndexModeDocs(ctx context.Context, sidxDir string, ruleByID map[uint32]indexRuleInfo,
+func readIndexModeDocsNative(ctx context.Context, sidxDir string, ruleByID map[uint32]indexRuleInfo,
schemasBySubject map[string]*measureSchemaInfo,
) ([]index.Document, error) {
- r, err := bluge.OpenReader(bluge.DefaultConfig(sidxDir))
- if err != nil {
- // A sidx directory can exist on disk with no committed bluge snapshot
- // (a fresh segment whose writer never received a doc). Treat that as an
- // empty sidx instead of failing the whole copy.
- if strings.Contains(err.Error(), "unable to find a usable snapshot") {
- return nil, nil
- }
- return nil, fmt.Errorf("open sidx reader %s: %w", sidxDir, err)
- }
- defer func() { _ = r.Close() }()
- dmi, err := r.Search(ctx, bluge.NewAllMatches(bluge.NewMatchAllQuery()))
- if err != nil {
- return nil, fmt.Errorf("search sidx %s: %w", sidxDir, err)
- }
- var out []index.Document
+ var documents []index.Document
var timeless []uint64
tagNames := collectTagNames(schemasBySubject)
missingRules := map[uint32]int{}
- for {
- match, nextErr := dmi.Next()
- if nextErr != nil {
- return nil, fmt.Errorf("iterate sidx %s: %w", sidxDir, nextErr)
+ walkErr := inverted.ReadOnlyWalkDocuments(ctx, sidxDir, func(source inverted.StoredDocument) error {
+ document, rebuildErr := rebuildOneDoc(source, ruleByID, schemasBySubject, tagNames, missingRules)
+ if rebuildErr != nil {
+ return fmt.Errorf("rebuild doc in %s: %w", sidxDir, rebuildErr)
}
- if match == nil {
- break
- }
- doc, buildErr := rebuildOneDoc(match, ruleByID, schemasBySubject, tagNames, missingRules)
- if buildErr != nil {
- return nil, fmt.Errorf("rebuild doc in %s: %w", sidxDir, buildErr)
+ if document.Timestamp == 0 {
+ timeless = append(timeless, document.DocID)
+ return nil
}
- // The write path always stamps a checked, non-zero _timestamp on every
- // index-mode doc (write_standalone.go), so Timestamp==0 here means the
- // stored _timestamp is missing or undecodable — corrupt or unexpected
- // source data. Routing such a doc by a guessed segment boundary would
- // silently misplace it, so collect the offenders (without retaining the
- // corrupt docs) and surface them once the whole sidx has been scanned.
- if doc.Timestamp == 0 {
- timeless = append(timeless, doc.DocID)
- continue
+ documents = append(documents, document)
+ return nil
+ })
+ if walkErr != nil {
+ if errors.Is(walkErr, inverted.ErrNoCommittedIndex) {
+ return nil, nil
}
- out = append(out, doc)
+ return nil, fmt.Errorf("walk sidx %s: %w", sidxDir, walkErr)
}
- // Surface corrupt data before anything else: a missing-rule warning implies
- // "continuing", which would be misleading when the read is about to abort.
if len(timeless) > 0 {
sample := timeless
if len(sample) > 10 {
@@ -192,7 +157,7 @@ func readIndexModeDocs(ctx context.Context, sidxDir string, ruleByID map[uint32]
sidxDir, len(timeless), len(sample), sample)
}
warnMissingRules(sidxDir, missingRules)
- return out, nil
+ return documents, nil
}
// warnMissingRules logs a single warning enumerating every IndexRuleID that an
@@ -221,66 +186,6 @@ func warnMissingRules(sidxDir string, missingRules map[uint32]int) {
"catalog; kept indexed with default Analyzer/NoSort — restore the missing index rule(s) for exact fidelity")
}
-// rebuildOneDoc reconstructs a single index.Document from a bluge match,
-// combining the two sources documented on readIndexModeDocs. tagNames is the
-// group's known tag-name set (to tell a tag from a rule id). missingRules, when
-// non-nil, accumulates (rule-id -> count) for fields that decode to an
-// IndexRuleID NOT present in ruleByID, so the caller can warn.
-func rebuildOneDoc(match *blugesearch.DocumentMatch, ruleByID map[uint32]indexRuleInfo,
- schemasBySubject map[string]*measureSchemaInfo, tagNames map[string]struct{}, missingRules map[uint32]int,
-) (index.Document, error) {
- var entityValues []byte
- var ts, version int64
- var fields []index.Field
- visitErr := match.VisitStoredFields(func(name string, value []byte) bool {
- switch name {
- case imDocIDField:
- entityValues = append([]byte(nil), value...)
- case imTimestampField:
- if dt, decErr := bluge.DecodeDateTime(value); decErr == nil {
- ts = dt.UnixNano()
- }
- case imVersionField:
- version = convert.BytesToInt64(value)
- default:
- key, indexed, missingRuleID := classifyStoredField(name, tagNames, ruleByID)
- // NewBytesField clones value internally, so no extra copy of the
- // reusable bluge visit buffer is needed here.
- f := index.NewBytesField(key, value)
- f.Store = true
- f.Index = indexed
- if ri, ok := ruleByID[key.IndexRuleID]; indexed && ok {
- f.NoSort = ri.NoSort
- f.Key.Analyzer = ri.Analyzer
- }
- // A field that decodes to an IndexRuleID the catalog does not know: it
- // stays indexed (defaults), but the operator must be told rather than
- // have a possibly-searchable field silently degraded.
- if missingRuleID != 0 && missingRules != nil {
- missingRules[missingRuleID]++
- }
- fields = append(fields, f)
- }
- return true
- })
- if visitErr != nil {
- return index.Document{}, fmt.Errorf("visit stored fields: %w", visitErr)
- }
- // Source B: rebuild the index-only entity-derived fields from _id.
- var series pbv1.Series
- if err := series.Unmarshal(entityValues); err != nil {
- return index.Document{}, fmt.Errorf("unmarshal series from _id: %w", err)
- }
- fields = appendRegeneratedEntityFields(fields, &series, schemasBySubject[series.Subject])
- return index.Document{
- Fields: fields,
- EntityValues: entityValues,
- Timestamp: ts,
- DocID: uint64(series.ID),
- Version: version,
- }, nil
-}
-
// appendRegeneratedEntityFields rebuilds the index-only fields the production
// write path adds in appendEntityTagsToIndexFields (write_standalone.go:410):
// _im_name (subject) and _im_entity_tag_<tag> (entity values). These are never
@@ -485,7 +390,7 @@ func copyIndexModeGroup(ctx context.Context, in migration.EntryGroupInput,
if ctx.Err() != nil {
return res, ctx.Err()
}
- docs, readErr := readIndexModeDocs(ctx, srcSidx, ruleByID, schemasBySubject)
+ docs, readErr := readIndexModeDocsNative(ctx, srcSidx, ruleByID, schemasBySubject)
if readErr != nil {
return res, readErr
}
diff --git a/banyand/measure/migration_indexmode_copy_test.go b/banyand/measure/migration_indexmode_copy_test.go
index 88ef4f1e4..9ae94d8ea 100644
--- a/banyand/measure/migration_indexmode_copy_test.go
+++ b/banyand/measure/migration_indexmode_copy_test.go
@@ -294,6 +294,15 @@ func TestReadIndexModeDocs_EmptySidx(t *testing.T) {
require.Empty(t, docs)
}
+func TestReadIndexModeDocsNative_EmptySidx(t *testing.T) {
+ sidxDir := filepath.Join(t.TempDir(), directCopySidxDirName)
+ require.NoError(t, os.MkdirAll(sidxDir, storage.DirPerm))
+
+ docs, err := readIndexModeDocsNative(context.Background(), sidxDir, map[uint32]indexRuleInfo{}, nil)
+ require.NoError(t, err)
+ require.Empty(t, docs)
+}
+
// ── Phase 2: copy routing tests ──────────────────────────────────────────────.
// day20260621Nanos returns the local-time start-of-day nanos for 2026-06-21.
diff --git a/banyand/measure/migration_indexmode_native_source_test.go b/banyand/measure/migration_indexmode_native_source_test.go
new file mode 100644
index 000000000..1d9d44fca
--- /dev/null
+++ b/banyand/measure/migration_indexmode_native_source_test.go
@@ -0,0 +1,375 @@
+// 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 measure
+
+import (
+ "context"
+ "go/ast"
+ "go/parser"
+ "go/token"
+ "os"
+ "path/filepath"
+ "sort"
+ "strconv"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/migration"
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/index/inverted"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+)
+
+const (
+ // indexModeSourceReadFile is the file that holds index-mode copy's source
+ // read. Issue #14010 names index-mode copy's source side as one of the two
+ // production readers this milestone switches to the native visitor; the
+ // destination writer and every other reader in this package stay where they
+ // are, so the assertion is scoped to this file rather than to the package.
+ indexModeSourceReadFile = "migration_indexmode_copy.go"
+
+ // nativeVisitorPackage and nativeVisitorFunc name the BanyanDB-owned
+ // read-only document visitor issue #14010 introduces.
+ nativeVisitorPackage = "github.com/apache/skywalking-banyandb/pkg/index/inverted"
+ nativeVisitorFunc = "ReadOnlyWalkDocuments"
+
+ // retiredReaderEntryPoint is the call that opens a retired third-party
+ // index reader directly on a source directory. The whole point of the
+ // milestone is that index-mode copy's source read stops making it.
+ retiredReaderEntryPoint = "OpenReader"
+
+ // iceFooterLength is the fixed width of an ICE v3 segment footer. A segment
+ // file shorter than it cannot carry a valid one.
+ iceFooterLength = 60
+
+ // nativeSourceTagName is the stored, non-indexed tag the seeded sources
+ // carry, and publishedSeriesLimit bounds a destination lookup so a stray
+ // duplicate is reported rather than silently truncated away.
+ nativeSourceTagName = "properties"
+ publishedSeriesLimit = 100
+)
+
+// TestIndexModeCopySourceReadReachesNativeVisitor is the structural half of the
+// index-mode copy cutover. Copying the right documents is not enough on its
+// own: the point of the milestone is that the source read stops going through
+// the retired reader, and a copy that is right for the wrong reason would hide
+// that.
+//
+// Requirement proved here:
+//
+// R5 -- index-mode copy's source read reaches the BanyanDB-owned native
+// document visitor and no longer opens a retired index reader directly.
+func TestIndexModeCopySourceReadReachesNativeVisitor(t *testing.T) {
+ tester := require.New(t)
+
+ fileSet := token.NewFileSet()
+ file, err := parser.ParseFile(fileSet, indexModeSourceReadFile, nil, parser.ParseComments)
+ tester.NoError(err)
+
+ imports := importPathsByQualifier(t, file)
+ visitorQualifiers := qualifiersForImport(imports, nativeVisitorPackage)
+ tester.NotEmpty(visitorQualifiers, "%s must import %s", indexModeSourceReadFile, nativeVisitorPackage)
+
+ reachesVisitor := false
+ ast.Inspect(file, func(node ast.Node) bool {
+ selector, isSelector := node.(*ast.SelectorExpr)
+ if !isSelector {
+ return true
+ }
+ tester.NotEqual(retiredReaderEntryPoint, selector.Sel.Name,
+ "%s must not open a retired index reader directly at %s",
+ indexModeSourceReadFile, fileSet.Position(selector.Pos()))
+ qualifier, isIdent := selector.X.(*ast.Ident)
+ if !isIdent || selector.Sel.Name != nativeVisitorFunc {
+ return true
+ }
+ if _, viaVisitor := visitorQualifiers[qualifier.Name]; viaVisitor {
+ reachesVisitor = true
+ }
+ return true
+ })
+ tester.True(reachesVisitor, "%s must read its source documents through %s.%s",
+ indexModeSourceReadFile, nativeVisitorPackage, nativeVisitorFunc)
+}
+
+// importPathsByQualifier returns the imported package paths of one parsed file,
+// keyed by the local name each is referred to by.
+func importPathsByQualifier(t *testing.T, file *ast.File) map[string]string {
+ t.Helper()
+ imports := map[string]string{}
+ for _, spec := range file.Imports {
+ importPath, err := strconv.Unquote(spec.Path.Value)
+ require.NoError(t, err)
+ name := filepath.Base(importPath)
+ if spec.Name != nil {
+ name = spec.Name.Name
+ }
+ imports[name] = importPath
+ }
+ return imports
+}
+
+// qualifiersForImport returns the local names one import path is referred to by.
+func qualifiersForImport(imports map[string]string, wanted string) map[string]struct{} {
+ names := map[string]struct{}{}
+ for name, importPath := range imports {
+ if importPath == wanted {
+ names[name] = struct{}{}
+ }
+ }
+ return names
+}
+
+// nativeSourceSeries is one series a legacy-created source segment index holds.
+// The label is issue #14010's name for the document; a series identifier is a
+// hash of the marshaled entity buffer, so the entity value, not the label, is
+// what the fixture chooses.
+type nativeSourceSeries struct {
+ label string
+ entityValue string
+ tagValue string
+ timestamp int64
+ version int64
+ deleted bool
+}
+
+// indexModeNativeSourceSeries is the content issue #14010 declares for the
+// migration source: two live series and one deleted series that must not reach
+// the destination.
+func indexModeNativeSourceSeries(t *testing.T) []nativeSourceSeries {
+ t.Helper()
+ base := day20260621Nanos(t) + int64(time.Hour)
+ return []nativeSourceSeries{
+ {label: "101", entityValue: "series-101", tagValue: "blue", timestamp: base, version: 2},
+ {label: "202", entityValue: "series-202", tagValue: "red", timestamp: base, version: 1},
+ {label: "303", entityValue: "series-303", tagValue: "gray", timestamp: base, version: 3, deleted: true},
+ }
+}
+
+// seedNativeSourceSidx writes the declared series into "<root>/<segName>/sidx"
+// with the compatibility writer, then deletes the ones declared deleted, so the
+// source is a directory a released BanyanDB produced rather than one this test
+// hand-assembled.
+func seedNativeSourceSidx(t *testing.T, root, segName string, seriesSet []nativeSourceSeries) string {
+ t.Helper()
+ sidxDir := filepath.Join(root, segName, directCopySidxDirName)
+ require.NoError(t, os.MkdirAll(sidxDir, storage.DirPerm))
+
+ store, err := inverted.NewStore(inverted.StoreOpts{Path: sidxDir, BatchWaitSec: 0})
+ require.NoError(t, err)
+ documents := make(index.Documents, 0, len(seriesSet))
+ var deletedIdentities [][]byte
+ for _, declared := range seriesSet {
+ identity := nativeSourceIdentity(t, declared)
+ tag := index.NewBytesField(index.FieldKey{TagName: "properties"}, convert.StringToBytes(declared.tagValue))
+ tag.Store = true
+ tag.Index = false
+ subject := index.NewStringField(index.FieldKey{TagName: index.IndexModeName}, "svc")
+ subject.Index = true
+ subject.NoSort = true
+ entityTag := index.NewBytesField(index.FieldKey{TagName: index.IndexModeEntityTagPrefix + "id"},
+ convert.StringToBytes(declared.entityValue))
+ entityTag.Index = true
+ entityTag.NoSort = true
+ documents = append(documents, index.Document{
+ Fields: []index.Field{tag, subject, entityTag},
+ EntityValues: identity,
+ Timestamp: declared.timestamp,
+ DocID: convert.Hash(identity),
+ Version: declared.version,
+ })
+ if declared.deleted {
+ deletedIdentities = append(deletedIdentities, identity)
+ }
+ }
+ require.NoError(t, store.UpdateSeriesBatch(index.Batch{Documents: documents}))
+ if len(deletedIdentities) > 0 {
+ require.NoError(t, store.Delete(deletedIdentities))
+ }
+ require.NoError(t, store.Close())
+ return sidxDir
+}
+
+// nativeSourceIdentity returns the marshaled entity buffer the store records as
+// one declared series' identity.
+func nativeSourceIdentity(t *testing.T, declared nativeSourceSeries) []byte {
+ t.Helper()
+ series := &pbv1.Series{Subject: "svc", EntityValues: []*modelv1.TagValue{strTagValue(declared.entityValue)}}
+ require.NoError(t, series.Marshal())
+ return append([]byte(nil), series.Buffer...)
+}
+
+// observedSeries is what the independent destination oracle reports for one
+// published series: its timestamp, its version and its stored tag values. It is
+// read back through the retained series-store query path, which this milestone
+// does not touch, so it cannot agree with a faulty source read by construction.
+type observedSeries struct {
+ tagValues []string
+ timestamp int64
+ version int64
+}
+
+// openPublishedDestination opens a published destination sidx through the
+// retained series-store surface, which this milestone does not touch, so what
+// it reports cannot agree with a faulty source read by construction.
+func openPublishedDestination(t *testing.T, sidxDir string) index.SeriesStore {
+ t.Helper()
+ store, err := inverted.NewStore(inverted.StoreOpts{Path: sidxDir, BatchWaitSec: 0})
+ require.NoError(t, err)
+ return store
+}
+
+// publishedSeriesFor looks one identity up in a published destination and
+// reports what it holds, or reports that the destination holds no live document
+// under that identity. The lookup is an exact one so that the answer for one
+// series can never be contaminated by another's.
+func publishedSeriesFor(t *testing.T, store index.SeriesStore, identity []byte) (observedSeries, bool) {
+ t.Helper()
+ query, err := store.BuildQuery([]index.SeriesMatcher{
+ {Type: index.SeriesMatcherTypeExact, Match: identity},
+ }, nil, nil)
+ require.NoError(t, err)
+ documents, err := store.Search(context.Background(),
+ []index.FieldKey{{TagName: nativeSourceTagName}}, query, publishedSeriesLimit)
+ require.NoError(t, err)
+ if len(documents) == 0 {
+ return observedSeries{}, false
+ }
+ require.Len(t, documents, 1, "identity %q must resolve to at most one live destination document", identity)
+ var values []string
+ if value, stored := documents[0].Fields[nativeSourceTagName]; stored && value != nil {
+ values = append(values, string(value))
+ }
+ sort.Strings(values)
+ return observedSeries{
+ timestamp: documents[0].Timestamp,
+ version: documents[0].Version,
+ tagValues: values,
+ }, true
+}
+
+// truncateSourceSegment cuts a source segment file short of the footer its
+// on-disk grammar requires, which is what a segment published by a process that
+// died mid-write looks like. The directory still holds a committed manifest
+// naming that segment, so the damage is only visible to a reader that validates
+// the segment it is about to read.
+func truncateSourceSegment(t *testing.T, sidxDir string) {
+ t.Helper()
+ segments, err := filepath.Glob(filepath.Join(sidxDir, "*.seg"))
+ require.NoError(t, err)
+ require.NotEmpty(t, segments, "the seeded source must hold a committed segment")
+ sort.Strings(segments)
+ for _, segmentPath := range segments {
+ require.NoError(t, os.Truncate(segmentPath, iceFooterLength-8))
+ }
+}
+
+// TestE2EIndexModeCopyNativeSource walks the production situation this
+// milestone exists for: an operator migrates an index-mode measure group whose
+// segment indexes were written by a released BanyanDB. Index-mode copy reads
+// those sources through the native document visitor and publishes the rebuilt
+// documents through its retained destination writer, which this milestone does
+// not change.
+//
+// Two source roots feed the same target segment, which is what a group
+// migrated from replicated nodes looks like and what forces the copy down its
+// rebuild path: a source that exclusively owns its target segment is byte
+// copied whole and never reads a document at all.
+//
+// Requirements proved here:
+//
+// R2 -- index-mode copy rebuilds the two live series through the native source
+// visitor, preserving each one's declared timestamp, version, stored tag
+// value and identity, while its destination stays written by the
+// retained writer. The deleted series never reaches the destination.
+//
+// R4 -- a source whose committed segment is damaged fails the copy with the
+// native typed corruption error and publishes no destination, so a
+// migration aborts on unreadable input instead of silently writing a
+// partial target group. The byte-level stored-chunk failures R4 also
+// names are proved against the visitor itself in
+// pkg/index/inverted, where no retired reader stands in the way.
+func TestE2EIndexModeCopyNativeSource(t *testing.T) {
+ tester := require.New(t)
+ declared := indexModeNativeSourceSeries(t)
+ interval := storage.IntervalRule{Unit: storage.DAY, Num: 1}
+ const sourceSegName = "seg-20260621"
+
+ firstRoot := t.TempDir()
+ secondRoot := t.TempDir()
+ targetRoot := filepath.Join(t.TempDir(), "target")
+ seedNativeSourceSidx(t, firstRoot, sourceSegName, declared)
+ seedNativeSourceSidx(t, secondRoot, sourceSegName, declared[:1])
+
+ result, err := copyIndexModeGroup(context.Background(), migration.EntryGroupInput{
+ EntryTag: "[entry 1/1]",
+ Group: "sw_metadata",
+ TargetGroupRoot: targetRoot,
+ SrcRoots: []string{firstRoot, secondRoot},
+ Interval: interval,
+ }, map[uint32]indexRuleInfo{}, svcSchemas())
+ tester.NoError(err)
+ tester.Equal(int64(2), result.Rows,
+ "the two live series survive; the repeat of the first collapses onto it and the deleted one is never read")
+
+ targetSidx := filepath.Join(targetRoot, sourceSegName, directCopySidxDirName)
+ publishedCount, countErr := inverted.ReadOnlyDocCount(targetSidx)
+ tester.NoError(countErr)
+ tester.Equal(int64(2), publishedCount, "the destination must hold exactly the two live series")
+
+ destination := openPublishedDestination(t, targetSidx)
+ defer func() {
+ tester.NoError(destination.Close())
+ }()
+ for _, declaredSeries := range declared {
+ identity := nativeSourceIdentity(t, declaredSeries)
+ observed, found := publishedSeriesFor(t, destination, identity)
+ if declaredSeries.deleted {
+ tester.False(found, "the deleted series labeled %s must not reach the destination", declaredSeries.label)
+ continue
+ }
+ tester.True(found, "the series labeled %s must reach the destination", declaredSeries.label)
+ tester.Equal(declaredSeries.timestamp, observed.timestamp,
+ "series %s must keep its declared timestamp", declaredSeries.label)
+ tester.Equal(declaredSeries.version, observed.version,
+ "series %s must keep its declared version", declaredSeries.label)
+ tester.Equal([]string{declaredSeries.tagValue}, observed.tagValues,
+ "series %s must keep its declared stored tag value", declaredSeries.label)
+ }
+
+ corruptRoot := t.TempDir()
+ corruptTarget := filepath.Join(t.TempDir(), "corrupt-target")
+ truncateSourceSegment(t, seedNativeSourceSidx(t, corruptRoot, sourceSegName, declared))
+
+ _, corruptErr := copyIndexModeGroup(context.Background(), migration.EntryGroupInput{
+ EntryTag: "[entry 1/1]",
+ Group: "sw_metadata",
+ TargetGroupRoot: corruptTarget,
+ SrcRoots: []string{corruptRoot},
+ Interval: interval,
+ }, map[uint32]indexRuleInfo{}, svcSchemas())
+ tester.ErrorIs(corruptErr, inverted.ErrCorruptIndex,
+ "a damaged source must be reported as a corrupt index, not as an opaque read failure")
+ tester.NoDirExists(filepath.Join(corruptTarget, sourceSegName, directCopySidxDirName),
+ "a copy that failed on a damaged source must publish no destination sidx")
+}
diff --git a/banyand/measure/migration_verify.go b/banyand/measure/migration_verify.go
index 57969e2da..9e7aa0aa6 100644
--- a/banyand/measure/migration_verify.go
+++ b/banyand/measure/migration_verify.go
@@ -18,6 +18,7 @@
package measure
import (
+ "context"
"encoding/json"
"errors"
"fmt"
@@ -28,10 +29,13 @@ import (
"strings"
"time"
- "github.com/blugelabs/bluge"
+ compat "github.com/blugelabs/bluge"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
)
// SegmentReport summarizes one `seg-*` directory under a measure
@@ -189,7 +193,7 @@ func partRowCount(partID uint64, name, shardDir string, fileSystem fs.FileSystem
// total document count. Used to spot-check the broadcast union sidx
// that `migration copy` writes into every aligned target segment.
func CountBlugeDocs(path string) (uint64, error) {
- reader, err := bluge.OpenReader(bluge.DefaultConfig(path))
+ reader, err := compat.OpenReader(compat.DefaultConfig(path))
if err != nil {
return 0, err
}
@@ -203,6 +207,121 @@ func CountBlugeDocs(path string) (uint64, error) {
return reader.Count()
}
+type storedFieldDocument interface {
+ VisitStoredFields(visit func(name string, value []byte) bool) error
+}
+
+type retainedStoredDocument struct {
+ visitStoredFields func(func(name string, value []byte) bool) error
+}
+
+func (d retainedStoredDocument) VisitStoredFields(visit func(name string, value []byte) bool) error {
+ return d.visitStoredFields(visit)
+}
+
+func readIndexModeDocs(ctx context.Context, sidxDir string, ruleByID map[uint32]indexRuleInfo,
+ schemasBySubject map[string]*measureSchemaInfo,
+) ([]index.Document, error) {
+ reader, openErr := compat.OpenReader(compat.DefaultConfig(sidxDir))
+ if openErr != nil {
+ if strings.Contains(openErr.Error(), "unable to find a usable snapshot") {
+ return nil, nil
+ }
+ return nil, fmt.Errorf("open sidx reader %s: %w", sidxDir, openErr)
+ }
+ defer func() { _ = reader.Close() }()
+ matches, searchErr := reader.Search(ctx, compat.NewAllMatches(compat.NewMatchAllQuery()))
+ if searchErr != nil {
+ return nil, fmt.Errorf("search sidx %s: %w", sidxDir, searchErr)
+ }
+ var documents []index.Document
+ var timeless []uint64
+ tagNames := collectTagNames(schemasBySubject)
+ missingRules := map[uint32]int{}
+ for {
+ match, nextErr := matches.Next()
+ if nextErr != nil {
+ return nil, fmt.Errorf("iterate sidx %s: %w", sidxDir, nextErr)
+ }
+ if match == nil {
+ break
+ }
+ document, rebuildErr := rebuildOneDoc(retainedStoredDocument{visitStoredFields: func(visit func(name string, value []byte) bool) error {
+ return match.VisitStoredFields(func(name string, value []byte) bool {
+ return visit(name, value)
+ })
+ }}, ruleByID, schemasBySubject, tagNames, missingRules)
+ if rebuildErr != nil {
+ return nil, fmt.Errorf("rebuild doc in %s: %w", sidxDir, rebuildErr)
+ }
+ if document.Timestamp == 0 {
+ timeless = append(timeless, document.DocID)
+ continue
+ }
+ documents = append(documents, document)
+ }
+ if len(timeless) > 0 {
+ sample := timeless
+ if len(sample) > 10 {
+ sample = sample[:10]
+ }
+ return nil, fmt.Errorf("sidx %s: %d index-mode doc(s) have a missing or undecodable _timestamp (ts==0), "+
+ "which never occurs for valid data; the source is corrupt or unexpected — sample series IDs (first %d) %v",
+ sidxDir, len(timeless), len(sample), sample)
+ }
+ warnMissingRules(sidxDir, missingRules)
+ return documents, nil
+}
+
+func rebuildOneDoc(source storedFieldDocument, ruleByID map[uint32]indexRuleInfo,
+ schemasBySubject map[string]*measureSchemaInfo, tagNames map[string]struct{}, missingRules map[uint32]int,
+) (index.Document, error) {
+ var entityValues []byte
+ var timestamp, version int64
+ var fields []index.Field
+ visitErr := source.VisitStoredFields(func(name string, value []byte) bool {
+ switch name {
+ case imDocIDField:
+ entityValues = append([]byte(nil), value...)
+ case imTimestampField:
+ if decodedTime, decodeErr := compat.DecodeDateTime(value); decodeErr == nil {
+ timestamp = decodedTime.UnixNano()
+ }
+ case imVersionField:
+ version = convert.BytesToInt64(value)
+ default:
+ key, indexed, missingRuleID := classifyStoredField(name, tagNames, ruleByID)
+ field := index.NewBytesField(key, value)
+ field.Store = true
+ field.Index = indexed
+ if rule, found := ruleByID[key.IndexRuleID]; indexed && found {
+ field.NoSort = rule.NoSort
+ field.Key.Analyzer = rule.Analyzer
+ }
+ if missingRuleID != 0 && missingRules != nil {
+ missingRules[missingRuleID]++
+ }
+ fields = append(fields, field)
+ }
+ return true
+ })
+ if visitErr != nil {
+ return index.Document{}, fmt.Errorf("visit stored fields: %w", visitErr)
+ }
+ var series pbv1.Series
+ if unmarshalErr := series.Unmarshal(entityValues); unmarshalErr != nil {
+ return index.Document{}, fmt.Errorf("unmarshal series from _id: %w", unmarshalErr)
+ }
+ fields = appendRegeneratedEntityFields(fields, &series, schemasBySubject[series.Subject])
+ return index.Document{
+ Fields: fields,
+ EntityValues: entityValues,
+ Timestamp: timestamp,
+ DocID: uint64(series.ID),
+ Version: version,
+ }, nil
+}
+
// EnumerateGroupTarget walks <groupRoot>/seg-* and reports per-segment
// row total + sidx doc count + whether the segment's start time aligns
// to the supplied IntervalRule's standard grid. A missing groupRoot
diff --git a/pkg/index/inverted/internal/nativeice/nativeice.go b/pkg/index/inverted/internal/nativeice/nativeice.go
index e681c6c86..b4b81fd6b 100644
--- a/pkg/index/inverted/internal/nativeice/nativeice.go
+++ b/pkg/index/inverted/internal/nativeice/nativeice.go
@@ -28,6 +28,7 @@
package nativeice
import (
+ "context"
"encoding/binary"
"errors"
"fmt"
@@ -39,6 +40,7 @@ import (
"strconv"
roaringpkg "github.com/RoaringBitmap/roaring"
+ "github.com/klauspost/compress/s2"
)
const (
@@ -50,6 +52,18 @@ const (
maxDirectoryEntries = 1 << 16
directoryReadSize = maxDirectoryEntries + 1
maxOpenAttempts = 2
+
+ storedDocumentsPerChunk = 128
+ maxStoredChunkCount = 1 << 20
+ maxStoredChunkTableSize = 16 << 20
+ maxStoredCompressedChunkSize = 16 << 20
+ maxStoredDecodedChunkSize = 64 << 20
+ maxStoredFieldNameLength = 64 << 10
+ maxStoredFieldsPerDocument = 1 << 20
+ segmentFooterPayloadLength = segmentFooterLength - 4
+ storedChunkTableFooterLength = 8
+ storedDocumentOffsetByteWidth = 8
+ fieldsIndexAddressByteWidth = 8
)
// ErrCorrupt is the sentinel that every structural rejection wraps: a footer,
@@ -70,9 +84,52 @@ var errManifestTooLarge = errors.New("nativeice: snapshot exceeds read limit")
// directory. The generation is chosen at Open and fixed for the Reader's
// lifetime, so generations committed afterwards stay invisible to it.
type Reader struct {
+ segments []segmentRecord
visibleDocCount int64
}
+// StoredDocument is one live document of the pinned generation, borrowed for
+// the duration of a single walk callback.
+type StoredDocument interface {
+ // VisitStoredFields calls visit once for every stored value the document
+ // records, passing the field's name and its raw value bytes. A field the
+ // document records more than once is visited once per recorded value, in
+ // the order the document records them. Visiting stops early when visit
+ // returns false.
+ //
+ // The name and value handed to visit are borrowed from the reader's decode
+ // buffers and stay valid only until visit returns; a caller that keeps
+ // either beyond that copies it.
+ VisitStoredFields(visit func(name string, value []byte) bool) error
+}
+
+// VisitLiveDocuments streams the pinned generation's live documents to visit,
+// one at a time, in ascending segment and local document order. Documents the
+// pinned snapshot's deletion masks cover are skipped, so a deleted document is
+// never handed to visit.
+//
+// The StoredDocument handed to visit is borrowed: it, and every name and value
+// it yields, stay valid only until visit returns. At most one document plus the
+// reader's configured decode buffers are resident at a time, so the walk's
+// memory does not grow with the generation's size.
+//
+// The walk stops and returns ctx.Err() when ctx is canceled between two
+// documents or two stored chunks, and stops and returns visit's error when
+// visit fails. A stored section whose chunk table, offsets, lengths, varints or
+// field identifiers violate the ICE v3 grammar, or that would require decoding
+// past a configured bound, stops the walk with an error wrapping ErrCorrupt.
+func (r *Reader) VisitLiveDocuments(ctx context.Context, visit func(StoredDocument) error) error {
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ return ctxErr
+ }
+ for segmentIndex := range r.segments {
+ if visitErr := walkStoredSegment(ctx, r.segments[segmentIndex], visit); visitErr != nil {
+ return visitErr
+ }
+ }
+ return nil
+}
+
// Open selects the newest committed generation in the index directory at path,
// validates its snapshot manifest and every segment that manifest references
// against the grammar and the reader's configured bounds, and returns a Reader
@@ -112,13 +169,13 @@ func openWithSnapshots(path string, listSnapshots snapshotLister) (*Reader, erro
lastCandidatePath = snapshotPath
continue
}
- visibleDocCount, parseErr := parseSnapshot(segmentPaths, manifest)
+ visibleDocCount, segments, parseErr := parseSnapshotSegments(segmentPaths, manifest)
if parseErr != nil {
lastCandidateErr = parseErr
lastCandidatePath = snapshotPath
continue
}
- return &Reader{visibleDocCount: visibleDocCount}, nil
+ return &Reader{visibleDocCount: visibleDocCount, segments: segments}, nil
}
}
if lastCandidateErr != nil {
@@ -209,55 +266,57 @@ func readDirectoryEntries(path string) ([]os.DirEntry, error) {
return entries, nil
}
-func parseSnapshot(segmentPaths map[uint64]string, payload []byte) (int64, error) {
+func parseSnapshotSegments(segmentPaths map[uint64]string, payload []byte) (int64, []segmentRecord, error) {
if len(payload) < 4 {
- return 0, corruptError("snapshot is shorter than its reserved CRC32", nil)
+ return 0, nil, corruptError("snapshot is shorter than its reserved CRC32", nil)
}
decoder := byteDecoder{payload: payload[:len(payload)-4]}
version, versionErr := decoder.uvarint()
if versionErr != nil {
- return 0, versionErr
+ return 0, nil, versionErr
}
if version != snapshotVersion {
- return 0, corruptError("unsupported snapshot version %d", version)
+ return 0, nil, corruptError("unsupported snapshot version %d", version)
}
segmentCount, countErr := decoder.uvarint()
if countErr != nil {
- return 0, countErr
+ return 0, nil, countErr
}
if segmentCount > uint64(len(decoder.payload)) {
- return 0, corruptError("snapshot segment count %d exceeds remaining bytes", segmentCount)
+ return 0, nil, corruptError("snapshot segment count %d exceeds remaining bytes", segmentCount)
}
var visibleDocCount int64
+ segments := make([]segmentRecord, 0, int(segmentCount))
for segmentIndex := uint64(0); segmentIndex < segmentCount; segmentIndex++ {
- segmentRecord, recordErr := decoder.segmentRecord(segmentPaths)
+ record, recordErr := decoder.segmentRecord(segmentPaths)
if recordErr != nil {
- return 0, recordErr
+ return 0, nil, recordErr
}
- segmentDocCount, segmentErr := validateSegment(segmentRecord)
+ segmentDocCount, segmentErr := validateSegment(record)
if segmentErr != nil {
- return 0, segmentErr
+ return 0, nil, segmentErr
}
- if segmentDocCount != segmentRecord.documentCount {
- return 0, corruptError("segment %d document count differs from snapshot", segmentRecord.id)
+ if segmentDocCount != record.documentCount {
+ return 0, nil, corruptError("segment %d document count differs from snapshot", record.id)
}
- deletedCount, deletionErr := deletionCount(segmentRecord.deletionBitmap, segmentRecord.documentCount)
+ deletedCount, deletionErr := deletionCount(record.deletionBitmap, record.documentCount)
if deletionErr != nil {
- return 0, deletionErr
+ return 0, nil, deletionErr
}
- if segmentRecord.documentCount > uint64(math.MaxInt64) || deletedCount > segmentRecord.documentCount {
- return 0, corruptError("invalid document count for segment %d", segmentRecord.id)
+ if record.documentCount > uint64(math.MaxInt64) || deletedCount > record.documentCount {
+ return 0, nil, corruptError("invalid document count for segment %d", record.id)
}
- segmentVisibleCount := int64(segmentRecord.documentCount - deletedCount)
+ segmentVisibleCount := int64(record.documentCount - deletedCount)
if segmentVisibleCount > math.MaxInt64-visibleDocCount {
- return 0, corruptError("visible document count overflows int64", nil)
+ return 0, nil, corruptError("visible document count overflows int64", nil)
}
visibleDocCount += segmentVisibleCount
+ segments = append(segments, record)
}
if decoder.remaining() != 0 {
- return 0, corruptError("snapshot has trailing bytes before its reserved CRC32", nil)
+ return 0, nil, corruptError("snapshot has trailing bytes before its reserved CRC32", nil)
}
- return visibleDocCount, nil
+ return visibleDocCount, segments, nil
}
type segmentRecord struct {
@@ -369,6 +428,18 @@ func (d *byteDecoder) segmentRecord(segmentPaths map[uint64]string) (segmentReco
return segmentRecord{deletionBitmap: deletionBitmap, path: segmentPath, documentCount: documentCount, id: id, timeMin: timeMin, timeMax: timeMax}, nil
}
+type segmentFooter struct {
+ documentCount uint64
+ storedIndexOffset uint64
+ fieldsIndexOffset uint64
+ docValueOffset uint64
+ chunkMode uint32
+ timeMin uint64
+ timeMax uint64
+ footerOffset uint64
+ fieldsIndexEntries uint64
+}
+
func validateSegment(record segmentRecord) (uint64, error) {
file, openErr := os.Open(record.path)
if errors.Is(openErr, os.ErrNotExist) {
@@ -387,48 +458,483 @@ func validateSegment(record segmentRecord) (uint64, error) {
if !info.Mode().IsRegular() || info.Size() < segmentFooterLength {
return 0, corruptError("segment %q is shorter than its footer", record.path)
}
- footerOffset := info.Size() - segmentFooterLength
- var footer [segmentFooterLength]byte
- if _, readErr := file.ReadAt(footer[:], footerOffset); readErr != nil {
- return 0, corruptError("read footer from segment %q", record.path, readErr)
- }
- documentCount := binary.BigEndian.Uint64(footer[0:8])
- storedIndex := binary.BigEndian.Uint64(footer[8:16])
- fieldsIndex := binary.BigEndian.Uint64(footer[16:24])
- docValues := binary.BigEndian.Uint64(footer[24:32])
- chunkMode := binary.BigEndian.Uint32(footer[32:36])
- timeMin := binary.BigEndian.Uint64(footer[36:44])
- timeMax := binary.BigEndian.Uint64(footer[44:52])
- version := binary.BigEndian.Uint32(footer[52:56])
- if version != segmentVersion {
- return 0, corruptError("unsupported segment version %d", version)
- }
- if chunkMode == 0 || storedIndex > docValues || docValues > fieldsIndex || fieldsIndex > uint64(footerOffset) {
- return 0, corruptError("segment %q has invalid section roots", record.path)
- }
- if documentCount > uint64(math.MaxInt64) || documentCount > (docValues-storedIndex)/8 {
- return 0, corruptError("segment %q has invalid document count", record.path)
+ footer, footerErr := readSegmentFooter(file, uint64(info.Size()), record.path)
+ if footerErr != nil {
+ return 0, footerErr
}
- if (uint64(footerOffset)-fieldsIndex)%8 != 0 {
- return 0, corruptError("segment %q has a misaligned fields index", record.path)
- }
- if (uint64(footerOffset)-fieldsIndex)/8 > maxFieldsIndexCount {
- return 0, corruptError("segment %q has too many fields index entries", record.path)
- }
- for fieldIndexOffset := fieldsIndex; fieldIndexOffset < uint64(footerOffset); fieldIndexOffset += 8 {
- var fieldRecord [8]byte
+ for fieldIndexOffset := footer.fieldsIndexOffset; fieldIndexOffset < footer.footerOffset; fieldIndexOffset += fieldsIndexAddressByteWidth {
+ var fieldRecord [fieldsIndexAddressByteWidth]byte
if _, readErr := file.ReadAt(fieldRecord[:], int64(fieldIndexOffset)); readErr != nil {
return 0, corruptError("read fields index from segment %q", record.path, readErr)
}
fieldRecordOffset := binary.BigEndian.Uint64(fieldRecord[:])
- if fieldRecordOffset >= fieldsIndex {
+ if fieldRecordOffset >= footer.fieldsIndexOffset {
return 0, corruptError("segment %q has a field record outside its section", record.path)
}
}
- if timeMin != record.timeMin || timeMax != record.timeMax {
+ if footer.timeMin != record.timeMin || footer.timeMax != record.timeMax {
return 0, corruptError("segment %d time bounds differ from snapshot", record.id)
}
- return documentCount, nil
+ return footer.documentCount, nil
+}
+
+func readSegmentFooter(file *os.File, size uint64, path string) (segmentFooter, error) {
+ if size < segmentFooterLength {
+ return segmentFooter{}, corruptError("segment %q is shorter than its footer", path)
+ }
+ footerOffset := size - segmentFooterLength
+ var payload [segmentFooterPayloadLength]byte
+ if _, readErr := file.ReadAt(payload[:], int64(footerOffset)); readErr != nil {
+ return segmentFooter{}, corruptError("read footer from segment %q", path, readErr)
+ }
+ footer := segmentFooter{
+ documentCount: binary.BigEndian.Uint64(payload[0:8]),
+ storedIndexOffset: binary.BigEndian.Uint64(payload[8:16]),
+ fieldsIndexOffset: binary.BigEndian.Uint64(payload[16:24]),
+ docValueOffset: binary.BigEndian.Uint64(payload[24:32]),
+ chunkMode: binary.BigEndian.Uint32(payload[32:36]),
+ timeMin: binary.BigEndian.Uint64(payload[36:44]),
+ timeMax: binary.BigEndian.Uint64(payload[44:52]),
+ footerOffset: footerOffset,
+ }
+ if binary.BigEndian.Uint32(payload[52:56]) != segmentVersion {
+ return segmentFooter{}, corruptError("unsupported segment version %d", binary.BigEndian.Uint32(payload[52:56]))
+ }
+ if footer.chunkMode == 0 || footer.storedIndexOffset > footer.docValueOffset ||
+ footer.docValueOffset > footer.fieldsIndexOffset || footer.fieldsIndexOffset > footer.footerOffset {
+ return segmentFooter{}, corruptError("segment %q has invalid section roots", path)
+ }
+ if footer.documentCount > uint64(math.MaxInt64) ||
+ footer.documentCount > (footer.docValueOffset-footer.storedIndexOffset)/storedDocumentOffsetByteWidth {
+ return segmentFooter{}, corruptError("segment %q has invalid document count", path)
+ }
+ if (footer.footerOffset-footer.fieldsIndexOffset)%fieldsIndexAddressByteWidth != 0 {
+ return segmentFooter{}, corruptError("segment %q has a misaligned fields index", path)
+ }
+ footer.fieldsIndexEntries = (footer.footerOffset - footer.fieldsIndexOffset) / fieldsIndexAddressByteWidth
+ if footer.fieldsIndexEntries > maxFieldsIndexCount {
+ return segmentFooter{}, corruptError("segment %q has too many fields index entries", path)
+ }
+ return footer, nil
+}
+
+type storedSegmentReader struct {
+ file *os.File
+ path string
+ chunkOffsets []uint64
+ compressedBuffer []byte
+ decodedBuffer []byte
+ fieldNames []string
+ fieldNameBuffer []byte
+ footer segmentFooter
+ size uint64
+ storedDataEnd uint64
+}
+
+type storedField struct {
+ name string
+ value []byte
+}
+
+type storedDocument struct {
+ fields []storedField
+}
+
+func (d storedDocument) VisitStoredFields(visit func(name string, value []byte) bool) error {
+ for _, field := range d.fields {
+ if !visit(field.name, field.value) {
+ break
+ }
+ }
+ return nil
+}
+
+func walkStoredSegment(ctx context.Context, record segmentRecord, visit func(StoredDocument) error) error {
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ return ctxErr
+ }
+ file, openErr := os.Open(record.path)
+ if errors.Is(openErr, os.ErrNotExist) {
+ return corruptError("open missing segment %d", record.id)
+ }
+ if openErr != nil {
+ return corruptError("open segment %q", record.path, openErr)
+ }
+ defer func() {
+ _ = file.Close()
+ }()
+ info, statErr := file.Stat()
+ if statErr != nil {
+ return corruptError("stat segment %q", record.path, statErr)
+ }
+ if !info.Mode().IsRegular() || info.Size() < segmentFooterLength {
+ return corruptError("segment %q is shorter than its footer", record.path)
+ }
+ storedReader, readerErr := newStoredSegmentReader(file, uint64(info.Size()), record)
+ if readerErr != nil {
+ return readerErr
+ }
+ deleted, deletionErr := deletedDocuments(record)
+ if deletionErr != nil {
+ return deletionErr
+ }
+ return storedReader.visit(ctx, deleted, visit)
+}
+
+func newStoredSegmentReader(file *os.File, size uint64, record segmentRecord) (*storedSegmentReader, error) {
+ footer, footerErr := readSegmentFooter(file, size, record.path)
+ if footerErr != nil {
+ return nil, footerErr
+ }
+ if footer.documentCount != record.documentCount {
+ return nil, corruptError("segment %d document count differs from snapshot", record.id)
+ }
+ if footer.timeMin != record.timeMin || footer.timeMax != record.timeMax {
+ return nil, corruptError("segment %d time bounds differ from snapshot", record.id)
+ }
+ storedReader := &storedSegmentReader{file: file, path: record.path, size: size, footer: footer}
+ if chunkErr := storedReader.loadChunkOffsets(); chunkErr != nil {
+ return nil, chunkErr
+ }
+ if fieldsErr := storedReader.loadFieldNames(); fieldsErr != nil {
+ return nil, fieldsErr
+ }
+ return storedReader, nil
+}
+
+func (s *storedSegmentReader) visit(ctx context.Context, deleted *roaringpkg.Bitmap, visit func(StoredDocument) error) error {
+ if s.footer.documentCount == 0 {
+ return nil
+ }
+ chunkCount := (s.footer.documentCount + storedDocumentsPerChunk - 1) / storedDocumentsPerChunk
+ for chunkIndex := uint64(0); chunkIndex < chunkCount; chunkIndex++ {
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ return ctxErr
+ }
+ chunk, chunkErr := s.loadChunk(chunkIndex)
+ if chunkErr != nil {
+ return chunkErr
+ }
+ firstDocument := chunkIndex * storedDocumentsPerChunk
+ lastDocument := firstDocument + storedDocumentsPerChunk
+ if lastDocument > s.footer.documentCount {
+ lastDocument = s.footer.documentCount
+ }
+ for documentNumber := firstDocument; documentNumber < lastDocument; documentNumber++ {
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ return ctxErr
+ }
+ if documentNumber <= math.MaxUint32 && deleted.Contains(uint32(documentNumber)) {
+ continue
+ }
+ document, documentErr := s.decodeDocument(documentNumber, chunk)
+ if documentErr != nil {
+ return documentErr
+ }
+ if visitErr := visit(document); visitErr != nil {
+ return visitErr
+ }
+ }
+ }
+ return nil
+}
+
+func (s *storedSegmentReader) loadChunkOffsets() error {
+ if s.footer.storedIndexOffset < storedChunkTableFooterLength {
+ return corruptError("segment %q has a truncated stored chunk table", s.path)
+ }
+ var tableFooter [storedChunkTableFooterLength]byte
+ if readErr := s.readInto(s.footer.storedIndexOffset-storedChunkTableFooterLength, tableFooter[:]); readErr != nil {
+ return readErr
+ }
+ offsetLength := uint64(binary.BigEndian.Uint32(tableFooter[0:4]))
+ chunkCount := uint64(binary.BigEndian.Uint32(tableFooter[4:8]))
+ if offsetLength > maxStoredChunkTableSize || chunkCount == 0 || chunkCount > maxStoredChunkCount || chunkCount > offsetLength {
+ return corruptError("segment %q has an invalid stored chunk table", s.path)
+ }
+ if offsetLength > s.footer.storedIndexOffset-storedChunkTableFooterLength {
+ return corruptError("segment %q has a stored chunk table outside its section", s.path)
+ }
+ tableStart := s.footer.storedIndexOffset - storedChunkTableFooterLength - offsetLength
+ table, tableErr := s.readBytes(tableStart, offsetLength)
+ if tableErr != nil {
+ return tableErr
+ }
+ decoder := byteDecoder{payload: table}
+ offsets := make([]uint64, int(chunkCount))
+ for chunkIndex := range offsets {
+ offset, offsetErr := decoder.uvarint()
+ if offsetErr != nil {
+ return offsetErr
+ }
+ offsets[chunkIndex] = offset
+ }
+ if decoder.remaining() != 0 {
+ return corruptError("segment %q has trailing bytes in its stored chunk table", s.path)
+ }
+ if offsets[0] != 0 {
+ return corruptError("segment %q has a stored chunk table without a zero origin", s.path)
+ }
+ for offsetIndex := 1; offsetIndex < len(offsets); offsetIndex++ {
+ if offsets[offsetIndex] < offsets[offsetIndex-1] || offsets[offsetIndex] > tableStart {
+ return corruptError("segment %q has invalid stored chunk offsets", s.path)
+ }
+ }
+ dataChunks := (s.footer.documentCount + storedDocumentsPerChunk - 1) / storedDocumentsPerChunk
+ if s.footer.documentCount == 0 {
+ if chunkCount > 2 {
+ return corruptError("segment %q has too many empty stored chunks", s.path)
+ }
+ } else if chunkCount != dataChunks+1 {
+ return corruptError("segment %q has %d stored chunks for %d documents", s.path, chunkCount, s.footer.documentCount)
+ }
+ for chunkIndex := uint64(0); chunkIndex < dataChunks; chunkIndex++ {
+ if offsets[chunkIndex] >= offsets[chunkIndex+1] {
+ return corruptError("segment %q has an empty stored document chunk", s.path)
+ }
+ }
+ s.chunkOffsets = offsets
+ s.storedDataEnd = tableStart
+ return nil
+}
+
+func (s *storedSegmentReader) loadFieldNames() error {
+ fieldNames := make([]string, int(s.footer.fieldsIndexEntries))
+ for fieldID := range fieldNames {
+ fieldName, fieldErr := s.readFieldName(uint64(fieldID))
+ if fieldErr != nil {
+ return fieldErr
+ }
+ fieldNames[fieldID] = fieldName
+ }
+ s.fieldNames = fieldNames
+ return nil
+}
+
+func (s *storedSegmentReader) readFieldName(fieldID uint64) (string, error) {
+ if fieldID >= s.footer.fieldsIndexEntries {
+ return "", corruptError("segment %q has an out-of-range stored field identifier %d", s.path, fieldID)
+ }
+ indexOffset := s.footer.fieldsIndexOffset + fieldID*fieldsIndexAddressByteWidth
+ var addressData [fieldsIndexAddressByteWidth]byte
+ if readErr := s.readInto(indexOffset, addressData[:]); readErr != nil {
+ return "", readErr
+ }
+ offset := binary.BigEndian.Uint64(addressData[:])
+ if offset >= s.footer.fieldsIndexOffset {
+ return "", corruptError("segment %q has a field record outside its section", s.path)
+ }
+ if _, dictErr := s.readUvarint(&offset, s.footer.fieldsIndexOffset); dictErr != nil {
+ return "", dictErr
+ }
+ nameLength, lengthErr := s.readUvarint(&offset, s.footer.fieldsIndexOffset)
+ if lengthErr != nil {
+ return "", lengthErr
+ }
+ if nameLength > maxStoredFieldNameLength || nameLength > s.footer.fieldsIndexOffset-offset {
+ return "", corruptError("segment %q has an invalid stored field name length", s.path)
+ }
+ if cap(s.fieldNameBuffer) < int(nameLength) {
+ s.fieldNameBuffer = make([]byte, int(nameLength))
+ } else {
+ s.fieldNameBuffer = s.fieldNameBuffer[:int(nameLength)]
+ }
+ if readErr := s.readInto(offset, s.fieldNameBuffer); readErr != nil {
+ return "", readErr
+ }
+ offset += nameLength
+ if _, documentCountErr := s.readUvarint(&offset, s.footer.fieldsIndexOffset); documentCountErr != nil {
+ return "", documentCountErr
+ }
+ if _, frequencyErr := s.readUvarint(&offset, s.footer.fieldsIndexOffset); frequencyErr != nil {
+ return "", frequencyErr
+ }
+ return string(s.fieldNameBuffer), nil
+}
+
+func (s *storedSegmentReader) readUvarint(offset *uint64, end uint64) (uint64, error) {
+ if *offset >= end {
+ return 0, corruptError("segment %q has a truncated variable-length integer", s.path)
+ }
+ length := uint64(binary.MaxVarintLen64)
+ if remaining := end - *offset; remaining < length {
+ length = remaining
+ }
+ var encoded [binary.MaxVarintLen64]byte
+ if readErr := s.readInto(*offset, encoded[:int(length)]); readErr != nil {
+ return 0, readErr
+ }
+ value, width := binary.Uvarint(encoded[:int(length)])
+ if width <= 0 {
+ return 0, corruptError("segment %q has an invalid variable-length integer", s.path)
+ }
+ *offset += uint64(width)
+ return value, nil
+}
+
+func (s *storedSegmentReader) loadChunk(chunkIndex uint64) ([]byte, error) {
+ if chunkIndex+1 >= uint64(len(s.chunkOffsets)) {
+ return nil, corruptError("segment %q has no stored chunk %d", s.path, chunkIndex)
+ }
+ start := s.chunkOffsets[chunkIndex]
+ end := s.chunkOffsets[chunkIndex+1]
+ if start >= end || end > s.storedDataEnd {
+ return nil, corruptError("segment %q has invalid bounds for stored chunk %d", s.path, chunkIndex)
+ }
+ compressedLength := end - start
+ if compressedLength > maxStoredCompressedChunkSize {
+ return nil, corruptError("segment %q has an oversized stored chunk", s.path)
+ }
+ if cap(s.compressedBuffer) < int(compressedLength) {
+ s.compressedBuffer = make([]byte, int(compressedLength))
+ } else {
+ s.compressedBuffer = s.compressedBuffer[:int(compressedLength)]
+ }
+ if readErr := s.readInto(start, s.compressedBuffer); readErr != nil {
+ return nil, readErr
+ }
+ decodedLength, lengthErr := storedChunkDecodedLength(s.compressedBuffer)
+ if lengthErr != nil {
+ return nil, corruptError("decode stored chunk length in segment %q: %w", s.path, lengthErr)
+ }
+ if decodedLength < 0 || decodedLength > maxStoredDecodedChunkSize {
+ return nil, corruptError("segment %q has an oversized decoded stored chunk", s.path)
+ }
+ if cap(s.decodedBuffer) < decodedLength {
+ s.decodedBuffer = make([]byte, decodedLength)
+ } else {
+ s.decodedBuffer = s.decodedBuffer[:decodedLength]
+ }
+ decoded, decodeErr := decodeStoredChunk(s.decodedBuffer[:0], s.compressedBuffer)
+ if decodeErr != nil {
+ return nil, corruptError("decode stored chunk in segment %q: %w", s.path, decodeErr)
+ }
+ if len(decoded) != decodedLength {
+ return nil, corruptError("segment %q decoded a stored chunk to an unexpected length", s.path)
+ }
+ s.decodedBuffer = decoded
+ return decoded, nil
+}
+
+func storedChunkDecodedLength(compressed []byte) (decodedLength int, err error) {
+ defer func() {
+ if recovered := recover(); recovered != nil {
+ err = fmt.Errorf("stored chunk length decoder panicked: %v", recovered)
+ }
+ }()
+ return s2.DecodedLen(compressed)
+}
+
+func decodeStoredChunk(dst, compressed []byte) (decoded []byte, err error) {
+ defer func() {
+ if recovered := recover(); recovered != nil {
+ err = fmt.Errorf("stored chunk decoder panicked: %v", recovered)
+ }
+ }()
+ return s2.Decode(dst, compressed)
+}
+
+func (s *storedSegmentReader) decodeDocument(documentNumber uint64, chunk []byte) (storedDocument, error) {
+ documentOffset, offsetErr := s.documentOffset(documentNumber)
+ if offsetErr != nil {
+ return storedDocument{}, offsetErr
+ }
+ if documentOffset >= uint64(len(chunk)) {
+ return storedDocument{}, corruptError("segment %q has a stored document offset outside its chunk", s.path)
+ }
+ decoder := byteDecoder{payload: chunk[documentOffset:]}
+ metaLength, metaLengthErr := decoder.uvarint()
+ if metaLengthErr != nil {
+ return storedDocument{}, metaLengthErr
+ }
+ dataLength, dataLengthErr := decoder.uvarint()
+ if dataLengthErr != nil {
+ return storedDocument{}, dataLengthErr
+ }
+ meta, metaErr := decoder.bytes(metaLength)
+ if metaErr != nil {
+ return storedDocument{}, metaErr
+ }
+ data, dataErr := decoder.bytes(dataLength)
+ if dataErr != nil {
+ return storedDocument{}, dataErr
+ }
+ metaDecoder := byteDecoder{payload: meta}
+ document := storedDocument{}
+ for fieldCount := 0; metaDecoder.remaining() > 0; fieldCount++ {
+ if fieldCount >= maxStoredFieldsPerDocument {
+ return storedDocument{}, corruptError("segment %q has too many stored field values in one document", s.path)
+ }
+ fieldID, fieldIDErr := metaDecoder.uvarint()
+ if fieldIDErr != nil {
+ return storedDocument{}, fieldIDErr
+ }
+ valueOffset, valueOffsetErr := metaDecoder.uvarint()
+ if valueOffsetErr != nil {
+ return storedDocument{}, valueOffsetErr
+ }
+ valueLength, valueLengthErr := metaDecoder.uvarint()
+ if valueLengthErr != nil {
+ return storedDocument{}, valueLengthErr
+ }
+ if fieldID >= uint64(len(s.fieldNames)) || valueOffset > uint64(len(data)) || valueLength > uint64(len(data))-valueOffset {
+ return storedDocument{}, corruptError("segment %q has invalid stored field metadata", s.path)
+ }
+ document.fields = append(document.fields, storedField{
+ name: s.fieldNames[fieldID],
+ value: data[valueOffset : valueOffset+valueLength],
+ })
+ }
+ return document, nil
+}
+
+func (s *storedSegmentReader) documentOffset(documentNumber uint64) (uint64, error) {
+ if documentNumber >= s.footer.documentCount {
+ return 0, corruptError("segment %q has an out-of-range stored document number", s.path)
+ }
+ indexOffset := s.footer.storedIndexOffset + documentNumber*storedDocumentOffsetByteWidth
+ if indexOffset > s.footer.docValueOffset-storedDocumentOffsetByteWidth {
+ return 0, corruptError("segment %q has a stored document offset outside its index", s.path)
+ }
+ var offsetData [storedDocumentOffsetByteWidth]byte
+ if readErr := s.readInto(indexOffset, offsetData[:]); readErr != nil {
+ return 0, readErr
+ }
+ return binary.BigEndian.Uint64(offsetData[:]), nil
+}
+
+func (s *storedSegmentReader) readBytes(offset, length uint64) ([]byte, error) {
+ if length > maxStoredChunkTableSize {
+ return nil, corruptError("segment %q requested an oversized read", s.path)
+ }
+ data := make([]byte, int(length))
+ if readErr := s.readInto(offset, data); readErr != nil {
+ return nil, readErr
+ }
+ return data, nil
+}
+
+func (s *storedSegmentReader) readInto(offset uint64, data []byte) error {
+ length := uint64(len(data))
+ if offset > s.size || length > s.size-offset {
+ return corruptError("segment %q read exceeds file bounds", s.path)
+ }
+ read, readErr := s.file.ReadAt(data, int64(offset))
+ if readErr != nil || read != len(data) {
+ return corruptError("read segment %q", s.path, readErr)
+ }
+ return nil
+}
+
+func deletedDocuments(record segmentRecord) (*roaringpkg.Bitmap, error) {
+ deleted := roaringpkg.New()
+ if len(record.deletionBitmap) == 0 {
+ return deleted, nil
+ }
+ if unmarshalErr := deleted.UnmarshalBinary(record.deletionBitmap); unmarshalErr != nil {
+ return nil, corruptError("decode deletion bitmap", unmarshalErr)
+ }
+ return deleted, nil
}
func deletionCount(payload []byte, documentCount uint64) (uint64, error) {
diff --git a/pkg/index/inverted/inverted.go b/pkg/index/inverted/inverted.go
index 10ad009c4..4c5a68768 100644
--- a/pkg/index/inverted/inverted.go
+++ b/pkg/index/inverted/inverted.go
@@ -331,6 +331,57 @@ func ReadOnlyDocCount(path string) (int64, error) {
return count, nil
}
+// StoredDocument is one live document of a committed index generation,
+// borrowed for the duration of a single ReadOnlyWalkDocuments callback.
+//
+// It is the whole of what a read-only match-all walk exposes: repeated raw
+// stored (field name, value) pairs, and nothing about terms, dictionaries,
+// postings, doc values, sorting or the container format underneath.
+type StoredDocument interface {
+ // VisitStoredFields calls visit once for every stored value the document
+ // records, passing the field's name and its raw value bytes. A field the
+ // document records more than once is visited once per recorded value, in
+ // the order the document records them. Visiting stops early when visit
+ // returns false.
+ //
+ // The name and value handed to visit are borrowed and stay valid only until
+ // visit returns; a caller that keeps either beyond that copies it.
+ VisitStoredFields(visit func(name string, value []byte) bool) error
+}
+
+// ReadOnlyWalkDocuments opens the index directory at path read-only, pins its
+// newest structurally complete committed generation and calls visit once for
+// every live document that generation holds, streaming one document at a time.
+// Documents the pinned generation's deletion masks cover are skipped.
+//
+// Like ReadOnlyDocCount it never acquires the exclusive directory lock and
+// writes no bytes, so a directory a live writer owns can be walked while it is
+// being written, and the walk leaves file contents, modification times and
+// directory entries unchanged.
+//
+// The StoredDocument handed to visit is borrowed: it, and every name and value
+// it yields, stay valid only until visit returns. A caller that retains a value
+// beyond its callback copies it.
+//
+// A directory holding no committed generation reports an error wrapping
+// ErrNoCommittedIndex, which callers that treat a cold or unflushed index as
+// empty match on. Committed bytes that violate the on-disk grammar, or that
+// would require decoding past a configured bound, report an error wrapping
+// ErrCorruptIndex. Canceling ctx stops the walk between two documents and
+// returns ctx.Err(); an error from visit stops the walk and is returned as-is.
+func ReadOnlyWalkDocuments(ctx context.Context, path string, visit func(doc StoredDocument) error) error {
+ reader, err := nativeice.Open(path)
+ if err != nil {
+ return err
+ }
+ defer func() {
+ _ = reader.Close()
+ }()
+ return reader.VisitLiveDocuments(ctx, func(doc nativeice.StoredDocument) error {
+ return visit(doc)
+ })
+}
+
func (s *store) Iterator(ctx context.Context, fieldKey index.FieldKey, termRange index.RangeOpts, order modelv1.Sort,
preLoadSize int,
) (iter index.FieldIterator[*index.DocumentResult], err error) {
diff --git a/pkg/index/inverted/native_dependency_test.go b/pkg/index/inverted/native_dependency_test.go
index 3577d428b..dd671c90b 100644
--- a/pkg/index/inverted/native_dependency_test.go
+++ b/pkg/index/inverted/native_dependency_test.go
@@ -38,8 +38,13 @@ const (
readOnlyDocCountFunc = "ReadOnlyDocCount"
)
+// allowedNativeReaderExternalImports is the native reader's whole third-party
+// dependency budget: the bitmap decoder its deletion masks are encoded with,
+// and the block codec BDB-NIDX-SPEC-001 revision 0.2 DEC-008 fixes for the
+// stored-field chunks of the historical corpus.
var allowedNativeReaderExternalImports = map[string]struct{}{
"github.com/RoaringBitmap/roaring": {},
+ "github.com/klauspost/compress/s2": {},
}
// TestReadOnlyDocCountReachesNativeReader is the structural half of the
diff --git a/pkg/index/inverted/native_fixture_nidx01c_test.go b/pkg/index/inverted/native_fixture_nidx01c_test.go
new file mode 100644
index 000000000..95aa9476b
--- /dev/null
+++ b/pkg/index/inverted/native_fixture_nidx01c_test.go
@@ -0,0 +1,331 @@
+// 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 inverted
+
+import (
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+)
+
+// The NIDX-01C corpus extends the checked-in legacy corpus with the live series
+// documents issue #14010 declares. It is two independent index directories:
+// sourceA, holding the walk the milestone is specified against, and sourceB,
+// which repeats one of sourceA's documents so the series-union case has a
+// second source. Every byte is produced by BanyanDB's compatibility writer
+// through the pkg/index/inverted store boundary and checked in; production code
+// never generates it, and the generator lives in this test file behind
+// nidx01cGenerateEnv.
+const (
+ nidx01cRoot = "testdata/nidx01c"
+ nidx01cSourceADir = nidx01cRoot + "/sourceA"
+ nidx01cSourceBDir = nidx01cRoot + "/sourceB"
+ nidx01cManifest = nidx01cRoot + "/provenance.json"
+
+ // nidx01cGenerateEnv gates regeneration so an ordinary test run can never
+ // overwrite the checked-in bytes the contract is pinned to.
+ nidx01cGenerateEnv = "GENERATE_NIDX01C_FIXTURE"
+ nidx01cGenerateCmd = "GENERATE_NIDX01C_FIXTURE=1 go test ./pkg/index/inverted/ -run TestGenerateNIDX01CFixture -count=1"
+
+ // nidx01cTagName is the one repeated tag name issue #14010 declares. It is
+ // stored and not indexed, which is how the production write path records a
+ // non-indexed tag.
+ nidx01cTagName = "color"
+
+ // The literals below are the documents issue #14010 declares, not values
+ // any reader computes. Document labels 101, 202 and 303 are the issue's
+ // names for the three documents; they are not series identifiers, because
+ // pkg/pb/v1.Series derives its identifier by hashing the marshaled entity
+ // buffer and a fixture cannot choose that hash. Each label's identity is
+ // the raw byte string below, which the store records as the document's
+ // identity field.
+ nidx01cLabel101 = "101"
+ nidx01cLabel202 = "202"
+ nidx01cLabel303 = "303"
+
+ // nidx01cSourceAPhysical is the physical document count of sourceA's
+ // selected generation: 101, 202 and the deleted 303. One deletion leaves
+ // two visible.
+ nidx01cSourceAPhysical = uint64(3)
+ nidx01cSourceADeleted = 1
+ nidx01cSourceAVisible = int64(2)
+
+ // nidx01cSourceBPhysical is sourceB's physical document count: the repeat
+ // of 101 alone, with no deletion.
+ nidx01cSourceBPhysical = uint64(1)
+ nidx01cSourceBVisible = int64(1)
+
+ // The stored timestamp bytes below are the compatibility writer's encoding
+ // of Unix nanoseconds 100, 200 and 300, pinned here as literals. Each is a
+ // shift marker of 0x20 followed by the ten seven-bit groups, most
+ // significant first, of the sortable form of the signed value -- 100 is
+ // 0x8000000000000064, whose groups are 01 00 00 00 00 00 00 00 00 64 -- so
+ // a reviewer can re-derive them by hand rather than by running a decoder.
+ //
+ // They are the compatibility claim this milestone makes: the native walk
+ // must hand a caller the same bytes the retired reader hands it today, so a
+ // caller that decodes them keeps working unchanged.
+ nidx01cStoredTimestamp100Hex = "2001000000000000000064"
+ nidx01cStoredTimestamp200Hex = "2001000000000000000148"
+ nidx01cStoredTimestamp300Hex = "200100000000000000022c"
+)
+
+// nidx01cDocument is one declared document of the corpus: its issue label, the
+// raw identity bytes the store records, its repeated tag values in the declared
+// order, and its timestamp and version.
+type nidx01cDocument struct {
+ label string
+ // storedTimestampHex is the compatibility writer's encoding of timestamp as
+ // the segment records it. Unlike the identity, the tag values and the
+ // version, a stored timestamp is not a BanyanDB encoding, so the corpus
+ // pins the oracle's bytes rather than restating a BanyanDB one.
+ storedTimestampHex string
+ identity []byte
+ tagValues []string
+ timestamp int64
+ version int64
+ deleted bool
+}
+
+// nidx01cSourceADocuments is sourceA's declared content, in the order issue
+// #14010 lists it. 101 carries the binary identity 0x010203 the issue names,
+// which is deliberately not valid UTF-8 and not a marshaled series buffer, so
+// the walk's byte fidelity is observable rather than inferred from text.
+var nidx01cSourceADocuments = []nidx01cDocument{
+ {
+ label: nidx01cLabel101,
+ identity: []byte{0x01, 0x02, 0x03},
+ tagValues: []string{"blue", "green"},
+ timestamp: 100,
+ storedTimestampHex: nidx01cStoredTimestamp100Hex,
+ version: 2,
+ },
+ {
+ label: nidx01cLabel202,
+ identity: []byte{0x04, 0x05, 0x06},
+ tagValues: []string{"red"},
+ timestamp: 200,
+ storedTimestampHex: nidx01cStoredTimestamp200Hex,
+ version: 1,
+ },
+ {
+ label: nidx01cLabel303,
+ identity: []byte{0x07, 0x08, 0x09},
+ tagValues: []string{"gray"},
+ timestamp: 300,
+ storedTimestampHex: nidx01cStoredTimestamp300Hex,
+ version: 3,
+ deleted: true,
+ },
+}
+
+// nidx01cSourceBDocuments repeats 101 so the series-union case reads the same
+// document from a second source.
+var nidx01cSourceBDocuments = []nidx01cDocument{nidx01cSourceADocuments[0]}
+
+// nidx01cProvenance is the manifest checked in beside the corpus bytes. It
+// records which oracle produced the bytes, how to re-derive them, which
+// documents each source declares, and what every file hashes to.
+type nidx01cProvenance struct {
+ Oracle map[string]string `json:"oracle"`
+ FileSHA256 map[string]string `json:"file_sha256"`
+ ReservedCRC32 map[string]string `json:"reserved_crc32"`
+ GeneratorCommand string `json:"generator_command"`
+ Notes string `json:"notes"`
+ Sources []nidx01cProvenanceSource `json:"sources"`
+}
+
+// nidx01cProvenanceSource describes one source directory of the corpus.
+type nidx01cProvenanceSource struct {
+ Directory string `json:"directory"`
+ Documents []nidx01cProvenanceDocument `json:"documents"`
+ PhysicalDocuments uint64 `json:"physical_documents"`
+ Deletions int `json:"deletions"`
+ VisibleCount int64 `json:"visible_count"`
+}
+
+// nidx01cProvenanceDocument records one declared document as hexadecimal bytes,
+// so a reviewer can compare the corpus against issue #14010 without decoding a
+// segment.
+type nidx01cProvenanceDocument struct {
+ Label string `json:"label"`
+ IdentityHex string `json:"identity_hex"`
+ TagName string `json:"tag_name"`
+ TimestampHex string `json:"timestamp_stored_hex"`
+ VersionHex string `json:"version_stored_hex"`
+ TagValues []string `json:"tag_values"`
+ Timestamp int64 `json:"timestamp"`
+ Version int64 `json:"version"`
+ Deleted bool `json:"deleted"`
+}
+
+// TestGenerateNIDX01CFixture rebuilds the checked-in NIDX-01C corpus and its
+// provenance manifest with BanyanDB's compatibility writer. It runs only when
+// nidx01cGenerateEnv is set, so a normal test run reads the committed bytes
+// instead of producing fresh ones.
+//
+// The writer reclaims deletions by merging, so the generation carrying 303's
+// deletion is superseded shortly after the delete lands. The generator
+// therefore lifts the generation out of the live directory as soon as the
+// commit that produced it returns, rather than reading the directory's final
+// state.
+func TestGenerateNIDX01CFixture(t *testing.T) {
+ if os.Getenv(nidx01cGenerateEnv) != "1" {
+ t.Skipf("set %s=1 to regenerate the NIDX-01C corpus", nidx01cGenerateEnv)
+ }
+ tester := require.New(t)
+
+ tester.NoError(os.RemoveAll(nidx01cRoot))
+ tester.NoError(os.MkdirAll(nidx01cSourceADir, 0o755))
+ tester.NoError(os.MkdirAll(nidx01cSourceBDir, 0o755))
+
+ hashes := map[string]string{}
+ record := func(relativeDir, name string, payload []byte) {
+ tester.NoError(os.WriteFile(filepath.Join(nidx01cRoot, relativeDir, name), payload, 0o600))
+ sum := sha256.Sum256(payload)
+ hashes[relativeDir+"/"+name] = hex.EncodeToString(sum[:])
+ }
+
+ sourceA := generateNIDX01CSource(t, nidx01cSourceADocuments, nidx01cSourceAPhysical, nidx01cSourceADeleted)
+ writeNIDX01CGeneration(t, record, "sourceA", sourceA)
+ sourceB := generateNIDX01CSource(t, nidx01cSourceBDocuments, nidx01cSourceBPhysical, 0)
+ writeNIDX01CGeneration(t, record, "sourceB", sourceB)
+
+ manifest := nidx01cProvenance{
+ Oracle: readCompatibilityOracleIdentity(t),
+ GeneratorCommand: nidx01cGenerateCmd,
+ FileSHA256: hashes,
+ Sources: []nidx01cProvenanceSource{
+ describeNIDX01CSource("sourceA", nidx01cSourceADocuments, nidx01cSourceAPhysical,
+ nidx01cSourceADeleted, nidx01cSourceAVisible, sourceA),
+ describeNIDX01CSource("sourceB", nidx01cSourceBDocuments, nidx01cSourceBPhysical,
+ 0, nidx01cSourceBVisible, sourceB),
+ },
+ ReservedCRC32: map[string]string{
+ segExt: hex.EncodeToString(reservedCRC32FillSegment) + " (arbitrary, no writer computed it)",
+ snpExt: "as written by the oracle; the compatibility loader validates this slot, " +
+ "so the corpus keeps it readable and the reserved-field contract is proved by run-time mutation",
+ },
+ Notes: "sourceA holds documents 101, 202 and a deleted 303 in one committed generation, so two are visible " +
+ "and the declared walk is 101 then 202. sourceB repeats 101 alone for the series-union case. " +
+ "Document labels are issue #14010's names, not series identifiers: a series identifier is a hash of the " +
+ "marshaled entity buffer and cannot be chosen by a fixture.",
+ }
+ encoded, err := json.MarshalIndent(manifest, "", " ")
+ tester.NoError(err)
+ tester.NoError(os.WriteFile(nidx01cManifest, append(encoded, '\n'), 0o600))
+}
+
+// generateNIDX01CSource writes the given documents through the compatibility
+// writer, deletes the ones declared deleted, and lifts out the generation
+// holding the declared physical document and deletion counts.
+func generateNIDX01CSource(t *testing.T, documents []nidx01cDocument, physical uint64, deletions int) capturedGeneration {
+ t.Helper()
+ tester := require.New(t)
+
+ staging := t.TempDir()
+ writer, err := NewStore(StoreOpts{Path: staging})
+ tester.NoError(err)
+ tester.NoError(writer.UpdateSeriesBatch(nidx01cBatch(documents)))
+ var deletedIdentities [][]byte
+ for _, document := range documents {
+ if document.deleted {
+ deletedIdentities = append(deletedIdentities, document.identity)
+ }
+ }
+ if len(deletedIdentities) > 0 {
+ tester.NoError(writer.Delete(deletedIdentities))
+ }
+ captured := captureGeneration(t, staging, "the NIDX-01C generation", func(records []snapshotSegmentRecord) bool {
+ return totalPhysicalDocuments(records) == physical && totalDeletions(records) == deletions
+ })
+ tester.NoError(writer.Close())
+ return captured
+}
+
+// writeNIDX01CGeneration checks a captured generation's manifest and segments
+// into the corpus directory.
+func writeNIDX01CGeneration(t *testing.T, record func(relativeDir, name string, payload []byte), directory string,
+ captured capturedGeneration,
+) {
+ t.Helper()
+ record(directory, filepath.Base(captured.source), captured.manifest)
+ for _, segmentID := range sortedSegmentIDs(captured.segments) {
+ record(directory, segmentFileName(segmentID), withReservedCRC32Fill(captured.segments[segmentID]))
+ }
+}
+
+// nidx01cBatch turns the declared documents into one series commit. Each
+// document's identity becomes the store's entity values, and every declared tag
+// value becomes a separate stored, non-indexed field of the same name, which is
+// how a repeated tag reaches the segment.
+func nidx01cBatch(documents []nidx01cDocument) index.Batch {
+ batch := index.Batch{Documents: make(index.Documents, 0, len(documents))}
+ for _, document := range documents {
+ fields := make([]index.Field, 0, len(document.tagValues))
+ for _, tagValue := range document.tagValues {
+ field := index.NewBytesField(index.FieldKey{TagName: nidx01cTagName}, []byte(tagValue))
+ field.Store = true
+ field.Index = false
+ fields = append(fields, field)
+ }
+ batch.Documents = append(batch.Documents, index.Document{
+ Fields: fields,
+ EntityValues: document.identity,
+ Timestamp: document.timestamp,
+ Version: document.version,
+ })
+ }
+ return batch
+}
+
+// describeNIDX01CSource renders one source's declared content for the
+// provenance manifest.
+func describeNIDX01CSource(directory string, documents []nidx01cDocument, physical uint64, deletions int,
+ visible int64, captured capturedGeneration,
+) nidx01cProvenanceSource {
+ described := make([]nidx01cProvenanceDocument, 0, len(documents))
+ for _, document := range documents {
+ described = append(described, nidx01cProvenanceDocument{
+ Label: document.label,
+ IdentityHex: hex.EncodeToString(document.identity),
+ TagName: nidx01cTagName,
+ TagValues: document.tagValues,
+ Timestamp: document.timestamp,
+ TimestampHex: document.storedTimestampHex,
+ Version: document.version,
+ VersionHex: hex.EncodeToString(convert.Int64ToBytes(document.version)),
+ Deleted: document.deleted,
+ })
+ }
+ return nidx01cProvenanceSource{
+ Directory: directory + "/" + filepath.Base(captured.source),
+ Documents: described,
+ PhysicalDocuments: physical,
+ Deletions: deletions,
+ VisibleCount: visible,
+ }
+}
diff --git a/pkg/index/inverted/native_multichunk_walk_test.go b/pkg/index/inverted/native_multichunk_walk_test.go
new file mode 100644
index 000000000..4e727254d
--- /dev/null
+++ b/pkg/index/inverted/native_multichunk_walk_test.go
@@ -0,0 +1,97 @@
+// 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. The 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 inverted
+
+import (
+ "context"
+ "encoding/hex"
+ "fmt"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+)
+
+const nativeStoredChunkWalkDocumentCount = 129
+
+// TestNativeStoredDocumentWalksMultipleStoredChunks proves that a native walk
+// retains every document identity and stored value when a compatibility-writer
+// segment spans more than one stored chunk.
+func TestNativeStoredDocumentWalksMultipleStoredChunks(t *testing.T) {
+ tester := require.New(t)
+ indexDir := t.TempDir()
+ writer, newStoreErr := NewStore(StoreOpts{Path: indexDir})
+ tester.NoError(newStoreErr)
+ writerClosed := false
+ defer func() {
+ if !writerClosed {
+ tester.NoError(writer.Close())
+ }
+ }()
+
+ fieldKey := index.FieldKey{
+ Analyzer: index.AnalyzerKeyword,
+ SeriesID: common.SeriesID(1),
+ IndexRuleID: 1,
+ }
+ seriesIdentity := hex.EncodeToString(convert.Uint64ToBytes(1))
+ expected := make(map[string]map[string][]string, nativeStoredChunkWalkDocumentCount)
+ documents := make(index.Documents, 0, nativeStoredChunkWalkDocumentCount)
+ for documentNumber := uint64(1); documentNumber <= nativeStoredChunkWalkDocumentCount; documentNumber++ {
+ storedValue := []byte(fmt.Sprintf("stored-value-%03d", documentNumber))
+ field := index.NewBytesField(fieldKey, storedValue)
+ field.Store = true
+ field.Index = true
+ identity := convert.Uint64ToBytes(documentNumber)
+ identityHex := hex.EncodeToString(identity)
+ expected[identityHex] = map[string][]string{
+ docIDField: {identityHex},
+ seriesIDField: {seriesIdentity},
+ fieldKey.Marshal(): {hex.EncodeToString(storedValue)},
+ }
+ documents = append(documents, index.Document{DocID: documentNumber, Fields: []index.Field{field}})
+ }
+ tester.NoError(writer.Batch(index.Batch{Documents: documents}))
+ tester.NoError(writer.Close())
+ writerClosed = true
+
+ actual := make(map[string]map[string][]string, nativeStoredChunkWalkDocumentCount)
+ walkErr := ReadOnlyWalkDocuments(context.Background(), indexDir, func(document StoredDocument) error {
+ fields := make(map[string][]string)
+ identityHex := ""
+ if visitErr := document.VisitStoredFields(func(name string, value []byte) bool {
+ encodedValue := hex.EncodeToString(value)
+ fields[name] = append(fields[name], encodedValue)
+ if name == docIDField {
+ identityHex = encodedValue
+ }
+ return true
+ }); visitErr != nil {
+ return visitErr
+ }
+ if identityHex == "" {
+ return fmt.Errorf("walked document has no identity field")
+ }
+ actual[identityHex] = fields
+ return nil
+ })
+
+ tester.NoError(walkErr)
+ tester.Equal(expected, actual)
+}
diff --git a/pkg/index/inverted/native_nidx01b_boundary_test.go b/pkg/index/inverted/native_nidx01b_boundary_test.go
index 0c9f52c89..7f698d69e 100644
--- a/pkg/index/inverted/native_nidx01b_boundary_test.go
+++ b/pkg/index/inverted/native_nidx01b_boundary_test.go
@@ -52,12 +52,13 @@ var nidx01bBoundary committedGenerationCounter = ReadOnlyDocCount
// permitted to export. Methods appear as Type.Method.
//
// The list is the read-only slice NIDX-01 allows and nothing else: open a
-// committed generation, report its visible document count, close it, and
-// classify the two failures a caller must distinguish. Issue #14009 places
-// stored fields, dictionaries, term postings, document walk, doc values,
-// sort/search-after, writers and merge outside this milestone, so an entry
-// appearing here for any of them is the milestone growing surface it was
-// explicitly denied.
+// committed generation, report its visible document count, walk its live
+// documents' stored fields, close it, and classify the two failures a caller
+// must distinguish. Issue #14009 placed the document walk outside NIDX-01B;
+// issue #14010 adds it and nothing more. Dictionaries, term postings, doc
+// values, sort/search-after, writers and merge remain outside the milestone, so
+// an entry appearing here for any of them is the milestone growing surface it
+// was explicitly denied.
var nativeReaderSurface = []string{
"ErrCorrupt",
"ErrNoSnapshot",
@@ -65,6 +66,8 @@ var nativeReaderSurface = []string{
"Reader",
"Reader.Close",
"Reader.VisibleDocCount",
+ "Reader.VisitLiveDocuments",
+ "StoredDocument",
}
// TestNIDX01BBoundarySurface guards the boundary itself rather than any
diff --git a/pkg/index/inverted/native_nidx01c_walk_test.go b/pkg/index/inverted/native_nidx01c_walk_test.go
new file mode 100644
index 000000000..68e6b3f66
--- /dev/null
+++ b/pkg/index/inverted/native_nidx01c_walk_test.go
@@ -0,0 +1,463 @@
+// 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 inverted
+
+import (
+ "context"
+ "encoding/hex"
+ "errors"
+ "path/filepath"
+ "sort"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/pkg/convert"
+)
+
+// liveDocumentWalker is the whole of NIDX-01C's contract: one call from an
+// index directory path to a streamed visit of every live document the newest
+// structurally complete committed generation holds, each exposing its repeated
+// raw stored (field name, value) pairs, or to a classified error.
+//
+// Issue #14010 names one BanyanDB-owned read-only document visitor as the
+// milestone's only test seam, and BDB-NIDX-SPEC-001 revision 0.2 NIDX-01
+// forbids exposing a general reader beside it. Everything the milestone adds --
+// the fields index, the stored chunk table, block decompression, the stored
+// meta triples, deletion-aware iteration -- sits behind this signature and
+// stays private. The coder is free to move any of it; the coder may not move
+// this.
+type liveDocumentWalker func(ctx context.Context, path string, visit func(doc StoredDocument) error) error
+
+// nidx01cBoundary binds the contract to the production symbol that satisfies
+// it, so a change to ReadOnlyWalkDocuments's signature fails to compile here
+// rather than quietly redefining the milestone.
+var nidx01cBoundary liveDocumentWalker = ReadOnlyWalkDocuments
+
+// walkedField is one stored value a document handed the walk, with its bytes
+// rendered as hexadecimal so a mismatch reads as bytes rather than as mojibake.
+type walkedField struct {
+ name string
+ value string
+}
+
+// walkNIDX01C walks path through the boundary and returns each visited
+// document's stored fields, in visit order, together with the walk's error.
+func walkNIDX01C(ctx context.Context, path string) ([][]walkedField, error) {
+ var documents [][]walkedField
+ err := nidx01cBoundary(ctx, path, func(doc StoredDocument) error {
+ var fields []walkedField
+ if visitErr := doc.VisitStoredFields(func(name string, value []byte) bool {
+ fields = append(fields, walkedField{name: name, value: hex.EncodeToString(value)})
+ return true
+ }); visitErr != nil {
+ return visitErr
+ }
+ documents = append(documents, fields)
+ return nil
+ })
+ return documents, err
+}
+
+// declaredFieldsOf renders one declared corpus document as the unordered set of
+// stored fields the walk must hand back for it: its identity, its stored
+// timestamp and version, and one entry per repeated tag value. Every value here
+// comes from issue #14010's declaration or from a BanyanDB encoder, never from
+// the reader under test.
+func declaredFieldsOf(document nidx01cDocument) []walkedField {
+ fields := []walkedField{
+ {name: docIDField, value: hex.EncodeToString(document.identity)},
+ {name: timestampField, value: document.storedTimestampHex},
+ {name: versionField, value: hex.EncodeToString(convert.Int64ToBytes(document.version))},
+ }
+ for _, tagValue := range document.tagValues {
+ fields = append(fields, walkedField{name: nidx01cTagName, value: hex.EncodeToString([]byte(tagValue))})
+ }
+ return fields
+}
+
+// sortedFields orders stored fields by name then value so a document's content
+// can be compared without pinning the order the writer happened to lay its
+// fields out in. Repeated values of one name keep a stable relative order,
+// which valuesOf checks separately.
+func sortedFields(fields []walkedField) []walkedField {
+ ordered := append([]walkedField(nil), fields...)
+ sort.Slice(ordered, func(left, right int) bool {
+ if ordered[left].name != ordered[right].name {
+ return ordered[left].name < ordered[right].name
+ }
+ return ordered[left].value < ordered[right].value
+ })
+ return ordered
+}
+
+// valuesOf returns the values a document carried for one field name, in the
+// order the walk visited them.
+func valuesOf(fields []walkedField, name string) []string {
+ var values []string
+ for _, field := range fields {
+ if field.name == name {
+ values = append(values, field.value)
+ }
+ }
+ return values
+}
+
+// hexValues renders declared tag values as hexadecimal, in the declared order.
+func hexValues(values []string) []string {
+ encoded := make([]string, 0, len(values))
+ for _, value := range values {
+ encoded = append(encoded, hex.EncodeToString([]byte(value)))
+ }
+ return encoded
+}
+
+// liveDeclaredDocuments returns the corpus documents a walk must visit, in
+// declaration order, dropping the ones the corpus deletes.
+func liveDeclaredDocuments(documents []nidx01cDocument) []nidx01cDocument {
+ var live []nidx01cDocument
+ for _, document := range documents {
+ if !document.deleted {
+ live = append(live, document)
+ }
+ }
+ return live
+}
+
+// TestNativeStoredDocumentWalkDeclaredDocuments is the boundary contract for
+// NIDX-01C. It exercises inverted.ReadOnlyWalkDocuments, and nothing behind it,
+// against the checked-in NIDX-01C corpus.
+//
+// Requirement proved here:
+//
+// R1 -- the visitor streams only live documents of the pinned committed
+// generation and preserves every repeated field name and value byte
+// sequence exactly. The corpus declares documents 101 and 202 live and
+// 303 deleted, so the walk is exactly 101 then 202; 101's repeated tag
+// values stay in the declared order blue then green; and every stored
+// value arrives as the bytes the compatibility writer recorded,
+// including an identity that is not valid UTF-8.
+func TestNativeStoredDocumentWalkDeclaredDocuments(t *testing.T) {
+ tester := require.New(t)
+
+ documents, err := walkNIDX01C(context.Background(), nidx01cSourceADir)
+ tester.NoError(err)
+
+ declared := liveDeclaredDocuments(nidx01cSourceADocuments)
+ tester.Len(documents, len(declared),
+ "the walk must visit every live document of the pinned generation and no deleted one")
+
+ deletedIdentities := map[string]string{}
+ for _, document := range nidx01cSourceADocuments {
+ if document.deleted {
+ deletedIdentities[hex.EncodeToString(document.identity)] = document.label
+ }
+ }
+
+ for position, expected := range declared {
+ visited := documents[position]
+ tester.Equal(sortedFields(declaredFieldsOf(expected)), sortedFields(visited),
+ "document at walk position %d must be the corpus document labeled %s", position, expected.label)
+ tester.Equal(hexValues(expected.tagValues), valuesOf(visited, nidx01cTagName),
+ "document %s must keep its repeated %q values in the declared order", expected.label, nidx01cTagName)
+ for _, identity := range valuesOf(visited, docIDField) {
+ label, deleted := deletedIdentities[identity]
+ tester.False(deleted, "the walk visited the deleted document labeled %s", label)
+ }
+ }
+}
+
+// TestNativeStoredDocumentWalkSecondSource walks the corpus' second source, the
+// one the series-union case reads alongside the first.
+//
+// Requirement proved here:
+//
+// R1 -- a source holding a single live document yields exactly that document,
+// with the same repeated values as the copy of it in the first source,
+// so a union reading both sources sees the same document twice rather
+// than two different ones.
+func TestNativeStoredDocumentWalkSecondSource(t *testing.T) {
+ tester := require.New(t)
+
+ documents, err := walkNIDX01C(context.Background(), nidx01cSourceBDir)
+ tester.NoError(err)
+
+ declared := liveDeclaredDocuments(nidx01cSourceBDocuments)
+ tester.Len(documents, len(declared))
+ tester.Equal(sortedFields(declaredFieldsOf(declared[0])), sortedFields(documents[0]))
+ tester.Equal(hexValues(declared[0].tagValues), valuesOf(documents[0], nidx01cTagName))
+}
+
+// TestNativeStoredDocumentWalkStopsOnCancellation cancels the walk from inside
+// the first document's callback.
+//
+// Requirement proved here:
+//
+// R4 -- cancellation stops the walk between two documents. The corpus source
+// holds two live documents; a walk canceled while the first is being
+// visited never delivers the second and reports the cancellation, which
+// is also what makes the walk streaming rather than a materialized list.
+func TestNativeStoredDocumentWalkStopsOnCancellation(t *testing.T) {
+ tester := require.New(t)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ visited := 0
+ err := nidx01cBoundary(ctx, nidx01cSourceADir, func(_ StoredDocument) error {
+ visited++
+ cancel()
+ return nil
+ })
+
+ tester.ErrorIs(err, context.Canceled, "a canceled walk must report the cancellation")
+ tester.Equal(1, visited, "a walk canceled during the first document must not deliver a second one")
+}
+
+// TestNativeStoredDocumentWalkStopsOnVisitError returns an error from the
+// caller's document callback.
+//
+// Requirement proved here:
+//
+// R4 -- the document callback owns the walk: an error it returns stops the
+// walk at that document and reaches the caller unchanged, so a source
+// reader that fails to rebuild one document aborts instead of
+// publishing a partial destination.
+func TestNativeStoredDocumentWalkStopsOnVisitError(t *testing.T) {
+ tester := require.New(t)
+
+ rebuildFailure := errors.New("rebuild refused this document")
+ visited := 0
+ err := nidx01cBoundary(context.Background(), nidx01cSourceADir, func(_ StoredDocument) error {
+ visited++
+ return rebuildFailure
+ })
+
+ tester.ErrorIs(err, rebuildFailure)
+ tester.Equal(1, visited, "the walk must stop at the document whose callback failed")
+}
+
+// TestNativeStoredDocumentWalkDamagedStoredRegionIsCorrupt walks a copy of the
+// corpus whose stored-document region -- the chunk table, the chunk offsets and
+// the compressed blocks the fields index root sits above -- has been damaged.
+// Counting documents never reads that region, so NIDX-01B reports this
+// directory as healthy; walking them cannot.
+//
+// Requirement proved here:
+//
+// R4 -- a truncated stored chunk, an invalid length and a decode-limit
+// overflow are rejected with the native typed corruption error rather
+// than a panic, a hang or an unbounded allocation.
+func TestNativeStoredDocumentWalkDamagedStoredRegionIsCorrupt(t *testing.T) {
+ tester := require.New(t)
+
+ damaged := copyIndexDir(t, nidx01cSourceADir)
+ damageStoredDocumentRegion(t, newestSegmentFile(t, damaged))
+
+ visited := 0
+ err := nidx01cBoundary(context.Background(), damaged, func(_ StoredDocument) error {
+ visited++
+ return nil
+ })
+
+ tester.ErrorIs(err, ErrCorruptIndex, "damaged stored bytes must be classified as a corrupt index")
+ tester.NotErrorIs(err, ErrNoCommittedIndex,
+ "a damaged generation is not an absent one; callers classify the two differently")
+ tester.Zero(visited, "no document may be handed to the caller out of a damaged stored region")
+}
+
+// TestNativeStoredDocumentWalkIgnoresReservedCRC32 rewrites the reserved CRC32
+// slot of every file in a copy of the corpus.
+//
+// Requirement proved here:
+//
+// R4 -- CRC32 remains ignored. BDB-NIDX-SPEC-001 revision 0.2 DEC-007 keeps
+// the historical CRC32 fields as carried bytes that are never
+// calculated, validated or used to classify corruption, so a walk over
+// a directory whose CRC32 slots hold values no writer computed returns
+// exactly the declared documents.
+func TestNativeStoredDocumentWalkIgnoresReservedCRC32(t *testing.T) {
+ tester := require.New(t)
+
+ rewritten := copyIndexDir(t, nidx01cSourceADir)
+ fillReservedCRC32(t, rewritten, []byte{0x01, 0x23, 0x45, 0x67})
+
+ documents, err := walkNIDX01C(context.Background(), rewritten)
+ tester.NoError(err)
+
+ declared := liveDeclaredDocuments(nidx01cSourceADocuments)
+ tester.Len(documents, len(declared))
+ for position, expected := range declared {
+ tester.Equal(sortedFields(declaredFieldsOf(expected)), sortedFields(documents[position]))
+ }
+}
+
+// TestNativeStoredDocumentWalkLeavesDirectoryUnchanged inventories the corpus
+// directory before and after a full walk.
+//
+// Requirement proved here:
+//
+// R7 -- the walk writes no bytes. Every entry's name, size, mode,
+// modification time and content hash is unchanged afterwards, which is
+// what lets a source read run against a directory another process owns.
+func TestNativeStoredDocumentWalkLeavesDirectoryUnchanged(t *testing.T) {
+ tester := require.New(t)
+
+ before := dirInventory(t, nidx01cSourceADir)
+ documents, err := walkNIDX01C(context.Background(), nidx01cSourceADir)
+ tester.NoError(err)
+ tester.Len(documents, len(liveDeclaredDocuments(nidx01cSourceADocuments)))
+ tester.Equal(before, dirInventory(t, nidx01cSourceADir),
+ "a read-only walk must leave every file's bytes, mode and modification time alone")
+}
+
+// TestNativeStoredDocumentWalkBesideOpenWriter walks a directory a live writer
+// holds open.
+//
+// Requirement proved here:
+//
+// R7 -- the walk takes no exclusive directory lock, so a migration source
+// read can inspect a segment index whose writer is still open. The
+// writer's own runtime lock file sits outside the on-disk grammar and
+// is ignored rather than mistaken for a generation.
+func TestNativeStoredDocumentWalkBesideOpenWriter(t *testing.T) {
+ tester := require.New(t)
+
+ shared := copyIndexDir(t, nidx01cSourceADir)
+ writer, err := NewStore(StoreOpts{Path: shared})
+ tester.NoError(err)
+ defer func() {
+ tester.NoError(writer.Close())
+ }()
+
+ documents, walkErr := walkNIDX01C(context.Background(), shared)
+ tester.NoError(walkErr)
+ tester.Len(documents, len(liveDeclaredDocuments(nidx01cSourceADocuments)),
+ "an open writer must not change what the pinned committed generation holds")
+}
+
+// TestNativeStoredDocumentWalkNoCommittedGeneration walks a directory that has
+// never been flushed.
+//
+// Requirement proved here:
+//
+// R7 -- a source directory holding no committed generation is classified as
+// absent, not damaged. A fresh segment whose index writer never
+// received a document is an ordinary state during migration, and the
+// source readers treat it as an empty source rather than aborting.
+func TestNativeStoredDocumentWalkNoCommittedGeneration(t *testing.T) {
+ tester := require.New(t)
+
+ visited := 0
+ err := nidx01cBoundary(context.Background(), t.TempDir(), func(_ StoredDocument) error {
+ visited++
+ return nil
+ })
+
+ tester.ErrorIs(err, ErrNoCommittedIndex)
+ tester.NotErrorIs(err, ErrCorruptIndex,
+ "an unflushed index is not a damaged one; callers classify the two differently")
+ tester.Zero(visited)
+}
+
+// TestNativeStoredDocumentWalkHistoricalCorpusStillReads walks the corpora the
+// preceding read-only milestones checked in, which were written by an older
+// shape of the store and hold no tag, timestamp or version field at all.
+//
+// The NIDX-01B corpus is also this milestone's crash case: its newest
+// generation references a segment the directory does not hold, which is what a
+// directory looks like when a process died between publishing a segment and
+// committing the manifest that adopts it. A walk must fall back to the newest
+// structurally complete generation and expose exactly that generation's live
+// documents.
+//
+// Requirement proved here:
+//
+// R7 -- historical directories still read. The NIDX-01A corpus yields its two
+// declared documents; the NIDX-01B corpus yields the four live
+// documents of its complete generation, with the deleted document
+// absent and the incomplete newer generation's document invisible.
+func TestNativeStoredDocumentWalkHistoricalCorpusStillReads(t *testing.T) {
+ tester := require.New(t)
+
+ firstCorpus, err := walkNIDX01C(context.Background(), nidx01aIndexDir)
+ tester.NoError(err)
+ tester.Equal(identityHexOf(nidx01aDocIDs), walkedIdentities(firstCorpus),
+ "the NIDX-01A corpus must still expose the documents its manifest declares")
+
+ secondCorpus, err := walkNIDX01C(context.Background(), nidx01bIndexDir)
+ tester.NoError(err)
+ visibleDocIDs := []uint64{21, 23, 24, 25}
+ tester.Equal(identityHexOf(visibleDocIDs), walkedIdentities(secondCorpus),
+ "the NIDX-01B corpus must expose the live documents of its newest complete generation: "+
+ "doc-%d is deleted and doc-%d belongs to the incomplete newer generation",
+ nidx01bDeletedDocID, nidx01bPublishedDocID)
+}
+
+// walkedIdentities returns the identity value of each walked document, in walk
+// order, sorted so the comparison does not pin the order segments are laid out
+// in.
+func walkedIdentities(documents [][]walkedField) []string {
+ identities := make([]string, 0, len(documents))
+ for _, document := range documents {
+ identities = append(identities, valuesOf(document, docIDField)...)
+ }
+ sort.Strings(identities)
+ return identities
+}
+
+// identityHexOf renders the identities the element-shaped historical corpora
+// record for a set of numeric document identifiers, sorted to match
+// walkedIdentities.
+func identityHexOf(docIDs []uint64) []string {
+ identities := make([]string, 0, len(docIDs))
+ for _, docID := range docIDs {
+ identities = append(identities, hex.EncodeToString(convert.Uint64ToBytes(docID)))
+ }
+ sort.Strings(identities)
+ return identities
+}
+
+// TestNativeStoredDocumentWalkBoundarySurface guards the boundary itself rather
+// than any behavior behind it.
+//
+// Requirement proved here:
+//
+// R6 -- the milestone is delivered entirely behind
+// inverted.ReadOnlyWalkDocuments and the StoredDocument it hands out.
+// The corpus lives where the corpus of every preceding read-only
+// milestone lives, the two sentinels callers classify with remain the
+// ones the boundary already publishes, and the private native reader
+// exports no operation beyond opening a committed generation, counting
+// it and walking its live documents. An entry appearing there for
+// dictionaries, term postings, doc values, sorting, search-after or any
+// writer is the milestone growing surface NIDX-01 explicitly denied it.
+func TestNativeStoredDocumentWalkBoundarySurface(t *testing.T) {
+ tester := require.New(t)
+
+ tester.NotNil(nidx01cBoundary, "ReadOnlyWalkDocuments must satisfy the live document walker contract")
+ tester.ErrorIs(ErrCorruptIndex, ErrCorruptIndex)
+ tester.ErrorIs(ErrNoCommittedIndex, ErrNoCommittedIndex)
+
+ for _, source := range []string{nidx01cSourceADir, nidx01cSourceBDir} {
+ tester.NotEmpty(dirInventory(t, source), "the NIDX-01C corpus source %s must be checked in", source)
+ }
+ tester.NotEmpty(dirInventory(t, filepath.Dir(nidx01cManifest)))
+
+ tester.Equal(nativeReaderSurface, exportedSurfaceOf(t, nativeReaderDir),
+ "the native reader's exported surface changed; NIDX-01C may only add the live document walk")
+}
diff --git a/pkg/index/inverted/testdata/nidx01c/provenance.json b/pkg/index/inverted/testdata/nidx01c/provenance.json
new file mode 100644
index 000000000..7849ea10d
--- /dev/null
+++ b/pkg/index/inverted/testdata/nidx01c/provenance.json
@@ -0,0 +1,90 @@
+{
+ "oracle": {
+ "go_mod_sha256": "b7faaa8e1fcef81607b38a6fa593b54381e11fa466f496e4cb10d4599b657ac1",
+ "writer": "pkg/index/inverted.NewStore"
+ },
+ "file_sha256": {
+ "sourceA/000000000002.seg": "ddaf8424ce113ca0fcda05f3365344803a672ba6cce26289be929b0225c6c6fa",
+ "sourceA/000000000003.snp": "59578576f4a18f77c82747b338ecf7ba441bea5cff3c40e6832158cae257bd6a",
+ "sourceB/000000000001.snp": "e457dc4d4410e458d8f427874fea937ea5002eb57d5eb5ea51cf2efb6f4e7e7c",
+ "sourceB/000000000002.seg": "92acfc30a432174549c6f370f182df21d38d22f9c810419dba6ad7dd4640e512"
+ },
+ "reserved_crc32": {
+ ".seg": "deadbeef (arbitrary, no writer computed it)",
+ ".snp": "as written by the oracle; the compatibility loader validates this slot, so the corpus keeps it readable and the reserved-field contract is proved by run-time mutation"
+ },
+ "generator_command": "GENERATE_NIDX01C_FIXTURE=1 go test ./pkg/index/inverted/ -run TestGenerateNIDX01CFixture -count=1",
+ "notes": "sourceA holds documents 101, 202 and a deleted 303 in one committed generation, so two are visible and the declared walk is 101 then 202. sourceB repeats 101 alone for the series-union case. Document labels are issue #14010's names, not series identifiers: a series identifier is a hash of the marshaled entity buffer and cannot be chosen by a fixture.",
+ "sources": [
+ {
+ "directory": "sourceA/000000000003.snp",
+ "documents": [
+ {
+ "label": "101",
+ "identity_hex": "010203",
+ "tag_name": "color",
+ "tag_values": [
+ "blue",
+ "green"
+ ],
+ "timestamp_stored_hex": "2001000000000000000064",
+ "version_stored_hex": "8000000000000002",
+ "timestamp": 100,
+ "version": 2,
+ "deleted": false
+ },
+ {
+ "label": "202",
+ "identity_hex": "040506",
+ "tag_name": "color",
+ "tag_values": [
+ "red"
+ ],
+ "timestamp_stored_hex": "2001000000000000000148",
+ "version_stored_hex": "8000000000000001",
+ "timestamp": 200,
+ "version": 1,
+ "deleted": false
+ },
+ {
+ "label": "303",
+ "identity_hex": "070809",
+ "tag_name": "color",
+ "tag_values": [
+ "gray"
+ ],
+ "timestamp_stored_hex": "200100000000000000022c",
+ "version_stored_hex": "8000000000000003",
+ "timestamp": 300,
+ "version": 3,
+ "deleted": true
+ }
+ ],
+ "physical_documents": 3,
+ "deletions": 1,
+ "visible_count": 2
+ },
+ {
+ "directory": "sourceB/000000000001.snp",
+ "documents": [
+ {
+ "label": "101",
+ "identity_hex": "010203",
+ "tag_name": "color",
+ "tag_values": [
+ "blue",
+ "green"
+ ],
+ "timestamp_stored_hex": "2001000000000000000064",
+ "version_stored_hex": "8000000000000002",
+ "timestamp": 100,
+ "version": 2,
+ "deleted": false
+ }
+ ],
+ "physical_documents": 1,
+ "deletions": 0,
+ "visible_count": 1
+ }
+ ]
+}
diff --git a/pkg/index/inverted/testdata/nidx01c/sourceA/000000000002.seg b/pkg/index/inverted/testdata/nidx01c/sourceA/000000000002.seg
new file mode 100644
index 000000000..87ae6c00a
Binary files /dev/null and b/pkg/index/inverted/testdata/nidx01c/sourceA/000000000002.seg differ
diff --git a/pkg/index/inverted/testdata/nidx01c/sourceA/000000000003.snp b/pkg/index/inverted/testdata/nidx01c/sourceA/000000000003.snp
new file mode 100644
index 000000000..97a958594
Binary files /dev/null and b/pkg/index/inverted/testdata/nidx01c/sourceA/000000000003.snp differ
diff --git a/pkg/index/inverted/testdata/nidx01c/sourceB/000000000001.snp b/pkg/index/inverted/testdata/nidx01c/sourceB/000000000001.snp
new file mode 100644
index 000000000..345389a67
Binary files /dev/null and b/pkg/index/inverted/testdata/nidx01c/sourceB/000000000001.snp differ
diff --git a/pkg/index/inverted/testdata/nidx01c/sourceB/000000000002.seg b/pkg/index/inverted/testdata/nidx01c/sourceB/000000000002.seg
new file mode 100644
index 000000000..e2b0a6392
Binary files /dev/null and b/pkg/index/inverted/testdata/nidx01c/sourceB/000000000002.seg differ