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/11 05:23:32 UTC
(skywalking-banyandb) branch main updated: feat(index): page repair tuples natively and cut Property repair over to them (#1335)
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 31b32ca25 feat(index): page repair tuples natively and cut Property repair over to them (#1335)
31b32ca25 is described below
commit 31b32ca2510a4c62e9fb2c9f9417a5805037b876
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Fri Sep 11 13:23:26 2026 +0800
feat(index): page repair tuples natively and cut Property repair over to them (#1335)
---
.../schema/schemaserver/repair_service_test.go | 2 +-
banyand/property/db/e2e_native_repair_test.go | 407 +++++++++++++
banyand/property/db/repair.go | 237 ++++----
banyand/property/db/repair_test.go | 44 +-
pkg/index/inverted/internal/nativeice/nativeice.go | 41 +-
.../inverted/internal/nativeice/repair_page.go | 673 +++++++++++++++++++++
.../internal/nativeice/repair_page_test.go | 79 +++
pkg/index/inverted/native_fixture_nidx01e_test.go | 342 +++++++++++
pkg/index/inverted/native_nidx01b_boundary_test.go | 32 +-
pkg/index/inverted/native_nidx01c_walk_test.go | 2 +-
.../inverted/native_nidx01d_selection_test.go | 2 +-
.../inverted/native_nidx01e_tuple_page_test.go | 663 ++++++++++++++++++++
pkg/index/inverted/repair_page.go | 170 ++++++
pkg/index/inverted/repair_page_tdd_test.go | 135 +++++
.../inverted/testdata/nidx01e/provenance.json | 103 ++++
.../testdata/nidx01e/shard-0/000000000002.seg | Bin 0 -> 2508 bytes
.../testdata/nidx01e/shard-0/000000000003.snp | Bin 0 -> 66 bytes
17 files changed, 2760 insertions(+), 172 deletions(-)
diff --git a/banyand/metadata/schema/schemaserver/repair_service_test.go b/banyand/metadata/schema/schemaserver/repair_service_test.go
index 68d671aff..f2793ddd1 100644
--- a/banyand/metadata/schema/schemaserver/repair_service_test.go
+++ b/banyand/metadata/schema/schemaserver/repair_service_test.go
@@ -283,7 +283,7 @@ func writeProperty(t *testing.T, d db.Database, name, id string, version int64)
func ensureShard(t *testing.T, nodes []*testNode) {
t.Helper()
for _, n := range nodes {
- writeProperty(t, n.srv.db, "seed", "seed-init", 0)
+ writeProperty(t, n.srv.db, "seed", "seed-init", 1)
}
}
diff --git a/banyand/property/db/e2e_native_repair_test.go b/banyand/property/db/e2e_native_repair_test.go
new file mode 100644
index 000000000..723542752
--- /dev/null
+++ b/banyand/property/db/e2e_native_repair_test.go
@@ -0,0 +1,407 @@
+// 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 db
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "go/ast"
+ "go/parser"
+ "go/token"
+ "os"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/index/inverted"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+const (
+ // nidx01eCorpus is the checked-in Property shard issue #14012 declares, and
+ // nidx01eProvenanceFile is the manifest recording what its bytes hold. The
+ // corpus lives beside the reader that pages it because the reader's own
+ // boundary suite reads the same bytes; this package reads them through the
+ // repair build instead.
+ nidx01eCorpus = "../../../pkg/index/inverted/testdata/nidx01e/shard-0"
+ nidx01eProvenanceFile = "../../../pkg/index/inverted/testdata/nidx01e/provenance.json"
+
+ // nidx01ePageSize is the page size issue #14012 declares the end-to-end
+ // repair build runs at, and nidx01eExpectedPages is how many calls it takes
+ // to exhaust the corpus's four visible rows at that size: two full pages and
+ // one empty page that ends the order.
+ nidx01ePageSize = 2
+ nidx01eExpectedPages = 3
+
+ // nidx01eTreeSlotCount is the slot count the repair tree is composed over,
+ // matching the count the Property repair suite already builds trees at.
+ nidx01eTreeSlotCount = 32
+
+ // repairSourceFile is the file whose read path this leaf cuts over.
+ repairSourceFile = "repair.go"
+
+ // retiredReaderEntryPoint and retiredSearcherEntryPoint are the calls that
+ // open a retired third-party index reader and build its ranked searcher.
+ // Naming the calls rather than the module keeps this assertion free of the
+ // dependency's own name.
+ retiredReaderEntryPoint = "OpenReader"
+ retiredSearcherEntryPoint = "NewTopNSearch"
+
+ banyanDBModulePrefix = "github.com/apache/skywalking-banyandb/"
+)
+
+// nidx01eDeclaredLeaves is the completed repair tree issue #14012 declares for
+// the corpus: one leaf per entity, carrying that entity's newest revision's SHA.
+// The corpus holds two revisions of e-1 and one deleted revision between them,
+// so a build that collapsed revisions the wrong way, or that let a deleted
+// revision win, produces a different SHA here.
+var nidx01eDeclaredLeaves = map[string]string{
+ "g-a/n-a/e-1": "sha-b",
+ "g-a/n-b/e-2": "sha-c",
+ "g-b/n-a/e-3": "sha-d",
+}
+
+// repairExternalImports is the whole third-party dependency budget of the
+// Property repair build once its reads are native: a clock it schedules
+// against, a hash its tree slots are assigned with, an error wrapper, a cron
+// parser, an error combiner and the RPC runtime its gossip server registers on.
+//
+// None of them reads an index. An entry appearing here for a search engine, a
+// segment format or a query language is the repair build keeping a retired
+// reader that BDB-NIDX-SPEC-001 revision 0.2 NIDX-01 names for replacement.
+var repairExternalImports = map[string]struct{}{
+ "github.com/benbjohnson/clock": {},
+ "github.com/cespare/xxhash/v2": {},
+ "github.com/pkg/errors": {},
+ "github.com/robfig/cron/v3": {},
+ "go.uber.org/multierr": {},
+ "google.golang.org/grpc": {},
+}
+
+// TestE2EPropertyRepairNativeBuildsTheDeclaredTree runs one complete Property
+// repair build over the checked-in shard at the declared page size and reads
+// back the tree and generation state it persisted.
+//
+// This is the milestone's real use case end to end: an operator's repair
+// schedule hands a committed shard snapshot to the build, and the build has to
+// produce the Merkle tree its peers compare against over gossip. The tree's
+// leaves are compared against issue #14012's declaration, not against anything
+// recomputed from the corpus.
+//
+// Requirements proved here:
+//
+// R3 -- the existing repair build pages natively. It reads the shard through
+// the pinned generation, in ascending (group, name, entity, revision)
+// order, projecting the stored SHA, at the page size it is configured
+// with, and it produces exactly the three declared entity leaves with
+// the declared SHA values.
+// R4 -- the generation state the build records names the generation it
+// actually paged, so a later build cannot skip a generation this one
+// never read.
+func TestE2EPropertyRepairNativeBuildsTheDeclaredTree(t *testing.T) {
+ tester := require.New(t)
+ shard := nidx01eShardCopy(t)
+ repairState, observed := nidx01eRepair(t, shard, nil)
+
+ tester.NoError(repairState.buildStatus(context.Background(), shard))
+
+ tester.Equal(nidx01eExpectedPages, len(observed.requests),
+ "a four-row shard at page size %d must be exhausted in %d native pages", nidx01ePageSize, nidx01eExpectedPages)
+ for pageIndex, request := range observed.requests {
+ tester.Equal(nidx01ePageSize, request.PageSize, "page %d must honor the configured page size", pageIndex)
+ }
+ tester.Empty(observed.requests[0].After, "the first page must start the order rather than resume into it")
+ tester.NotNil(observed.requests[1].After,
+ "a later page must resume after a complete prior tuple")
+
+ tester.Equal(nidx01eDeclaredLeaves, nidx01eTreeLeaves(t, repairState))
+
+ state, err := repairState.readState()
+ tester.NoError(err)
+ tester.NotNil(state)
+ tester.Equal(nidx01eDeclaredSnapshotID(t), state.LastSnpID)
+}
+
+// TestE2EPropertyRepairNativeIgnoresAGenerationPublishedBetweenPages publishes
+// a Property revision into the shard while a repair build is between its first
+// and second page.
+//
+// A repair build runs for as long as a shard is large, and the Property writer
+// does not stop while it does. A build that saw part of one generation and part
+// of the next would hash a tree that never existed on disk, and its peers would
+// repair against it.
+//
+// Requirements proved here:
+//
+// R4 -- a generation published after the build pinned its own does not enter
+// a later page. The tree holds exactly the pinned generation's declared
+// leaves, and the generation state records the pinned generation rather
+// than the one published underneath it.
+func TestE2EPropertyRepairNativeIgnoresAGenerationPublishedBetweenPages(t *testing.T) {
+ tester := require.New(t)
+ shard := nidx01eShardCopy(t)
+ published := false
+ repairState, observed := nidx01eRepair(t, shard, func(pageIndex int, _ []inverted.RepairRow, _ error) error {
+ if pageIndex != 1 || published {
+ return nil
+ }
+ published = true
+ nidx01ePublishRevision(t, shard, "g-a", "n-b", "e-4", 6, "sha-published")
+ return nil
+ })
+
+ tester.NoError(repairState.buildStatus(context.Background(), shard))
+ tester.True(published, "the build must read more than one page, or the pin is untested")
+ tester.Greater(len(observed.requests), 1)
+
+ tester.Equal(nidx01eDeclaredLeaves, nidx01eTreeLeaves(t, repairState),
+ "a revision published between two pages must not reach the tree the build was already composing")
+
+ state, err := repairState.readState()
+ tester.NoError(err)
+ tester.NotNil(state)
+ tester.Equal(nidx01eDeclaredSnapshotID(t), state.LastSnpID,
+ "the build must record the generation it paged, not the one published underneath it")
+
+ newest, err := inverted.OpenReadOnlyGeneration(shard)
+ tester.NoError(err)
+ defer func() {
+ tester.NoError(newest.Close())
+ }()
+ tester.NotEqual(nidx01eDeclaredSnapshotID(t), newest.SnapshotID(),
+ "the publication must have committed a newer generation, or nothing was raced")
+}
+
+// TestE2EPropertyRepairNativePublishesNoStateWhenAPageFails fails the build's
+// second page and inspects what it left behind.
+//
+// Requirement proved here:
+//
+// R4 -- a page that fails part way through a build publishes no repair state.
+// Neither the tree nor the generation marker appears, so the next build
+// redoes the work instead of trusting a tree that covers half a shard.
+func TestE2EPropertyRepairNativePublishesNoStateWhenAPageFails(t *testing.T) {
+ tester := require.New(t)
+ shard := nidx01eShardCopy(t)
+ failure := errors.New("page failed")
+ repairState, observed := nidx01eRepair(t, shard, func(pageIndex int, _ []inverted.RepairRow, _ error) error {
+ if pageIndex < 2 {
+ return nil
+ }
+ return failure
+ })
+
+ buildErr := repairState.buildStatus(context.Background(), shard)
+ tester.ErrorIs(buildErr, failure)
+ tester.Greater(len(observed.requests), 1, "the build must have reached the failing page")
+
+ _, statErr := os.Stat(repairState.statePath)
+ tester.True(os.IsNotExist(statErr), "a failed build must record no generation state: %v", statErr)
+ _, treeErr := os.Stat(repairState.composeTreeFilePath)
+ tester.True(os.IsNotExist(treeErr), "a failed build must publish no repair tree: %v", treeErr)
+}
+
+// TestE2EPropertyRepairNativeKeepsNoRetiredReader parses the repair build's own
+// source.
+//
+// Producing the right tree is not enough on its own: the point of the milestone
+// is that the repair read stops going through the retired reader, and a tree
+// that is right for the wrong reason would hide that.
+//
+// Requirement proved here:
+//
+// R7 -- the Property repair build opens no retired index reader and builds no
+// retired ranked searcher. Its third-party dependency set holds nothing
+// that reads an index, so the only way it reaches a shard's bytes is the
+// BanyanDB-owned pinned generation.
+func TestE2EPropertyRepairNativeKeepsNoRetiredReader(t *testing.T) {
+ tester := require.New(t)
+ fileSet := token.NewFileSet()
+ file, err := parser.ParseFile(fileSet, repairSourceFile, nil, parser.SkipObjectResolution)
+ tester.NoError(err)
+
+ for _, spec := range file.Imports {
+ importPath, unquoteErr := strconv.Unquote(spec.Path.Value)
+ tester.NoError(unquoteErr)
+ if isStandardImport(importPath) || strings.HasPrefix(importPath, banyanDBModulePrefix) {
+ continue
+ }
+ _, budgeted := repairExternalImports[importPath]
+ tester.True(budgeted, "%s imports %s, which is outside the repair build's dependency budget",
+ repairSourceFile, importPath)
+ }
+
+ for _, called := range selectorCallsIn(file) {
+ tester.NotEqual(retiredReaderEntryPoint, called, "%s still opens a retired index reader", repairSourceFile)
+ tester.NotEqual(retiredSearcherEntryPoint, called, "%s still builds a retired ranked searcher", repairSourceFile)
+ }
+}
+
+// nidx01eShardCopy copies the checked-in corpus into a writable shard directory
+// so a test may publish over it without touching the pinned bytes.
+func nidx01eShardCopy(t *testing.T) string {
+ t.Helper()
+ shard := filepath.Join(t.TempDir(), "shard-0")
+ require.NoError(t, os.MkdirAll(shard, 0o755))
+ require.NoError(t, copyDirRecursive(nidx01eCorpus, shard))
+ return shard
+}
+
+// nidx01eRepair builds a repair state over a shard directory at the declared
+// page size, and reports every page its build asks the pinned generation for.
+// The optional hook runs after each page, which is where a test publishes a
+// competing generation or fails the build part way through.
+func nidx01eRepair(t *testing.T, shard string, hook nidx01ePageHook) (*repair, *nidx01eObservedGeneration) {
+ t.Helper()
+ factory := observability.BypassRegistry.With(observability.RootScope.SubScope("nidx01e"))
+ repairState := newRepair(shard, t.TempDir(), logger.GetLogger("nidx01e"), factory,
+ nidx01ePageSize, nidx01eTreeSlotCount, &repairScheduler{})
+ observed := &nidx01eObservedGeneration{hook: hook}
+ repairState.openGeneration = func(shardPath string) (repairGeneration, error) {
+ generation, err := openNativeRepairGeneration(shardPath)
+ if err != nil {
+ return nil, err
+ }
+ observed.repairGeneration = generation
+ return observed, nil
+ }
+ return repairState, observed
+}
+
+// nidx01ePageHook observes one page a build read, and may fail the build.
+type nidx01ePageHook func(pageIndex int, rows []inverted.RepairRow, pageErr error) error
+
+// nidx01eObservedGeneration wraps the pinned generation a build pages so a test
+// can see which pages the build asked for and act between them.
+type nidx01eObservedGeneration struct {
+ repairGeneration
+ hook nidx01ePageHook
+ requests []inverted.RepairPageRequest
+}
+
+func (g *nidx01eObservedGeneration) RepairTuplePage(
+ ctx context.Context, request inverted.RepairPageRequest,
+) ([]inverted.RepairRow, error) {
+ rows, err := g.repairGeneration.RepairTuplePage(ctx, request)
+ g.requests = append(g.requests, request)
+ if g.hook == nil {
+ return rows, err
+ }
+ if hookErr := g.hook(len(g.requests), rows, err); hookErr != nil {
+ return nil, hookErr
+ }
+ return rows, err
+}
+
+// nidx01eTreeLeaves reads the repair tree a build persisted and returns its
+// leaves as entity to SHA.
+func nidx01eTreeLeaves(t *testing.T, repairState *repair) map[string]string {
+ t.Helper()
+ tree := newRepairData(repairState, nil).readTree(t, defaultGroupName)
+ require.NotNil(t, tree, "the build must have persisted a repair tree")
+ leaves := map[string]string{}
+ for _, slot := range tree.root.children {
+ for _, leaf := range slot.children {
+ _, duplicate := leaves[leaf.id]
+ require.False(t, duplicate, "entity %s appears in the tree more than once", leaf.id)
+ leaves[leaf.id] = leaf.shaValue
+ }
+ }
+ return leaves
+}
+
+// nidx01ePublishRevision commits one more Property revision into a shard
+// directory, which publishes a new generation over it. The revision is shaped
+// the way the Property writer shapes one, so a build that read it would gain a
+// leaf for it.
+func nidx01ePublishRevision(t *testing.T, shard, group, name, entity string, revision int64, sha string) {
+ t.Helper()
+ writer, err := inverted.NewStore(inverted.StoreOpts{Path: shard})
+ require.NoError(t, err)
+ document := index.Document{
+ EntityValues: []byte(group + "/" + name + "/" + entity + "/" + strconv.FormatInt(revision, 10)),
+ Timestamp: revision,
+ Fields: []index.Field{
+ nidx01eIndexedField(entityID, entity),
+ nidx01eIndexedField(groupField, group),
+ nidx01eIndexedField(nameField, name),
+ nidx01eStoredField(sourceField, "source-"+entity),
+ nidx01eStoredField(shaValueField, sha),
+ },
+ }
+ require.NoError(t, writer.UpdateSeriesBatch(index.Batch{Documents: index.Documents{document}}))
+ require.NoError(t, writer.Close())
+}
+
+func nidx01eIndexedField(name, value string) index.Field {
+ field := index.NewBytesField(index.FieldKey{TagName: name}, []byte(value))
+ field.Index = true
+ return field
+}
+
+func nidx01eStoredField(name, value string) index.Field {
+ field := index.NewBytesField(index.FieldKey{TagName: name}, []byte(value))
+ field.Store = true
+ field.NoSort = true
+ return field
+}
+
+// nidx01eDeclaredSnapshotID reads the generation the corpus commits out of the
+// manifest checked in beside its bytes, so the expected value comes from the
+// fixture's own declaration rather than from anything the build computes.
+func nidx01eDeclaredSnapshotID(t *testing.T) uint64 {
+ t.Helper()
+ raw, err := os.ReadFile(nidx01eProvenanceFile)
+ require.NoError(t, err)
+ var manifest struct {
+ SnapshotID uint64 `json:"snapshot_id"`
+ }
+ require.NoError(t, json.Unmarshal(raw, &manifest))
+ require.NotZero(t, manifest.SnapshotID, "the corpus manifest must declare the generation its bytes commit")
+ return manifest.SnapshotID
+}
+
+// isStandardImport reports whether an import path names a standard library
+// package, which has no dot in its first path element.
+func isStandardImport(importPath string) bool {
+ return !strings.Contains(strings.SplitN(importPath, "/", 2)[0], ".")
+}
+
+// selectorCallsIn lists the names of every package-qualified call the file
+// makes, so a call the build must no longer contain can be named without naming
+// the package it came from.
+func selectorCallsIn(file *ast.File) []string {
+ var called []string
+ ast.Inspect(file, func(node ast.Node) bool {
+ call, isCall := node.(*ast.CallExpr)
+ if !isCall {
+ return true
+ }
+ if selector, isSelector := call.Fun.(*ast.SelectorExpr); isSelector {
+ called = append(called, selector.Sel.Name)
+ }
+ return true
+ })
+ return called
+}
diff --git a/banyand/property/db/repair.go b/banyand/property/db/repair.go
index e079c239a..d9bf732bb 100644
--- a/banyand/property/db/repair.go
+++ b/banyand/property/db/repair.go
@@ -29,7 +29,6 @@ import (
"os"
"path"
"path/filepath"
- "sort"
"strconv"
"strings"
"sync"
@@ -37,8 +36,6 @@ import (
"time"
"github.com/benbjohnson/clock"
- "github.com/blugelabs/bluge"
- "github.com/blugelabs/bluge/index"
"github.com/cespare/xxhash/v2"
"github.com/pkg/errors"
"github.com/robfig/cron/v3"
@@ -52,6 +49,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/property/gossip"
"github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/encoding"
+ "github.com/apache/skywalking-banyandb/pkg/index/inverted"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/meter"
"github.com/apache/skywalking-banyandb/pkg/run"
@@ -62,10 +60,45 @@ const (
repairBatchSearchSize = 100
)
+// repairGeneration is the pinned committed generation one repair build pages
+// through. Pinning is what makes a build reproducible: every page a build reads
+// comes from the same generation, so a generation published while the build
+// runs contributes no leaf to the tree the build writes, and the generation
+// state the build records names the generation it actually read.
+//
+// Its production implementation is *inverted.ReadOnlyGeneration.
+type repairGeneration interface {
+ // RepairTuplePage returns one bounded page of the pinned generation's
+ // repair rows, ordered ascending and resuming strictly after the request's
+ // cursor.
+ RepairTuplePage(ctx context.Context, request inverted.RepairPageRequest) ([]inverted.RepairRow, error)
+ // SnapshotID identifies the pinned generation.
+ SnapshotID() uint64
+ // Close releases the files the pinned generation holds.
+ Close() error
+}
+
+// repairGenerationOpener pins the committed generation of a shard directory
+// that one repair build reads. It is the seam a build's source of pages is
+// chosen at, so a build can be pointed at an instrumented or a retained
+// generation without changing the build itself.
+type repairGenerationOpener func(shardPath string) (repairGeneration, error)
+
+// openNativeRepairGeneration pins the newest committed generation of a shard
+// directory through BanyanDB's own read-only reader.
+func openNativeRepairGeneration(shardPath string) (repairGeneration, error) {
+ generation, err := inverted.OpenReadOnlyGeneration(shardPath)
+ if err != nil {
+ return nil, err
+ }
+ return generation, nil
+}
+
type repair struct {
metrics *repairMetrics
l *logger.Logger
scheduler *repairScheduler
+ openGeneration repairGenerationOpener
shardPath string
repairBasePath string
snapshotDir string
@@ -97,26 +130,27 @@ func newRepair(
metrics: newRepairMetrics(metricsFactory),
scheduler: scheduler,
treeSlotCount: treeSlotCount,
+ openGeneration: openNativeRepairGeneration,
}
}
func (r *repair) checkHasUpdates() (bool, error) {
- state, err := r.readState()
- if err != nil {
- return false, fmt.Errorf("reading state failure: %w", err)
- }
- indexConfig := index.DefaultConfig(r.shardPath)
- items, err := indexConfig.DirectoryFunc().List(index.ItemKindSnapshot)
- if err != nil {
- return false, fmt.Errorf("reading item kind snapshot failure: %w", err)
+ state, stateErr := r.readState()
+ if stateErr != nil {
+ return false, fmt.Errorf("reading state failure: %w", stateErr)
}
- sort.Sort(snapshotIDList(items))
- // check the snapshot ID have any updated
- // if no updates, the building Trees should be skipped
- if state != nil && len(items) != 0 && items[len(items)-1] == state.LastSnpID {
+ generation, generationErr := r.openGeneration(r.shardPath)
+ if errors.Is(generationErr, inverted.ErrNoCommittedIndex) {
return false, nil
}
- if len(items) == 0 {
+ if generationErr != nil {
+ return false, fmt.Errorf("opening pinned generation failure: %w", generationErr)
+ }
+ snapshotID := generation.SnapshotID()
+ if closeErr := generation.Close(); closeErr != nil {
+ return false, fmt.Errorf("closing pinned generation failure: %w", closeErr)
+ }
+ if state != nil && snapshotID == state.LastSnpID {
return false, nil
}
return true, nil
@@ -132,95 +166,79 @@ func (r *repair) buildStatus(ctx context.Context, snapshotPath string) (err erro
}
r.metrics.totalBuildTreeDuration.Inc(time.Since(startTime).Seconds())
}()
- indexConfig := index.DefaultConfig(snapshotPath)
- items, err := indexConfig.DirectoryFunc().List(index.ItemKindSnapshot)
- if err != nil {
- return fmt.Errorf("reading item kind segment failure: %w", err)
+ generation, openErr := r.openGeneration(snapshotPath)
+ if errors.Is(openErr, inverted.ErrNoCommittedIndex) {
+ return nil
}
- sort.Sort(snapshotIDList(items))
-
- blugeConf := bluge.DefaultConfig(snapshotPath)
- err = r.buildTree(ctx, blugeConf)
- if err != nil {
- return fmt.Errorf("building trees failure: %w", err)
+ if openErr != nil {
+ return fmt.Errorf("opening pinned generation failure: %w", openErr)
}
-
- var latestSnapshotID uint64
- if len(items) > 0 {
- latestSnapshotID = items[len(items)-1]
+ defer func() {
+ if closeErr := generation.Close(); err == nil && closeErr != nil {
+ err = fmt.Errorf("closing pinned generation failure: %w", closeErr)
+ }
+ }()
+ if buildErr := r.buildTree(ctx, generation); buildErr != nil {
+ return fmt.Errorf("building trees failure: %w", buildErr)
}
- // save the Trees to the state
+
state := &repairStatus{
- LastSnpID: latestSnapshotID,
+ LastSnpID: generation.SnapshotID(),
LastSyncTime: time.Now(),
}
- stateVal, err := json.Marshal(state)
- if err != nil {
- return fmt.Errorf("marshall state failure: %w", err)
+ stateVal, marshalErr := json.Marshal(state)
+ if marshalErr != nil {
+ return fmt.Errorf("marshall state failure: %w", marshalErr)
}
- if err = os.MkdirAll(filepath.Dir(r.statePath), storage.DirPerm); err != nil {
- return fmt.Errorf("creating state directory failure: %w", err)
+ if mkdirErr := os.MkdirAll(filepath.Dir(r.statePath), storage.DirPerm); mkdirErr != nil {
+ return fmt.Errorf("creating state directory failure: %w", mkdirErr)
}
- err = os.WriteFile(r.statePath, stateVal, storage.FilePerm)
- if err != nil {
- return fmt.Errorf("writing state file failure: %w", err)
+ if writeErr := os.WriteFile(r.statePath, stateVal, storage.FilePerm); writeErr != nil {
+ return fmt.Errorf("writing state file failure: %w", writeErr)
}
return nil
}
-func (r *repair) buildTree(ctx context.Context, conf bluge.Config) error {
- reader, err := bluge.OpenReader(conf)
- if err != nil {
- // means no data found
- if strings.Contains(err.Error(), "unable to find a usable snapshot") {
- return nil
- }
- return fmt.Errorf("opening index reader failure: %w", err)
- }
- defer func() {
- _ = reader.Close()
- }()
- query := bluge.Query(bluge.NewMatchAllQuery())
- topNSearch := bluge.NewTopNSearch(r.batchSearchSize, query)
- topNSearch.SortBy([]string{
- fmt.Sprintf("+%s", groupField),
- fmt.Sprintf("+%s", nameField),
- fmt.Sprintf("+%s", entityID),
- fmt.Sprintf("+%s", timestampField),
- })
-
+func (r *repair) buildTree(ctx context.Context, generation repairGeneration) error {
var latestProperty *searchingProperty
treeComposer := newRepairTreeComposer(r.composeSlotAppendFilePath, r.composeTreeFilePath, r.treeSlotCount, r.l)
- err = r.pageSearch(ctx, reader, topNSearch, func(sortValue [][]byte, shaValue string) error {
- if len(sortValue) != 4 {
- return fmt.Errorf("unexpected sort value length: %d", len(sortValue))
+ var after *inverted.RepairCursor
+ for {
+ rows, pageErr := generation.RepairTuplePage(ctx, inverted.RepairPageRequest{
+ After: after,
+ PageSize: r.batchSearchSize,
+ })
+ if pageErr != nil {
+ return fmt.Errorf("paging pinned generation failure: %w", pageErr)
+ }
+ if len(rows) == 0 {
+ break
}
- groupName := convert.BytesToString(sortValue[0])
- name := convert.BytesToString(sortValue[1])
- entity := r.buildLeafNodeEntity(groupName, name, convert.BytesToString(sortValue[2]))
-
- s := newSearchingProperty(groupName, shaValue, entity)
- if latestProperty != nil && latestProperty.entityID != entity {
- if appendErr := treeComposer.append(latestProperty.entityID, latestProperty.shaValue); appendErr != nil {
- return fmt.Errorf("appending property to tree composer failure: %w", appendErr)
+ for rowIndex, row := range rows {
+ if len(row.SortValues) != inverted.RepairSortFieldCount {
+ return fmt.Errorf("unexpected sort value length at row %d: %d", rowIndex, len(row.SortValues))
+ }
+ groupName := convert.BytesToString(row.SortValues[0])
+ name := convert.BytesToString(row.SortValues[1])
+ entity := r.buildLeafNodeEntity(groupName, name, convert.BytesToString(row.SortValues[2]))
+ property := newSearchingProperty(groupName, convert.BytesToString(row.Value), entity)
+ if latestProperty != nil && latestProperty.entityID != entity {
+ if appendErr := treeComposer.append(latestProperty.entityID, latestProperty.shaValue); appendErr != nil {
+ return fmt.Errorf("appending property to tree composer failure: %w", appendErr)
+ }
}
+ latestProperty = property
}
- latestProperty = s
- return nil
- })
- if err != nil {
- return err
+ after = rows[len(rows)-1].Cursor
}
- // if the latestProperty is not nil, it means the latest property need to be saved
if latestProperty != nil {
- if err = treeComposer.append(latestProperty.entityID, latestProperty.shaValue); err != nil {
- return fmt.Errorf("appending latest property to tree composer failure: %w", err)
+ if appendErr := treeComposer.append(latestProperty.entityID, latestProperty.shaValue); appendErr != nil {
+ return fmt.Errorf("appending latest property to tree composer failure: %w", appendErr)
}
- if err = treeComposer.composeAndSave(); err != nil {
- return fmt.Errorf("composing tree failure: %w", err)
+ if composeErr := treeComposer.composeAndSave(); composeErr != nil {
+ return fmt.Errorf("composing tree failure: %w", composeErr)
}
}
-
return nil
}
@@ -236,53 +254,6 @@ func (r *repair) parseLeafNodeEntity(entity string) (string, string, string, err
return parts[0], parts[1], parts[2], nil
}
-func (r *repair) pageSearch(
- ctx context.Context,
- reader *bluge.Reader,
- searcher *bluge.TopNSearch,
- each func(sortValue [][]byte, shaValue string) error,
-) error {
- var latestDocValues [][]byte
- for {
- searcher.After(latestDocValues)
- result, err := reader.Search(ctx, searcher)
- if err != nil {
- return fmt.Errorf("searching index failure: %w", err)
- }
-
- next, err := result.Next()
- var hitNumber int
- if err != nil {
- return errors.WithMessage(err, "iterate document match iterator")
- }
- // if next is nil, it means no more documents to process
- if next == nil {
- return nil
- }
- var shaValue string
- for err == nil && next != nil {
- hitNumber = next.HitNumber
- var errTime error
- err = next.VisitStoredFields(func(field string, value []byte) bool {
- if field == shaValueField {
- shaValue = convert.BytesToString(value)
- return false
- }
- return true
- })
- if err = multierr.Combine(err, errTime); err != nil {
- return errors.WithMessagef(err, "visit stored fields, hit: %d", hitNumber)
- }
- err = each(next.SortValue, shaValue)
- if err != nil {
- return errors.WithMessagef(err, "processing source failure, hit: %d", hitNumber)
- }
- latestDocValues = next.SortValue
- next, err = result.Next()
- }
- }
-}
-
func (r *repair) buildShaValue(source []byte, deleteTime int64) (string, error) {
var err error
hash := sha512.New()
@@ -644,12 +615,6 @@ type repairTreeFooter struct {
rootNodeLen int64
}
-type snapshotIDList []uint64
-
-func (s snapshotIDList) Len() int { return len(s) }
-func (s snapshotIDList) Less(i, j int) bool { return s[i] < s[j] }
-func (s snapshotIDList) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
-
type repairTreeComposer struct {
l *logger.Logger
appendFileFmt string
diff --git a/banyand/property/db/repair_test.go b/banyand/property/db/repair_test.go
index 3376c0323..37fbf78b3 100644
--- a/banyand/property/db/repair_test.go
+++ b/banyand/property/db/repair_test.go
@@ -19,6 +19,7 @@ package db
import (
"context"
+ "errors"
"fmt"
"io"
"os"
@@ -35,6 +36,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/index/inverted"
"github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
)
@@ -69,12 +71,12 @@ func TestBuildTree(t *testing.T) {
name: "single property",
existingDoc: func(s *shard) ([]index.Document, error) {
return buildPropertyDocuments(s,
- propertyBuilder{id: "test1"},
+ propertyBuilder{id: "test1", version: 1},
)
},
statusVerify: func(t *testing.T, s *shard, data *repairData) {
basicStatusVerify(t, data, defaultGroupName, 1)
- verifyContainsProperty(t, s, data, defaultGroupName, propertyBuilder{id: "test1"})
+ verifyContainsProperty(t, s, data, defaultGroupName, propertyBuilder{id: "test1", version: 1})
},
},
{
@@ -99,26 +101,26 @@ func TestBuildTree(t *testing.T) {
propertyBuilder{id: "test1", version: 4, value: 2},
propertyBuilder{id: "test2", version: 1, value: 1},
propertyBuilder{id: "test2", version: 2, value: 2},
- propertyBuilder{id: "test3"},
+ propertyBuilder{id: "test3", version: 1},
)
},
statusVerify: func(t *testing.T, s *shard, data *repairData) {
basicStatusVerify(t, data, defaultGroupName, 3)
verifyContainsProperty(t, s, data, defaultGroupName, propertyBuilder{id: "test1", version: 4, value: 2})
verifyContainsProperty(t, s, data, defaultGroupName, propertyBuilder{id: "test2", version: 2, value: 2})
- verifyContainsProperty(t, s, data, defaultGroupName, propertyBuilder{id: "test3"})
+ verifyContainsProperty(t, s, data, defaultGroupName, propertyBuilder{id: "test3", version: 1})
},
},
{
name: "build multiple times",
existingDoc: func(s *shard) ([]index.Document, error) {
return buildPropertyDocuments(s,
- propertyBuilder{id: "test1"},
+ propertyBuilder{id: "test1", version: 1},
)
},
statusVerify: func(t *testing.T, s *shard, data *repairData) {
basicStatusVerify(t, data, defaultGroupName, 1)
- verifyContainsProperty(t, s, data, defaultGroupName, propertyBuilder{id: "test1"})
+ verifyContainsProperty(t, s, data, defaultGroupName, propertyBuilder{id: "test1", version: 1})
},
nextStatusVerify: func(t *testing.T, _ *shard, before, after *repairData) {
basicStatusVerify(t, before, defaultGroupName, 1)
@@ -133,23 +135,23 @@ func TestBuildTree(t *testing.T) {
name: "build multiple times with new properties",
existingDoc: func(s *shard) ([]index.Document, error) {
return buildPropertyDocuments(s,
- propertyBuilder{id: "test1"},
+ propertyBuilder{id: "test1", version: 1},
)
},
statusVerify: func(t *testing.T, s *shard, data *repairData) {
basicStatusVerify(t, data, defaultGroupName, 1)
- verifyContainsProperty(t, s, data, defaultGroupName, propertyBuilder{id: "test1"})
+ verifyContainsProperty(t, s, data, defaultGroupName, propertyBuilder{id: "test1", version: 1})
},
afterFirstBuild: func(s *shard) ([]index.Document, error) {
return buildPropertyDocuments(s,
- propertyBuilder{id: "test2"},
+ propertyBuilder{id: "test2", version: 1},
)
},
nextStatusVerify: func(t *testing.T, s *shard, before, after *repairData) {
basicStatusVerify(t, after, defaultGroupName, 2)
verifyContainsProperty(t, s, after, defaultGroupName,
- propertyBuilder{id: "test1"},
- propertyBuilder{id: "test2"},
+ propertyBuilder{id: "test1", version: 1},
+ propertyBuilder{id: "test2", version: 1},
)
if before.LastSnpID == after.LastSnpID {
t.Fatalf("expected last snapshot ID to be incremented by 1, got before: %d, after: %d",
@@ -318,7 +320,7 @@ func TestDocumentUpdatesNotify(t *testing.T) {
t.Fatal(err)
}
- p1 := buildProperties(propertyBuilder{id: "1"})
+ p1 := buildProperties(propertyBuilder{id: "1", version: 1})
err = newShard.update(GetPropertyID(p1), p1)
if err != nil {
t.Fatalf("failed to update property: %v", err)
@@ -653,3 +655,21 @@ func copyDirRecursive(srcDir, dstDir string) error {
return copyFile(path, dstPath)
})
}
+
+// TestRepairRejectsZeroRevision verifies that a missing timestamp doc value is
+// rejected without publishing a repair tree or generation marker.
+func TestRepairRejectsZeroRevision(t *testing.T) {
+ shardPath := t.TempDir()
+ nidx01ePublishRevision(t, shardPath, "group", "name", "entity", 0, "sha-zero")
+ repairState, _ := nidx01eRepair(t, shardPath, nil)
+ buildErr := repairState.buildStatus(context.Background(), shardPath)
+ if !errors.Is(buildErr, inverted.ErrCorruptIndex) {
+ t.Fatalf("expected corrupt repair input for a zero revision, got %v", buildErr)
+ }
+ for _, unpublishedPath := range []string{repairState.statePath, repairState.composeTreeFilePath} {
+ _, statErr := os.Stat(unpublishedPath)
+ if !os.IsNotExist(statErr) {
+ t.Fatalf("failed repair must not publish %s: %v", unpublishedPath, statErr)
+ }
+ }
+}
diff --git a/pkg/index/inverted/internal/nativeice/nativeice.go b/pkg/index/inverted/internal/nativeice/nativeice.go
index ec26a02df..6f99332e3 100644
--- a/pkg/index/inverted/internal/nativeice/nativeice.go
+++ b/pkg/index/inverted/internal/nativeice/nativeice.go
@@ -85,10 +85,14 @@ 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 {
- closeErr error
- segments []pinnedSegment
- visibleDocCount int64
- closeOnce sync.Once
+ closeErr error
+ repairPageSortFields [repairSortFieldCount]string
+ segments []pinnedSegment
+ repairPageReaders []*repairSegmentPageReader
+ snapshotID uint64
+ visibleDocCount int64
+ repairPageMu sync.Mutex
+ closeOnce sync.Once
}
// StoredDocument is one live document of the pinned generation, borrowed for
@@ -178,7 +182,11 @@ func openWithSnapshots(path string, listSnapshots snapshotLister) (*Reader, erro
lastCandidatePath = snapshotPath
continue
}
- return &Reader{visibleDocCount: visibleDocCount, segments: segments}, nil
+ snapshotID, validID := parseFinalName(filepath.Base(snapshotPath), ".snp")
+ if !validID {
+ return nil, corruptError("snapshot %q has an invalid identifier", snapshotPath)
+ }
+ return &Reader{snapshotID: snapshotID, visibleDocCount: visibleDocCount, segments: segments}, nil
}
}
if lastCandidateErr != nil {
@@ -194,10 +202,17 @@ func (r *Reader) VisibleDocCount() (int64, error) {
return r.visibleDocCount, nil
}
+// SnapshotID returns the identifier of the committed generation the Reader pins.
+func (r *Reader) SnapshotID() uint64 {
+ return r.snapshotID
+}
+
// Close releases the file handles the Reader pinned at Open. It is idempotent.
// Callers must not close a Reader concurrently with a document visit.
func (r *Reader) Close() error {
r.closeOnce.Do(func() {
+ r.repairPageReaders = nil
+ r.repairPageSortFields = [repairSortFieldCount]string{}
r.closeErr = closePinnedSegments(r.segments)
})
return r.closeErr
@@ -527,12 +542,16 @@ func readSegmentFooter(file *os.File, size uint64, path string) (segmentFooter,
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 {
+ storedIndexEnd := footer.docValueOffset
+ if storedIndexEnd == math.MaxUint64 {
+ storedIndexEnd = footer.fieldsIndexOffset
+ }
+ if footer.chunkMode == 0 || footer.storedIndexOffset > storedIndexEnd ||
+ storedIndexEnd > 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 {
+ footer.documentCount > (storedIndexEnd-footer.storedIndexOffset)/storedDocumentOffsetByteWidth {
return segmentFooter{}, corruptError("segment %q has invalid document count", path)
}
if (footer.footerOffset-footer.fieldsIndexOffset)%fieldsIndexAddressByteWidth != 0 {
@@ -902,8 +921,12 @@ func (s *storedSegmentReader) documentOffset(documentNumber uint64) (uint64, err
if documentNumber >= s.footer.documentCount {
return 0, corruptError("segment %q has an out-of-range stored document number", s.path)
}
+ storedIndexEnd := s.footer.docValueOffset
+ if storedIndexEnd == math.MaxUint64 {
+ storedIndexEnd = s.footer.fieldsIndexOffset
+ }
indexOffset := s.footer.storedIndexOffset + documentNumber*storedDocumentOffsetByteWidth
- if indexOffset > s.footer.docValueOffset-storedDocumentOffsetByteWidth {
+ if indexOffset > storedIndexEnd-storedDocumentOffsetByteWidth {
return 0, corruptError("segment %q has a stored document offset outside its index", s.path)
}
var offsetData [storedDocumentOffsetByteWidth]byte
diff --git a/pkg/index/inverted/internal/nativeice/repair_page.go b/pkg/index/inverted/internal/nativeice/repair_page.go
new file mode 100644
index 000000000..80df70cff
--- /dev/null
+++ b/pkg/index/inverted/internal/nativeice/repair_page.go
@@ -0,0 +1,673 @@
+// 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 nativeice
+
+import (
+ "bytes"
+ "container/heap"
+ "context"
+ "encoding/binary"
+ "errors"
+ "fmt"
+ "math"
+ "os"
+ "sort"
+
+ roaringpkg "github.com/RoaringBitmap/roaring"
+)
+
+const (
+ repairSortFieldCount = 4
+ maxRepairPageSize = 1 << 16
+ maxRepairSortValueLength = 64 << 10
+ docValueDocumentsPerChunk = 1024
+ docValueChunkTableFooterLength = 16
+ maxDocValueChunkCount = 1 << 20
+ maxDocValueCompressedChunkSize = 16 << 20
+ maxDocValueDecodedChunkSize = 64 << 20
+)
+
+// ErrInvalidRepairPage reports a bounded repair page request that the native
+// reader cannot serve.
+var ErrInvalidRepairPage = errors.New("nativeice: invalid repair page request")
+
+// RepairCursor identifies a row in the pinned generation, including tuple ties.
+type RepairCursor struct {
+ SortValues [][]byte
+ SegmentID uint64
+ DocumentNumber uint64
+}
+
+// RepairPageRequest describes one bounded repair page.
+type RepairPageRequest struct {
+ After *RepairCursor
+ SortFields [repairSortFieldCount]string
+ ProjectField string
+ PageSize int
+}
+
+// RepairTupleRow is one bounded repair row returned by Reader.RepairTuplePage.
+type RepairTupleRow struct {
+ SortValues [][]byte
+ Value []byte
+ Cursor RepairCursor
+}
+
+// RepairTuplePage returns a bounded page from the Reader's pinned generation.
+func (r *Reader) RepairTuplePage(ctx context.Context, request RepairPageRequest) ([]RepairTupleRow, error) {
+ if requestErr := validateRepairPageRequest(request); requestErr != nil {
+ return nil, requestErr
+ }
+
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ return nil, ctxErr
+ }
+
+ r.repairPageMu.Lock()
+ defer r.repairPageMu.Unlock()
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ return nil, ctxErr
+ }
+ segmentReaders, readersErr := r.repairPageReadersFor(request.SortFields)
+ if readersErr != nil {
+ return nil, readersErr
+ }
+
+ candidates := make(repairTupleHeap, 0, request.PageSize)
+ heap.Init(&candidates)
+ for segmentIndex, segmentReader := range segmentReaders {
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ return nil, ctxErr
+ }
+ documentCount := r.segments[segmentIndex].record.documentCount
+ for documentNumber := uint64(0); documentNumber < documentCount; documentNumber++ {
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ return nil, ctxErr
+ }
+ if documentNumber <= math.MaxUint32 && segmentReader.deleted.Contains(uint32(documentNumber)) {
+ continue
+ }
+ candidate, candidateErr := segmentReader.candidate(documentNumber, r.segments[segmentIndex].record.id)
+ if candidateErr != nil {
+ return nil, candidateErr
+ }
+ if compareRepairCandidateCursor(candidate, request.After) <= 0 {
+ continue
+ }
+ if len(candidates) == request.PageSize && compareRepairCandidates(candidate, candidates[0]) >= 0 {
+ continue
+ }
+ stableCandidate := repairTupleCandidate{}
+ if len(candidates) == request.PageSize {
+ stableCandidate = heap.Pop(&candidates).(repairTupleCandidate)
+ }
+ copyRepairTupleCandidate(&stableCandidate, candidate)
+ projectedValue, projectedErr := segmentReader.projectedValue(documentNumber, request.ProjectField, stableCandidate.value)
+ if projectedErr != nil {
+ return nil, projectedErr
+ }
+ stableCandidate.value = projectedValue
+ heap.Push(&candidates, stableCandidate)
+ }
+ }
+
+ sort.Slice(candidates, func(leftIndex, rightIndex int) bool {
+ return compareRepairCandidates(candidates[leftIndex], candidates[rightIndex]) < 0
+ })
+ rows := make([]RepairTupleRow, len(candidates))
+ for rowIndex, candidate := range candidates {
+ sortValues := make([][]byte, repairSortFieldCount)
+ copy(sortValues, candidate.sortValues[:])
+ rows[rowIndex] = RepairTupleRow{
+ SortValues: sortValues, Value: candidate.value,
+ Cursor: RepairCursor{SortValues: sortValues, SegmentID: candidate.segmentID, DocumentNumber: candidate.documentNumber},
+ }
+ }
+ return rows, nil
+}
+
+func (r *Reader) repairPageReadersFor(
+ sortFields [repairSortFieldCount]string,
+) ([]*repairSegmentPageReader, error) {
+ if r.repairPageReaders != nil && r.repairPageSortFields == sortFields {
+ return r.repairPageReaders, nil
+ }
+ readers := make([]*repairSegmentPageReader, len(r.segments))
+ for segmentIndex := range r.segments {
+ segmentReader, readerErr := newRepairSegmentPageReader(r.segments[segmentIndex], sortFields)
+ if readerErr != nil {
+ return nil, readerErr
+ }
+ readers[segmentIndex] = segmentReader
+ }
+ r.repairPageReaders = readers
+ r.repairPageSortFields = sortFields
+ return readers, nil
+}
+
+func validateRepairPageRequest(request RepairPageRequest) error {
+ if request.PageSize <= 0 || request.PageSize > maxRepairPageSize {
+ return repairPageRequestError("page size %d is outside the allowed range", request.PageSize)
+ }
+ if request.ProjectField == "" {
+ return repairPageRequestError("projection field is empty")
+ }
+ if request.After != nil && len(request.After.SortValues) != repairSortFieldCount {
+ return repairPageRequestError("cursor has %d components", len(request.After.SortValues))
+ }
+ for fieldIndex, fieldName := range request.SortFields {
+ if fieldName == "" {
+ return repairPageRequestError("sort field %d is empty", fieldIndex)
+ }
+ for previousIndex := 0; previousIndex < fieldIndex; previousIndex++ {
+ if fieldName == request.SortFields[previousIndex] {
+ return repairPageRequestError("sort field %q is repeated", fieldName)
+ }
+ }
+ }
+ if request.After == nil {
+ return nil
+ }
+ for componentIndex, component := range request.After.SortValues {
+ if component == nil {
+ return repairPageRequestError("cursor component %d is missing", componentIndex)
+ }
+ if len(component) > maxRepairSortValueLength {
+ return repairPageRequestError("cursor component %d exceeds %d bytes", componentIndex, maxRepairSortValueLength)
+ }
+ }
+ return nil
+}
+
+func repairPageRequestError(format string, arguments ...any) error {
+ return fmt.Errorf("nativeice: invalid repair page: "+format+": %w", append(arguments, ErrInvalidRepairPage)...)
+}
+
+type repairTupleCandidate struct {
+ sortValues [repairSortFieldCount][]byte
+ value []byte
+ documentNumber uint64
+ segmentID uint64
+}
+
+func copyRepairTupleCandidate(destination *repairTupleCandidate, source repairTupleCandidate) {
+ destination.documentNumber = source.documentNumber
+ destination.segmentID = source.segmentID
+ for valueIndex, value := range source.sortValues {
+ destination.sortValues[valueIndex] = copyRepairPageValueInto(destination.sortValues[valueIndex], value)
+ }
+}
+
+func compareRepairCandidates(left, right repairTupleCandidate) int {
+ for valueIndex := range left.sortValues {
+ if comparison := compareRepairValues(left.sortValues[valueIndex], right.sortValues[valueIndex]); comparison != 0 {
+ return comparison
+ }
+ }
+ switch {
+ case left.segmentID < right.segmentID:
+ return -1
+ case left.segmentID > right.segmentID:
+ return 1
+ case left.documentNumber < right.documentNumber:
+ return -1
+ case left.documentNumber > right.documentNumber:
+ return 1
+ default:
+ return 0
+ }
+}
+
+func compareRepairCandidateCursor(candidate repairTupleCandidate, cursor *RepairCursor) int {
+ if cursor == nil {
+ return 1
+ }
+ cursorCandidate := repairTupleCandidate{segmentID: cursor.SegmentID, documentNumber: cursor.DocumentNumber}
+ copy(cursorCandidate.sortValues[:], cursor.SortValues)
+ return compareRepairCandidates(candidate, cursorCandidate)
+}
+
+func compareRepairValues(left, right []byte) int {
+ switch {
+ case left == nil && right == nil:
+ return 0
+ case left == nil:
+ return 1
+ case right == nil:
+ return -1
+ default:
+ return bytes.Compare(left, right)
+ }
+}
+
+type repairTupleHeap []repairTupleCandidate
+
+func (h repairTupleHeap) Len() int {
+ return len(h)
+}
+
+func (h repairTupleHeap) Less(leftIndex, rightIndex int) bool {
+ return compareRepairCandidates(h[leftIndex], h[rightIndex]) > 0
+}
+
+func (h repairTupleHeap) Swap(leftIndex, rightIndex int) {
+ h[leftIndex], h[rightIndex] = h[rightIndex], h[leftIndex]
+}
+
+func (h *repairTupleHeap) Push(value any) {
+ *h = append(*h, value.(repairTupleCandidate))
+}
+
+func (h *repairTupleHeap) Pop() any {
+ current := *h
+ lastIndex := len(current) - 1
+ value := current[lastIndex]
+ *h = current[:lastIndex]
+ return value
+}
+
+type repairSegmentPageReader struct {
+ deleted *roaringpkg.Bitmap
+ sortReaders [repairSortFieldCount]*repairDocValueReader
+ stored *storedSegmentReader
+ storedChunk []byte
+ storedChunkNumber uint64
+}
+
+func newRepairSegmentPageReader(
+ segment pinnedSegment,
+ sortFields [repairSortFieldCount]string,
+) (*repairSegmentPageReader, error) {
+ storedReader, readerErr := newStoredSegmentReader(segment.file, segment.size, segment.record)
+ if readerErr != nil {
+ return nil, readerErr
+ }
+ deleted, deletionErr := deletedDocuments(segment.record)
+ if deletionErr != nil {
+ return nil, deletionErr
+ }
+ pageReader := &repairSegmentPageReader{
+ deleted: deleted,
+ stored: storedReader,
+ storedChunkNumber: math.MaxUint64,
+ }
+ if segment.record.documentCount == 0 || storedReader.footer.docValueOffset == math.MaxUint64 {
+ return pageReader, nil
+ }
+
+ locationOffset := storedReader.footer.docValueOffset
+ for fieldIndex, fieldName := range storedReader.fieldNames {
+ fieldStart, startErr := storedReader.readUvarint(&locationOffset, storedReader.footer.fieldsIndexOffset)
+ if startErr != nil {
+ return nil, startErr
+ }
+ fieldEnd, endErr := storedReader.readUvarint(&locationOffset, storedReader.footer.fieldsIndexOffset)
+ if endErr != nil {
+ return nil, endErr
+ }
+ if fieldStart == math.MaxUint64 || fieldEnd == math.MaxUint64 {
+ if fieldStart != math.MaxUint64 || fieldEnd != math.MaxUint64 {
+ return nil, corruptError("segment %q has an incomplete doc-value location for field %d", storedReader.path, fieldIndex)
+ }
+ continue
+ }
+ fieldReader, fieldErr := newRepairDocValueReader(storedReader, fieldStart, fieldEnd)
+ if fieldErr != nil {
+ return nil, fieldErr
+ }
+ for sortIndex, sortField := range sortFields {
+ if sortField == fieldName {
+ pageReader.sortReaders[sortIndex] = fieldReader
+ }
+ }
+ }
+ return pageReader, nil
+}
+
+func (r *repairSegmentPageReader) candidate(documentNumber, segmentID uint64) (repairTupleCandidate, error) {
+ candidate := repairTupleCandidate{documentNumber: documentNumber, segmentID: segmentID}
+ for sortIndex, fieldReader := range r.sortReaders {
+ if fieldReader == nil {
+ return repairTupleCandidate{}, corruptError("segment %d document %d is missing sort field %d", segmentID, documentNumber, sortIndex)
+ }
+ value, valueErr := fieldReader.smallestValue(documentNumber)
+ if valueErr != nil {
+ return repairTupleCandidate{}, valueErr
+ }
+ if value == nil {
+ return repairTupleCandidate{}, corruptError("segment %d document %d is missing sort value %d", segmentID, documentNumber, sortIndex)
+ }
+ candidate.sortValues[sortIndex] = value
+ }
+ return candidate, nil
+}
+
+func (r *repairSegmentPageReader) projectedValue(documentNumber uint64, projectField string, destination []byte) ([]byte, error) {
+ chunkNumber := documentNumber / storedDocumentsPerChunk
+ if r.storedChunkNumber != chunkNumber {
+ chunk, chunkErr := r.stored.loadChunk(chunkNumber)
+ if chunkErr != nil {
+ return nil, chunkErr
+ }
+ r.storedChunk = chunk
+ r.storedChunkNumber = chunkNumber
+ }
+ document, documentErr := r.stored.decodeDocument(documentNumber, r.storedChunk)
+ if documentErr != nil {
+ return nil, documentErr
+ }
+ for _, field := range document.fields {
+ if field.name == projectField {
+ return copyRepairPageValueInto(destination, field.value), nil
+ }
+ }
+ return nil, nil
+}
+
+type repairDocValueReader struct {
+ path string
+ file *os.File
+ chunkOffsets []uint64
+ compressedBuffer []byte
+ decodedBuffer []byte
+ header []repairDocValueMeta
+ chunkNumber uint64
+ fieldDataOffset uint64
+ size uint64
+ totalDocumentCount uint64
+}
+
+type repairDocValueMeta struct {
+ documentNumber uint64
+ valueEnd uint64
+}
+
+func newRepairDocValueReader(storedReader *storedSegmentReader, fieldStart, fieldEnd uint64) (*repairDocValueReader, error) {
+ if storedReader.footer.documentCount > math.MaxUint64/storedDocumentOffsetByteWidth {
+ return nil, corruptError("segment %q has an oversized stored document index", storedReader.path)
+ }
+ minimumDataOffset := storedReader.footer.storedIndexOffset + storedReader.footer.documentCount*storedDocumentOffsetByteWidth
+ if minimumDataOffset < storedReader.footer.storedIndexOffset || fieldStart < minimumDataOffset ||
+ fieldStart >= fieldEnd || fieldEnd > storedReader.footer.docValueOffset {
+ return nil, corruptError("segment %q has invalid doc-value bounds", storedReader.path)
+ }
+ if fieldEnd-fieldStart <= docValueChunkTableFooterLength {
+ return nil, corruptError("segment %q has a truncated doc-value chunk table", storedReader.path)
+ }
+
+ var tableFooter [docValueChunkTableFooterLength]byte
+ if readErr := storedReader.readInto(fieldEnd-docValueChunkTableFooterLength, tableFooter[:]); readErr != nil {
+ return nil, readErr
+ }
+ chunkOffsetsLength := binary.BigEndian.Uint64(tableFooter[0:8])
+ chunkCount := binary.BigEndian.Uint64(tableFooter[8:16])
+ expectedChunkCount := (storedReader.footer.documentCount + docValueDocumentsPerChunk - 1) / docValueDocumentsPerChunk
+ if chunkCount == 0 || chunkCount != expectedChunkCount || chunkCount > maxDocValueChunkCount ||
+ chunkOffsetsLength < chunkCount || chunkOffsetsLength > chunkCount*binary.MaxVarintLen64 {
+ return nil, corruptError("segment %q has an invalid doc-value chunk table", storedReader.path)
+ }
+ tableStart := fieldEnd - docValueChunkTableFooterLength
+ if chunkOffsetsLength > tableStart-fieldStart {
+ return nil, corruptError("segment %q has a doc-value chunk table outside its field data", storedReader.path)
+ }
+ tableStart -= chunkOffsetsLength
+ table, tableErr := storedReader.readBytes(tableStart, chunkOffsetsLength)
+ if tableErr != nil {
+ return nil, tableErr
+ }
+ decoder := byteDecoder{payload: table}
+ offsets := make([]uint64, int(chunkCount))
+ for offsetIndex := range offsets {
+ offset, offsetErr := decoder.uvarint()
+ if offsetErr != nil {
+ return nil, offsetErr
+ }
+ offsets[offsetIndex] = offset
+ }
+ if decoder.remaining() != 0 {
+ return nil, corruptError("segment %q has trailing bytes in a doc-value chunk table", storedReader.path)
+ }
+ fieldDataLength := tableStart - fieldStart
+ for offsetIndex, offset := range offsets {
+ if offset > fieldDataLength || (offsetIndex > 0 && offset < offsets[offsetIndex-1]) {
+ return nil, corruptError("segment %q has invalid doc-value chunk offsets", storedReader.path)
+ }
+ }
+ return &repairDocValueReader{
+ chunkNumber: math.MaxUint64,
+ chunkOffsets: offsets,
+ fieldDataOffset: fieldStart,
+ path: storedReader.path,
+ file: storedReader.file,
+ size: storedReader.size,
+ totalDocumentCount: storedReader.footer.documentCount,
+ }, nil
+}
+
+func (r *repairDocValueReader) smallestValue(documentNumber uint64) ([]byte, error) {
+ chunkNumber := documentNumber / docValueDocumentsPerChunk
+ if r.chunkNumber != chunkNumber {
+ if chunkErr := r.loadChunk(chunkNumber); chunkErr != nil {
+ return nil, chunkErr
+ }
+ }
+ headerIndex := sort.Search(len(r.header), func(index int) bool {
+ return r.header[index].documentNumber >= documentNumber
+ })
+ if headerIndex == len(r.header) || r.header[headerIndex].documentNumber != documentNumber {
+ return nil, nil
+ }
+ start := uint64(0)
+ if headerIndex > 0 {
+ start = r.header[headerIndex-1].valueEnd
+ }
+ end := r.header[headerIndex].valueEnd
+ if end < start || end > uint64(len(r.decodedBuffer)) {
+ return nil, corruptError("segment %q has invalid doc-value bounds in a chunk", r.path)
+ }
+ return decodeRepairDocValueTerms(r.decodedBuffer[start:end], r.path)
+}
+
+func (r *repairDocValueReader) loadChunk(chunkNumber uint64) error {
+ if chunkNumber >= uint64(len(r.chunkOffsets)) {
+ return corruptError("segment %q has no doc-value chunk %d", r.path, chunkNumber)
+ }
+ start := uint64(0)
+ if chunkNumber > 0 {
+ start = r.chunkOffsets[chunkNumber-1]
+ }
+ end := r.chunkOffsets[chunkNumber]
+ if start > end {
+ return corruptError("segment %q has unordered doc-value chunk bounds", r.path)
+ }
+ if start == end {
+ r.chunkNumber = chunkNumber
+ r.header = r.header[:0]
+ r.compressedBuffer = r.compressedBuffer[:0]
+ r.decodedBuffer = r.decodedBuffer[:0]
+ return nil
+ }
+ compressedLength := end - start
+ if compressedLength > maxDocValueCompressedChunkSize {
+ return corruptError("segment %q has an oversized doc-value chunk", r.path)
+ }
+ if cap(r.compressedBuffer) < int(compressedLength) {
+ r.compressedBuffer = make([]byte, int(compressedLength))
+ } else {
+ r.compressedBuffer = r.compressedBuffer[:int(compressedLength)]
+ }
+ if readErr := readRepairDocValueBytes(r.file, r.size, r.fieldDataOffset+start, r.compressedBuffer, r.path); readErr != nil {
+ return readErr
+ }
+
+ decoder := byteDecoder{payload: r.compressedBuffer}
+ documentCount, countErr := decoder.uvarint()
+ if countErr != nil {
+ return countErr
+ }
+ if documentCount == 0 || documentCount > docValueDocumentsPerChunk {
+ return corruptError("segment %q has an invalid doc-value chunk document count", r.path)
+ }
+ if cap(r.header) < int(documentCount) {
+ r.header = make([]repairDocValueMeta, int(documentCount))
+ } else {
+ r.header = r.header[:int(documentCount)]
+ }
+ chunkFirstDocument := chunkNumber * docValueDocumentsPerChunk
+ chunkLastDocument := chunkFirstDocument + docValueDocumentsPerChunk
+ if chunkLastDocument > r.totalDocumentCount {
+ chunkLastDocument = r.totalDocumentCount
+ }
+ var previousDocument, previousValueEnd uint64
+ for headerIndex := range r.header {
+ documentDelta, documentErr := decoder.uvarint()
+ if documentErr != nil {
+ return documentErr
+ }
+ if headerIndex > 0 && documentDelta == 0 {
+ return corruptError("segment %q has repeated doc-value document numbers", r.path)
+ }
+ if previousDocument > math.MaxUint64-documentDelta {
+ return corruptError("segment %q has overflowing doc-value document numbers", r.path)
+ }
+ documentNumber := previousDocument + documentDelta
+ if documentNumber < chunkFirstDocument || documentNumber >= chunkLastDocument {
+ return corruptError("segment %q has a doc-value document outside its chunk", r.path)
+ }
+ valueDelta, valueErr := decoder.uvarint()
+ if valueErr != nil {
+ return valueErr
+ }
+ if valueDelta == 0 || previousValueEnd > math.MaxUint64-valueDelta {
+ return corruptError("segment %q has invalid doc-value value offsets", r.path)
+ }
+ valueEnd := previousValueEnd + valueDelta
+ r.header[headerIndex] = repairDocValueMeta{documentNumber: documentNumber, valueEnd: valueEnd}
+ previousDocument = documentNumber
+ previousValueEnd = valueEnd
+ }
+ compressedValues := decoder.payload[decoder.offset:]
+ decodedLength, lengthErr := storedChunkDecodedLength(compressedValues)
+ if lengthErr != nil {
+ return corruptError("decode doc-value chunk length in segment %q: %w", r.path, lengthErr)
+ }
+ if decodedLength < 0 || decodedLength > maxDocValueDecodedChunkSize {
+ return corruptError("segment %q has an oversized decoded doc-value chunk", r.path)
+ }
+ if cap(r.decodedBuffer) < decodedLength {
+ r.decodedBuffer = make([]byte, decodedLength)
+ } else {
+ r.decodedBuffer = r.decodedBuffer[:decodedLength]
+ }
+ decoded, decodeErr := decodeStoredChunk(r.decodedBuffer[:0], compressedValues)
+ if decodeErr != nil {
+ return corruptError("decode doc-value chunk in segment %q: %w", r.path, decodeErr)
+ }
+ if len(decoded) != decodedLength || previousValueEnd > uint64(len(decoded)) {
+ return corruptError("segment %q has invalid decoded doc-value offsets", r.path)
+ }
+ r.decodedBuffer = decoded
+ r.chunkNumber = chunkNumber
+ return nil
+}
+
+func readRepairDocValueBytes(file *os.File, size, offset uint64, data []byte, path string) error {
+ length := uint64(len(data))
+ if offset > size || length > size-offset {
+ return corruptError("segment %q reads outside its doc-value data", path)
+ }
+ readCount, readErr := file.ReadAt(data, int64(offset))
+ if readErr != nil || readCount != len(data) {
+ return corruptError("read doc-value data from segment %q", path, readErr)
+ }
+ return nil
+}
+
+func decodeRepairDocValueTerms(encoded []byte, path string) ([]byte, error) {
+ var smallest []byte
+ found := false
+ for len(encoded) > 0 {
+ value, remaining, valueErr := decodeRepairDocValueTerm(encoded)
+ if valueErr != nil {
+ return nil, corruptError("decode doc-value term in segment %q: %w", path, valueErr)
+ }
+ if !found || bytes.Compare(value, smallest) < 0 {
+ smallest = value
+ found = true
+ }
+ encoded = remaining
+ }
+ if !found {
+ return nil, corruptError("segment %q has an empty doc-value entry", path)
+ }
+ return smallest, nil
+}
+
+func decodeRepairDocValueTerm(encoded []byte) ([]byte, []byte, error) {
+ if len(encoded) == 0 {
+ return nil, nil, errors.New("empty term")
+ }
+ var unescaped []byte
+ for sourceIndex := 0; sourceIndex < len(encoded); sourceIndex++ {
+ switch encoded[sourceIndex] {
+ case 0xff:
+ if unescaped == nil {
+ if sourceIndex > maxRepairSortValueLength {
+ return nil, nil, fmt.Errorf("term exceeds %d bytes", maxRepairSortValueLength)
+ }
+ return encoded[:sourceIndex], encoded[sourceIndex+1:], nil
+ }
+ return unescaped, encoded[sourceIndex+1:], nil
+ case '\\':
+ if sourceIndex+1 >= len(encoded) {
+ return nil, nil, errors.New("truncated term escape")
+ }
+ if encoded[sourceIndex+1] != '\\' && encoded[sourceIndex+1] != 0xff {
+ return nil, nil, errors.New("invalid term escape")
+ }
+ if unescaped == nil {
+ if sourceIndex > maxRepairSortValueLength {
+ return nil, nil, fmt.Errorf("term exceeds %d bytes", maxRepairSortValueLength)
+ }
+ unescaped = make([]byte, sourceIndex, maxRepairSortValueLength)
+ copy(unescaped, encoded[:sourceIndex])
+ }
+ sourceIndex++
+ }
+ if unescaped != nil {
+ if len(unescaped) >= maxRepairSortValueLength {
+ return nil, nil, fmt.Errorf("term exceeds %d bytes", maxRepairSortValueLength)
+ }
+ unescaped = append(unescaped, encoded[sourceIndex])
+ }
+ }
+ return nil, nil, errors.New("unterminated term")
+}
+
+func copyRepairPageValueInto(destination, value []byte) []byte {
+ if value == nil {
+ return nil
+ }
+ if destination == nil || cap(destination) < len(value) {
+ destination = make([]byte, len(value))
+ } else {
+ destination = destination[:len(value)]
+ }
+ copy(destination, value)
+ return destination
+}
diff --git a/pkg/index/inverted/internal/nativeice/repair_page_test.go b/pkg/index/inverted/internal/nativeice/repair_page_test.go
new file mode 100644
index 000000000..7f4789aa1
--- /dev/null
+++ b/pkg/index/inverted/internal/nativeice/repair_page_test.go
@@ -0,0 +1,79 @@
+// 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 nativeice
+
+import (
+ "bytes"
+ "errors"
+ "testing"
+)
+
+func TestRepairCursorOrdersTupleTies(t *testing.T) {
+ values := [repairSortFieldCount][]byte{[]byte("group"), []byte("name"), []byte("entity"), []byte("timestamp")}
+ cursor := &RepairCursor{SortValues: values[:], SegmentID: 2, DocumentNumber: 3}
+ for _, testCase := range []struct {
+ name string
+ segmentID uint64
+ documentNumber uint64
+ want int
+ }{
+ {"same row", 2, 3, 0},
+ {"next document", 2, 4, 1},
+ {"previous document", 2, 2, -1},
+ {"next segment", 3, 0, 1},
+ {"previous segment", 1, 9, -1},
+ } {
+ t.Run(testCase.name, func(t *testing.T) {
+ candidate := repairTupleCandidate{sortValues: values, segmentID: testCase.segmentID, documentNumber: testCase.documentNumber}
+ if got := compareRepairCandidateCursor(candidate, cursor); got != testCase.want {
+ t.Fatalf("cursor comparison = %d, want %d", got, testCase.want)
+ }
+ })
+ }
+}
+
+func TestRepairCandidateRejectsMissingSortField(t *testing.T) {
+ reader := &repairSegmentPageReader{}
+ _, candidateErr := reader.candidate(0, 1)
+ if !errors.Is(candidateErr, ErrCorrupt) {
+ t.Fatalf("missing field error = %v, want ErrCorrupt", candidateErr)
+ }
+}
+
+func TestDecodeRepairDocValueTermEscapes(t *testing.T) {
+ for _, testCase := range []struct {
+ name string
+ encoded []byte
+ want []byte
+ wantError bool
+ }{
+ {"invalid escape", []byte{'\\', 'A', 0xff}, nil, true},
+ {"truncated escape", []byte{'\\'}, nil, true},
+ {"escaped backslash", []byte{'\\', '\\', 0xff}, []byte{'\\'}, false},
+ {"escaped separator", []byte{'\\', 0xff, 0xff}, []byte{0xff}, false},
+ {"empty value", []byte{0xff}, []byte{}, false},
+ } {
+ t.Run(testCase.name, func(t *testing.T) {
+ value, remaining, decodeErr := decodeRepairDocValueTerm(testCase.encoded)
+ if (decodeErr != nil) != testCase.wantError {
+ t.Fatalf("decode error = %v, want error %t", decodeErr, testCase.wantError)
+ }
+ if decodeErr == nil && (!bytes.Equal(value, testCase.want) || len(remaining) != 0) {
+ t.Fatalf("decoded value = %x, remaining = %x, want %x", value, remaining, testCase.want)
+ }
+ })
+ }
+}
diff --git a/pkg/index/inverted/native_fixture_nidx01e_test.go b/pkg/index/inverted/native_fixture_nidx01e_test.go
new file mode 100644
index 000000000..928aa2716
--- /dev/null
+++ b/pkg/index/inverted/native_fixture_nidx01e_test.go
@@ -0,0 +1,342 @@
+// 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"
+ "sort"
+ "strconv"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/pkg/index"
+)
+
+// The NIDX-01E corpus is the Property shard issue #14012 declares: one shard
+// directory whose newest committed generation holds the four visible repair
+// rows the issue lists, in the ascending order it lists them, plus one deleted
+// row that sorts between the first two so a page that ignored the deletion
+// masks would be caught at the very first page boundary.
+//
+// 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
+// nidx01eGenerateEnv.
+//
+// It is an independent fixture in the sense the milestone needs: the expected
+// rows, cursor and repair-tree leaves are issue #14012's own declaration, and
+// the bytes are an oracle's output that no reader under test produced.
+const (
+ nidx01eRoot = "testdata/nidx01e"
+ nidx01eShardDir = nidx01eRoot + "/shard-0"
+ nidx01eManifest = nidx01eRoot + "/provenance.json"
+
+ // nidx01eGenerateEnv gates regeneration so an ordinary test run can never
+ // overwrite the checked-in bytes the contract is pinned to.
+ nidx01eGenerateEnv = "GENERATE_NIDX01E_FIXTURE"
+ nidx01eGenerateCmd = "GENERATE_NIDX01E_FIXTURE=1 go test ./pkg/index/inverted/ -run TestGenerateNIDX01EFixture -count=1"
+
+ // The field names below are the Property shard's own, pinned here as
+ // literals rather than imported: pkg/index/inverted sits underneath
+ // banyand/property/db, so the corpus declares the names it was written with
+ // instead of borrowing the constants the consumer happens to hold.
+ nidx01eGroupField = "_group"
+ nidx01eNameField = index.IndexModeName
+ nidx01eEntityField = "_entity_id"
+ nidx01eSourceField = "_source"
+ nidx01eSHAField = "_sha_value"
+
+ // nidx01eSnapshotID is the identifier of the committed generation the
+ // corpus's newest snapshot manifest names. It is a property of the
+ // checked-in bytes, recorded when they were produced.
+ nidx01eSnapshotID = uint64(3)
+
+ // nidx01eVisibleRowCount is how many rows of the corpus survive the pinned
+ // generation's deletion masks: the four the issue declares.
+ nidx01eVisibleRowCount = int64(4)
+)
+
+// nidx01eRow is one physical document of the corpus: the four ascending sort
+// components the repair order is built from, the stored SHA a repair leaf
+// carries, an unrelated stored payload that must stay unread, and whether the
+// row is deleted in the pinned generation.
+type nidx01eRow struct {
+ group string
+ name string
+ entityID string
+ sha string
+ source string
+ timestamp int64
+ deleted bool
+}
+
+// nidx01eRows is the corpus issue #14012 declares, in insertion order. The four
+// visible rows are the issue's rows 1 to 4; row "e-1@15" is the extra deleted
+// row, placed between visible rows 1 and 2 in the ascending order so its
+// exclusion is observable inside the first page rather than only at the end.
+var nidx01eRows = []nidx01eRow{
+ {group: "g-a", name: "n-a", entityID: "e-1", timestamp: 10, sha: "sha-a", source: "source-a"},
+ {group: "g-a", name: "n-a", entityID: "e-1", timestamp: 15, sha: "sha-x", source: "source-x", deleted: true},
+ {group: "g-a", name: "n-a", entityID: "e-1", timestamp: 20, sha: "sha-b", source: "source-b"},
+ {group: "g-a", name: "n-b", entityID: "e-2", timestamp: 5, sha: "sha-c", source: "source-c"},
+ {group: "g-b", name: "n-a", entityID: "e-3", timestamp: 7, sha: "sha-d", source: "source-d"},
+}
+
+// nidx01eProvenance is the manifest checked in beside the corpus bytes. It
+// records which oracle produced them, how to re-derive them, what every row
+// declares, which row is deleted, which generation the bytes commit, and what
+// every file hashes to.
+type nidx01eProvenance struct {
+ Oracle map[string]string `json:"oracle"`
+ FileSHA256 map[string]string `json:"file_sha256"`
+ GeneratorCommand string `json:"generator_command"`
+ Notes string `json:"notes"`
+ Rows []nidx01eProvenanceRow `json:"rows"`
+ DeclaredPages [][]nidx01eProvenanceRow `json:"declared_pages"`
+ SnapshotID uint64 `json:"snapshot_id"`
+ VisibleRowCount int64 `json:"visible_row_count"`
+}
+
+// nidx01eProvenanceRow records one physical document of the corpus so a
+// reviewer can compare it against issue #14012 without decoding a segment.
+type nidx01eProvenanceRow struct {
+ Group string `json:"group"`
+ Name string `json:"name"`
+ EntityID string `json:"entity_id"`
+ SHA string `json:"sha_value"`
+ DocID string `json:"doc_id"`
+ Timestamp int64 `json:"timestamp"`
+ Deleted bool `json:"deleted"`
+}
+
+// TestGenerateNIDX01EFixture rebuilds the checked-in NIDX-01E corpus and its
+// provenance manifest with BanyanDB's compatibility writer. It runs only when
+// nidx01eGenerateEnv is set, so a normal test run reads the committed bytes
+// instead of producing fresh ones.
+//
+// The corpus is written as one commit followed by one deletion, which is how a
+// Property shard reaches a generation that carries both live and masked rows
+// for the same entity. Every visible row keeps its own document identifier --
+// the Property writer keys a document by entity and revision -- so the two
+// revisions of e-1 are two live rows rather than one replacing the other, which
+// is the case the repair build's own per-entity collapse exists to handle.
+func TestGenerateNIDX01EFixture(t *testing.T) {
+ if os.Getenv(nidx01eGenerateEnv) != "1" {
+ t.Skipf("set %s=1 to regenerate the NIDX-01E corpus", nidx01eGenerateEnv)
+ }
+ tester := require.New(t)
+
+ staging := t.TempDir()
+ store, err := NewStore(StoreOpts{Path: staging})
+ tester.NoError(err)
+ tester.NoError(store.UpdateSeriesBatch(index.Batch{Documents: nidx01eBatch()}))
+ for _, row := range nidx01eRows {
+ if !row.deleted {
+ continue
+ }
+ tester.NoError(store.Delete([][]byte{[]byte(nidx01eDocID(row))}))
+ }
+ tester.NoError(store.Close())
+
+ tester.NoError(os.RemoveAll(nidx01eRoot))
+ tester.NoError(os.MkdirAll(nidx01eShardDir, 0o755))
+ entries, err := os.ReadDir(staging)
+ tester.NoError(err)
+ segments, snapshots := 0, uint64(0)
+ for _, entry := range entries {
+ extension := filepath.Ext(entry.Name())
+ if extension != segExt && extension != snpExt {
+ continue
+ }
+ if extension == segExt {
+ segments++
+ } else if identifier, parsed := parseFixtureIdentifier(entry.Name(), snpExt); parsed && identifier > snapshots {
+ snapshots = identifier
+ }
+ payload, readErr := os.ReadFile(filepath.Join(staging, entry.Name()))
+ tester.NoError(readErr)
+ tester.NoError(os.WriteFile(filepath.Join(nidx01eShardDir, entry.Name()), payload, 0o600))
+ }
+ tester.NotZero(segments, "the corpus must hold at least one sealed segment")
+
+ // The corpus is only evidence if its own declaration holds on the bytes
+ // just written, so the two facts every later assertion rests on -- which
+ // generation is newest, and that exactly the four declared rows survive its
+ // deletion masks -- are checked here with operations already merged.
+ tester.Equal(nidx01eSnapshotID, snapshots, "regenerating moved the corpus's newest generation; update nidx01eSnapshotID")
+ visible, err := ReadOnlyDocCount(nidx01eShardDir)
+ tester.NoError(err)
+ tester.Equal(nidx01eVisibleRowCount, visible, "the corpus must leave exactly the declared visible rows undeleted")
+
+ tester.NoError(os.WriteFile(nidx01eManifest, nidx01eProvenanceBytes(t), 0o600))
+}
+
+// nidx01eBatch turns the declared rows into one commit shaped like the Property
+// shard's own writes: the entity identifier, the group and the name as indexed,
+// sortable fields, an unrelated stored payload, and the repair SHA as a stored
+// field, with the property's modification revision as the document timestamp.
+func nidx01eBatch() index.Documents {
+ documents := make(index.Documents, 0, len(nidx01eRows))
+ for _, row := range nidx01eRows {
+ documents = append(documents, index.Document{
+ EntityValues: []byte(nidx01eDocID(row)),
+ Timestamp: row.timestamp,
+ Fields: []index.Field{
+ nidx01eSortableField(nidx01eEntityField, []byte(row.entityID)),
+ nidx01eSortableField(nidx01eGroupField, []byte(row.group)),
+ nidx01eSortableField(nidx01eNameField, []byte(row.name)),
+ nidx01eStoredField(nidx01eSourceField, []byte(row.source)),
+ nidx01eStoredField(nidx01eSHAField, []byte(row.sha)),
+ },
+ })
+ }
+ return documents
+}
+
+// nidx01eDocID renders the document identifier the Property writer gives a
+// revision: the entity it belongs to followed by that revision.
+func nidx01eDocID(row nidx01eRow) string {
+ return row.group + "/" + row.name + "/" + row.entityID + "/" + strconv.FormatInt(row.timestamp, 10)
+}
+
+// nidx01eSortableField builds a field the segment indexes and records a doc
+// value for.
+func nidx01eSortableField(name string, value []byte) index.Field {
+ field := index.NewBytesField(index.FieldKey{TagName: name}, value)
+ field.Index = true
+ return field
+}
+
+// nidx01eStoredField builds a field the segment stores but neither indexes nor
+// records a doc value for.
+func nidx01eStoredField(name string, value []byte) index.Field {
+ field := index.NewBytesField(index.FieldKey{TagName: name}, value)
+ field.Store = true
+ field.NoSort = true
+ return field
+}
+
+// nidx01eVisibleRows lists the corpus's rows that survive its deletion masks,
+// in the ascending order issue #14012 declares them.
+func nidx01eVisibleRows() []nidx01eRow {
+ visible := make([]nidx01eRow, 0, len(nidx01eRows))
+ for _, row := range nidx01eRows {
+ if !row.deleted {
+ visible = append(visible, row)
+ }
+ }
+ return visible
+}
+
+// parseFixtureIdentifier reads the numeric identifier out of a fixed-width
+// hexadecimal index file name.
+func parseFixtureIdentifier(name, extension string) (uint64, bool) {
+ base := filepath.Base(name)
+ if filepath.Ext(base) != extension {
+ return 0, false
+ }
+ identifier, err := strconv.ParseUint(base[:len(base)-len(extension)], 16, 64)
+ if err != nil {
+ return 0, false
+ }
+ return identifier, true
+}
+
+// nidx01eProvenanceBytes renders the manifest checked in beside the corpus.
+func nidx01eProvenanceBytes(t *testing.T) []byte {
+ t.Helper()
+ described := make([]nidx01eProvenanceRow, 0, len(nidx01eRows))
+ for _, row := range nidx01eRows {
+ described = append(described, nidx01eDescribeRow(row))
+ }
+ visible := nidx01eVisibleRows()
+ manifest := nidx01eProvenance{
+ Oracle: map[string]string{
+ "writer": "pkg/index/inverted.NewStore",
+ "go_mod_sha256": nidx01eGoModSHA256(t),
+ },
+ GeneratorCommand: nidx01eGenerateCmd,
+ FileSHA256: nidx01eFileHashes(t),
+ Rows: described,
+ DeclaredPages: [][]nidx01eProvenanceRow{
+ {nidx01eDescribeRow(visible[0]), nidx01eDescribeRow(visible[1])},
+ {nidx01eDescribeRow(visible[2]), nidx01eDescribeRow(visible[3])},
+ },
+ SnapshotID: nidx01eSnapshotID,
+ VisibleRowCount: nidx01eVisibleRowCount,
+ Notes: "One Property shard generation holding five physical rows of three entities. " +
+ "Ascending by (_group, _im_name, _entity_id, _timestamp) the visible rows are " +
+ "(g-a,n-a,e-1,10,sha-a), (g-a,n-a,e-1,20,sha-b), (g-a,n-b,e-2,5,sha-c), (g-b,n-a,e-3,7,sha-d). " +
+ "Row (g-a,n-a,e-1,15,sha-x) is deleted in this generation and sorts between the first two visible rows, " +
+ "so a page that ignored the deletion masks is caught inside the first page rather than at the end. " +
+ "At page size 2 the first page is the first two visible rows, the strict cursor is (g-a,n-a,e-1,20), " +
+ "and the second page is the last two. The repair tree these rows complete maps " +
+ "g-a/n-a/e-1 to sha-b, g-a/n-b/e-2 to sha-c and g-b/n-a/e-3 to sha-d. " +
+ "Every row also stores a _source payload that no repair page decodes.",
+ }
+ encoded, err := json.MarshalIndent(manifest, "", " ")
+ require.NoError(t, err)
+ return append(encoded, '\n')
+}
+
+func nidx01eDescribeRow(row nidx01eRow) nidx01eProvenanceRow {
+ return nidx01eProvenanceRow{
+ Group: row.group,
+ Name: row.name,
+ EntityID: row.entityID,
+ Timestamp: row.timestamp,
+ SHA: row.sha,
+ DocID: nidx01eDocID(row),
+ Deleted: row.deleted,
+ }
+}
+
+// nidx01eGoModSHA256 pins the dependency set the oracle ran with, so the corpus
+// is identified by an immutable content hash rather than by a module name.
+func nidx01eGoModSHA256(t *testing.T) string {
+ t.Helper()
+ raw, err := os.ReadFile(filepath.Join("..", "..", "..", "go.mod"))
+ require.NoError(t, err)
+ sum := sha256.Sum256(raw)
+ return hex.EncodeToString(sum[:])
+}
+
+// nidx01eFileHashes hashes every checked-in corpus file.
+func nidx01eFileHashes(t *testing.T) map[string]string {
+ t.Helper()
+ entries, err := os.ReadDir(nidx01eShardDir)
+ require.NoError(t, err)
+ names := make([]string, 0, len(entries))
+ for _, entry := range entries {
+ names = append(names, entry.Name())
+ }
+ sort.Strings(names)
+ hashes := make(map[string]string, len(names))
+ for _, name := range names {
+ payload, readErr := os.ReadFile(filepath.Join(nidx01eShardDir, name))
+ require.NoError(t, readErr)
+ sum := sha256.Sum256(payload)
+ hashes["shard-0/"+name] = hex.EncodeToString(sum[:])
+ }
+ return hashes
+}
diff --git a/pkg/index/inverted/native_nidx01b_boundary_test.go b/pkg/index/inverted/native_nidx01b_boundary_test.go
index a8206a9a4..01fdeb477 100644
--- a/pkg/index/inverted/native_nidx01b_boundary_test.go
+++ b/pkg/index/inverted/native_nidx01b_boundary_test.go
@@ -52,26 +52,32 @@ 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, walk its live
-// documents' stored fields, walk the subset of them one field's exact terms
-// select, close it, and classify the three failures a caller must distinguish.
+// committed generation, report its identifier and visible document count, walk
+// its live documents' stored fields, walk the subset of them one field's exact
+// terms select, serve the bounded repair tuple page, close it, and classify the
+// four failures a caller must distinguish.
// Issue #14009 placed the document walk outside NIDX-01B; issue #14010 added it;
-// issue #14011 adds the exact-term selection and nothing more. Term dictionaries
-// and postings are what that selection is built from, so they stay private:
-// doc values, sort/search-after, ranges, prefix and wildcard expansion,
-// analyzers, writers and merge remain outside the milestone, and an entry
-// appearing here for any of them is the milestone growing surface it was
-// explicitly denied.
+// issue #14011 added exact-term selection; and issue #14012 adds the bounded
+// repair tuple page. Term dictionaries and postings are what those operations
+// are built from, so they stay private: ranges, prefix and wildcard expansion,
+// analyzers, writers and merge remain outside the milestone, and an entry here
+// for any of them is the milestone growing surface it was explicitly denied.
var nativeReaderSurface = []string{
"ErrCorrupt",
+ "ErrInvalidRepairPage",
"ErrInvalidSelection",
"ErrNoSnapshot",
"Open",
"Reader",
"Reader.Close",
+ "Reader.RepairTuplePage",
+ "Reader.SnapshotID",
"Reader.VisibleDocCount",
"Reader.VisitLiveDocuments",
"Reader.VisitSelectedDocuments",
+ "RepairCursor",
+ "RepairPageRequest",
+ "RepairTupleRow",
"StoredDocument",
}
@@ -94,15 +100,17 @@ func TestNIDX01BBoundarySurface(t *testing.T) {
tester.NotErrorIs(ErrNoCommittedIndex, ErrCorruptIndex,
"an absent committed generation and damaged committed bytes must stay separately classifiable")
- observed := exportedSurfaceOf(t, nativeReaderDir)
+ observed := exportedSurfaceOf(t)
tester.Equal(nativeReaderSurface, observed,
"the native reader's exported surface changed; NIDX-01B may only extend private generation selection")
}
// exportedSurfaceOf lists the exported top-level identifiers and exported
-// methods on exported types declared by the non-test Go sources in dir, sorted.
-func exportedSurfaceOf(t *testing.T, dir string) []string {
+// methods on exported types declared by the non-test Go sources of the private
+// native reader package, sorted.
+func exportedSurfaceOf(t *testing.T) []string {
t.Helper()
+ dir := nativeReaderDir
entries, err := os.ReadDir(dir)
require.NoError(t, err, "the native reader package must exist at %s", dir)
var surface []string
diff --git a/pkg/index/inverted/native_nidx01c_walk_test.go b/pkg/index/inverted/native_nidx01c_walk_test.go
index 68e6b3f66..c2bfc922a 100644
--- a/pkg/index/inverted/native_nidx01c_walk_test.go
+++ b/pkg/index/inverted/native_nidx01c_walk_test.go
@@ -458,6 +458,6 @@ func TestNativeStoredDocumentWalkBoundarySurface(t *testing.T) {
}
tester.NotEmpty(dirInventory(t, filepath.Dir(nidx01cManifest)))
- tester.Equal(nativeReaderSurface, exportedSurfaceOf(t, nativeReaderDir),
+ tester.Equal(nativeReaderSurface, exportedSurfaceOf(t),
"the native reader's exported surface changed; NIDX-01C may only add the live document walk")
}
diff --git a/pkg/index/inverted/native_nidx01d_selection_test.go b/pkg/index/inverted/native_nidx01d_selection_test.go
index 7acbdbe44..b4ef6d154 100644
--- a/pkg/index/inverted/native_nidx01d_selection_test.go
+++ b/pkg/index/inverted/native_nidx01d_selection_test.go
@@ -504,7 +504,7 @@ func TestNativeExactTermsBoundarySurface(t *testing.T) {
tester.Equal([]string{"Field:string", "Terms:[][]uint8"}, selectionShape(),
"TermSelection is one field name and its literal terms; anything more is a query language")
- tester.Equal(nativeReaderSurface, exportedSurfaceOf(t, nativeReaderDir),
+ tester.Equal(nativeReaderSurface, exportedSurfaceOf(t),
"the native reader's exported surface changed; NIDX-01D may only add the exact-term selection")
}
diff --git a/pkg/index/inverted/native_nidx01e_tuple_page_test.go b/pkg/index/inverted/native_nidx01e_tuple_page_test.go
new file mode 100644
index 000000000..f1b75633f
--- /dev/null
+++ b/pkg/index/inverted/native_nidx01e_tuple_page_test.go
@@ -0,0 +1,663 @@
+// 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/binary"
+ "fmt"
+ "os"
+ "reflect"
+ "runtime"
+ "strconv"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/pkg/index"
+)
+
+const (
+ // iceFooterDocValueStart and iceFooterDocValueEnd bracket the doc-value
+ // section root in the 60-byte ICE v3 segment footer BDB-NIDX-SPEC-001
+ // revision 0.2 section 08 fixes the widths of.
+ iceFooterDocValueStart = 24
+ iceFooterDocValueEnd = 32
+
+ // nidx01eDeclaredPageSize is the page size issue #14012 declares its two
+ // pages at.
+ nidx01eDeclaredPageSize = 2
+
+ // The generation sizes and page size the bounded-page requirement is
+ // measured over, and the resident growth a bounded pager is allowed across
+ // all of them.
+ nidx01eBoundedRowCount = 2000
+ nidx01eBoundedPageSize = 10
+ nidx01eBoundedGrowthBytes = 8 << 20
+)
+
+// nidx01eEncodedTimestamps are the exact doc values the corpus records for each
+// declared revision. They are literals lifted from the checked-in generation,
+// not a re-encoding: an eleven-byte historical prefix-coded signed int64 whose
+// leading byte is the shift and whose sign bit is inverted so byte order
+// matches numeric order.
+var nidx01eEncodedTimestamps = map[int64][]byte{
+ 5: {0x20, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05},
+ 7: {0x20, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x07},
+ 10: {0x20, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x0a},
+ 20: {0x20, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x14},
+}
+
+// nidx01ePermittedReaderSurface is the ceiling on what the private native
+// reader may export once this leaf lands: the NIDX-01D surface, plus opening a
+// pinned generation's identifier, the bounded repair tuple page, the row that
+// page yields, and the sentinel an out-of-bounds page request is classified
+// with.
+//
+// Nothing else may appear. An entry for a collector, a sort expression, a
+// comparator, a heap, a doc-value reader, a chunk decoder, a range, a prefix,
+// a wildcard, an analyzer or any writer is the milestone growing the surface
+// BDB-NIDX-SPEC-001 revision 0.2 NIDX-01 explicitly denied it, however
+// convenient the entry is.
+var nidx01ePermittedReaderSurface = map[string]struct{}{
+ "ErrCorrupt": {},
+ "ErrInvalidRepairPage": {},
+ "ErrInvalidSelection": {},
+ "ErrNoSnapshot": {},
+ "Open": {},
+ "Reader": {},
+ "Reader.Close": {},
+ "Reader.RepairTuplePage": {},
+ "Reader.SnapshotID": {},
+ "Reader.VisibleDocCount": {},
+ "Reader.VisitLiveDocuments": {},
+ "Reader.VisitSelectedDocuments": {},
+ "RepairCursor": {},
+ "RepairPageRequest": {},
+ "RepairTupleRow": {},
+ "StoredDocument": {},
+}
+
+// TestNativeRepairTuplePageReturnsTheDeclaredFirstPage asks the corpus for the
+// first page issue #14012 declares.
+//
+// Requirement proved here:
+//
+// R1 -- a page orders the pinned generation's live rows ascending by all
+// four declared components and returns the first PageSize of them. The
+// declared row that the generation deletes sorts between the two rows
+// of this page, so a page built without the deletion masks would return
+// it here rather than at the end of the order.
+func TestNativeRepairTuplePageReturnsTheDeclaredFirstPage(t *testing.T) {
+ tester := require.New(t)
+ generation := nidx01eOpen(t, nidx01eShardDir)
+
+ page, err := generation.RepairTuplePage(context.Background(), nidx01eRequest(nidx01eDeclaredPageSize, nil))
+ tester.NoError(err)
+
+ declared := nidx01eVisibleRows()
+ tester.Equal(nidx01eDeclaredRows(declared[0], declared[1]), nidx01eWithoutCursors(page))
+}
+
+// TestNativeRepairTuplePageResumesStrictlyAfterTheDeclaredCursor asks for the
+// second declared page with the cursor issue #14012 declares.
+//
+// Requirement proved here:
+//
+// R1 -- a page resumes strictly after the complete prior tuple: the row the
+// cursor names is not repeated, and no row between it and the next in
+// the order is skipped.
+func TestNativeRepairTuplePageResumesStrictlyAfterTheDeclaredCursor(t *testing.T) {
+ tester := require.New(t)
+ generation := nidx01eOpen(t, nidx01eShardDir)
+ declared := nidx01eVisibleRows()
+ first, firstErr := generation.RepairTuplePage(context.Background(), nidx01eRequest(nidx01eDeclaredPageSize, nil))
+ tester.NoError(firstErr)
+
+ page, err := generation.RepairTuplePage(context.Background(),
+ nidx01eRequest(nidx01eDeclaredPageSize, first[len(first)-1].Cursor))
+ tester.NoError(err)
+
+ tester.Equal(nidx01eDeclaredRows(declared[2], declared[3]), nidx01eWithoutCursors(page))
+}
+
+// TestNativeRepairTuplePagePagesTheWholeGenerationExactlyOnce walks the corpus
+// page by page until the order is exhausted.
+//
+// Requirement proved here:
+//
+// R1 -- paging a generation end to end yields every visible row exactly once
+// and in the declared order, with no duplicate across a page boundary
+// and no row skipped at one, and the page after the last row is empty
+// rather than an error.
+func TestNativeRepairTuplePagePagesTheWholeGenerationExactlyOnce(t *testing.T) {
+ tester := require.New(t)
+ generation := nidx01eOpen(t, nidx01eShardDir)
+
+ for _, pageSize := range []int{1, nidx01eDeclaredPageSize, 3, len(nidx01eRows) + 1} {
+ walked, pages := nidx01eWalk(t, generation, pageSize)
+ declared := nidx01eVisibleRows()
+ tester.Equal(nidx01eDeclaredRows(declared...), walked,
+ "page size %d must yield the declared rows once each, in order", pageSize)
+ expectedPages := (len(declared)+pageSize-1)/pageSize + 1
+ tester.Equal(expectedPages, pages,
+ "page size %d must exhaust the order in %d calls", pageSize, expectedPages)
+ }
+}
+
+// TestNativeRepairTuplePageProjectsTheRequestedStoredField reads a page whose
+// rows each store two fields.
+//
+// Requirement proved here:
+//
+// R1 -- a row carries the value of the stored field the request projects.
+// Every corpus row also stores an unrelated payload that is recorded
+// first, so a page that returned a row's first stored value instead of
+// the projected one is caught here.
+func TestNativeRepairTuplePageProjectsTheRequestedStoredField(t *testing.T) {
+ tester := require.New(t)
+ generation := nidx01eOpen(t, nidx01eShardDir)
+
+ page, err := generation.RepairTuplePage(context.Background(), nidx01eRequest(len(nidx01eRows), nil))
+ tester.NoError(err)
+ tester.Len(page, len(nidx01eVisibleRows()))
+
+ for rowIndex, row := range nidx01eVisibleRows() {
+ tester.Equal(row.sha, string(page[rowIndex].Value),
+ "row %d must project %s, not its stored %s", rowIndex, nidx01eSHAField, nidx01eSourceField)
+ }
+}
+
+// TestNativeRepairTuplePagePinsTheDeclaredGeneration reads the identifier of
+// the generation the corpus commits.
+//
+// Requirement proved here:
+//
+// R4 -- a pinned view names the generation it reads, so a caller can record
+// which generation a completed page sequence covered instead of asking
+// the directory again afterwards and recording one it never read.
+func TestNativeRepairTuplePagePinsTheDeclaredGeneration(t *testing.T) {
+ tester := require.New(t)
+ generation := nidx01eOpen(t, nidx01eShardDir)
+
+ tester.Equal(nidx01eSnapshotID, generation.SnapshotID())
+}
+
+// TestNativeRepairTuplePageIgnoresAGenerationPublishedBetweenPages publishes a
+// row that would head the second page, after the first page has been read.
+//
+// Requirement proved here:
+//
+// R4 -- a generation published after a view is pinned does not enter a later
+// page of that view. The published row sorts strictly between the
+// declared cursor and the next declared row, so an unpinned pager would
+// return it as the first row of page two.
+func TestNativeRepairTuplePageIgnoresAGenerationPublishedBetweenPages(t *testing.T) {
+ tester := require.New(t)
+ shard := copyIndexDir(t, nidx01eShardDir)
+ generation := nidx01eOpen(t, shard)
+ declared := nidx01eVisibleRows()
+
+ first, err := generation.RepairTuplePage(context.Background(), nidx01eRequest(nidx01eDeclaredPageSize, nil))
+ tester.NoError(err)
+ tester.Equal(nidx01eDeclaredRows(declared[0], declared[1]), nidx01eWithoutCursors(first))
+
+ published := nidx01eRow{group: "g-a", name: "n-a", entityID: "e-1", timestamp: 30, sha: "sha-published", source: "source-published"}
+ nidx01ePublish(t, shard, published)
+
+ second, err := generation.RepairTuplePage(context.Background(),
+ nidx01eRequest(nidx01eDeclaredPageSize, first[len(first)-1].Cursor))
+ tester.NoError(err)
+ tester.Equal(nidx01eDeclaredRows(declared[2], declared[3]), nidx01eWithoutCursors(second))
+
+ republished := nidx01eOpen(t, shard)
+ tester.NotEqual(nidx01eSnapshotID, republished.SnapshotID(),
+ "the publication must have committed a newer generation, or the pin is untested")
+ after, err := republished.RepairTuplePage(context.Background(), nidx01eRequest(len(nidx01eRows)+1, nil))
+ tester.NoError(err)
+ tester.Len(after, len(declared)+1, "a view opened after the publication must see the published row")
+}
+
+// TestNativeRepairTuplePageRejectsMissingSortValues verifies R4: a missing
+// sort component is corrupt repair input and never produces a partial page.
+func TestNativeRepairTuplePageRejectsMissingSortValues(t *testing.T) {
+ tester := require.New(t)
+ shard := nidx01eSeed(t,
+ nidx01eRow{group: "g-a", name: "n-a", entityID: "e-1", timestamp: 10, sha: "sha-present"},
+ nidx01eRow{group: "g-a", name: "n-a", entityID: "e-1", sha: "sha-absent"},
+ )
+ generation := nidx01eOpen(t, shard)
+ page, err := generation.RepairTuplePage(context.Background(), nidx01eRequest(1, nil))
+ tester.ErrorIs(err, ErrCorruptIndex)
+ tester.Empty(page)
+}
+
+// TestNativeRepairTuplePageRejectsRequestsOutsideItsBounds asks for pages the
+// reader will not serve.
+//
+// Requirement proved here:
+//
+// R4 -- an invalid cursor and a page-size overflow are rejected as bounded,
+// typed failures before any doc value is read, and stay distinguishable
+// from damaged committed bytes. No page is returned, so a caller cannot
+// mistake a rejection for the end of the order and record a generation
+// it never finished reading.
+func TestNativeRepairTuplePageRejectsRequestsOutsideItsBounds(t *testing.T) {
+ generation := nidx01eOpen(t, nidx01eShardDir)
+ first, firstErr := generation.RepairTuplePage(context.Background(), nidx01eRequest(1, nil))
+ require.NoError(t, firstErr)
+ require.Len(t, first, 1)
+ cases := []struct {
+ mutate func(*RepairPageRequest)
+ name string
+ }{
+ {name: "short cursor", mutate: func(r *RepairPageRequest) { r.After.cursor.SortValues = r.After.cursor.SortValues[:3] }},
+ {name: "long cursor", mutate: func(r *RepairPageRequest) {
+ r.After.cursor.SortValues = append(r.After.cursor.SortValues, []byte("extra"))
+ }},
+ {name: "oversize cursor component", mutate: func(r *RepairPageRequest) { r.After.cursor.SortValues[0] = make([]byte, MaxRepairSortValueLength+1) }},
+ {name: "zero cursor", mutate: func(r *RepairPageRequest) { r.After = &RepairCursor{} }},
+ {name: "page size zero", mutate: func(r *RepairPageRequest) { r.PageSize = 0 }},
+ {name: "page size negative", mutate: func(r *RepairPageRequest) { r.PageSize = -1 }},
+ {name: "page size over the bound", mutate: func(r *RepairPageRequest) { r.PageSize = MaxRepairPageSize + 1 }},
+ }
+
+ for _, testCase := range cases {
+ t.Run(testCase.name, func(t *testing.T) {
+ tester := require.New(t)
+ cursor := *first[0].Cursor
+ cursor.cursor.SortValues = append([][]byte(nil), cursor.cursor.SortValues...)
+ request := nidx01eRequest(nidx01eDeclaredPageSize, &cursor)
+ testCase.mutate(&request)
+
+ page, err := generation.RepairTuplePage(context.Background(), request)
+ tester.ErrorIs(err, ErrInvalidRepairPage)
+ tester.NotErrorIs(err, ErrCorruptIndex,
+ "an out-of-bounds request and damaged committed bytes must stay separately classifiable")
+ tester.Empty(page)
+ })
+ }
+}
+
+// TestNativeRepairTuplePageStopsOnCancellation asks for a page with a context
+// that is already canceled.
+//
+// Requirement proved here:
+//
+// R4 -- cancellation stops a page and is reported as cancellation rather than
+// as damage or as the end of the order, and no rows are returned.
+func TestNativeRepairTuplePageStopsOnCancellation(t *testing.T) {
+ tester := require.New(t)
+ generation := nidx01eOpen(t, nidx01eShardDir)
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+
+ page, err := generation.RepairTuplePage(ctx, nidx01eRequest(nidx01eDeclaredPageSize, nil))
+ tester.ErrorIs(err, context.Canceled)
+ tester.Empty(page)
+}
+
+// TestNativeRepairTuplePageReportsDamagedDocValuesAsCorruption pages a copy of
+// the corpus whose doc-value section root table has been overwritten.
+//
+// Requirement proved here:
+//
+// R4 -- a doc-value section that violates the ICE v3 grammar fails the page
+// as bounded, typed corruption. The damage is confined to the
+// doc-value locations, so a generation whose stored records and
+// postings are intact still fails here rather than returning a
+// partially ordered page.
+func TestNativeRepairTuplePageReportsDamagedDocValuesAsCorruption(t *testing.T) {
+ tester := require.New(t)
+ damaged := copyIndexDir(t, nidx01eShardDir)
+ damageDocValueLocations(t, newestSegmentFile(t, damaged))
+
+ generation := nidx01eOpen(t, damaged)
+ page, err := generation.RepairTuplePage(context.Background(), nidx01eRequest(nidx01eDeclaredPageSize, nil))
+ tester.ErrorIs(err, ErrCorruptIndex)
+ tester.NotErrorIs(err, ErrNoCommittedIndex)
+ tester.Empty(page)
+}
+
+// TestNativeRepairTuplePageClassifiesAnAbsentGeneration opens a directory that
+// holds no committed generation.
+//
+// Requirement proved here:
+//
+// R4 -- a shard that was never flushed is absent, not damaged. The repair
+// build treats an absent generation as nothing to do, so the two
+// failures must stay separately classifiable at the boundary.
+func TestNativeRepairTuplePageClassifiesAnAbsentGeneration(t *testing.T) {
+ tester := require.New(t)
+
+ generation, err := OpenReadOnlyGeneration(t.TempDir())
+ tester.ErrorIs(err, ErrNoCommittedIndex)
+ tester.NotErrorIs(err, ErrCorruptIndex)
+ tester.Nil(generation)
+}
+
+// TestNativeRepairTuplePageIgnoresReservedCRC32 pages a copy of the corpus
+// whose reserved CRC32 slots hold bytes no writer computed.
+//
+// Requirement proved here:
+//
+// R4 -- BDB-NIDX-SPEC-001 revision 0.2 DEC-007 preserves the historical
+// CRC32 fields without calculating, validating or using them, so a
+// generation carrying arbitrary values in those slots pages exactly as
+// the untouched corpus does.
+func TestNativeRepairTuplePageIgnoresReservedCRC32(t *testing.T) {
+ tester := require.New(t)
+ rewritten := copyIndexDir(t, nidx01eShardDir)
+ fillReservedCRC32(t, rewritten, []byte{0x01, 0x23, 0x45, 0x67})
+
+ generation := nidx01eOpen(t, rewritten)
+ page, err := generation.RepairTuplePage(context.Background(), nidx01eRequest(nidx01eDeclaredPageSize, nil))
+ tester.NoError(err)
+
+ declared := nidx01eVisibleRows()
+ tester.Equal(nidx01eDeclaredRows(declared[0], declared[1]), nidx01eWithoutCursors(page))
+}
+
+// TestNativeRepairTuplePageLeavesTheDirectoryUnchanged inventories the corpus
+// directory before and after paging it end to end, and pages a directory a live
+// writer holds open.
+//
+// Requirement proved here:
+//
+// R5 -- paging writes no bytes and takes no exclusive directory lock. Every
+// entry's name, size, mode, modification time and content hash is
+// unchanged afterwards, and a shard a live writer owns still serves its
+// pinned generation, which is what lets a repair build run beside the
+// Property writer instead of draining it.
+func TestNativeRepairTuplePageLeavesTheDirectoryUnchanged(t *testing.T) {
+ tester := require.New(t)
+ before := dirInventory(t, nidx01eShardDir)
+
+ generation := nidx01eOpen(t, nidx01eShardDir)
+ walked, _ := nidx01eWalk(t, generation, nidx01eDeclaredPageSize)
+ tester.Equal(nidx01eDeclaredRows(nidx01eVisibleRows()...), walked)
+ tester.Equal(before, dirInventory(t, nidx01eShardDir),
+ "a read-only page must leave every file's bytes, mode and modification time alone")
+
+ shared := copyIndexDir(t, nidx01eShardDir)
+ writer, err := NewStore(StoreOpts{Path: shared})
+ tester.NoError(err)
+ defer func() {
+ tester.NoError(writer.Close())
+ }()
+
+ beside := nidx01eOpen(t, shared)
+ page, besideErr := beside.RepairTuplePage(context.Background(), nidx01eRequest(nidx01eDeclaredPageSize, nil))
+ tester.NoError(besideErr)
+ declared := nidx01eVisibleRows()
+ tester.Equal(nidx01eDeclaredRows(declared[0], declared[1]), nidx01eWithoutCursors(page),
+ "an open writer must not change what the pinned committed generation pages")
+}
+
+// TestNativeRepairTuplePageBoundsResidentStateAcrossPages pages a generation
+// far larger than one page, many pages deep.
+//
+// Requirement proved here:
+//
+// R2 -- a page holds at most the requested number of rows, and paging does
+// not accumulate: the memory a view retains after two hundred pages of
+// a two-thousand-row generation is not materially above what it
+// retained before the first, so the pager is neither materializing the
+// shard nor carrying a growing offset scan from page to page.
+func TestNativeRepairTuplePageBoundsResidentStateAcrossPages(t *testing.T) {
+ tester := require.New(t)
+ shard := nidx01eSeed(t, nidx01eBoundedRows(nidx01eBoundedRowCount)...)
+ generation := nidx01eOpen(t, shard)
+
+ baseline := nidx01eRetainedBytes()
+ var after *RepairCursor
+ rows, pages := 0, 0
+ for {
+ page, err := generation.RepairTuplePage(context.Background(), nidx01eRequest(nidx01eBoundedPageSize, after))
+ tester.NoError(err)
+ tester.LessOrEqual(len(page), nidx01eBoundedPageSize, "a page must never exceed the size it was asked for")
+ if len(page) == 0 {
+ break
+ }
+ rows += len(page)
+ pages++
+ after = page[len(page)-1].Cursor
+ }
+
+ tester.Equal(nidx01eBoundedRowCount, rows)
+ tester.Equal(nidx01eBoundedRowCount/nidx01eBoundedPageSize, pages)
+
+ // Compared as a ceiling rather than as a difference: retained heap may end
+ // below the baseline when a collection during the walk releases something
+ // the baseline still counted, and a pager that gave memory back has met
+ // this requirement, not failed it. Differencing two unsigned readings turns
+ // that case into a near-maximal "growth" instead.
+ retained := nidx01eRetainedBytes()
+ tester.Less(retained, baseline+uint64(nidx01eBoundedGrowthBytes),
+ "paging %d pages must not accumulate resident state: retained %d bytes against a %d byte baseline",
+ pages, retained, baseline)
+}
+
+// TestNativeRepairTuplePageBoundarySurface guards the boundary itself rather
+// than any behavior behind it.
+//
+// Requirement proved here:
+//
+// R2 -- the milestone is delivered entirely behind
+// inverted.OpenReadOnlyGeneration and the bounded page it serves. The
+// request carries one opaque cursor and one page size and nothing else; the sentinel an
+// out-of-bounds request is classified with stays distinct from the two
+// the boundary already publishes; and the private native reader gains
+// no export beyond the pinned generation's identifier, the page, and
+// the row it yields. An entry for a collector, an arbitrary sort
+// expression, a comparator, an offset or a relevance order is the
+// milestone growing surface NIDX-01 explicitly denied it.
+func TestNativeRepairTuplePageBoundarySurface(t *testing.T) {
+ tester := require.New(t)
+
+ tester.ErrorIs(ErrInvalidRepairPage, ErrInvalidRepairPage)
+ tester.NotErrorIs(ErrInvalidRepairPage, ErrCorruptIndex)
+ tester.NotErrorIs(ErrInvalidRepairPage, ErrNoCommittedIndex)
+ tester.NotErrorIs(ErrInvalidRepairPage, ErrInvalidSelection)
+
+ tester.Equal([]string{
+ "After:*inverted.RepairCursor",
+ "PageSize:int",
+ }, structShape(RepairPageRequest{}),
+ "a repair page request is an opaque cursor and page size; sort fields and projection are fixed")
+ tester.Equal([]string{"Cursor:*inverted.RepairCursor", "SortValues:[][]uint8", "Value:[]uint8"}, structShape(RepairRow{}),
+ "a repair row carries its encoded sort values, stored SHA value and opaque continuation cursor")
+
+ for _, exported := range exportedSurfaceOf(t) {
+ _, permitted := nidx01ePermittedReaderSurface[exported]
+ tester.True(permitted, "the native reader exports %s, which NIDX-01E does not admit", exported)
+ }
+}
+
+// nidx01eOpen pins a directory's newest committed generation for the duration
+// of the test.
+func nidx01eOpen(t *testing.T, dir string) *ReadOnlyGeneration {
+ t.Helper()
+ generation, err := OpenReadOnlyGeneration(dir)
+ require.NoError(t, err)
+ require.NotNil(t, generation)
+ t.Cleanup(func() {
+ require.NoError(t, generation.Close())
+ })
+ return generation
+}
+
+// nidx01eRequest builds the repair page request the Property repair build
+// issues, at the given page size and cursor.
+func nidx01eRequest(pageSize int, after *RepairCursor) RepairPageRequest {
+ return RepairPageRequest{
+ After: after,
+ PageSize: pageSize,
+ }
+}
+
+// nidx01eDeclaredTuple renders the four ascending components issue #14012
+// declares for one row: the group, name and entity identifier as the literal
+// bytes their keyword doc values hold, and the revision as the literal encoded
+// doc value the corpus records.
+func nidx01eDeclaredTuple(row nidx01eRow) [][]byte {
+ encoded, declared := nidx01eEncodedTimestamps[row.timestamp]
+ if !declared {
+ return [][]byte{[]byte(row.group), []byte(row.name), []byte(row.entityID), nil}
+ }
+ return [][]byte{[]byte(row.group), []byte(row.name), []byte(row.entityID), encoded}
+}
+
+// nidx01eDeclaredRows renders the rows a page must hold for the given declared
+// rows, in order.
+func nidx01eDeclaredRows(rows ...nidx01eRow) []RepairRow {
+ declared := make([]RepairRow, 0, len(rows))
+ for _, row := range rows {
+ declared = append(declared, RepairRow{SortValues: nidx01eDeclaredTuple(row), Value: []byte(row.sha)})
+ }
+ return declared
+}
+
+// nidx01eWalk pages a generation end to end at the given page size and returns
+// every row it yielded together with the number of calls it took.
+func nidx01eWalk(t *testing.T, generation *ReadOnlyGeneration, pageSize int) ([]RepairRow, int) {
+ t.Helper()
+ var walked []RepairRow
+ var after *RepairCursor
+ for pages := 1; ; pages++ {
+ page, err := generation.RepairTuplePage(context.Background(), nidx01eRequest(pageSize, after))
+ require.NoError(t, err)
+ if len(page) == 0 {
+ return walked, pages
+ }
+ walked = append(walked, nidx01eWithoutCursors(page)...)
+ after = page[len(page)-1].Cursor
+ require.LessOrEqual(t, len(walked), len(nidx01eRows), "paging must terminate rather than repeat rows")
+ }
+}
+
+// nidx01eSeed writes the given rows into a fresh shard directory through the
+// compatibility writer and returns it.
+func nidx01eSeed(t *testing.T, rows ...nidx01eRow) string {
+ t.Helper()
+ shard := t.TempDir()
+ writer, err := NewStore(StoreOpts{Path: shard})
+ require.NoError(t, err)
+ require.NoError(t, writer.UpdateSeriesBatch(index.Batch{Documents: nidx01eDocumentsOf(rows)}))
+ require.NoError(t, writer.Close())
+ return shard
+}
+
+// nidx01ePublish commits one more row into an existing shard directory, which
+// publishes a new generation over it.
+func nidx01ePublish(t *testing.T, shard string, row nidx01eRow) {
+ t.Helper()
+ writer, err := NewStore(StoreOpts{Path: shard})
+ require.NoError(t, err)
+ require.NoError(t, writer.UpdateSeriesBatch(index.Batch{Documents: nidx01eDocumentsOf([]nidx01eRow{row})}))
+ require.NoError(t, writer.Close())
+}
+
+// nidx01eDocumentsOf shapes rows the way the Property shard writes them.
+func nidx01eDocumentsOf(rows []nidx01eRow) index.Documents {
+ documents := make(index.Documents, 0, len(rows))
+ for _, row := range rows {
+ documents = append(documents, index.Document{
+ EntityValues: []byte(nidx01eDocID(row)),
+ Timestamp: row.timestamp,
+ Fields: []index.Field{
+ nidx01eSortableField(nidx01eEntityField, []byte(row.entityID)),
+ nidx01eSortableField(nidx01eGroupField, []byte(row.group)),
+ nidx01eSortableField(nidx01eNameField, []byte(row.name)),
+ nidx01eStoredField(nidx01eSourceField, []byte(row.source)),
+ nidx01eStoredField(nidx01eSHAField, []byte(row.sha)),
+ },
+ })
+ }
+ return documents
+}
+
+// nidx01eBoundedRows builds a generation whose rows all differ in their least
+// significant sort component, so every page boundary falls on a distinct
+// cursor.
+func nidx01eBoundedRows(count int) []nidx01eRow {
+ rows := make([]nidx01eRow, 0, count)
+ for ordinal := 1; ordinal <= count; ordinal++ {
+ rows = append(rows, nidx01eRow{
+ group: "g-bounded",
+ name: "n-bounded",
+ entityID: "e-" + strconv.Itoa(ordinal),
+ timestamp: int64(ordinal),
+ sha: "sha-" + strconv.Itoa(ordinal),
+ source: "source-" + strconv.Itoa(ordinal),
+ })
+ }
+ return rows
+}
+
+// nidx01eRetainedBytes reports the heap a full collection leaves live, which is
+// the memory a pager is still holding rather than the memory it has churned
+// through.
+func nidx01eRetainedBytes() uint64 {
+ runtime.GC()
+ var stats runtime.MemStats
+ runtime.ReadMemStats(&stats)
+ return stats.HeapAlloc
+}
+
+// damageDocValueLocations overwrites a segment's doc-value location table, the
+// section the footer's doc-value root addresses, leaving its stored records,
+// term dictionaries and postings intact. It is the smallest damage that is
+// invisible to every operation NIDX-01 merged before this leaf and fatal to a
+// page that orders by doc values.
+func damageDocValueLocations(t *testing.T, segmentPath string) {
+ t.Helper()
+ payload, err := os.ReadFile(segmentPath)
+ require.NoError(t, err)
+ require.Greater(t, len(payload), iceFooterLength)
+ footer := payload[len(payload)-iceFooterLength:]
+ docValues := binary.BigEndian.Uint64(footer[iceFooterDocValueStart:iceFooterDocValueEnd])
+ fieldsIndex := binary.BigEndian.Uint64(footer[iceFooterFieldsIndexStart:iceFooterFieldsIndexEnd])
+ require.Less(t, docValues, fieldsIndex, "segment %s records no doc-value locations to damage", segmentPath)
+ for offset := docValues; offset < fieldsIndex; offset++ {
+ payload[offset] = 0xFF
+ }
+ require.NoError(t, os.WriteFile(segmentPath, payload, 0o600))
+}
+
+// structShape renders every field a struct declares as name:type, in
+// declaration order, so a field added to carry a direction, an offset, a filter
+// or a score fails the boundary test instead of quietly widening the milestone.
+func structShape(value any) []string {
+ valueType := reflect.TypeOf(value)
+ shape := make([]string, 0, valueType.NumField())
+ for fieldIndex := range valueType.NumField() {
+ field := valueType.Field(fieldIndex)
+ shape = append(shape, fmt.Sprintf("%s:%s", field.Name, field.Type))
+ }
+ return shape
+}
+
+// nidx01eWithoutCursors isolates the declared row values from continuation state.
+func nidx01eWithoutCursors(rows []RepairRow) []RepairRow {
+ values := make([]RepairRow, len(rows))
+ for rowIndex, row := range rows {
+ values[rowIndex] = RepairRow{SortValues: row.SortValues, Value: row.Value}
+ }
+ return values
+}
diff --git a/pkg/index/inverted/repair_page.go b/pkg/index/inverted/repair_page.go
new file mode 100644
index 000000000..f87109813
--- /dev/null
+++ b/pkg/index/inverted/repair_page.go
@@ -0,0 +1,170 @@
+// 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"
+ "errors"
+ "fmt"
+
+ "github.com/apache/skywalking-banyandb/pkg/index"
+ "github.com/apache/skywalking-banyandb/pkg/index/inverted/internal/nativeice"
+)
+
+// RepairSortFieldCount is the number of fixed Property repair sort components.
+const RepairSortFieldCount = 4
+
+const (
+ // MaxRepairPageSize bounds the number of rows returned by a repair page.
+ MaxRepairPageSize = 1 << 16
+ // MaxRepairSortValueLength bounds each encoded repair sort component.
+ MaxRepairSortValueLength = 64 << 10
+)
+
+// ErrInvalidRepairPage identifies invalid page sizes or resume cursors.
+var ErrInvalidRepairPage = errors.New("inverted: invalid repair page request")
+
+// RepairCursor is an opaque position in one pinned generation, including the
+// document identity needed to resume across equal four-component tuples.
+type RepairCursor struct {
+ generation *ReadOnlyGeneration
+ cursor nativeice.RepairCursor
+}
+
+// RepairPageRequest selects a page in the fixed ascending Property repair order.
+// Sorting and projection cannot be customized.
+type RepairPageRequest struct {
+ // After must be a cursor returned by this generation; nil starts the scan.
+ After *RepairCursor
+ // PageSize must be positive and no greater than MaxRepairPageSize.
+ PageSize int
+}
+
+// RepairRow contains encoded (group, name, entity ID, timestamp) values and stored SHA.
+type RepairRow struct {
+ Cursor *RepairCursor
+ SortValues [][]byte
+ Value []byte
+}
+
+// ReadOnlyGeneration is a pinned, immutable read-only view of exactly one
+// committed generation of an index directory.
+//
+// The generation is chosen when the view is opened and fixed for its lifetime,
+// so a generation committed afterwards -- by a live writer, a merge or a
+// restore -- stays invisible to every operation the view serves. That is what
+// lets a caller page through an order across several calls and see one
+// consistent set of documents, which a sequence of independent opens could not
+// promise.
+//
+// A view takes no exclusive directory lock and writes no bytes, so a directory
+// a live writer owns can be read while it is being written, and reading leaves
+// file contents, modification times and directory entries unchanged.
+type ReadOnlyGeneration struct {
+ reader *nativeice.Reader
+ path string
+}
+
+// OpenReadOnlyGeneration opens the index directory at path, selects its newest
+// structurally complete committed generation and returns a view pinned to 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. A directory whose committed bytes violate the on-disk
+// grammar, and which therefore offers no generation to pin, reports an error
+// wrapping ErrCorruptIndex.
+func OpenReadOnlyGeneration(path string) (*ReadOnlyGeneration, error) {
+ reader, openErr := nativeice.Open(path)
+ if openErr != nil {
+ switch {
+ case errors.Is(openErr, nativeice.ErrNoSnapshot):
+ return nil, fmt.Errorf("open pinned generation %q: %w", path, errors.Join(ErrNoCommittedIndex, openErr))
+ case errors.Is(openErr, nativeice.ErrCorrupt):
+ return nil, fmt.Errorf("open pinned generation %q: %w", path, errors.Join(ErrCorruptIndex, openErr))
+ default:
+ return nil, fmt.Errorf("open pinned generation %q: %w", path, openErr)
+ }
+ }
+ return &ReadOnlyGeneration{reader: reader, path: path}, nil
+}
+
+// SnapshotID returns the identifier of the committed generation the view is
+// pinned to. It identifies the same generation the directory's own snapshot
+// manifests are numbered by, so a caller can record which generation a read
+// covered and later decide whether the directory has moved on.
+func (g *ReadOnlyGeneration) SnapshotID() uint64 {
+ if g == nil || g.reader == nil {
+ return 0
+ }
+ return g.reader.SnapshotID()
+}
+
+// RepairTuplePage returns a bounded page in ascending Property repair order.
+// Equal tuples are ordered by segment and local document identity. Missing sort
+// values and malformed sections return ErrCorruptIndex without a partial page.
+func (g *ReadOnlyGeneration) RepairTuplePage(ctx context.Context, request RepairPageRequest) ([]RepairRow, error) {
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ return nil, ctxErr
+ }
+ if g == nil || g.reader == nil {
+ return nil, fmt.Errorf("page unopened generation: %w", ErrInvalidRepairPage)
+ }
+ var after *nativeice.RepairCursor
+ if request.After != nil {
+ if request.After.generation != g {
+ return nil, fmt.Errorf("cursor belongs to another generation: %w", ErrInvalidRepairPage)
+ }
+ after = &request.After.cursor
+ }
+ nativeRows, pageErr := g.reader.RepairTuplePage(ctx, nativeice.RepairPageRequest{
+ SortFields: [RepairSortFieldCount]string{"_group", index.IndexModeName, "_entity_id", timestampField},
+ ProjectField: "_sha_value", After: after, PageSize: request.PageSize,
+ })
+ if pageErr != nil {
+ switch {
+ case errors.Is(pageErr, nativeice.ErrInvalidRepairPage):
+ return nil, errors.Join(ErrInvalidRepairPage, pageErr)
+ case errors.Is(pageErr, nativeice.ErrCorrupt):
+ return nil, errors.Join(ErrCorruptIndex, pageErr)
+ default:
+ return nil, pageErr
+ }
+ }
+ rows := make([]RepairRow, len(nativeRows))
+ for rowIndex, nativeRow := range nativeRows {
+ cursor := nativeRow.Cursor
+ cursor.SortValues = make([][]byte, len(nativeRow.SortValues))
+ for valueIndex, value := range nativeRow.SortValues {
+ cursor.SortValues[valueIndex] = append([]byte{}, value...)
+ }
+ rows[rowIndex] = RepairRow{SortValues: nativeRow.SortValues, Value: nativeRow.Value, Cursor: &RepairCursor{cursor: cursor, generation: g}}
+ }
+ return rows, nil
+}
+
+// Close releases the segment files and mapped regions the pinned generation
+// holds. A view that has been closed serves no further page.
+func (g *ReadOnlyGeneration) Close() error {
+ if g == nil || g.reader == nil {
+ return nil
+ }
+ if closeErr := g.reader.Close(); closeErr != nil {
+ return fmt.Errorf("close pinned generation %q: %w", g.path, closeErr)
+ }
+ return nil
+}
diff --git a/pkg/index/inverted/repair_page_tdd_test.go b/pkg/index/inverted/repair_page_tdd_test.go
new file mode 100644
index 000000000..1f5dbb403
--- /dev/null
+++ b/pkg/index/inverted/repair_page_tdd_test.go
@@ -0,0 +1,135 @@
+// 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"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/pkg/index"
+)
+
+func TestRepairTuplePageOrdersEqualTuplesByLocalDocumentNumber(t *testing.T) {
+ tester := require.New(t)
+ shardPath := t.TempDir()
+ writer, writerErr := NewStore(StoreOpts{Path: shardPath})
+ tester.NoError(writerErr)
+ documents := index.Documents{
+ repairTupleTieDocument("document-1", "sha-first"),
+ repairTupleTieDocument("document-2", "sha-second"),
+ }
+ tester.NoError(writer.UpdateSeriesBatch(index.Batch{Documents: documents}))
+ tester.NoError(writer.Close())
+ generation, generationErr := OpenReadOnlyGeneration(shardPath)
+ tester.NoError(generationErr)
+ t.Cleanup(func() {
+ tester.NoError(generation.Close())
+ })
+ page, pageErr := generation.RepairTuplePage(context.Background(), RepairPageRequest{
+ PageSize: 2,
+ })
+ tester.NoError(pageErr)
+ tester.Equal([]RepairRow{
+ {SortValues: [][]byte{[]byte("g-tie"), []byte("n-tie"), []byte("e-tie"), nidx01eEncodedTimestamps[10]}, Value: []byte("sha-first")},
+ {SortValues: [][]byte{[]byte("g-tie"), []byte("n-tie"), []byte("e-tie"), nidx01eEncodedTimestamps[10]}, Value: []byte("sha-second")},
+ }, nidx01eWithoutCursors(page))
+}
+
+func repairTupleTieDocument(identifier, shaValue string) index.Document {
+ return index.Document{
+ EntityValues: []byte(identifier),
+ Timestamp: 10,
+ Fields: []index.Field{
+ nidx01eSortableField(nidx01eEntityField, []byte("e-tie")),
+ nidx01eSortableField(nidx01eGroupField, []byte("g-tie")),
+ nidx01eSortableField(nidx01eNameField, []byte("n-tie")),
+ nidx01eStoredField(nidx01eSHAField, []byte(shaValue)),
+ },
+ }
+}
+
+// TestRepairTuplePageRejectsFieldsWithoutDocValueLocations verifies that stored
+// fields cannot substitute for the four required sortable doc values.
+func TestRepairTuplePageRejectsFieldsWithoutDocValueLocations(t *testing.T) {
+ tester := require.New(t)
+ shardPath := t.TempDir()
+ writer, writerErr := NewStore(StoreOpts{Path: shardPath})
+ tester.NoError(writerErr)
+ tester.NoError(writer.UpdateSeriesBatch(index.Batch{Documents: index.Documents{
+ repairTupleNoSortDocument("document-1", "sha-first"),
+ }}))
+ tester.NoError(writer.Close())
+ generation := nidx01eOpen(t, shardPath)
+ page, pageErr := generation.RepairTuplePage(context.Background(), RepairPageRequest{PageSize: 1})
+ tester.ErrorIs(pageErr, ErrCorruptIndex)
+ tester.Empty(page)
+}
+
+// TestRepairTuplePageResumesEqualTuples verifies that continuation includes the
+// physical row identity instead of skipping rows with the same visible tuple.
+func TestRepairTuplePageResumesEqualTuples(t *testing.T) {
+ tester := require.New(t)
+ shardPath := t.TempDir()
+ writer, writerErr := NewStore(StoreOpts{Path: shardPath})
+ tester.NoError(writerErr)
+ tester.NoError(writer.UpdateSeriesBatch(index.Batch{Documents: index.Documents{
+ repairTupleTieDocument("document-1", "sha-first"),
+ repairTupleTieDocument("document-2", "sha-second"),
+ }}))
+ tester.NoError(writer.Close())
+ generation := nidx01eOpen(t, shardPath)
+ first, firstErr := generation.RepairTuplePage(context.Background(), RepairPageRequest{PageSize: 1})
+ tester.NoError(firstErr)
+ tester.Len(first, 1)
+ tester.NotNil(first[0].Cursor)
+ second, secondErr := generation.RepairTuplePage(context.Background(), RepairPageRequest{PageSize: 1, After: first[0].Cursor})
+ tester.NoError(secondErr)
+ tester.Len(second, 1)
+ tester.Equal(first[0].SortValues, second[0].SortValues)
+ tester.Equal("sha-first", string(first[0].Value))
+ tester.Equal("sha-second", string(second[0].Value))
+ last, lastErr := generation.RepairTuplePage(context.Background(), RepairPageRequest{PageSize: 1, After: second[0].Cursor})
+ tester.NoError(lastErr)
+ tester.Empty(last)
+
+ otherGeneration := nidx01eOpen(t, shardPath)
+ foreign, foreignErr := otherGeneration.RepairTuplePage(context.Background(), RepairPageRequest{PageSize: 1, After: first[0].Cursor})
+ tester.ErrorIs(foreignErr, ErrInvalidRepairPage)
+ tester.Empty(foreign)
+}
+
+func repairTupleNoSortDocument(identifier, shaValue string) index.Document {
+ return index.Document{
+ EntityValues: []byte(identifier),
+ Timestamp: 0,
+ Fields: []index.Field{
+ repairTupleNoSortStoredField(nidx01eGroupField, []byte("g-no-sort")),
+ repairTupleNoSortStoredField(nidx01eNameField, []byte("n-no-sort")),
+ repairTupleNoSortStoredField(nidx01eEntityField, []byte("e-no-sort")),
+ repairTupleNoSortStoredField(nidx01eSHAField, []byte(shaValue)),
+ },
+ }
+}
+
+func repairTupleNoSortStoredField(tagName string, value []byte) index.Field {
+ field := index.NewBytesField(index.FieldKey{TagName: tagName}, value)
+ field.Store = true
+ field.NoSort = true
+ field.Index = false
+ return field
+}
diff --git a/pkg/index/inverted/testdata/nidx01e/provenance.json b/pkg/index/inverted/testdata/nidx01e/provenance.json
new file mode 100644
index 000000000..43c01b999
--- /dev/null
+++ b/pkg/index/inverted/testdata/nidx01e/provenance.json
@@ -0,0 +1,103 @@
+{
+ "oracle": {
+ "go_mod_sha256": "23076127a0a072d3a06b157fc063c483fe0d8293e0e77bea7fed76658423c28d",
+ "writer": "pkg/index/inverted.NewStore"
+ },
+ "file_sha256": {
+ "shard-0/000000000002.seg": "26e6b5f513816a2413e958ac33a8ebdbdaf19f9efb820c77c5314516ff211ab2",
+ "shard-0/000000000003.snp": "9637119e2fb9100019dd9822f3f64a22393de2937708f746d598b5a2a4fbccf5"
+ },
+ "generator_command": "GENERATE_NIDX01E_FIXTURE=1 go test ./pkg/index/inverted/ -run TestGenerateNIDX01EFixture -count=1",
+ "notes": "One Property shard generation holding five physical rows of three entities. Ascending by (_group, _im_name, _entity_id, _timestamp) the visible rows are (g-a,n-a,e-1,10,sha-a), (g-a,n-a,e-1,20,sha-b), (g-a,n-b,e-2,5,sha-c), (g-b,n-a,e-3,7,sha-d). Row (g-a,n-a,e-1,15,sha-x) is deleted in this generation and sorts between the first two visible rows, so a page that ignored the deletion masks is caught inside the first page rather than at the end. At page size 2 the first page is [...]
+ "rows": [
+ {
+ "group": "g-a",
+ "name": "n-a",
+ "entity_id": "e-1",
+ "sha_value": "sha-a",
+ "doc_id": "g-a/n-a/e-1/10",
+ "timestamp": 10,
+ "deleted": false
+ },
+ {
+ "group": "g-a",
+ "name": "n-a",
+ "entity_id": "e-1",
+ "sha_value": "sha-x",
+ "doc_id": "g-a/n-a/e-1/15",
+ "timestamp": 15,
+ "deleted": true
+ },
+ {
+ "group": "g-a",
+ "name": "n-a",
+ "entity_id": "e-1",
+ "sha_value": "sha-b",
+ "doc_id": "g-a/n-a/e-1/20",
+ "timestamp": 20,
+ "deleted": false
+ },
+ {
+ "group": "g-a",
+ "name": "n-b",
+ "entity_id": "e-2",
+ "sha_value": "sha-c",
+ "doc_id": "g-a/n-b/e-2/5",
+ "timestamp": 5,
+ "deleted": false
+ },
+ {
+ "group": "g-b",
+ "name": "n-a",
+ "entity_id": "e-3",
+ "sha_value": "sha-d",
+ "doc_id": "g-b/n-a/e-3/7",
+ "timestamp": 7,
+ "deleted": false
+ }
+ ],
+ "declared_pages": [
+ [
+ {
+ "group": "g-a",
+ "name": "n-a",
+ "entity_id": "e-1",
+ "sha_value": "sha-a",
+ "doc_id": "g-a/n-a/e-1/10",
+ "timestamp": 10,
+ "deleted": false
+ },
+ {
+ "group": "g-a",
+ "name": "n-a",
+ "entity_id": "e-1",
+ "sha_value": "sha-b",
+ "doc_id": "g-a/n-a/e-1/20",
+ "timestamp": 20,
+ "deleted": false
+ }
+ ],
+ [
+ {
+ "group": "g-a",
+ "name": "n-b",
+ "entity_id": "e-2",
+ "sha_value": "sha-c",
+ "doc_id": "g-a/n-b/e-2/5",
+ "timestamp": 5,
+ "deleted": false
+ },
+ {
+ "group": "g-b",
+ "name": "n-a",
+ "entity_id": "e-3",
+ "sha_value": "sha-d",
+ "doc_id": "g-b/n-a/e-3/7",
+ "timestamp": 7,
+ "deleted": false
+ }
+ ]
+ ],
+ "snapshot_id": 3,
+ "visible_row_count": 4
+}
diff --git a/pkg/index/inverted/testdata/nidx01e/shard-0/000000000002.seg b/pkg/index/inverted/testdata/nidx01e/shard-0/000000000002.seg
new file mode 100644
index 000000000..f7b115a3a
Binary files /dev/null and b/pkg/index/inverted/testdata/nidx01e/shard-0/000000000002.seg differ
diff --git a/pkg/index/inverted/testdata/nidx01e/shard-0/000000000003.snp b/pkg/index/inverted/testdata/nidx01e/shard-0/000000000003.snp
new file mode 100644
index 000000000..802bffadb
Binary files /dev/null and b/pkg/index/inverted/testdata/nidx01e/shard-0/000000000003.snp differ