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 3a438a59 Add more test cases and fix some issues found in the tests
(#426)
3a438a59 is described below
commit 3a438a5940d344684d56f0c56ef905efc35621a7
Author: Gao Hongtao <hanahm...@gmail.com>
AuthorDate: Wed Apr 3 15:37:49 2024 +0800
Add more test cases and fix some issues found in the tests (#426)
---
banyand/measure/block.go | 5 ++-
banyand/measure/block_metadata.go | 12 +++++-
banyand/stream/block.go | 5 ++-
banyand/stream/block_metadata.go | 11 ++++-
pkg/test/query/trace.go | 79 +++++++++++++++++++++++++++++++++--
test/stress/trace/trace_suite_test.go | 8 ++++
6 files changed, 110 insertions(+), 10 deletions(-)
diff --git a/banyand/measure/block.go b/banyand/measure/block.go
index cf3c2eab..b4792d85 100644
--- a/banyand/measure/block.go
+++ b/banyand/measure/block.go
@@ -610,10 +610,13 @@ func (bc *blockCursor) loadData(tmpBlock *block) bool {
}
bc.bm.field.columnMetadata = cfm
bc.bm.tagProjection = bc.tagProjection
- tf := make(map[string]*dataBlock, len(bc.tagProjection))
+ var tf map[string]*dataBlock
for i := range bc.tagProjection {
for tfName, block := range bc.bm.tagFamilies {
if bc.tagProjection[i].Family == tfName {
+ if tf == nil {
+ tf = make(map[string]*dataBlock,
len(bc.tagProjection))
+ }
tf[tfName] = block
}
}
diff --git a/banyand/measure/block_metadata.go
b/banyand/measure/block_metadata.go
index 82e44ca1..6a066329 100644
--- a/banyand/measure/block_metadata.go
+++ b/banyand/measure/block_metadata.go
@@ -176,7 +176,7 @@ func (bm *blockMetadata) unmarshal(src []byte) ([]byte,
error) {
if err != nil {
return nil, fmt.Errorf("cannot unmarshal
tagFamily dataBlock: %w", err)
}
- bm.tagFamilies[convert.BytesToString(nameBytes)] = tf
+ bm.tagFamilies[string(nameBytes)] = tf
}
}
src, err = bm.field.unmarshal(src)
@@ -212,6 +212,13 @@ type blockMetadataArray struct {
arr []blockMetadata
}
+func (bma *blockMetadataArray) reset() {
+ for i := range bma.arr {
+ bma.arr[i].reset()
+ }
+ bma.arr = bma.arr[:0]
+}
+
var blockMetadataArrayPool sync.Pool
func generateBlockMetadataArray() *blockMetadataArray {
@@ -223,7 +230,7 @@ func generateBlockMetadataArray() *blockMetadataArray {
}
func releaseBlockMetadataArray(bma *blockMetadataArray) {
- bma.arr = bma.arr[:0]
+ bma.reset()
blockMetadataArrayPool.Put(bma)
}
@@ -279,6 +286,7 @@ func unmarshalBlockMetadata(dst []blockMetadata, src
[]byte) ([]blockMetadata, e
dst = append(dst, blockMetadata{})
}
bm := &dst[len(dst)-1]
+ bm.reset()
tail, err := bm.unmarshal(src)
if err != nil {
return dstOrig, fmt.Errorf("cannot unmarshal
blockMetadata entries: %w", err)
diff --git a/banyand/stream/block.go b/banyand/stream/block.go
index e8f63200..63342b11 100644
--- a/banyand/stream/block.go
+++ b/banyand/stream/block.go
@@ -529,10 +529,13 @@ func (bc *blockCursor) copyTo(r *pbv1.StreamResult) {
func (bc *blockCursor) loadData(tmpBlock *block) bool {
tmpBlock.reset()
bc.bm.tagProjection = bc.tagProjection
- tf := make(map[string]*dataBlock, len(bc.tagProjection))
+ var tf map[string]*dataBlock
for i := range bc.tagProjection {
for tfName, block := range bc.bm.tagFamilies {
if bc.tagProjection[i].Family == tfName {
+ if tf == nil {
+ tf = make(map[string]*dataBlock,
len(bc.tagProjection))
+ }
tf[tfName] = block
}
}
diff --git a/banyand/stream/block_metadata.go b/banyand/stream/block_metadata.go
index 8c8e2771..3906f603 100644
--- a/banyand/stream/block_metadata.go
+++ b/banyand/stream/block_metadata.go
@@ -181,7 +181,7 @@ func (bm *blockMetadata) unmarshal(src []byte) ([]byte,
error) {
if err != nil {
return nil, fmt.Errorf("cannot unmarshal
tagFamily dataBlock: %w", err)
}
- bm.tagFamilies[convert.BytesToString(nameBytes)] = tf
+ bm.tagFamilies[string(nameBytes)] = tf
}
}
if err != nil {
@@ -216,6 +216,13 @@ type blockMetadataArray struct {
arr []blockMetadata
}
+func (bma *blockMetadataArray) reset() {
+ for i := range bma.arr {
+ bma.arr[i].reset()
+ }
+ bma.arr = bma.arr[:0]
+}
+
var blockMetadataArrayPool sync.Pool
func generateBlockMetadataArray() *blockMetadataArray {
@@ -227,7 +234,7 @@ func generateBlockMetadataArray() *blockMetadataArray {
}
func releaseBlockMetadataArray(bma *blockMetadataArray) {
- bma.arr = bma.arr[:0]
+ bma.reset()
blockMetadataArrayPool.Put(bma)
}
diff --git a/pkg/test/query/trace.go b/pkg/test/query/trace.go
index 3c4834f1..18da1bf8 100644
--- a/pkg/test/query/trace.go
+++ b/pkg/test/query/trace.go
@@ -43,7 +43,51 @@ func TraceListOrderByDuration(basePath string, timeout
time.Duration, fs *flag.F
close(stopCh)
}()
collect(basePath, func() ([]float64, error) {
- d, err := traceList(api.QueryOrderByDuration, fs)
+ d, _, err := traceList(api.QueryOrderByDuration, fs)
+ if err != nil {
+ return nil, err
+ }
+ return []float64{d.Seconds()}, nil
+ }, 500*time.Millisecond, stopCh)
+ analyze([]string{"result"}, basePath)
+}
+
+// TraceListOrderByTime verifies the trace list order by time.
+func TraceListOrderByTime(basePath string, timeout time.Duration, fs
*flag.FlagSet) {
+ basePath = path.Join(basePath, "trace-time")
+ err := os.MkdirAll(basePath, 0o755)
+ if err != nil {
+ panic(err)
+ }
+ stopCh := make(chan struct{})
+ go func() {
+ time.Sleep(timeout)
+ close(stopCh)
+ }()
+ collect(basePath, func() ([]float64, error) {
+ d, _, err := traceList(api.QueryOrderByStartTime, fs)
+ if err != nil {
+ return nil, err
+ }
+ return []float64{d.Seconds()}, nil
+ }, 500*time.Millisecond, stopCh)
+ analyze([]string{"result"}, basePath)
+}
+
+// TraceByID verifies the trace by ID.
+func TraceByID(basePath string, timeout time.Duration, fs *flag.FlagSet) {
+ basePath = path.Join(basePath, "trace-by-id")
+ err := os.MkdirAll(basePath, 0o755)
+ if err != nil {
+ panic(err)
+ }
+ stopCh := make(chan struct{})
+ go func() {
+ time.Sleep(timeout)
+ close(stopCh)
+ }()
+ collect(basePath, func() ([]float64, error) {
+ d, err := traceByID(api.QueryOrderByDuration, fs)
if err != nil {
return nil, err
}
@@ -56,7 +100,7 @@ const (
defaultPageSize = 15
)
-func traceList(order api.QueryOrder, fs *flag.FlagSet) (time.Duration, error) {
+func traceList(order api.QueryOrder, fs *flag.FlagSet) (time.Duration,
[]*api.BasicTrace, error) {
ctx := cli.NewContext(cli.NewApp(), fs, nil)
duration := api.Duration{
Start: time.Now().Add(-30 *
time.Minute).Format(utils.StepFormats[api.StepMinute]),
@@ -79,10 +123,37 @@ func traceList(order api.QueryOrder, fs *flag.FlagSet)
(time.Duration, error) {
result, err := trace.Traces(ctx, condition)
elapsed := time.Since(start)
if err != nil {
- return 0, err
+ return 0, nil, err
}
if len(result.Traces) < 1 {
+ return 0, nil, fmt.Errorf("no result")
+ }
+ return elapsed, result.Traces, nil
+}
+
+func traceByID(order api.QueryOrder, fs *flag.FlagSet) (time.Duration, error) {
+ _, traces, err := traceList(order, fs)
+ if err != nil {
+ return 0, err
+ }
+ if len(traces) < 1 {
return 0, fmt.Errorf("no result")
}
- return elapsed, nil
+ start := time.Now()
+ numTraces := int64(0)
+ for _, bt := range traces {
+ for _, id := range bt.TraceIds {
+ ctx := cli.NewContext(cli.NewApp(), fs, nil)
+ t, err := trace.Trace(ctx, id)
+ if err != nil {
+ return 0, err
+ }
+ if len(t.Spans) < 1 {
+ return 0, fmt.Errorf("no result span")
+ }
+ numTraces++
+ }
+ }
+ elapsed := time.Since(start)
+ return time.Duration(int64(elapsed) / numTraces), nil
}
diff --git a/test/stress/trace/trace_suite_test.go
b/test/stress/trace/trace_suite_test.go
index a749a236..75d7ddf8 100644
--- a/test/stress/trace/trace_suite_test.go
+++ b/test/stress/trace/trace_suite_test.go
@@ -54,4 +54,12 @@ var _ = Describe("Query", func() {
It("TraceListOrderByDuration", func() {
query.TraceListOrderByDuration(basePath, timeout, fs)
})
+
+ It("TraceListOrderByTime", func() {
+ query.TraceListOrderByTime(basePath, timeout, fs)
+ })
+
+ It("TraceByID", func() {
+ query.TraceByID(basePath, timeout, fs)
+ })
})
