github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2024-12-31 15:06:26 +00:00
Code Issues Projects Releases Packages Wiki Activity

wip

Browse source
This commit is contained in:
Aliaksandr Valialkin 2024-05-13 16:45:34 +02:00
parent 9673da2578
commit 900e558678
No known key found for this signature in database
GPG key ID: 52C003EE2BCDB9EB
7 changed files with 26 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
lib/logstorage/block_result.go
View file

@ -4,6 +4,7 @@ import (
"encoding/binary"
"math"
"slices"
"sync/atomic"
"time"
"unsafe"
@ -12,6 +13,7 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fastnum"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
)
// blockResult holds results for a single block of log entries.
@ -1512,7 +1514,7 @@ func (c *blockResultColumn) getFloatValueAtRow(rowIdx int) float64 {
}
}
func (c *blockResultColumn) getMaxValue(_ *blockResult) float64 {
func (c *blockResultColumn) getMaxValue() float64 {
if c.isConst {
v := c.encodedValues[0]
f, ok := tryParseFloat64(v)
@ -1620,7 +1622,7 @@ func (c *blockResultColumn) getMaxValue(_ *blockResult) float64 {
}
}
func (c *blockResultColumn) getMinValue(_ *blockResult) float64 {
func (c *blockResultColumn) getMinValue() float64 {
if c.isConst {
v := c.encodedValues[0]
f, ok := tryParseFloat64(v)
@ -1870,5 +1872,18 @@ func truncateTimestampToYear(timestamp int64) int64 {
return time.Date(t.Year(), time.January, 1, 0, 0, 0, 0, time.UTC).UnixNano()
}
func getEmptyStrings(rowsCount int) []string {
p := emptyStrings.Load()
if p == nil {
values := make([]string, rowsCount)
emptyStrings.Store(&values)
return values
}
values := *p
return slicesutil.SetLength(values, rowsCount)
}
var emptyStrings atomic.Pointer[[]string]
var nan = math.NaN()
var inf = math.Inf(1)

4
lib/logstorage/stats_count_uniq.go
View file

@ -175,6 +175,7 @@ func (sup *statsCountUniqProcessor) updateStatsForAllRows(br *blockResult) int {
stateSizeIncrease += len(keyBuf) + int(unsafe.Sizeof(""))
}
}
sup.keyBuf = keyBuf
return stateSizeIncrease
}
@ -307,7 +308,7 @@ func (sup *statsCountUniqProcessor) updateStatsForRow(br *blockResult, rowIdx in
m[string(keyBuf)] = struct{}{}
stateSizeIncrease += len(keyBuf) + int(unsafe.Sizeof(""))
}
//sup.keyBuf = keyBuf
sup.keyBuf = keyBuf
return stateSizeIncrease
}
@ -324,6 +325,7 @@ func (sup *statsCountUniqProcessor) updateStatsForRow(br *blockResult, rowIdx in
m[string(keyBuf)] = struct{}{}
stateSizeIncrease += len(keyBuf) + int(unsafe.Sizeof(""))
}
sup.keyBuf = keyBuf
return stateSizeIncrease
}

4
lib/logstorage/stats_max.go
View file

@ -38,7 +38,7 @@ func (smp *statsMaxProcessor) updateStatsForAllRows(br *blockResult) int {
if smp.sm.containsStar {
// Find the maximum value across all the columns
for _, c := range br.getColumns() {
f := c.getMaxValue(br)
f := c.getMaxValue()
if f > smp.max || math.IsNaN(smp.max) {
smp.max = f
}
@ -47,7 +47,7 @@ func (smp *statsMaxProcessor) updateStatsForAllRows(br *blockResult) int {
// Find the maximum value across the requested columns
for _, field := range smp.sm.fields {
c := br.getColumnByName(field)
f := c.getMaxValue(br)
f := c.getMaxValue()
if f > smp.max || math.IsNaN(smp.max) {
smp.max = f
}

4
lib/logstorage/stats_min.go
View file

@ -38,7 +38,7 @@ func (smp *statsMinProcessor) updateStatsForAllRows(br *blockResult) int {
if smp.sm.containsStar {
// Find the minimum value across all the columns
for _, c := range br.getColumns() {
f := c.getMinValue(br)
f := c.getMinValue()
if f < smp.min || math.IsNaN(smp.min) {
smp.min = f
}
@ -47,7 +47,7 @@ func (smp *statsMinProcessor) updateStatsForAllRows(br *blockResult) int {
// Find the minimum value across the requested columns
for _, field := range smp.sm.fields {
c := br.getColumnByName(field)
f := c.getMinValue(br)
f := c.getMinValue()
if f < smp.min || math.IsNaN(smp.min) {
smp.min = f
}

15
lib/logstorage/storage_search.go
View file

@ -6,10 +6,8 @@ import (
"slices"
"sort"
"sync"
"sync/atomic"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
)
// genericSearchOptions contain options used for search.
@ -164,19 +162,6 @@ func (c *BlockColumn) reset() {
c.Values = nil
}
func getEmptyStrings(rowsCount int) []string {
p := emptyStrings.Load()
if p == nil {
values := make([]string, rowsCount)
emptyStrings.Store(&values)
return values
}
values := *p
return slicesutil.SetLength(values, rowsCount)
}
var emptyStrings atomic.Pointer[[]string]
// The number of blocks to search at once by a single worker
//
// This number must be increased on systems with many CPU cores in order to amortize

6
lib/logstorage/values_encoder.go
View file

@ -1096,11 +1096,7 @@ func (vd *valuesDict) copyFrom(a *arena, src *valuesDict) {
func (vd *valuesDict) copyFromNoArena(src *valuesDict) {
vd.reset()
dstValues := vd.values
for _, v := range src.values {
dstValues = append(dstValues, v)
}
vd.values = dstValues
vd.values = append(vd.values[:0], src.values...)
}
func (vd *valuesDict) getOrAdd(k string) (byte, bool) {

2
lib/streamaggr/rate.go
View file

@ -103,7 +103,7 @@ func (as *rateAggrState) pushSamples(samples []pushSample) {
}
}
func (as *rateAggrState) flushState(ctx *flushCtx, resetState bool) {
func (as *rateAggrState) flushState(ctx *flushCtx, _ bool) {
currentTime := fasttime.UnixTimestamp()
currentTimeMsec := int64(currentTime) * 1000