You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2025/08/06 01:39:28 UTC
(skywalking-banyandb) branch main updated: Adjusted TopNTagNames to include "source" and updated associated logic in GetTopNSchema. This improves the handling of node-specific data in the top N processing. (#718)
This is an automated email from the ASF dual-hosted git repository.
wusheng 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 16229c5e Adjusted TopNTagNames to include "source" and updated associated logic in GetTopNSchema. This improves the handling of node-specific data in the top N processing. (#718)
16229c5e is described below
commit 16229c5eeab44ab30894288cb3041d8fa20a5bae
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Aug 6 09:39:24 2025 +0800
Adjusted TopNTagNames to include "source" and updated associated logic in GetTopNSchema. This improves the handling of node-specific data in the top N processing. (#718)
---
CHANGES.md | 1 +
banyand/measure/metadata.go | 13 +++++++++----
banyand/measure/svc_standalone.go | 2 +-
banyand/measure/topn.go | 19 ++++++++++++++++---
pkg/query/logical/measure/topn_plan_localscan.go | 3 ++-
5 files changed, 29 insertions(+), 9 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 2274e135..62653027 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -26,6 +26,7 @@ Release Notes.
- Add trace module metadata management.
- Add chunked data sync to improve memory efficiency and performance during data transfer operations, supporting configurable chunk sizes, retry mechanisms, and out-of-order handling for both measure and stream services.
- Implement comprehensive migration system for both measure and stream data with file-based approach and enhanced progress tracking
+- Improve TopN processing by adding "source" tag to track node-specific data, enhancing data handling across distributed nodes
### Bug Fixes
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index fd4155f6..9402b37a 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -67,7 +67,7 @@ var (
CompressionMethod: databasev1.CompressionMethod_COMPRESSION_METHOD_ZSTD,
}}
// TopNTagNames is the tag names of the topN result measure.
- TopNTagNames = []string{"name", "direction", "group"}
+ TopNTagNames = []string{"name", "direction", "group", "source"}
)
// SchemaService allows querying schema information.
@@ -81,20 +81,22 @@ type schemaRepo struct {
pipeline queue.Client
l *logger.Logger
topNProcessorMap sync.Map
+ nodeID string
path string
}
-func newSchemaRepo(path string, svc *standalone, nodeLabels map[string]string) *schemaRepo {
+func newSchemaRepo(path string, svc *standalone, nodeLabels map[string]string, nodeID string) *schemaRepo {
sr := &schemaRepo{
path: path,
l: svc.l,
metadata: svc.metadata,
pipeline: svc.localPipeline,
+ nodeID: nodeID,
}
sr.Repository = resourceSchema.NewRepository(
svc.metadata,
svc.l,
- newSupplier(path, svc, sr, nodeLabels),
+ newSupplier(path, svc, sr, nodeLabels, nodeID),
resourceSchema.NewMetrics(svc.omr.With(metadataScope)),
)
sr.start()
@@ -390,11 +392,12 @@ type supplier struct {
l *logger.Logger
schemaRepo *schemaRepo
nodeLabels map[string]string
+ nodeID string
path string
option option
}
-func newSupplier(path string, svc *standalone, sr *schemaRepo, nodeLabels map[string]string) *supplier {
+func newSupplier(path string, svc *standalone, sr *schemaRepo, nodeLabels map[string]string, nodeID string) *supplier {
if svc.pm == nil {
svc.l.Panic().Msg("CRITICAL: svc.pm is nil in newSupplier")
}
@@ -415,6 +418,7 @@ func newSupplier(path string, svc *standalone, sr *schemaRepo, nodeLabels map[st
pm: svc.pm,
schemaRepo: sr,
nodeLabels: nodeLabels,
+ nodeID: nodeID,
}
}
@@ -625,6 +629,7 @@ func GetTopNSchema(md *commonv1.Metadata) *databasev1.Measure {
{Name: TopNTagNames[0], Type: databasev1.TagType_TAG_TYPE_STRING},
{Name: TopNTagNames[1], Type: databasev1.TagType_TAG_TYPE_INT},
{Name: TopNTagNames[2], Type: databasev1.TagType_TAG_TYPE_STRING},
+ {Name: TopNTagNames[3], Type: databasev1.TagType_TAG_TYPE_STRING},
},
},
},
diff --git a/banyand/measure/svc_standalone.go b/banyand/measure/svc_standalone.go
index 8f8a1780..b6504730 100644
--- a/banyand/measure/svc_standalone.go
+++ b/banyand/measure/svc_standalone.go
@@ -163,7 +163,7 @@ func (s *standalone) PreRun(ctx context.Context) error {
}
s.c = storage.NewServiceCacheWithConfig(s.cc)
node := val.(common.Node)
- s.schemaRepo = newSchemaRepo(s.dataPath, s, node.Labels)
+ s.schemaRepo = newSchemaRepo(s.dataPath, s, node.Labels, node.NodeID)
s.cm = newCacheMetrics(s.omr)
observability.MetricsCollector.Register("measure_cache", s.collectCacheMetrics)
diff --git a/banyand/measure/topn.go b/banyand/measure/topn.go
index 2af2bae1..7776fb41 100644
--- a/banyand/measure/topn.go
+++ b/banyand/measure/topn.go
@@ -86,6 +86,7 @@ func (sr *schemaRepo) getSteamingManager(source *commonv1.Metadata, pipeline que
m, _ := sr.topNProcessorMap.LoadOrStore(key, &topNProcessorManager{
l: sr.l,
pipeline: pipeline,
+ nodeID: sr.nodeID,
})
manager = m.(*topNProcessorManager)
return manager
@@ -99,6 +100,7 @@ func (sr *schemaRepo) getSteamingManager(source *commonv1.Metadata, pipeline que
manager = &topNProcessorManager{
l: sr.l,
pipeline: pipeline,
+ nodeID: sr.nodeID,
}
manager.registeredTasks = append(manager.registeredTasks, pre.registeredTasks...)
} else {
@@ -109,6 +111,7 @@ func (sr *schemaRepo) getSteamingManager(source *commonv1.Metadata, pipeline que
manager = &topNProcessorManager{
l: sr.l,
pipeline: pipeline,
+ nodeID: sr.nodeID,
}
}
manager.init(sourceMeasure.GetSchema())
@@ -141,6 +144,7 @@ type topNStreamingProcessor struct {
m *databasev1.Measure
errCh <-chan error
stopCh chan struct{}
+ nodeID string
flow.ComponentState
interval time.Duration
sortDirection modelv1.Sort
@@ -250,6 +254,13 @@ func (t *topNStreamingProcessor) writeStreamRecord(record flow.StreamRecord, buf
},
},
},
+ {
+ Value: &modelv1.TagValue_Str{
+ Str: &modelv1.Str{
+ Value: t.nodeID,
+ },
+ },
+ },
}
buf = buf[:0]
if buf, err = topNValue.marshal(buf); err != nil {
@@ -329,14 +340,15 @@ func (t *topNStreamingProcessor) handleError() {
// topNProcessorManager manages multiple topNStreamingProcessor(s) belonging to a single measure.
type topNProcessorManager struct {
- l *logger.Logger
pipeline queue.Client
- m *databasev1.Measure
s logical.TagSpecRegistry
+ l *logger.Logger
+ m *databasev1.Measure
+ nodeID string
registeredTasks []*databasev1.TopNAggregation
processorList []*topNStreamingProcessor
- closed bool
sync.RWMutex
+ closed bool
}
func (manager *topNProcessorManager) init(m *databasev1.Measure) {
@@ -479,6 +491,7 @@ func (manager *topNProcessorManager) start(topNSchema *databasev1.TopNAggregatio
stopCh: make(chan struct{}),
streamingFlow: streamingFlow,
pipeline: manager.pipeline,
+ nodeID: manager.nodeID,
}
processorList[i] = processor.start()
}
diff --git a/pkg/query/logical/measure/topn_plan_localscan.go b/pkg/query/logical/measure/topn_plan_localscan.go
index f7c85323..23907764 100644
--- a/pkg/query/logical/measure/topn_plan_localscan.go
+++ b/pkg/query/logical/measure/topn_plan_localscan.go
@@ -78,10 +78,11 @@ func (uls *unresolvedLocalScan) Analyze(s logical.Schema) (logical.Plan, error)
},
},
pbv1.AnyTagValue,
+ pbv1.AnyTagValue,
},
}
if len(groupByTags) > 0 {
- entities[0][len(entities[0])-1] = &modelv1.TagValue{
+ entities[0][2] = &modelv1.TagValue{
Value: &modelv1.TagValue_Str{
Str: &modelv1.Str{
Value: measure.GroupName(groupByTags),