Merge branch 'main' into feature/add-optional-time-range-manifest

Author: wu-sheng

CHANGES.md

@@ -19,6 +19,9 @@ Release Notes.
 - Add replication integration test for measure.
 - Activate the property repair mechanism by default.
 - Add snapshot time retention policy to ensure the snapshot only can be deleted after the configured minimum age(time).
+- **Breaking Change**: Change the data storage path structure for property model:
+  - From: `<data-dir>/property/data/shard-<id>/...`
+  - To: `<data-dir>/property/data/<group>/shard-<id>/...`
 
 ### Bug Fixes
 

banyand/backup/lifecycle/current_node.go

@@ -0,0 +1,30 @@
+// 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 lifecycle
+
+import (
+	"context"
+
+	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+)
+
+func (l *lifecycleService) GetCurrentNode(_ context.Context, _ *databasev1.GetCurrentNodeRequest) (*databasev1.GetCurrentNodeResponse, error) {
+	return &databasev1.GetCurrentNodeResponse{
+		Node: l.currentNode,
+	}, nil
+}

banyand/backup/lifecycle/service.go

@@ -41,6 +41,7 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/health"
 	"google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/apache/skywalking-banyandb/api/common"
 	commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
@@ -70,6 +71,7 @@ var _ service = (*lifecycleService)(nil)
 
 type lifecycleService struct {
 	databasev1.UnimplementedClusterStateServiceServer
+	databasev1.UnimplementedNodeQueryServiceServer
 	metadata          metadata.Repo
 	omr               observability.MetricsRegistry
 	pm                protector.Memory
@@ -79,6 +81,7 @@ type lifecycleService struct {
 	grpcServer        *grpclib.Server
 	httpSrv           *http.Server
 	tlsReloader       *pkgtls.Reloader
+	currentNode       *databasev1.Node
 	clientCloser      context.CancelFunc
 	stopCh            chan struct{}
 	measureRoot       string
@@ -165,6 +168,16 @@ func (l *lifecycleService) Validate() error {
 				return errors.New("missing key file when TLS is enabled")
 			}
 		}
+		l.currentNode = &databasev1.Node{
+			Metadata: &commonv1.Metadata{
+				Name: l.lifecycleGRPCAddr,
+			},
+			GrpcAddress: l.lifecycleGRPCAddr,
+			HttpAddress: l.lifecycleHTTPAddr,
+			Roles:       make([]databasev1.Role, 0),
+			Labels:      common.ParseNodeFlags(),
+			CreatedAt:   timestamppb.Now(),
+		}
 	}
 	return nil
 }
