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:53 UTC
(skywalking-banyandb) 04/08: Encode int
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 083a621f944536027657a355d921a77300bfd389
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue May 28 06:50:02 2024 +0000
Encode int
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/dquery/topn.go | 9 ++---
banyand/measure/block_metadata_test.go | 14 ++++----
banyand/measure/block_writer.go | 24 ++++++-------
banyand/measure/column.go | 16 ++++-----
banyand/measure/topn.go | 36 +++++--------------
banyand/stream/block_metadata.go | 64 +++++++++++++++++-----------------
banyand/stream/block_metadata_test.go | 14 ++++----
banyand/stream/block_writer.go | 24 ++++++-------
banyand/stream/tag.go | 16 ++++-----
pkg/pb/v1/series_test.go | 2 +-
pkg/pb/v1/value.go | 3 +-
pkg/schema/init.go | 12 +++----
test/cases/topn/topn.go | 2 +-
13 files changed, 103 insertions(+), 133 deletions(-)
diff --git a/banyand/dquery/topn.go b/banyand/dquery/topn.go
index 508851ba..9aefc08c 100644
--- a/banyand/dquery/topn.go
+++ b/banyand/dquery/topn.go
@@ -18,7 +18,6 @@
package dquery
import (
- "fmt"
"time"
"go.uber.org/multierr"
@@ -66,7 +65,7 @@ func (t *topNQueryProcessor) Rev(message bus.Message) (resp bus.Message) {
aggregator := query.CreateTopNPostAggregator(request.GetTopN(),
agg, request.GetFieldValueSort())
var tags []string
- for i, f := range ff {
+ for _, f := range ff {
if m, getErr := f.Get(); getErr != nil {
allErr = multierr.Append(allErr, getErr)
} else {
@@ -87,13 +86,11 @@ func (t *topNQueryProcessor) Rev(message bus.Message) (resp bus.Message) {
for _, e := range tn.Entity {
entityValues = append(entityValues, e.Value)
}
- fmt.Printf("%d entityValues: %v value: %d\n", i, entityValues, tn.Value.GetInt().GetValue())
_ = aggregator.Put(entityValues, tn.Value.GetInt().GetValue(), uint64(l.Timestamp.AsTime().UnixMilli()))
}
}
}
}
- fmt.Println("--------------------")
if allErr != nil {
resp = bus.NewMessage(now, common.NewError("execute the query %s: %v", request.GetName(), allErr))
return
@@ -102,10 +99,8 @@ func (t *topNQueryProcessor) Rev(message bus.Message) (resp bus.Message) {
resp = bus.NewMessage(now, &measurev1.TopNResponse{})
return
}
- lst := aggregator.Val(tags)
- fmt.Println(lst)
resp = bus.NewMessage(now, &measurev1.TopNResponse{
- Lists: lst,
+ Lists: aggregator.Val(tags),
})
return
}
diff --git a/banyand/measure/block_metadata_test.go b/banyand/measure/block_metadata_test.go
index 81da440b..59709b73 100644
--- a/banyand/measure/block_metadata_test.go
+++ b/banyand/measure/block_metadata_test.go
@@ -75,7 +75,7 @@ func Test_dataBlock_marshal_unmarshal(t *testing.T) {
}
func Test_timestampsMetadata_reset(t *testing.T) {
- th := ×tampsMetadata{
+ tm := ×tampsMetadata{
dataBlock: dataBlock{
offset: 1,
size: 1,
@@ -85,13 +85,13 @@ func Test_timestampsMetadata_reset(t *testing.T) {
encodeType: encoding.EncodeTypeConst,
}
- th.reset()
+ tm.reset()
- assert.Equal(t, uint64(0), th.dataBlock.offset)
- assert.Equal(t, uint64(0), th.dataBlock.size)
- assert.Equal(t, int64(0), th.min)
- assert.Equal(t, int64(0), th.max)
- assert.Equal(t, encoding.EncodeTypeUnknown, th.encodeType)
+ assert.Equal(t, uint64(0), tm.dataBlock.offset)
+ assert.Equal(t, uint64(0), tm.dataBlock.size)
+ assert.Equal(t, int64(0), tm.min)
+ assert.Equal(t, int64(0), tm.max)
+ assert.Equal(t, encoding.EncodeTypeUnknown, tm.encodeType)
}
func Test_timestampsMetadata_copyFrom(t *testing.T) {
diff --git a/banyand/measure/block_writer.go b/banyand/measure/block_writer.go
index 8f97c066..760a63ad 100644
--- a/banyand/measure/block_writer.go
+++ b/banyand/measure/block_writer.go
@@ -214,23 +214,23 @@ func (bw *blockWriter) mustWriteBlock(sid common.SeriesID, b *block) {
bm := generateBlockMetadata()
b.mustWriteTo(sid, bm, &bw.writers)
- th := &bm.timestamps
- if bw.totalCount == 0 || th.min < bw.totalMinTimestamp {
- bw.totalMinTimestamp = th.min
+ tm := &bm.timestamps
+ if bw.totalCount == 0 || tm.min < bw.totalMinTimestamp {
+ bw.totalMinTimestamp = tm.min
}
- if bw.totalCount == 0 || th.max > bw.totalMaxTimestamp {
- bw.totalMaxTimestamp = th.max
+ if bw.totalCount == 0 || tm.max > bw.totalMaxTimestamp {
+ bw.totalMaxTimestamp = tm.max
}
- if !hasWrittenBlocks || th.min < bw.minTimestamp {
- bw.minTimestamp = th.min
+ if !hasWrittenBlocks || tm.min < bw.minTimestamp {
+ bw.minTimestamp = tm.min
}
- if !hasWrittenBlocks || th.max > bw.maxTimestamp {
- bw.maxTimestamp = th.max
+ if !hasWrittenBlocks || tm.max > bw.maxTimestamp {
+ bw.maxTimestamp = tm.max
}
- if isSeenSid && th.min < bw.minTimestampLast {
- logger.Panicf("the block for sid=%d cannot contain timestamp smaller than %d, but it contains timestamp %d", sid, bw.minTimestampLast, th.min)
+ if isSeenSid && tm.min < bw.minTimestampLast {
+ logger.Panicf("the block for sid=%d cannot contain timestamp smaller than %d, but it contains timestamp %d", sid, bw.minTimestampLast, tm.min)
}
- bw.minTimestampLast = th.min
+ bw.minTimestampLast = tm.min
bw.totalUncompressedSizeBytes += bm.uncompressedSizeBytes
bw.totalCount += bm.count
diff --git a/banyand/measure/column.go b/banyand/measure/column.go
index 2f17d186..f208786c 100644
--- a/banyand/measure/column.go
+++ b/banyand/measure/column.go
@@ -51,11 +51,11 @@ func (c *column) resizeValues(valuesLen int) [][]byte {
return values
}
-func (c *column) mustWriteTo(ch *columnMetadata, columnWriter *writer) {
- ch.reset()
+func (c *column) mustWriteTo(cm *columnMetadata, columnWriter *writer) {
+ cm.reset()
- ch.name = c.name
- ch.valueType = c.valueType
+ cm.name = c.name
+ cm.valueType = c.valueType
// TODO: encoding values based on value type
@@ -64,11 +64,11 @@ func (c *column) mustWriteTo(ch *columnMetadata, columnWriter *writer) {
// marshal values
bb.Buf = encoding.EncodeBytesBlock(bb.Buf[:0], c.values)
- ch.size = uint64(len(bb.Buf))
- if ch.size > maxValuesBlockSize {
- logger.Panicf("too valuesSize: %d bytes; mustn't exceed %d bytes", ch.size, maxValuesBlockSize)
+ cm.size = uint64(len(bb.Buf))
+ if cm.size > maxValuesBlockSize {
+ logger.Panicf("too valuesSize: %d bytes; mustn't exceed %d bytes", cm.size, maxValuesBlockSize)
}
- ch.offset = columnWriter.bytesWritten
+ cm.offset = columnWriter.bytesWritten
columnWriter.MustWrite(bb.Buf)
}
diff --git a/banyand/measure/topn.go b/banyand/measure/topn.go
index e7a260cb..01bd61e8 100644
--- a/banyand/measure/topn.go
+++ b/banyand/measure/topn.go
@@ -132,7 +132,6 @@ func (t *topNStreamingProcessor) writeStreamRecord(record flow.StreamRecord) err
}
// down-sample the start of the timeWindow to a time-bucket
eventTime := t.downSampleTimeBucket(record.TimestampMillis())
- timeBucket := eventTime.Format(timeBucketFormat)
var err error
publisher := t.pipeline.NewBatchPublisher(resultPersistencyTimeout)
defer publisher.Close()
@@ -146,14 +145,14 @@ func (t *topNStreamingProcessor) writeStreamRecord(record flow.StreamRecord) err
for rankNum, tuple := range tuples {
fieldValue := tuple.V1.(int64)
data := tuple.V2.(flow.StreamRecord).Data().(flow.Data)
- err = multierr.Append(err, t.writeData(publisher, eventTime, timeBucket, fieldValue, group, data, rankNum))
+ err = multierr.Append(err, t.writeData(publisher, eventTime, fieldValue, data, rankNum))
}
}
return err
}
-func (t *topNStreamingProcessor) writeData(publisher queue.BatchPublisher, eventTime time.Time, timeBucket string, fieldValue int64,
- group string, data flow.Data, rankNum int,
+func (t *topNStreamingProcessor) writeData(publisher queue.BatchPublisher, eventTime time.Time, fieldValue int64,
+ data flow.Data, rankNum int,
) error {
var tagValues []*modelv1.TagValue
if len(t.topNSchema.GetGroupByTagNames()) > 0 {
@@ -172,11 +171,6 @@ func (t *topNStreamingProcessor) writeData(publisher queue.BatchPublisher, event
return err
}
- // measureID is consist of three parts,
- // 1. groupValues
- // 2. rankNumber
- // 3. timeBucket
- measureID := group + "_" + strconv.Itoa(rankNum) + "_" + timeBucket
iwr := &measurev1.InternalWriteRequest{
Request: &measurev1.WriteRequest{
MessageId: uint64(time.Now().UnixNano()),
@@ -185,17 +179,10 @@ func (t *topNStreamingProcessor) writeData(publisher queue.BatchPublisher, event
Timestamp: timestamppb.New(eventTime),
TagFamilies: []*modelv1.TagFamilyForWrite{
{
- Tags: append([]*modelv1.TagValue{
- // MeasureID
- {
- Value: &modelv1.TagValue_Str{
- Str: &modelv1.Str{
- Value: measureID,
- },
- },
- },
+ Tags: append(
+ data[0].([]*modelv1.TagValue),
// SortDirection
- {
+ &modelv1.TagValue{
Value: &modelv1.TagValue_Int{
Int: &modelv1.Int{
Value: int64(t.sortDirection),
@@ -203,14 +190,14 @@ func (t *topNStreamingProcessor) writeData(publisher queue.BatchPublisher, event
},
},
// RankNumber
- {
+ &modelv1.TagValue{
Value: &modelv1.TagValue_Int{
Int: &modelv1.Int{
Value: int64(rankNum),
},
},
},
- }, data[0].([]*modelv1.TagValue)...),
+ ),
},
},
Fields: []*modelv1.FieldValue{
@@ -240,13 +227,6 @@ func (t *topNStreamingProcessor) locate(tagValues []*modelv1.TagValue, rankNum i
if len(tagValues) != 0 && len(t.topNSchema.GetGroupByTagNames()) != len(tagValues) {
return nil, 0, errors.New("no enough tag values for the entity")
}
- // Subject: topN aggregation Name
- //
- // entity prefix
- //
- // 1) sort direction
- // 2) rank number
- // >3) group tag values if needed
series := &pbv1.Series{
Subject: t.topNSchema.GetMetadata().GetName(),
EntityValues: make([]*modelv1.TagValue, 1+1+len(tagValues)),
diff --git a/banyand/stream/block_metadata.go b/banyand/stream/block_metadata.go
index 3906f603..41527229 100644
--- a/banyand/stream/block_metadata.go
+++ b/banyand/stream/block_metadata.go
@@ -245,38 +245,38 @@ type timestampsMetadata struct {
encodeType encoding.EncodeType
}
-func (th *timestampsMetadata) reset() {
- th.dataBlock.reset()
- th.min = 0
- th.max = 0
- th.encodeType = 0
+func (tm *timestampsMetadata) reset() {
+ tm.dataBlock.reset()
+ tm.min = 0
+ tm.max = 0
+ tm.encodeType = 0
}
-func (th *timestampsMetadata) copyFrom(src *timestampsMetadata) {
- th.dataBlock.copyFrom(&src.dataBlock)
- th.min = src.min
- th.max = src.max
- th.encodeType = src.encodeType
+func (tm *timestampsMetadata) copyFrom(src *timestampsMetadata) {
+ tm.dataBlock.copyFrom(&src.dataBlock)
+ tm.min = src.min
+ tm.max = src.max
+ tm.encodeType = src.encodeType
}
-func (th *timestampsMetadata) marshal(dst []byte) []byte {
- dst = th.dataBlock.marshal(dst)
- dst = encoding.Uint64ToBytes(dst, uint64(th.min))
- dst = encoding.Uint64ToBytes(dst, uint64(th.max))
- dst = append(dst, byte(th.encodeType))
+func (tm *timestampsMetadata) marshal(dst []byte) []byte {
+ dst = tm.dataBlock.marshal(dst)
+ dst = encoding.Uint64ToBytes(dst, uint64(tm.min))
+ dst = encoding.Uint64ToBytes(dst, uint64(tm.max))
+ dst = append(dst, byte(tm.encodeType))
return dst
}
-func (th *timestampsMetadata) unmarshal(src []byte) ([]byte, error) {
- src, err := th.dataBlock.unmarshal(src)
+func (tm *timestampsMetadata) unmarshal(src []byte) ([]byte, error) {
+ src, err := tm.dataBlock.unmarshal(src)
if err != nil {
return nil, fmt.Errorf("cannot unmarshal dataBlock: %w", err)
}
- th.min = int64(encoding.BytesToUint64(src))
+ tm.min = int64(encoding.BytesToUint64(src))
src = src[8:]
- th.max = int64(encoding.BytesToUint64(src))
+ tm.max = int64(encoding.BytesToUint64(src))
src = src[8:]
- th.encodeType = encoding.EncodeType(src[0])
+ tm.encodeType = encoding.EncodeType(src[0])
return src[1:], nil
}
@@ -285,28 +285,28 @@ type elementIDsMetadata struct {
encodeType encoding.EncodeType
}
-func (th *elementIDsMetadata) reset() {
- th.dataBlock.reset()
- th.encodeType = 0
+func (em *elementIDsMetadata) reset() {
+ em.dataBlock.reset()
+ em.encodeType = 0
}
-func (th *elementIDsMetadata) copyFrom(src *elementIDsMetadata) {
- th.dataBlock.copyFrom(&src.dataBlock)
- th.encodeType = src.encodeType
+func (em *elementIDsMetadata) copyFrom(src *elementIDsMetadata) {
+ em.dataBlock.copyFrom(&src.dataBlock)
+ em.encodeType = src.encodeType
}
-func (th *elementIDsMetadata) marshal(dst []byte) []byte {
- dst = th.dataBlock.marshal(dst)
- dst = append(dst, byte(th.encodeType))
+func (em *elementIDsMetadata) marshal(dst []byte) []byte {
+ dst = em.dataBlock.marshal(dst)
+ dst = append(dst, byte(em.encodeType))
return dst
}
-func (th *elementIDsMetadata) unmarshal(src []byte) ([]byte, error) {
- src, err := th.dataBlock.unmarshal(src)
+func (em *elementIDsMetadata) unmarshal(src []byte) ([]byte, error) {
+ src, err := em.dataBlock.unmarshal(src)
if err != nil {
return nil, fmt.Errorf("cannot unmarshal dataBlock: %w", err)
}
- th.encodeType = encoding.EncodeType(src[0])
+ em.encodeType = encoding.EncodeType(src[0])
return src[1:], nil
}
diff --git a/banyand/stream/block_metadata_test.go b/banyand/stream/block_metadata_test.go
index 67c7e8e3..5c721b7d 100644
--- a/banyand/stream/block_metadata_test.go
+++ b/banyand/stream/block_metadata_test.go
@@ -74,7 +74,7 @@ func Test_dataBlock_marshal_unmarshal(t *testing.T) {
}
func Test_timestampsMetadata_reset(t *testing.T) {
- th := ×tampsMetadata{
+ tm := ×tampsMetadata{
dataBlock: dataBlock{
offset: 1,
size: 1,
@@ -84,13 +84,13 @@ func Test_timestampsMetadata_reset(t *testing.T) {
encodeType: encoding.EncodeTypeConst,
}
- th.reset()
+ tm.reset()
- assert.Equal(t, uint64(0), th.dataBlock.offset)
- assert.Equal(t, uint64(0), th.dataBlock.size)
- assert.Equal(t, int64(0), th.min)
- assert.Equal(t, int64(0), th.max)
- assert.Equal(t, encoding.EncodeTypeUnknown, th.encodeType)
+ assert.Equal(t, uint64(0), tm.dataBlock.offset)
+ assert.Equal(t, uint64(0), tm.dataBlock.size)
+ assert.Equal(t, int64(0), tm.min)
+ assert.Equal(t, int64(0), tm.max)
+ assert.Equal(t, encoding.EncodeTypeUnknown, tm.encodeType)
}
func Test_timestampsMetadata_copyFrom(t *testing.T) {
diff --git a/banyand/stream/block_writer.go b/banyand/stream/block_writer.go
index ad6a69e3..2c981c3d 100644
--- a/banyand/stream/block_writer.go
+++ b/banyand/stream/block_writer.go
@@ -213,23 +213,23 @@ func (bw *blockWriter) mustWriteBlock(sid common.SeriesID, b *block) {
bm := generateBlockMetadata()
b.mustWriteTo(sid, bm, &bw.writers)
- th := &bm.timestamps
- if bw.totalCount == 0 || th.min < bw.totalMinTimestamp {
- bw.totalMinTimestamp = th.min
+ tm := &bm.timestamps
+ if bw.totalCount == 0 || tm.min < bw.totalMinTimestamp {
+ bw.totalMinTimestamp = tm.min
}
- if bw.totalCount == 0 || th.max > bw.totalMaxTimestamp {
- bw.totalMaxTimestamp = th.max
+ if bw.totalCount == 0 || tm.max > bw.totalMaxTimestamp {
+ bw.totalMaxTimestamp = tm.max
}
- if !hasWrittenBlocks || th.min < bw.minTimestamp {
- bw.minTimestamp = th.min
+ if !hasWrittenBlocks || tm.min < bw.minTimestamp {
+ bw.minTimestamp = tm.min
}
- if !hasWrittenBlocks || th.max > bw.maxTimestamp {
- bw.maxTimestamp = th.max
+ if !hasWrittenBlocks || tm.max > bw.maxTimestamp {
+ bw.maxTimestamp = tm.max
}
- if isSeenSid && th.min < bw.minTimestampLast {
- logger.Panicf("the block for sid=%d cannot contain timestamp smaller than %d, but it contains timestamp %d", sid, bw.minTimestampLast, th.min)
+ if isSeenSid && tm.min < bw.minTimestampLast {
+ logger.Panicf("the block for sid=%d cannot contain timestamp smaller than %d, but it contains timestamp %d", sid, bw.minTimestampLast, tm.min)
}
- bw.minTimestampLast = th.min
+ bw.minTimestampLast = tm.min
bw.totalUncompressedSizeBytes += bm.uncompressedSizeBytes
bw.totalCount += bm.count
diff --git a/banyand/stream/tag.go b/banyand/stream/tag.go
index f9ad8740..5d595043 100644
--- a/banyand/stream/tag.go
+++ b/banyand/stream/tag.go
@@ -51,11 +51,11 @@ func (t *tag) resizeValues(valuesLen int) [][]byte {
return values
}
-func (t *tag) mustWriteTo(ch *tagMetadata, tagWriter *writer) {
- ch.reset()
+func (t *tag) mustWriteTo(tm *tagMetadata, tagWriter *writer) {
+ tm.reset()
- ch.name = t.name
- ch.valueType = t.valueType
+ tm.name = t.name
+ tm.valueType = t.valueType
// TODO: encoding values based on value type
@@ -64,11 +64,11 @@ func (t *tag) mustWriteTo(ch *tagMetadata, tagWriter *writer) {
// marshal values
bb.Buf = encoding.EncodeBytesBlock(bb.Buf[:0], t.values)
- ch.size = uint64(len(bb.Buf))
- if ch.size > maxValuesBlockSize {
- logger.Panicf("too valuesSize: %d bytes; mustn't exceed %d bytes", ch.size, maxValuesBlockSize)
+ tm.size = uint64(len(bb.Buf))
+ if tm.size > maxValuesBlockSize {
+ logger.Panicf("too valuesSize: %d bytes; mustn't exceed %d bytes", tm.size, maxValuesBlockSize)
}
- ch.offset = tagWriter.bytesWritten
+ tm.offset = tagWriter.bytesWritten
tagWriter.MustWrite(bb.Buf)
}
diff --git a/pkg/pb/v1/series_test.go b/pkg/pb/v1/series_test.go
index 7003865b..0399d045 100644
--- a/pkg/pb/v1/series_test.go
+++ b/pkg/pb/v1/series_test.go
@@ -97,7 +97,7 @@ func TestMarshalAndUnmarshalSeries(t *testing.T) {
src: &Series{Subject: "subject", EntityValues: []*modelv1.TagValue{
{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "stringValue"}}},
{Value: &modelv1.TagValue_Int{Int: &modelv1.Int{Value: 1}}},
- {Value: &modelv1.TagValue_Int{Int: &modelv1.Int{Value: 1}}},
+ {Value: &modelv1.TagValue_Int{Int: &modelv1.Int{Value: 46}}},
}},
},
}
diff --git a/pkg/pb/v1/value.go b/pkg/pb/v1/value.go
index bcaabc23..064d2022 100644
--- a/pkg/pb/v1/value.go
+++ b/pkg/pb/v1/value.go
@@ -71,8 +71,7 @@ func marshalTagValue(dest []byte, tv *modelv1.TagValue) ([]byte, error) {
case *modelv1.TagValue_Str:
dest = marshalEntityValue(dest, convert.StringToBytes(tv.GetStr().Value))
case *modelv1.TagValue_Int:
- dest = encoding.Int64ToBytes(dest, tv.GetInt().Value)
- dest = marshalEntityValue(dest, nil)
+ dest = marshalEntityValue(dest, encoding.Int64ToBytes(nil, tv.GetInt().Value))
case *modelv1.TagValue_BinaryData:
dest = marshalEntityValue(dest, tv.GetBinaryData())
default:
diff --git a/pkg/schema/init.go b/pkg/schema/init.go
index afa530a0..a158daab 100644
--- a/pkg/schema/init.go
+++ b/pkg/schema/init.go
@@ -368,20 +368,16 @@ func buildTopNSourceMeasure(topNSchema *databasev1.TopNAggregation, sourceMeasur
TagFamilies: []*databasev1.TagFamilySpec{
{
Name: topNTagFamily,
- Tags: append([]*databasev1.TagSpec{
- {
- Name: "measure_id",
- Type: databasev1.TagType_TAG_TYPE_STRING,
- },
- {
+ Tags: append(seriesSpecs,
+ &databasev1.TagSpec{
Name: "sortDirection",
Type: databasev1.TagType_TAG_TYPE_INT,
},
- {
+ &databasev1.TagSpec{
Name: "rankNumber",
Type: databasev1.TagType_TAG_TYPE_INT,
},
- }, seriesSpecs...),
+ ),
},
},
Fields: []*databasev1.FieldSpec{topNValueFieldSpec},
diff --git a/test/cases/topn/topn.go b/test/cases/topn/topn.go
index 48f5ad9b..e1fc8c39 100644
--- a/test/cases/topn/topn.go
+++ b/test/cases/topn/topn.go
@@ -39,7 +39,7 @@ var (
}
)
-var _ = g.FDescribeTable("TopN Tests", verify,
+var _ = g.DescribeTable("TopN Tests", verify,
g.Entry("max top3 order by desc", helpers.Args{Input: "aggr_desc", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("max top3 with condition order by desc", helpers.Args{Input: "condition_aggr_desc", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),
g.Entry("max top3 for null group order by desc", helpers.Args{Input: "null_group", Duration: 25 * time.Minute, Offset: -20 * time.Minute}),