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 2024/05/29 14:21:57 UTC
(skywalking-banyandb) 08/08: Fix lint issues
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch topn-val
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 4f4bc4ba940b7a5ac3506ef7e33cf562062a9217
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed May 29 14:20:43 2024 +0000
Fix lint issues
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/query/processor_topn.go | 2 +-
pkg/flow/streaming/sliding_window.go | 8 ++++----
pkg/flow/streaming/streaming.go | 2 +-
pkg/flow/streaming/topn.go | 2 +-
pkg/pb/v1/series.go | 1 +
pkg/query/logical/measure/topn_analyzer.go | 2 +-
6 files changed, 9 insertions(+), 8 deletions(-)
diff --git a/banyand/query/processor_topn.go b/banyand/query/processor_topn.go
index 06720ce3..d734e18b 100644
--- a/banyand/query/processor_topn.go
+++ b/banyand/query/processor_topn.go
@@ -105,7 +105,7 @@ func (t *topNQueryProcessor) Rev(message bus.Message) (resp bus.Message) {
Str("topN", topNMetadata.GetName()).
Msg("fail to build schema")
}
- plan, err := logical_measure.TopNAnalyze(context.TODO(), request, topNResultMeasure.GetSchema(), sourceMeasure.GetSchema(), topNSchema, s)
+ plan, err := logical_measure.TopNAnalyze(context.TODO(), request, topNResultMeasure.GetSchema(), sourceMeasure.GetSchema(), s)
if err != nil {
resp = bus.NewMessage(bus.MessageID(now), common.NewError("fail to analyze the query request for topn %s: %v", topNMetadata.GetName(), err))
return
diff --git a/pkg/flow/streaming/sliding_window.go b/pkg/flow/streaming/sliding_window.go
index 1b652427..ed012084 100644
--- a/pkg/flow/streaming/sliding_window.go
+++ b/pkg/flow/streaming/sliding_window.go
@@ -74,20 +74,20 @@ func (s *windowedFlow) AllowedMaxWindows(windowCnt int) flow.WindowedFlow {
}
type tumblingTimeWindows struct {
- errorHandler func(error)
+ l *logger.Logger
snapshots *lru.Cache
timerHeap *flow.DedupPriorityQueue
aggregationFactory flow.AggregationOpFactory
in chan flow.StreamRecord
out chan flow.StreamRecord
+ errorHandler func(error)
flow.ComponentState
- windowSize int64
windowCount int
currentWatermark int64
lastFlushTime int64
- timerMu sync.Mutex
flushInterval int64
- l *logger.Logger
+ windowSize int64
+ timerMu sync.Mutex
}
func (s *tumblingTimeWindows) In() chan<- flow.StreamRecord {
diff --git a/pkg/flow/streaming/streaming.go b/pkg/flow/streaming/streaming.go
index 316b8f62..f144aebd 100644
--- a/pkg/flow/streaming/streaming.go
+++ b/pkg/flow/streaming/streaming.go
@@ -35,8 +35,8 @@ type streamingFlow struct {
source flow.Source
sink flow.Sink
drain chan error
- ops []flow.Operator
l *logger.Logger
+ ops []flow.Operator
}
// New returns a new streaming flow.
diff --git a/pkg/flow/streaming/topn.go b/pkg/flow/streaming/topn.go
index 54a2af16..eaaafe24 100644
--- a/pkg/flow/streaming/topn.go
+++ b/pkg/flow/streaming/topn.go
@@ -79,9 +79,9 @@ type topNAggregatorGroup struct {
sortKeyExtractor func(flow.StreamRecord) int64
groupKeyExtractor func(flow.StreamRecord) string
comparator utils.Comparator
+ l *logger.Logger
cacheSize int
sort TopNSort
- l *logger.Logger
}
type topNAggregator struct {
diff --git a/pkg/pb/v1/series.go b/pkg/pb/v1/series.go
index 441a5db3..f6d47397 100644
--- a/pkg/pb/v1/series.go
+++ b/pkg/pb/v1/series.go
@@ -51,6 +51,7 @@ func (s *Series) Marshal() error {
return nil
}
+// MarshalWithWildcard encodes series to internal Buffer and generates ID with wildcard.
func (s *Series) MarshalWithWildcard() error {
s.Buffer = marshalEntityValue(s.Buffer, convert.StringToBytes(s.Subject))
var err error
diff --git a/pkg/query/logical/measure/topn_analyzer.go b/pkg/query/logical/measure/topn_analyzer.go
index a2915fa9..4d72787f 100644
--- a/pkg/query/logical/measure/topn_analyzer.go
+++ b/pkg/query/logical/measure/topn_analyzer.go
@@ -53,7 +53,7 @@ func BuildTopNSchema(md *databasev1.Measure) (logical.Schema, error) {
// TopNAnalyze converts logical expressions to executable operation tree represented by Plan.
func TopNAnalyze(_ context.Context, criteria *measurev1.TopNRequest, schema *databasev1.Measure,
- sourceMeasureSchema *databasev1.Measure, topNAggregation *databasev1.TopNAggregation, s logical.Schema,
+ sourceMeasureSchema *databasev1.Measure, s logical.Schema,
) (logical.Plan, error) {
groupByProjectionTags := sourceMeasureSchema.GetEntity().GetTagNames()
groupByTags := make([][]*logical.Tag, len(schema.GetTagFamilies()))