@@ -309,6 +322,7 @@ func (l *lifecycleService) startServers() {
 
 	l.grpcServer = grpclib.NewServer(opts...)
 	databasev1.RegisterClusterStateServiceServer(l.grpcServer, l)
+	databasev1.RegisterNodeQueryServiceServer(l.grpcServer, l)
 	grpc_health_v1.RegisterHealthServer(l.grpcServer, health.NewServer())
 
 	// Setup HTTP server

banyand/property/db.go

@@ -52,14 +52,21 @@ var (
 	propertyScope = observability.RootScope.SubScope("property")
 )
 
+type groupShards struct {
+	shards   atomic.Pointer[[]*shard]
+	group    string
+	location string
+	mu       sync.RWMutex
+}
+
 type database struct {
 	metadata            metadata.Repo
 	omr                 observability.MetricsRegistry
 	lfs                 fs.FileSystem
 	lock                fs.File
 	logger              *logger.Logger
 	repairScheduler     *repairScheduler
-	sLst                atomic.Pointer[[]*shard]
+	groups              sync.Map
 	location            string
 	repairBaseDir       string
 	flushInterval       time.Duration
@@ -129,18 +136,29 @@ func (db *database) load(ctx context.Context) error {
 	if db.closed.Load() {
 		return errors.New("database is closed")
 	}
-	return walkDir(db.location, "shard-", func(suffix string) error {
-		id, err := strconv.Atoi(suffix)
-		if err != nil {
-			return err
+	for _, groupDir := range lfs.ReadDir(db.location) {
+		if !groupDir.IsDir() {
+			continue
 		}
-		_, err = db.loadShard(ctx, common.ShardID(id))
-		return err
-	})
+		groupName := groupDir.Name()
+		groupPath := filepath.Join(db.location, groupName)
+		walkErr := walkDir(groupPath, "shard-", func(suffix string) error {
+			id, parseErr := strconv.Atoi(suffix)
+			if parseErr != nil {
+				return parseErr
+			}
+			_, loadErr := db.loadShard(ctx, groupName, common.ShardID(id))
+			return loadErr
+		})
+		if walkErr != nil {
+			return walkErr
+		}
+	}
+	return nil
 }
 
 func (db *database) update(ctx context.Context, shardID common.ShardID, id []byte, property *propertyv1.Property) error {
-	sd, err := db.loadShard(ctx, shardID)
+	sd, err := db.loadShard(ctx, property.Metadata.Group, shardID)
 	if err != nil {
 		return err
 	}
@@ -152,14 +170,18 @@ func (db *database) update(ctx context.Context, shardID common.ShardID, id []byt
 }
 
 func (db *database) delete(ctx context.Context, docIDs [][]byte) error {
-	sLst := db.sLst.Load()
-	if sLst == nil {
-		return nil
-	}
 	var err error
-	for _, s := range *sLst {
-		multierr.AppendInto(&err, s.delete(ctx, docIDs))
-	}
+	db.groups.Range(func(_, value any) bool {
+		gs := value.(*groupShards)
+		sLst := gs.shards.Load()
+		if sLst == nil {
+			return true
+		}
+		for _, s := range *sLst {
+			multierr.AppendInto(&err, s.delete(ctx, docIDs))
+		}
+		return true
+	})
 	return err
 }
 
@@ -168,14 +190,18 @@ func (db *database) query(ctx context.Context, req *propertyv1.QueryRequest) ([]
 	if err != nil {
 		return nil, err
 	}
-	sLst := db.sLst.Load()
-	if sLst == nil {
+	requestedGroups := make(map[string]bool, len(req.Groups))
+	for _, g := range req.Groups {
+		requestedGroups[g] = true
+	}
+	shards := db.collectGroupShards(requestedGroups)
+	if len(shards) == 0 {
 		return nil, nil
 	}
 
 	if req.OrderBy == nil {
 		var res []*queryProperty
-		for _, s := range *sLst {
+		for _, s := range shards {
 			r, searchErr := s.search(ctx, iq, nil, int(req.Limit))
 			if searchErr != nil {
 				return nil, searchErr
@@ -185,8 +211,8 @@ func (db *database) query(ctx context.Context, req *propertyv1.QueryRequest) ([]
 		return res, nil
 	}
 
-	iters := make([]sort.Iterator[*queryProperty], 0, len(*sLst))
-	for _, s := range *sLst {
+	iters := make([]sort.Iterator[*queryProperty], 0, len(shards))
+	for _, s := range shards {
 		// Each shard returns pre-sorted results (via SeriesSort)
 		r, searchErr := s.search(ctx, iq, req.OrderBy, int(req.Limit))
 		if searchErr != nil {
@@ -216,34 +242,77 @@ func (db *database) query(ctx context.Context, req *propertyv1.QueryRequest) ([]
 	return result, nil
 }
 
-func (db *database) loadShard(ctx context.Context, id common.ShardID) (*shard, error) {
+func (db *database) collectGroupShards(requestedGroups map[string]bool) []*shard {
+	var shards []*shard
+	db.groups.Range(func(key, value any) bool {
+		groupName := key.(string)
+		if len(requestedGroups) > 0 {
+			if _, ok := requestedGroups[groupName]; !ok {
+				return true
+			}
+		}
+		gs := value.(*groupShards)
+		sLst := gs.shards.Load()
+		if sLst == nil {
+			return true
+		}
+		shards = append(shards, *sLst...)
+		return true
+	})
+	return shards
+}
+
+func (db *database) loadShard(ctx context.Context, group string, id common.ShardID) (*shard, error) {
 	if db.closed.Load() {
 		return nil, errors.New("database is closed")
 	}
-	if s, ok := db.getShard(id); ok {
+	if s, ok := db.getShard(group, id); ok {
 		return s, nil
 	}
 	db.mu.Lock()
 	defer db.mu.Unlock()
-	if s, ok := db.getShard(id); ok {
+	if s, ok := db.getShard(group, id); ok {
 		return s, nil
 	}
-	sd, err := db.newShard(context.WithValue(ctx, logger.ContextKey, db.logger), id, int64(db.flushInterval.Seconds()),
+
+	gs := db.getOrCreateGroupShards(group)
+	sd, err := db.newShard(context.WithValue(ctx, logger.ContextKey, db.logger),
+		group, id, int64(db.flushInterval.Seconds()),
 		int64(db.expireDelete.Seconds()), db.repairBaseDir, db.repairTreeSlotCount)
 	if err != nil {
 		return nil, err
 	}
-	sLst := db.sLst.Load()
-	if sLst == nil {
-		sLst = &[]*shard{}
+
+	gs.mu.Lock()
+	sLst := gs.shards.Load()
+	var oldList []*shard
+	if sLst != nil {
+		oldList = *sLst
 	}
-	*sLst = append(*sLst, sd)
-	db.sLst.Store(sLst)
+	newList := make([]*shard, len(oldList)+1)
+	copy(newList, oldList)
+	newList[len(oldList)] = sd
+	gs.shards.Store(&newList)
+	gs.mu.Unlock()
 	return sd, nil
 }
 
-func (db *database) getShard(id common.ShardID) (*shard, bool) {
-	sLst := db.sLst.Load()
+func (db *database) getOrCreateGroupShards(group string) *groupShards {
+	gs := &groupShards{
+		group:    group,
+		location: filepath.Join(db.location, group),
+	}
+	actual, _ := db.groups.LoadOrStore(group, gs)
+	return actual.(*groupShards)
+}
+
+func (db *database) getShard(group string, id common.ShardID) (*shard, bool) {
+	value, ok := db.groups.Load(group)
+	if !ok {
+		return nil, false
+	}
+	gs := value.(*groupShards)
+	sLst := gs.shards.Load()
 	if sLst == nil {
 		return nil, false
 	}
@@ -262,13 +331,18 @@ func (db *database) close() error {
 	if db.repairScheduler != nil {
 		db.repairScheduler.close()
 	}
-	sLst := db.sLst.Load()
 	var err error
-	if sLst != nil {
+	db.groups.Range(func(_, value any) bool {
+		gs := value.(*groupShards)
+		sLst := gs.shards.Load()
+		if sLst == nil {
+			return true
+		}
 		for _, s := range *sLst {
 			multierr.AppendInto(&err, s.close())
 		}
-	}
+		return true
+	})
 	db.lock.Close()
 	return err
 }
@@ -277,17 +351,21 @@ func (db *database) collect() {
 	if db.closed.Load() {
 		return
 	}
-	sLst := db.sLst.Load()
-	if sLst == nil {
-		return
-	}
-	for _, s := range *sLst {
-		s.store.CollectMetrics()
-	}
+	db.groups.Range(func(_, value any) bool {
+		gs := value.(*groupShards)
+		sLst := gs.shards.Load()
+		if sLst == nil {
+			return true
+		}
+		for _, s := range *sLst {
+			s.store.CollectMetrics()
+		}
+		return true
+	})
 }
 
 func (db *database) repair(ctx context.Context, id []byte, shardID uint64, property *propertyv1.Property, deleteTime int64) error {
-	s, err := db.loadShard(ctx, common.ShardID(shardID))
+	s, err := db.loadShard(ctx, property.Metadata.Group, common.ShardID(shardID))
 	if err != nil {
 		return errors.WithMessagef(err, "failed to load shard %d", id)
 	}