Copilot commented on code in PR #967:
URL:
https://github.com/apache/skywalking-banyandb/pull/967#discussion_r2776690180
##########
banyand/internal/sidx/sidx.go:
##########
@@ -248,6 +248,79 @@ func (s *sidx) currentSnapshot() *snapshot {
return nil
}
+// CurrentSnapshot returns the current snapshot with incremented reference
count.
+// Implements snapshot.Manager[*snapshot] interface.
Review Comment:
The doc comment says this implements `snapshot.Manager[*snapshot]`, but the
method returns `*Snapshot`. This looks like a stale type name from before
`snapshot` was renamed/exported; updating the comment would avoid confusion for
readers and future refactors.
```suggestion
// Implements snapshot.Manager[*Snapshot] interface.
```
##########
banyand/internal/snapshot/snapshot.go:
##########
@@ -0,0 +1,219 @@
+// 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 snapshot provides generic transaction coordination for
snapshot-based systems.
+// It enables atomic updates across multiple heterogeneous snapshot managers
using Go generics.
+package snapshot
+
+import (
+ "reflect"
+ "sync"
+
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+)
+
+// Snapshot is a type constraint for snapshot types that support reference
counting.
+// Any type implementing this interface can participate in atomic transactions.
+type Snapshot interface {
+ // IncRef increments the reference count.
+ IncRef()
+ // DecRef decrements the reference count and releases resources when
zero.
+ DecRef()
+}
+
+// Manager is a generic interface for managing snapshots of type S.
+// Both trace and sidx implement this interface for their respective snapshot
types.
+type Manager[S Snapshot] interface {
+ // CurrentSnapshot returns the current snapshot with incremented ref
count.
+ // Returns nil if no snapshot exists.
+ CurrentSnapshot() S
+ // ReplaceSnapshot atomically replaces the current snapshot with next.
+ // The old snapshot's DecRef is called automatically.
+ ReplaceSnapshot(next S)
+}
+
+// Transition represents a prepared but uncommitted snapshot change.
+// It holds both the current and next snapshots, allowing atomic commit or
rollback.
+type Transition[S Snapshot] struct {
+ manager Manager[S]
+ current S
+ next S
+ committed bool
+}
+
+// getTransitionPool returns or creates a pool for the given snapshot type.
+func getTransitionPool[S Snapshot]() *pool.Synced[any] {
+ var zero S
+ typ := reflect.TypeOf(zero)
+
+ if p, ok := transitionPools.Load(typ); ok {
+ return p.(*pool.Synced[any])
+ }
+
+ // Create new pool for this type
+ poolName := "snapshot.Transition[" + typ.String() + "]"
+ p := pool.Register[any](poolName)
+ actual, _ := transitionPools.LoadOrStore(typ, p)
+ return actual.(*pool.Synced[any])
+}
+
+// NewTransition creates a transition by preparing the next snapshot from the
pool.
+// The prepareNext function receives the current snapshot and returns the next.
+func NewTransition[S Snapshot](manager Manager[S], prepareNext func(current S)
S) *Transition[S] {
+ p := getTransitionPool[S]()
+
+ var t *Transition[S]
+ if pooled := p.Get(); pooled != nil {
+ t = pooled.(*Transition[S])
+ } else {
+ t = &Transition[S]{}
+ }
+
+ current := manager.CurrentSnapshot()
+ next := prepareNext(current)
+
+ t.manager = manager
+ t.current = current
+ t.next = next
+ t.committed = false
+
+ return t
+}
+
+// Commit commits the transition, replacing the current snapshot with next.
+// ReplaceSnapshot satisfies the Manager contract by calling DecRef on the old
snapshot.
+func (t *Transition[S]) Commit() {
+ if t.committed {
+ return
+ }
+ t.committed = true
+ t.manager.ReplaceSnapshot(t.next)
+}
+
+// Rollback discards the prepared next snapshot without committing.
+func (t *Transition[S]) Rollback() {
+ if t.committed {
+ return
+ }
+ // Release the prepared next snapshot
+ // Use reflection to check if underlying value is nil (Go interface
quirk)
+ if !reflect.ValueOf(t.next).IsNil() {
+ t.next.DecRef()
+ }
+ // Release reference to current (acquired in NewTransition)
+ if !reflect.ValueOf(t.current).IsNil() {
Review Comment:
Same issue as above for `t.current`: `reflect.ValueOf(t.current).IsNil()`
can panic if `S` isn’t nil-able or if it’s a nil interface. This makes the
generic snapshot transaction fragile outside the current pointer-only usages.
```suggestion
// isNilSnapshot safely determines whether a snapshot-like value is nil.
// It avoids calling reflect.Value.IsNil on non-nilable kinds, which would
panic.
func isNilSnapshot(v any) bool {
if v == nil {
return true
}
rv := reflect.ValueOf(v)
switch rv.Kind() {
case reflect.Ptr, reflect.Slice, reflect.Map, reflect.Chan,
reflect.Func, reflect.Interface:
return rv.IsNil()
default:
// Non-nilable kinds (struct, array, etc.) are never nil.
return false
}
}
// Rollback discards the prepared next snapshot without committing.
func (t *Transition[S]) Rollback() {
if t.committed {
return
}
// Release the prepared next snapshot
if !isNilSnapshot(t.next) {
t.next.DecRef()
}
// Release reference to current (acquired in NewTransition)
if !isNilSnapshot(t.current) {
```
##########
banyand/internal/snapshot/snapshot.go:
##########
@@ -0,0 +1,219 @@
+// 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 snapshot provides generic transaction coordination for
snapshot-based systems.
+// It enables atomic updates across multiple heterogeneous snapshot managers
using Go generics.
+package snapshot
+
+import (
+ "reflect"
+ "sync"
+
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+)
+
+// Snapshot is a type constraint for snapshot types that support reference
counting.
+// Any type implementing this interface can participate in atomic transactions.
+type Snapshot interface {
+ // IncRef increments the reference count.
+ IncRef()
+ // DecRef decrements the reference count and releases resources when
zero.
+ DecRef()
+}
+
+// Manager is a generic interface for managing snapshots of type S.
+// Both trace and sidx implement this interface for their respective snapshot
types.
+type Manager[S Snapshot] interface {
+ // CurrentSnapshot returns the current snapshot with incremented ref
count.
+ // Returns nil if no snapshot exists.
+ CurrentSnapshot() S
+ // ReplaceSnapshot atomically replaces the current snapshot with next.
+ // The old snapshot's DecRef is called automatically.
+ ReplaceSnapshot(next S)
+}
+
+// Transition represents a prepared but uncommitted snapshot change.
+// It holds both the current and next snapshots, allowing atomic commit or
rollback.
+type Transition[S Snapshot] struct {
+ manager Manager[S]
+ current S
+ next S
+ committed bool
+}
+
+// getTransitionPool returns or creates a pool for the given snapshot type.
+func getTransitionPool[S Snapshot]() *pool.Synced[any] {
+ var zero S
+ typ := reflect.TypeOf(zero)
+
+ if p, ok := transitionPools.Load(typ); ok {
+ return p.(*pool.Synced[any])
+ }
+
+ // Create new pool for this type
+ poolName := "snapshot.Transition[" + typ.String() + "]"
+ p := pool.Register[any](poolName)
+ actual, _ := transitionPools.LoadOrStore(typ, p)
+ return actual.(*pool.Synced[any])
+}
+
+// NewTransition creates a transition by preparing the next snapshot from the
pool.
+// The prepareNext function receives the current snapshot and returns the next.
+func NewTransition[S Snapshot](manager Manager[S], prepareNext func(current S)
S) *Transition[S] {
+ p := getTransitionPool[S]()
+
+ var t *Transition[S]
+ if pooled := p.Get(); pooled != nil {
+ t = pooled.(*Transition[S])
+ } else {
+ t = &Transition[S]{}
+ }
+
+ current := manager.CurrentSnapshot()
+ next := prepareNext(current)
+
+ t.manager = manager
+ t.current = current
+ t.next = next
+ t.committed = false
+
+ return t
+}
+
+// Commit commits the transition, replacing the current snapshot with next.
+// ReplaceSnapshot satisfies the Manager contract by calling DecRef on the old
snapshot.
+func (t *Transition[S]) Commit() {
+ if t.committed {
+ return
+ }
+ t.committed = true
+ t.manager.ReplaceSnapshot(t.next)
+}
+
+// Rollback discards the prepared next snapshot without committing.
+func (t *Transition[S]) Rollback() {
+ if t.committed {
+ return
+ }
+ // Release the prepared next snapshot
+ // Use reflection to check if underlying value is nil (Go interface
quirk)
+ if !reflect.ValueOf(t.next).IsNil() {
+ t.next.DecRef()
+ }
+ // Release reference to current (acquired in NewTransition)
+ if !reflect.ValueOf(t.current).IsNil() {
Review Comment:
`Transition.Rollback` uses `reflect.ValueOf(t.next).IsNil()` as a nil check.
`IsNil()` panics for non-nilable kinds (e.g., structs) and for invalid
`reflect.Value` (e.g., nil interface values), but the `Snapshot` constraint
doesn’t enforce nil-able types. Consider constraining `Snapshot`/`Manager` to
nil-able types or guarding the reflection check with `v.IsValid()` + kind
checks before calling `IsNil()`.
```suggestion
// isNilSnapshot safely determines whether a Snapshot value is nil.
// It avoids calling reflect.Value.IsNil on non-nilable kinds or invalid
values.
func isNilSnapshot[S Snapshot](s S) bool {
v := reflect.ValueOf(s)
if !v.IsValid() {
return true
}
switch v.Kind() {
case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map,
reflect.Ptr, reflect.Slice:
return v.IsNil()
default:
// Non-nilable kinds are never considered nil here.
return false
}
}
// Rollback discards the prepared next snapshot without committing.
func (t *Transition[S]) Rollback() {
if t.committed {
return
}
// Release the prepared next snapshot
if !isNilSnapshot(t.next) {
t.next.DecRef()
}
// Release reference to current (acquired in NewTransition)
if !isNilSnapshot(t.current) {
```
##########
banyand/internal/snapshot/snapshot.go:
##########
@@ -0,0 +1,219 @@
+// 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 snapshot provides generic transaction coordination for
snapshot-based systems.
+// It enables atomic updates across multiple heterogeneous snapshot managers
using Go generics.
+package snapshot
+
+import (
+ "reflect"
+ "sync"
+
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+)
+
+// Snapshot is a type constraint for snapshot types that support reference
counting.
+// Any type implementing this interface can participate in atomic transactions.
+type Snapshot interface {
+ // IncRef increments the reference count.
+ IncRef()
+ // DecRef decrements the reference count and releases resources when
zero.
+ DecRef()
+}
+
+// Manager is a generic interface for managing snapshots of type S.
+// Both trace and sidx implement this interface for their respective snapshot
types.
+type Manager[S Snapshot] interface {
+ // CurrentSnapshot returns the current snapshot with incremented ref
count.
+ // Returns nil if no snapshot exists.
+ CurrentSnapshot() S
+ // ReplaceSnapshot atomically replaces the current snapshot with next.
+ // The old snapshot's DecRef is called automatically.
+ ReplaceSnapshot(next S)
+}
+
+// Transition represents a prepared but uncommitted snapshot change.
+// It holds both the current and next snapshots, allowing atomic commit or
rollback.
+type Transition[S Snapshot] struct {
+ manager Manager[S]
+ current S
+ next S
+ committed bool
+}
+
+// getTransitionPool returns or creates a pool for the given snapshot type.
+func getTransitionPool[S Snapshot]() *pool.Synced[any] {
+ var zero S
+ typ := reflect.TypeOf(zero)
+
+ if p, ok := transitionPools.Load(typ); ok {
+ return p.(*pool.Synced[any])
+ }
+
+ // Create new pool for this type
+ poolName := "snapshot.Transition[" + typ.String() + "]"
+ p := pool.Register[any](poolName)
+ actual, _ := transitionPools.LoadOrStore(typ, p)
+ return actual.(*pool.Synced[any])
+}
+
+// NewTransition creates a transition by preparing the next snapshot from the
pool.
+// The prepareNext function receives the current snapshot and returns the next.
+func NewTransition[S Snapshot](manager Manager[S], prepareNext func(current S)
S) *Transition[S] {
+ p := getTransitionPool[S]()
+
+ var t *Transition[S]
+ if pooled := p.Get(); pooled != nil {
+ t = pooled.(*Transition[S])
+ } else {
+ t = &Transition[S]{}
+ }
+
+ current := manager.CurrentSnapshot()
+ next := prepareNext(current)
+
+ t.manager = manager
+ t.current = current
+ t.next = next
+ t.committed = false
+
+ return t
+}
+
+// Commit commits the transition, replacing the current snapshot with next.
+// ReplaceSnapshot satisfies the Manager contract by calling DecRef on the old
snapshot.
+func (t *Transition[S]) Commit() {
+ if t.committed {
+ return
+ }
+ t.committed = true
+ t.manager.ReplaceSnapshot(t.next)
+}
+
+// Rollback discards the prepared next snapshot without committing.
+func (t *Transition[S]) Rollback() {
+ if t.committed {
+ return
+ }
+ // Release the prepared next snapshot
+ // Use reflection to check if underlying value is nil (Go interface
quirk)
+ if !reflect.ValueOf(t.next).IsNil() {
+ t.next.DecRef()
+ }
+ // Release reference to current (acquired in NewTransition)
+ if !reflect.ValueOf(t.current).IsNil() {
+ t.current.DecRef()
+ }
+}
+
+// Release returns the transition to the pool for reuse.
+// This should be called after Commit or Rollback to reduce allocations.
+func (t *Transition[S]) Release() {
+ t.reset()
+ p := getTransitionPool[S]()
+ p.Put(any(t))
+}
+
+// reset clears the transition state for reuse.
+// When the transition was committed, the ref acquired in NewTransition
(t.current) was not
+// decremented by ReplaceSnapshot (which only decrements the manager's ref),
so we release it here.
+func (t *Transition[S]) reset() {
+ var zero S
+ if t.committed && !reflect.ValueOf(t.current).IsNil() {
+ t.current.DecRef()
+ }
Review Comment:
`reset` also relies on `reflect.ValueOf(t.current).IsNil()`; this has the
same panic risk for non-nilable `S` or nil interfaces. If the API intends `S`
to always be a pointer, it would be safer to enforce that in the type
constraint / doc, otherwise harden the nil check as in `Rollback`.
##########
banyand/internal/snapshot/snapshot.go:
##########
@@ -0,0 +1,219 @@
+// 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 snapshot provides generic transaction coordination for
snapshot-based systems.
+// It enables atomic updates across multiple heterogeneous snapshot managers
using Go generics.
+package snapshot
+
+import (
+ "reflect"
+ "sync"
+
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+)
+
+// Snapshot is a type constraint for snapshot types that support reference
counting.
+// Any type implementing this interface can participate in atomic transactions.
+type Snapshot interface {
+ // IncRef increments the reference count.
+ IncRef()
+ // DecRef decrements the reference count and releases resources when
zero.
+ DecRef()
+}
+
+// Manager is a generic interface for managing snapshots of type S.
+// Both trace and sidx implement this interface for their respective snapshot
types.
+type Manager[S Snapshot] interface {
+ // CurrentSnapshot returns the current snapshot with incremented ref
count.
+ // Returns nil if no snapshot exists.
+ CurrentSnapshot() S
+ // ReplaceSnapshot atomically replaces the current snapshot with next.
+ // The old snapshot's DecRef is called automatically.
+ ReplaceSnapshot(next S)
+}
+
+// Transition represents a prepared but uncommitted snapshot change.
+// It holds both the current and next snapshots, allowing atomic commit or
rollback.
+type Transition[S Snapshot] struct {
+ manager Manager[S]
+ current S
+ next S
+ committed bool
+}
+
+// getTransitionPool returns or creates a pool for the given snapshot type.
+func getTransitionPool[S Snapshot]() *pool.Synced[any] {
+ var zero S
+ typ := reflect.TypeOf(zero)
+
+ if p, ok := transitionPools.Load(typ); ok {
+ return p.(*pool.Synced[any])
+ }
+
+ // Create new pool for this type
+ poolName := "snapshot.Transition[" + typ.String() + "]"
+ p := pool.Register[any](poolName)
+ actual, _ := transitionPools.LoadOrStore(typ, p)
+ return actual.(*pool.Synced[any])
Review Comment:
`typ.String()` will panic if `typ` is nil (possible when
`reflect.TypeOf(zero)` returns nil). Even if current usages pass pointer types,
this makes the generic package unsafe for other `Snapshot` implementations.
Consider switching to a non-nil type derivation (e.g.,
`reflect.TypeOf((*S)(nil)).Elem()`) before constructing the pool name.
##########
banyand/trace/merger.go:
##########
@@ -164,10 +165,19 @@ func (tst *tsTable)
mergePartsThenSendIntroduction(creator snapshotCreator, part
mergerIntroductionMap := make(map[string]*sidx.MergerIntroduction)
for sidxName, sidxInstance := range tst.getAllSidx() {
start = time.Now()
- mergerIntroduction, err := sidxInstance.Merge(closeCh,
partIDMap, newPartID)
- if err != nil {
- tst.l.Warn().Err(err).Msg("sidx merge mem parts failed")
- return nil, err
+ mergerIntroduction, mergeErr := sidxInstance.Merge(closeCh,
partIDMap, newPartID)
+ if mergeErr != nil {
+ tst.l.Warn().Err(mergeErr).Msg("sidx merge mem parts
failed")
+ tst.removeTracePartOnFailure(newPart)
+ for doneSidxName, intro := range mergerIntroductionMap {
+ intro.ReleaseNewPart()
Review Comment:
On `sidxInstance.Merge` error, cleanup removes the newly created trace part
and previously successful sidx parts, but it doesn’t attempt to remove the
*failing* `sidxName`’s `newPartID` directory. Since `sidx.Merge` can create the
destination directory before returning an error, this can still leave on-disk
trash and contribute to disk exhaustion; consider removing `sidxName`’s path
best-effort on `mergeErr` as well.
##########
banyand/internal/snapshot/snapshot.go:
##########
@@ -0,0 +1,219 @@
+// 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 snapshot provides generic transaction coordination for
snapshot-based systems.
+// It enables atomic updates across multiple heterogeneous snapshot managers
using Go generics.
+package snapshot
+
+import (
+ "reflect"
+ "sync"
+
+ "github.com/apache/skywalking-banyandb/pkg/pool"
+)
+
+// Snapshot is a type constraint for snapshot types that support reference
counting.
+// Any type implementing this interface can participate in atomic transactions.
+type Snapshot interface {
+ // IncRef increments the reference count.
+ IncRef()
+ // DecRef decrements the reference count and releases resources when
zero.
+ DecRef()
+}
+
+// Manager is a generic interface for managing snapshots of type S.
+// Both trace and sidx implement this interface for their respective snapshot
types.
+type Manager[S Snapshot] interface {
+ // CurrentSnapshot returns the current snapshot with incremented ref
count.
+ // Returns nil if no snapshot exists.
+ CurrentSnapshot() S
+ // ReplaceSnapshot atomically replaces the current snapshot with next.
+ // The old snapshot's DecRef is called automatically.
+ ReplaceSnapshot(next S)
+}
+
+// Transition represents a prepared but uncommitted snapshot change.
+// It holds both the current and next snapshots, allowing atomic commit or
rollback.
+type Transition[S Snapshot] struct {
+ manager Manager[S]
+ current S
+ next S
+ committed bool
+}
+
+// getTransitionPool returns or creates a pool for the given snapshot type.
+func getTransitionPool[S Snapshot]() *pool.Synced[any] {
+ var zero S
+ typ := reflect.TypeOf(zero)
Review Comment:
`getTransitionPool` derives the pool key from `reflect.TypeOf(zero)`. If `S`
can be an interface type, `zero` may be a nil interface and `TypeOf` returns
nil, which later leads to a panic. Using `reflect.TypeOf((*S)(nil)).Elem()`
avoids this edge case and provides a stable type key regardless of the current
value.
```suggestion
typ := reflect.TypeOf((*S)(nil)).Elem()
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
