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/10/22 08:12:54 UTC
(skywalking-banyandb) 01/01: fix(index): preserve compaction across persistence promotion
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch fix/e2e-docker-image-store
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit a7990cde152d0e1e50c54cb84e7fff975d89e0cf
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Oct 22 08:12:38 2026 +0000
fix(index): preserve compaction across persistence promotion
---
CHANGES.md | 1 +
pkg/index/native/compaction_promotion_test.go | 132 ++++++++++++++++++++++++++
pkg/index/native/owner.go | 31 +++---
3 files changed, 148 insertions(+), 16 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 9b807bf54..06b3e3438 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -41,6 +41,7 @@ Release Notes.
### Bug Fixes
+- Keep native index compaction progressing when persistence replaces unchanged in-memory segments with file-backed readers.
- Upgrade the Go toolchain to 1.26.10 for the crypto/x509 security fix (GO-2026-6809).
- Return zero for empty scalar QL `COUNT(DISTINCT)` and apply its pagination after the complete count instead of pruning native shard partials.
- Preserve indexed-array measure values, empty-entity selector semantics and retained merge-tail buffers in shared batch execution.
diff --git a/pkg/index/native/compaction_promotion_test.go b/pkg/index/native/compaction_promotion_test.go
new file mode 100644
index 000000000..5f8bcd3ab
--- /dev/null
+++ b/pkg/index/native/compaction_promotion_test.go
@@ -0,0 +1,132 @@
+// 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 native
+
+import (
+ "context"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestOwnerCompactionSurvivesPersistencePromotion(t *testing.T) {
+ cases := []struct {
+ name string
+ automatic bool
+ deleteDuringMerge bool
+ }{
+ {name: "manual"},
+ {name: "automatic", automatic: true},
+ {name: "concurrent delete remains stale", deleteDuringMerge: true},
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ path := t.TempDir()
+ lease := &gatedPathLease{reached: make(chan struct{}), release: make(chan struct{})}
+ var releaseOnce sync.Once
+ release := func() { releaseOnce.Do(func() { close(lease.release) }) }
+ persisted := make(chan error, 1)
+ threshold := -1
+ if tc.automatic {
+ threshold = 2
+ }
+ var owner *Owner
+ var prepareOnce sync.Once
+ var prepareErr error
+ var err error
+ owner, err = NewOwner(OwnerOptions{
+ Lease: lease, Path: path, CompactionThreshold: threshold,
+ PrepareMergeCallback: func(mergeCtx context.Context, _ MergeDocument) (bool, error) {
+ prepareOnce.Do(func() {
+ // Compaction has captured the inputs. Complete the first persist now,
+ // forcing its file-backed replacements into the current root before
+ // compaction checks whether those inputs are still live.
+ release()
+ select {
+ case prepareErr = <-persisted:
+ case <-ctx.Done():
+ prepareErr = ctx.Err()
+ case <-mergeCtx.Done():
+ prepareErr = mergeCtx.Err()
+ }
+ if prepareErr == nil && tc.deleteDuringMerge {
+ prepareErr = owner.Batch(mergeCtx, Batch{Deletes: [][]byte{[]byte("series-a")}})
+ }
+ })
+ return false, prepareErr
+ },
+ })
+ require.NoError(t, err)
+ t.Cleanup(func() {
+ cancel()
+ release()
+ require.NoError(t, owner.Close())
+ })
+ require.NoError(t, owner.Batch(ctx, Batch{
+ Documents: []Document{{Identifier: []byte("series-a")}, {Identifier: []byte("deleted")}},
+ PersistentCallback: func(persistErr error) { persisted <- persistErr },
+ }))
+ // The third path validation is the first persist: construction and the
+ // first admission ran before it. Hold that persist while admitting b.
+ select {
+ case <-lease.reached:
+ case <-ctx.Done():
+ t.Fatal("persistence did not reach its lease barrier")
+ }
+ require.NoError(t, owner.Batch(ctx, Batch{
+ Documents: []Document{{Identifier: []byte("series-b")}},
+ Deletes: [][]byte{[]byte("deleted")},
+ }))
+ if !tc.automatic {
+ compactErr := owner.Compact(ctx)
+ if tc.deleteDuringMerge {
+ require.ErrorIs(t, compactErr, ErrStaleCompaction)
+ } else {
+ require.NoError(t, compactErr)
+ }
+ }
+ if !tc.deleteDuringMerge {
+ require.Eventually(t, func() bool {
+ owner.mu.Lock()
+ defer owner.mu.Unlock()
+ return len(owner.root.segments) == 1
+ }, 5*time.Second, time.Millisecond, "promotion must not strand compaction")
+ }
+ checkDocuments := func(current *Owner) {
+ view, acquireErr := current.Acquire(ctx)
+ require.NoError(t, acquireErr)
+ defer func() { require.NoError(t, view.Close()) }()
+ for id, want := range map[string]bool{"series-a": !tc.deleteDuringMerge, "series-b": true, "deleted": false} {
+ _, found, lookupErr := view.Lookup(ctx, []byte(id))
+ require.NoError(t, lookupErr)
+ require.Equal(t, want, found, id)
+ }
+ }
+ checkDocuments(owner)
+ require.NoError(t, owner.Close())
+ reopened, reopenErr := NewOwner(OwnerOptions{Lease: testLease{}, Path: path, CompactionThreshold: -1})
+ require.NoError(t, reopenErr)
+ defer func() { require.NoError(t, reopened.Close()) }()
+ checkDocuments(reopened)
+ })
+ }
+}
diff --git a/pkg/index/native/owner.go b/pkg/index/native/owner.go
index 4334d21a4..ae2e4cdf2 100644
--- a/pkg/index/native/owner.go
+++ b/pkg/index/native/owner.go
@@ -20,6 +20,7 @@ import (
"context"
"errors"
"fmt"
+ "maps"
"path/filepath"
"sort"
"strconv"
@@ -1621,11 +1622,11 @@ func (o *Owner) compact(ctx context.Context, plan func([]mergeCandidate) []merge
task := tasks[0]
inputs := make([]nativeice.MergeInput, 0, len(task.candidates))
- taskSegments := make(map[rootSegment]struct{}, len(task.candidates))
+ taskSegments := make(map[uint64]*memorySegment, len(task.candidates))
mergeDroppedAny := false
for _, candidate := range task.candidates {
segment := candidate.segment.(*memorySegment)
- taskSegments[candidate.segment] = struct{}{}
+ taskSegments[segment.handle.id] = segment
drop, prepareErr := o.prepareMergeDrop(ctx, segment)
if prepareErr != nil {
base.release()
@@ -1705,14 +1706,11 @@ func (o *Owner) compact(ctx context.Context, plan func([]mergeCandidate) []merge
// preload), some newer root almost always exists by the time this merge
// finishes; discarding the merge outright on every such write would
// starve compaction indefinitely. remainder/found below check only the
- // segments this task actually merged, by identity: any of them replaced
- // by a concurrent Delete (which clones and replaces its segment rather
- // than mutating it in place) is detected and the whole task is discarded
- // as stale, but any other concurrent admission -- which only adds
- // segments or replaces segments outside this task -- cannot invalidate
- // it. That lets the compacted result replace exactly the segments it
- // merged while every other segment, old or newly admitted, carries
- // forward untouched.
+ // segments this task actually merged, by stable ID and deletion mask.
+ // Persistence replaces their wrappers and readers when promoting them to
+ // files, but preserves both the ID and live documents. That must not make
+ // a merge stale. A concurrent Delete changes the mask and still invalidates
+ // the task; admissions outside the task carry forward untouched.
// A concurrent compaction may already have merged away some of this
// task's segments, leaving current with fewer segments than the task;
// the capacity must not go negative (a panic here, under o.mu, would
@@ -1721,9 +1719,11 @@ func (o *Owner) compact(ctx context.Context, plan func([]mergeCandidate) []merge
remaining := make([]rootSegment, 0, max(len(current.segments)-len(taskSegments)+1, 1))
found := 0
for _, segment := range current.segments {
- if _, ok := taskSegments[segment]; ok {
- found++
- continue
+ if live, ok := segment.(*memorySegment); ok {
+ if input, merged := taskSegments[live.handle.id]; merged && (live == input || maps.Equal(live.deleted, input.deleted)) {
+ found++
+ continue
+ }
}
remaining = append(remaining, segment)
}
@@ -1753,9 +1753,8 @@ func (o *Owner) compact(ctx context.Context, plan func([]mergeCandidate) []merge
o.nextSegmentID++
old.release()
o.pruneRootsLocked()
- for segment := range taskSegments {
- handle := segment.(*memorySegment).handle
- o.unindexSegmentLocked(handle.id)
+ for mergedID := range taskSegments {
+ o.unindexSegmentLocked(mergedID)
}
if mergeDroppedAny {
// A merge-dropped identifier is not individually known here (the