Re: [PR] feat(index): complete the Property native cutover and serve Q1-Q3 from the native index [skywalking-banyandb]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #1390:
URL: https://github.com/apache/skywalking-banyandb/pull/1390#discussion_r4193985729


##########
banyand/internal/storage/tsdb.go:
##########
@@ -244,17 +254,30 @@ func OpenTSDB[T TSTable, O any](ctx context.Context, opts TSDBOpts[T, O], cache
 	}
 	db.lock = lock
 	// Release the lock fd if any subsequent step in OpenTSDB returns an error.
-	// Otherwise the file descriptor leaks until process exit and a retry on the
-	// same data dir hits "resource temporarily unavailable" from the next
+	// Otherwise the file descriptor leaks until process exit and a retry on
+	// the same data dir hits "resource temporarily unavailable" from the next
 	// CreateLockFile call. Ownership transfers to (*database).Close() only on
 	// success — released = true is set on the success return below.
 	released := false
 	defer func() {
 		if !released {
+			if revoker, ok := db.rootLease.(RootLeaseRevoker); ok {
+				_ = revoker.Revoke()
+			}

Review Comment:
   If startup opens one segment successfully and then fails on a later segment, the earlier segment remains in `segmentController.lst` with its tables and native owner workers running. This defer revokes their lease and releases the root lock without draining them. Close the successfully opened segments before revoking the lease, matching the normal database shutdown order. Add a regression test with a valid first segment and invalid metadata in a later segment.



##########
banyand/stream/write_standalone.go:
##########
@@ -340,7 +340,7 @@ func (w *writeCallback) Rev(_ context.Context, message bus.Message) (resp bus.Me
 			releaseElements(es.elements)
 			if len(es.docs) > 0 {
 				index := es.tsTable.Index()
-				if err := index.Write(es.docs); err != nil {
+				if err := index.WriteContext(ctx, es.docs); err != nil {

Review Comment:
   `mustAddElements` has already admitted the raw rows before this call. The local batch publisher retains the RPC context and passes it here when closing the batch. If that context is canceled before native `Owner.Batch` starts, index admission is rejected, but this handler only logs the error and keeps the raw rows. Indexed queries then miss those rows. Once raw admission has happened, finish index admission independently of request cancellation, while retaining the owner's shutdown checks, or provide coordinated rollback.



##########
pkg/index/internal/nativeice/dictionary_iterator.go:
##########
@@ -132,6 +189,53 @@ func (i *DictionaryTermIterator) Next() ([]byte, uint64, error) {
 	return term, count, nil
 }
 
+// NextTerm returns the next dictionary term without decoding its posting bitmap.
+// A nil term and nil error indicate exhaustion.
+func (i *DictionaryTermIterator) NextTerm() ([]byte, error) {
+	if i.closed || i.iterator == nil {
+		return nil, nil
+	}
+	term, postingOffset := i.iterator.Current()
+	if term == nil && postingOffset == 0 {
+		return nil, i.closeIterator()
+	}
+	owned := append([]byte(nil), term...)

Review Comment:
   A valid empty dictionary key becomes nil here, so `NextTerm` returns the same `(nil, nil)` result as exhaustion even when more keys follow. Preserve a non-nil empty slice, as `nextEntry` already does, and add a regression test that reads an empty key followed by a nonempty key.



##########
pkg/index/nativeadapter/native_dependency_test.go:
##########
@@ -0,0 +1,40 @@
+// Licensed to the 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 nativeadapter
+
+import (
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAdapterHasNoRetiredTransitiveDependencies(t *testing.T) {
+	_, sourceFile, _, ok := runtime.Caller(0)
+	require.True(t, ok)
+	command := exec.Command("go", "list", "-deps", ".")
+	command.Dir = filepath.Dir(sourceFile)
+	output, err := command.Output()
+	require.NoError(t, err)
+	for _, dependency := range strings.Split(strings.TrimSpace(string(output)), "\n") {
+		require.NotContains(t, dependency, "/pkg/index/inverted")
+		require.NotContains(t, dependency, "github.com/blugelabs/bluge")
+		require.NotContains(t, dependency, "github.com/blugelabs/bluge_segment_api")

Review Comment:
   The PR description exempts dependency-guard and compatibility tests from its lexical claim, but the linked [NIDX-02 acceptance gate](https://github.com/apache/skywalking/issues/14007) explicitly includes tests. These new literals, the corresponding literals in `pkg/index/native/native_dependency_test.go`, and the imports in `pkg/index/inverted/legacy_merge_field_order_test.go` add the forbidden token. Reuse existing neutral compatibility helpers to meet the gate, or obtain and document an explicit exception covering these tests.



##########
pkg/index/native/lifecycle.go:
##########
@@ -0,0 +1,157 @@
+// 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 (
+	"bytes"
+	"fmt"
+	"path/filepath"
+
+	roaringpkg "github.com/RoaringBitmap/roaring"
+
+	"github.com/apache/skywalking-banyandb/pkg/index/internal/nativeice"
+)
+
+// Stats returns the current live document count and immutable segment bytes.
+// It observes the current in-memory root and does not reopen a cold directory.
+func (o *Owner) Stats() (dataCount int64, dataSizeBytes int64) {
+	if o == nil {
+		return 0, 0
+	}
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	if o.root == nil {
+		return 0, 0
+	}
+	for _, current := range o.root.segments {
+		segment, ok := current.(*memorySegment)
+		if !ok {
+			continue
+		}
+		live := segment.handle.count - uint64(len(segment.deleted))
+		dataCount += int64(live)
+		dataSizeBytes += int64(segment.handle.size)
+	}
+	return dataCount, dataSizeBytes
+}
+
+// Reset publishes an empty root. Existing pinned views remain readable until
+// they close; the reset itself is persisted through the same background
+// worker as normal writes.
+func (o *Owner) Reset() error {
+	if o == nil {
+		return nil
+	}
+	queue := o.persistQ
+	o.mu.Lock()
+	if o.closed || o.closing {
+		o.mu.Unlock()
+		return ErrOwnerClosed
+	}
+	if o.durabilityFault != nil {
+		fault := o.durabilityFault
+		o.mu.Unlock()
+		return fault
+	}
+	if o.root.generation == ^uint64(0) {
+		o.mu.Unlock()
+		return fmt.Errorf("native generation exhausted: %w", ErrInvalidDocument)
+	}
+	next := &publishedRoot{generation: o.root.generation + 1}
+	next.refs.Store(1)
+	old := o.root
+	o.root = next
+	o.admittedIdentifiers, o.admittedSegments = nil, nil
+	o.roots[next] = struct{}{}
+	old.release()
+	o.pruneRootsLocked()
+	if queue == nil {
+		o.mu.Unlock()
+		return nil
+	}
+	o.schedulePersistLocked(queue, nil, false)
+	o.mu.Unlock()
+	return nil
+}
+
+// TakeFileSnapshot publishes a copy of the current pinned root. Unlike a
+// committed-directory copy, this includes admitted NRT segments even when
+// their asynchronous persistence has not completed. The destination manifest
+// is installed last by nativeice.PublishSnapshot.
+func (o *Owner) TakeFileSnapshot(destination string) error {
+	if o == nil || o.options.Path == "" {
+		return ErrPersistenceConfiguration
+	}
+	o.mu.Lock()
+	if o.closed || o.closing {
+		o.mu.Unlock()
+		return ErrOwnerClosed
+	}
+	if err := o.validateLease(); err != nil {
+		o.mu.Unlock()
+		return fmt.Errorf("validate native root lease for snapshot: %w", err)
+	}
+	o.activeOps++
+	root := o.root
+	root.refs.Add(1)
+	o.mu.Unlock()
+	defer o.endOperation()
+	defer root.release()
+
+	segments := make([]nativeice.SnapshotSegmentPayload, 0, len(root.segments))
+	for _, current := range root.segments {
+		segment, ok := current.(*memorySegment)
+		if !ok {
+			return fmt.Errorf("snapshot native root: unsupported segment type %T", current)
+		}
+		if segmentHasNoLiveDocuments(segment) {
+			continue
+		}
+		metadata := nativeice.SnapshotSegment{
+			ID: segment.handle.id, Size: segment.handle.size, DocumentCount: segment.handle.count,
+			TimeMin: segment.handle.timeMin, TimeMax: segment.handle.timeMax,
+		}
+		if len(segment.deleted) != 0 {
+			deleted := roaringpkg.New()
+			for number := range segment.deleted {
+				if number > uint64(^uint32(0)) {
+					return fmt.Errorf("snapshot deletion %d exceeds mask range: %w", number, ErrInvalidDocument)
+				}
+				deleted.Add(uint32(number))
+			}
+			bitmap, marshalErr := deleted.MarshalBinary()
+			if marshalErr != nil {
+				return fmt.Errorf("snapshot deletion mask: %w", marshalErr)
+			}
+			metadata.DeletionBitmap = bitmap
+		}
+		payload := segment.handle.payload
+		sourcePath := segment.handle.sourcePath
+		if payload == nil && sourcePath == "" {
+			var readErr error
+			payload, readErr = fileSystem.Read(filepath.Join(o.options.Path, fmt.Sprintf("%012x.seg", segment.handle.id)))
+			if readErr != nil {
+				return fmt.Errorf("read persisted native segment %d for snapshot: %w", segment.handle.id, readErr)
+			}
+		}

Review Comment:
   Reopened and promoted disk-backed handles have neither a payload nor a source path. This branch reads each segment in full, then retains its clone until all segments are ready for publication. Snapshot heap usage therefore grows with the entire persisted index and can exhaust memory during backup or repair. Pass the persisted filename as `SourcePath` instead; `PublishSnapshot` already supports streaming that file, and the pinned root protects it during the copy.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org