2024-05-12 16:33:29 +02:00
|
|
|
package logstorage
|
|
|
|
|
|
|
|
import (
|
|
|
|
"fmt"
|
|
|
|
"slices"
|
2024-05-20 04:08:30 +02:00
|
|
|
"strings"
|
2024-10-17 23:44:38 +02:00
|
|
|
"sync"
|
2024-05-12 16:33:29 +02:00
|
|
|
"sync/atomic"
|
|
|
|
"unsafe"
|
|
|
|
|
2024-10-17 23:44:38 +02:00
|
|
|
"github.com/cespare/xxhash/v2"
|
|
|
|
|
2024-05-12 16:33:29 +02:00
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
|
|
|
|
)
|
|
|
|
|
|
|
|
// pipeUniq processes '| uniq ...' queries.
|
|
|
|
//
|
|
|
|
// See https://docs.victoriametrics.com/victorialogs/logsql/#uniq-pipe
|
|
|
|
type pipeUniq struct {
|
|
|
|
// fields contains field names for returning unique values
|
|
|
|
byFields []string
|
|
|
|
|
2024-05-24 03:06:55 +02:00
|
|
|
// if hitsFieldName isn't empty, then the number of hits per each unique value is stored in this field.
|
|
|
|
hitsFieldName string
|
|
|
|
|
2024-05-12 16:33:29 +02:00
|
|
|
limit uint64
|
|
|
|
}
|
|
|
|
|
|
|
|
func (pu *pipeUniq) String() string {
|
|
|
|
s := "uniq"
|
|
|
|
if len(pu.byFields) > 0 {
|
|
|
|
s += " by (" + fieldNamesString(pu.byFields) + ")"
|
|
|
|
}
|
2024-05-24 03:06:55 +02:00
|
|
|
if pu.hitsFieldName != "" {
|
2024-05-25 21:36:16 +02:00
|
|
|
s += " with hits"
|
2024-05-24 03:06:55 +02:00
|
|
|
}
|
2024-05-12 16:33:29 +02:00
|
|
|
if pu.limit > 0 {
|
|
|
|
s += fmt.Sprintf(" limit %d", pu.limit)
|
|
|
|
}
|
|
|
|
return s
|
|
|
|
}
|
|
|
|
|
2024-06-27 14:18:42 +02:00
|
|
|
func (pu *pipeUniq) canLiveTail() bool {
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
2024-05-12 16:33:29 +02:00
|
|
|
func (pu *pipeUniq) updateNeededFields(neededFields, unneededFields fieldsSet) {
|
|
|
|
neededFields.reset()
|
|
|
|
unneededFields.reset()
|
|
|
|
|
|
|
|
if len(pu.byFields) == 0 {
|
|
|
|
neededFields.add("*")
|
|
|
|
} else {
|
2024-05-20 04:08:30 +02:00
|
|
|
neededFields.addFields(pu.byFields)
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2024-05-25 21:36:16 +02:00
|
|
|
func (pu *pipeUniq) optimize() {
|
|
|
|
// nothing to do
|
|
|
|
}
|
|
|
|
|
|
|
|
func (pu *pipeUniq) hasFilterInWithQuery() bool {
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
2024-05-26 02:01:32 +02:00
|
|
|
func (pu *pipeUniq) initFilterInValues(_ map[string][]string, _ getFieldValuesFunc) (pipe, error) {
|
2024-05-25 21:36:16 +02:00
|
|
|
return pu, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (pu *pipeUniq) newPipeProcessor(workersCount int, stopCh <-chan struct{}, cancel func(), ppNext pipeProcessor) pipeProcessor {
|
2024-05-12 16:33:29 +02:00
|
|
|
maxStateSize := int64(float64(memory.Allowed()) * 0.2)
|
|
|
|
|
|
|
|
shards := make([]pipeUniqProcessorShard, workersCount)
|
|
|
|
for i := range shards {
|
2024-05-20 04:08:30 +02:00
|
|
|
shards[i] = pipeUniqProcessorShard{
|
|
|
|
pipeUniqProcessorShardNopad: pipeUniqProcessorShardNopad{
|
2024-09-29 10:16:14 +02:00
|
|
|
pu: pu,
|
2024-05-20 04:08:30 +02:00
|
|
|
},
|
|
|
|
}
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
pup := &pipeUniqProcessor{
|
|
|
|
pu: pu,
|
|
|
|
stopCh: stopCh,
|
|
|
|
cancel: cancel,
|
2024-05-25 21:36:16 +02:00
|
|
|
ppNext: ppNext,
|
2024-05-12 16:33:29 +02:00
|
|
|
|
|
|
|
shards: shards,
|
|
|
|
|
|
|
|
maxStateSize: maxStateSize,
|
|
|
|
}
|
|
|
|
pup.stateSizeBudget.Store(maxStateSize)
|
|
|
|
|
|
|
|
return pup
|
|
|
|
}
|
|
|
|
|
|
|
|
type pipeUniqProcessor struct {
|
|
|
|
pu *pipeUniq
|
|
|
|
stopCh <-chan struct{}
|
|
|
|
cancel func()
|
2024-05-25 21:36:16 +02:00
|
|
|
ppNext pipeProcessor
|
2024-05-12 16:33:29 +02:00
|
|
|
|
|
|
|
shards []pipeUniqProcessorShard
|
|
|
|
|
|
|
|
maxStateSize int64
|
|
|
|
stateSizeBudget atomic.Int64
|
|
|
|
}
|
|
|
|
|
|
|
|
type pipeUniqProcessorShard struct {
|
|
|
|
pipeUniqProcessorShardNopad
|
|
|
|
|
|
|
|
// The padding prevents false sharing on widespread platforms with 128 mod (cache line size) = 0 .
|
|
|
|
_ [128 - unsafe.Sizeof(pipeUniqProcessorShardNopad{})%128]byte
|
|
|
|
}
|
|
|
|
|
|
|
|
type pipeUniqProcessorShardNopad struct {
|
|
|
|
// pu points to the parent pipeUniq.
|
|
|
|
pu *pipeUniq
|
|
|
|
|
2024-05-24 03:06:55 +02:00
|
|
|
// m holds per-row hits.
|
|
|
|
m map[string]*uint64
|
2024-05-12 16:33:29 +02:00
|
|
|
|
|
|
|
// keyBuf is a temporary buffer for building keys for m.
|
|
|
|
keyBuf []byte
|
|
|
|
|
|
|
|
// columnValues is a temporary buffer for the processed column values.
|
|
|
|
columnValues [][]string
|
|
|
|
|
|
|
|
// stateSizeBudget is the remaining budget for the whole state size for the shard.
|
|
|
|
// The per-shard budget is provided in chunks from the parent pipeUniqProcessor.
|
|
|
|
stateSizeBudget int
|
|
|
|
}
|
|
|
|
|
|
|
|
// writeBlock writes br to shard.
|
|
|
|
//
|
|
|
|
// It returns false if the block cannot be written because of the exceeded limit.
|
|
|
|
func (shard *pipeUniqProcessorShard) writeBlock(br *blockResult) bool {
|
2024-09-29 10:50:25 +02:00
|
|
|
if limit := shard.pu.limit; limit > 0 && uint64(len(shard.m)) > limit {
|
2024-05-12 16:33:29 +02:00
|
|
|
return false
|
|
|
|
}
|
|
|
|
|
2024-05-24 03:06:55 +02:00
|
|
|
needHits := shard.pu.hitsFieldName != ""
|
2024-05-12 16:33:29 +02:00
|
|
|
byFields := shard.pu.byFields
|
|
|
|
if len(byFields) == 0 {
|
|
|
|
// Take into account all the columns in br.
|
|
|
|
keyBuf := shard.keyBuf
|
|
|
|
cs := br.getColumns()
|
2024-09-25 16:16:53 +02:00
|
|
|
for i := 0; i < br.rowsLen; i++ {
|
2024-05-12 16:33:29 +02:00
|
|
|
keyBuf = keyBuf[:0]
|
|
|
|
for _, c := range cs {
|
|
|
|
v := c.getValueAtRow(br, i)
|
|
|
|
keyBuf = encoding.MarshalBytes(keyBuf, bytesutil.ToUnsafeBytes(c.name))
|
|
|
|
keyBuf = encoding.MarshalBytes(keyBuf, bytesutil.ToUnsafeBytes(v))
|
|
|
|
}
|
2024-05-24 03:06:55 +02:00
|
|
|
shard.updateState(bytesutil.ToUnsafeString(keyBuf), 1)
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
|
|
|
shard.keyBuf = keyBuf
|
|
|
|
return true
|
|
|
|
}
|
2024-05-20 04:08:30 +02:00
|
|
|
if len(byFields) == 1 {
|
|
|
|
// Fast path for a single field.
|
|
|
|
c := br.getColumnByName(byFields[0])
|
|
|
|
if c.isConst {
|
|
|
|
v := c.valuesEncoded[0]
|
2024-09-25 16:16:53 +02:00
|
|
|
shard.updateState(v, uint64(br.rowsLen))
|
2024-05-20 04:08:30 +02:00
|
|
|
return true
|
|
|
|
}
|
|
|
|
if c.valueType == valueTypeDict {
|
2024-10-01 13:29:07 +02:00
|
|
|
c.forEachDictValueWithHits(br, shard.updateState)
|
2024-05-20 04:08:30 +02:00
|
|
|
return true
|
|
|
|
}
|
|
|
|
|
|
|
|
values := c.getValues(br)
|
|
|
|
for i, v := range values {
|
2024-05-24 03:06:55 +02:00
|
|
|
if needHits || i == 0 || values[i-1] != values[i] {
|
|
|
|
shard.updateState(v, 1)
|
2024-05-20 04:08:30 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
return true
|
|
|
|
}
|
2024-05-12 16:33:29 +02:00
|
|
|
|
|
|
|
// Take into account only the selected columns.
|
|
|
|
columnValues := shard.columnValues[:0]
|
|
|
|
for _, f := range byFields {
|
|
|
|
c := br.getColumnByName(f)
|
2024-05-20 04:08:30 +02:00
|
|
|
values := c.getValues(br)
|
|
|
|
columnValues = append(columnValues, values)
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
|
|
|
shard.columnValues = columnValues
|
|
|
|
|
|
|
|
keyBuf := shard.keyBuf
|
2024-09-25 16:16:53 +02:00
|
|
|
for i := 0; i < br.rowsLen; i++ {
|
2024-05-12 16:33:29 +02:00
|
|
|
seenValue := true
|
|
|
|
for _, values := range columnValues {
|
2024-05-24 03:06:55 +02:00
|
|
|
if needHits || i == 0 || values[i-1] != values[i] {
|
2024-05-12 16:33:29 +02:00
|
|
|
seenValue = false
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if seenValue {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
keyBuf = keyBuf[:0]
|
|
|
|
for _, values := range columnValues {
|
|
|
|
keyBuf = encoding.MarshalBytes(keyBuf, bytesutil.ToUnsafeBytes(values[i]))
|
|
|
|
}
|
2024-05-24 03:06:55 +02:00
|
|
|
shard.updateState(bytesutil.ToUnsafeString(keyBuf), 1)
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
|
|
|
shard.keyBuf = keyBuf
|
|
|
|
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
|
2024-05-24 03:06:55 +02:00
|
|
|
func (shard *pipeUniqProcessorShard) updateState(v string, hits uint64) {
|
|
|
|
m := shard.getM()
|
|
|
|
pHits, ok := m[v]
|
|
|
|
if !ok {
|
2024-05-20 04:08:30 +02:00
|
|
|
vCopy := strings.Clone(v)
|
2024-05-24 03:06:55 +02:00
|
|
|
hits := uint64(0)
|
|
|
|
pHits = &hits
|
|
|
|
m[vCopy] = pHits
|
|
|
|
shard.stateSizeBudget -= len(vCopy) + int(unsafe.Sizeof(vCopy)+unsafe.Sizeof(hits)+unsafe.Sizeof(pHits))
|
2024-05-20 04:08:30 +02:00
|
|
|
}
|
2024-05-24 03:06:55 +02:00
|
|
|
*pHits += hits
|
|
|
|
}
|
|
|
|
|
|
|
|
func (shard *pipeUniqProcessorShard) getM() map[string]*uint64 {
|
|
|
|
if shard.m == nil {
|
|
|
|
shard.m = make(map[string]*uint64)
|
|
|
|
}
|
|
|
|
return shard.m
|
2024-05-20 04:08:30 +02:00
|
|
|
}
|
|
|
|
|
2024-05-12 16:33:29 +02:00
|
|
|
func (pup *pipeUniqProcessor) writeBlock(workerID uint, br *blockResult) {
|
2024-09-25 16:16:53 +02:00
|
|
|
if br.rowsLen == 0 {
|
2024-05-12 16:33:29 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
shard := &pup.shards[workerID]
|
|
|
|
|
|
|
|
for shard.stateSizeBudget < 0 {
|
|
|
|
// steal some budget for the state size from the global budget.
|
|
|
|
remaining := pup.stateSizeBudget.Add(-stateSizeBudgetChunk)
|
|
|
|
if remaining < 0 {
|
|
|
|
// The state size is too big. Stop processing data in order to avoid OOM crash.
|
|
|
|
if remaining+stateSizeBudgetChunk >= 0 {
|
|
|
|
// Notify worker goroutines to stop calling writeBlock() in order to save CPU time.
|
|
|
|
pup.cancel()
|
|
|
|
}
|
|
|
|
return
|
|
|
|
}
|
|
|
|
shard.stateSizeBudget += stateSizeBudgetChunk
|
|
|
|
}
|
|
|
|
|
|
|
|
if !shard.writeBlock(br) {
|
|
|
|
pup.cancel()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (pup *pipeUniqProcessor) flush() error {
|
|
|
|
if n := pup.stateSizeBudget.Load(); n <= 0 {
|
|
|
|
return fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", pup.pu.String(), pup.maxStateSize/(1<<20))
|
|
|
|
}
|
|
|
|
|
2024-10-17 23:44:38 +02:00
|
|
|
// merge state across shards in parallel
|
|
|
|
ms, err := pup.mergeShardsParallel()
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if needStop(pup.stopCh) {
|
|
|
|
return nil
|
|
|
|
}
|
2024-05-12 16:33:29 +02:00
|
|
|
|
2024-10-17 23:44:38 +02:00
|
|
|
resetHits := false
|
|
|
|
if limit := pup.pu.limit; limit > 0 {
|
|
|
|
// Trim the number of entries according to the given limit
|
|
|
|
entriesLen := 0
|
|
|
|
result := ms[:0]
|
|
|
|
for _, m := range ms {
|
|
|
|
entriesLen += len(m)
|
|
|
|
if uint64(entriesLen) <= limit {
|
|
|
|
result = append(result, m)
|
|
|
|
continue
|
2024-05-24 03:06:55 +02:00
|
|
|
}
|
2024-10-17 23:44:38 +02:00
|
|
|
|
|
|
|
// There is little sense in returning partial hits when the limit on the number of unique entries is reached,
|
|
|
|
// since arbitrary number of unique entries and hits for these entries could be skipped.
|
|
|
|
// It is better to return zero hits instead of misleading hits results.
|
|
|
|
resetHits = true
|
|
|
|
for k := range m {
|
|
|
|
delete(m, k)
|
|
|
|
entriesLen--
|
|
|
|
if uint64(entriesLen) <= limit {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if len(m) > 0 {
|
|
|
|
result = append(result, m)
|
|
|
|
}
|
|
|
|
break
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
2024-10-17 23:44:38 +02:00
|
|
|
ms = result
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
|
|
|
|
2024-10-17 23:44:38 +02:00
|
|
|
// Write the calculated stats in parallel to the next pipe.
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
for i, m := range ms {
|
|
|
|
wg.Add(1)
|
|
|
|
go func(workerID uint) {
|
|
|
|
defer wg.Done()
|
|
|
|
pup.writeShardData(workerID, m, resetHits)
|
|
|
|
}(uint(i))
|
|
|
|
}
|
|
|
|
wg.Wait()
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
2024-05-24 03:06:55 +02:00
|
|
|
|
2024-10-17 23:44:38 +02:00
|
|
|
func (pup *pipeUniqProcessor) writeShardData(workerID uint, m map[string]*uint64, resetHits bool) {
|
2024-05-12 16:33:29 +02:00
|
|
|
wctx := &pipeUniqWriteContext{
|
2024-10-17 23:44:38 +02:00
|
|
|
workerID: workerID,
|
|
|
|
pup: pup,
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
|
|
|
byFields := pup.pu.byFields
|
|
|
|
var rowFields []Field
|
|
|
|
|
2024-05-24 03:06:55 +02:00
|
|
|
addHitsFieldIfNeeded := func(dst []Field, hits uint64) []Field {
|
|
|
|
if pup.pu.hitsFieldName == "" {
|
|
|
|
return dst
|
|
|
|
}
|
|
|
|
if resetHits {
|
|
|
|
hits = 0
|
|
|
|
}
|
|
|
|
hitsStr := string(marshalUint64String(nil, hits))
|
|
|
|
dst = append(dst, Field{
|
|
|
|
Name: pup.pu.hitsFieldName,
|
|
|
|
Value: hitsStr,
|
|
|
|
})
|
|
|
|
return dst
|
|
|
|
}
|
|
|
|
|
2024-05-12 16:33:29 +02:00
|
|
|
if len(byFields) == 0 {
|
2024-05-24 03:06:55 +02:00
|
|
|
for k, pHits := range m {
|
2024-05-15 04:55:44 +02:00
|
|
|
if needStop(pup.stopCh) {
|
2024-10-17 23:44:38 +02:00
|
|
|
return
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
rowFields = rowFields[:0]
|
|
|
|
keyBuf := bytesutil.ToUnsafeBytes(k)
|
|
|
|
for len(keyBuf) > 0 {
|
2024-05-14 01:23:44 +02:00
|
|
|
name, nSize := encoding.UnmarshalBytes(keyBuf)
|
|
|
|
if nSize <= 0 {
|
|
|
|
logger.Panicf("BUG: cannot unmarshal field name")
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
2024-05-14 01:23:44 +02:00
|
|
|
keyBuf = keyBuf[nSize:]
|
2024-05-12 16:33:29 +02:00
|
|
|
|
2024-05-14 01:23:44 +02:00
|
|
|
value, nSize := encoding.UnmarshalBytes(keyBuf)
|
|
|
|
if nSize <= 0 {
|
|
|
|
logger.Panicf("BUG: cannot unmarshal field value")
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
2024-05-14 01:23:44 +02:00
|
|
|
keyBuf = keyBuf[nSize:]
|
2024-05-12 16:33:29 +02:00
|
|
|
|
|
|
|
rowFields = append(rowFields, Field{
|
|
|
|
Name: bytesutil.ToUnsafeString(name),
|
|
|
|
Value: bytesutil.ToUnsafeString(value),
|
|
|
|
})
|
|
|
|
}
|
2024-05-24 03:06:55 +02:00
|
|
|
rowFields = addHitsFieldIfNeeded(rowFields, *pHits)
|
2024-05-12 16:33:29 +02:00
|
|
|
wctx.writeRow(rowFields)
|
|
|
|
}
|
2024-05-20 04:08:30 +02:00
|
|
|
} else if len(byFields) == 1 {
|
|
|
|
fieldName := byFields[0]
|
2024-05-24 03:06:55 +02:00
|
|
|
for k, pHits := range m {
|
2024-05-20 04:08:30 +02:00
|
|
|
if needStop(pup.stopCh) {
|
2024-10-17 23:44:38 +02:00
|
|
|
return
|
2024-05-20 04:08:30 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
rowFields = append(rowFields[:0], Field{
|
|
|
|
Name: fieldName,
|
|
|
|
Value: k,
|
|
|
|
})
|
2024-05-24 03:06:55 +02:00
|
|
|
rowFields = addHitsFieldIfNeeded(rowFields, *pHits)
|
2024-05-20 04:08:30 +02:00
|
|
|
wctx.writeRow(rowFields)
|
|
|
|
}
|
2024-05-12 16:33:29 +02:00
|
|
|
} else {
|
2024-05-24 03:06:55 +02:00
|
|
|
for k, pHits := range m {
|
2024-05-15 04:55:44 +02:00
|
|
|
if needStop(pup.stopCh) {
|
2024-10-17 23:44:38 +02:00
|
|
|
return
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
rowFields = rowFields[:0]
|
|
|
|
keyBuf := bytesutil.ToUnsafeBytes(k)
|
|
|
|
fieldIdx := 0
|
|
|
|
for len(keyBuf) > 0 {
|
2024-05-14 01:23:44 +02:00
|
|
|
value, nSize := encoding.UnmarshalBytes(keyBuf)
|
|
|
|
if nSize <= 0 {
|
|
|
|
logger.Panicf("BUG: cannot unmarshal field value")
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
2024-05-14 01:23:44 +02:00
|
|
|
keyBuf = keyBuf[nSize:]
|
2024-05-12 16:33:29 +02:00
|
|
|
|
|
|
|
rowFields = append(rowFields, Field{
|
|
|
|
Name: byFields[fieldIdx],
|
|
|
|
Value: bytesutil.ToUnsafeString(value),
|
|
|
|
})
|
|
|
|
fieldIdx++
|
|
|
|
}
|
2024-05-24 03:06:55 +02:00
|
|
|
rowFields = addHitsFieldIfNeeded(rowFields, *pHits)
|
2024-05-12 16:33:29 +02:00
|
|
|
wctx.writeRow(rowFields)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
wctx.flush()
|
2024-10-17 23:44:38 +02:00
|
|
|
}
|
2024-05-12 16:33:29 +02:00
|
|
|
|
2024-10-17 23:44:38 +02:00
|
|
|
func (pup *pipeUniqProcessor) mergeShardsParallel() ([]map[string]*uint64, error) {
|
|
|
|
shards := pup.shards
|
|
|
|
shardsLen := len(shards)
|
|
|
|
if shardsLen == 1 {
|
|
|
|
m := shards[0].getM()
|
|
|
|
var ms []map[string]*uint64
|
|
|
|
if len(m) > 0 {
|
|
|
|
ms = append(ms, m)
|
|
|
|
}
|
|
|
|
return ms, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
perShardMaps := make([][]map[string]*uint64, shardsLen)
|
|
|
|
for i := range shards {
|
|
|
|
wg.Add(1)
|
|
|
|
go func(idx int) {
|
|
|
|
defer wg.Done()
|
|
|
|
|
|
|
|
shardMaps := make([]map[string]*uint64, shardsLen)
|
|
|
|
for i := range shardMaps {
|
|
|
|
shardMaps[i] = make(map[string]*uint64)
|
|
|
|
}
|
|
|
|
|
|
|
|
n := int64(0)
|
|
|
|
nTotal := int64(0)
|
|
|
|
for k, pHits := range shards[idx].getM() {
|
|
|
|
if needStop(pup.stopCh) {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
h := xxhash.Sum64(bytesutil.ToUnsafeBytes(k))
|
|
|
|
m := shardMaps[h%uint64(len(shardMaps))]
|
|
|
|
n += updatePipeUniqMap(m, k, pHits)
|
|
|
|
if n > stateSizeBudgetChunk {
|
|
|
|
if nRemaining := pup.stateSizeBudget.Add(-n); nRemaining < 0 {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
nTotal += n
|
|
|
|
n = 0
|
|
|
|
}
|
|
|
|
}
|
|
|
|
nTotal += n
|
|
|
|
pup.stateSizeBudget.Add(-n)
|
|
|
|
|
|
|
|
perShardMaps[idx] = shardMaps
|
|
|
|
|
|
|
|
// Clean the original map and return its state size budget back.
|
|
|
|
shards[idx].m = nil
|
|
|
|
pup.stateSizeBudget.Add(nTotal)
|
|
|
|
}(i)
|
|
|
|
}
|
|
|
|
wg.Wait()
|
|
|
|
if needStop(pup.stopCh) {
|
|
|
|
return nil, nil
|
|
|
|
}
|
|
|
|
if n := pup.stateSizeBudget.Load(); n < 0 {
|
|
|
|
return nil, fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", pup.pu.String(), pup.maxStateSize/(1<<20))
|
|
|
|
}
|
|
|
|
|
|
|
|
// Merge per-shard entries into perShardMaps[0]
|
|
|
|
for i := range perShardMaps {
|
|
|
|
wg.Add(1)
|
|
|
|
go func(idx int) {
|
|
|
|
defer wg.Done()
|
|
|
|
|
|
|
|
m := perShardMaps[0][idx]
|
|
|
|
for i := 1; i < len(perShardMaps); i++ {
|
|
|
|
n := int64(0)
|
|
|
|
nTotal := int64(0)
|
|
|
|
for k, psg := range perShardMaps[i][idx] {
|
|
|
|
if needStop(pup.stopCh) {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
n += updatePipeUniqMap(m, k, psg)
|
|
|
|
if n > stateSizeBudgetChunk {
|
|
|
|
if nRemaining := pup.stateSizeBudget.Add(-n); nRemaining < 0 {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
nTotal += n
|
|
|
|
n = 0
|
|
|
|
}
|
|
|
|
}
|
|
|
|
nTotal += n
|
|
|
|
pup.stateSizeBudget.Add(-n)
|
|
|
|
|
|
|
|
// Clean the original map and return its state size budget back.
|
|
|
|
perShardMaps[i][idx] = nil
|
|
|
|
pup.stateSizeBudget.Add(nTotal)
|
|
|
|
}
|
|
|
|
}(i)
|
|
|
|
}
|
|
|
|
wg.Wait()
|
|
|
|
if needStop(pup.stopCh) {
|
|
|
|
return nil, nil
|
|
|
|
}
|
|
|
|
if n := pup.stateSizeBudget.Load(); n < 0 {
|
|
|
|
return nil, fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", pup.pu.String(), pup.maxStateSize/(1<<20))
|
|
|
|
}
|
|
|
|
|
|
|
|
// Filter out maps without entries
|
|
|
|
ms := perShardMaps[0]
|
|
|
|
result := ms[:0]
|
|
|
|
for _, m := range ms {
|
|
|
|
if len(m) > 0 {
|
|
|
|
result = append(result, m)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return result, nil
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
|
|
|
|
2024-10-17 23:44:38 +02:00
|
|
|
func updatePipeUniqMap(m map[string]*uint64, k string, pHitsSrc *uint64) int64 {
|
|
|
|
pHitsDst := m[k]
|
|
|
|
if pHitsDst != nil {
|
|
|
|
*pHitsDst += *pHitsSrc
|
|
|
|
return 0
|
|
|
|
}
|
2024-05-12 16:33:29 +02:00
|
|
|
|
2024-10-17 23:44:38 +02:00
|
|
|
m[k] = pHitsSrc
|
|
|
|
return int64(unsafe.Sizeof(k) + unsafe.Sizeof(pHitsSrc))
|
|
|
|
}
|
|
|
|
|
|
|
|
type pipeUniqWriteContext struct {
|
|
|
|
workerID uint
|
|
|
|
pup *pipeUniqProcessor
|
|
|
|
rcs []resultColumn
|
|
|
|
br blockResult
|
2024-05-12 16:33:29 +02:00
|
|
|
|
2024-05-22 21:01:20 +02:00
|
|
|
// rowsCount is the number of rows in the current block
|
|
|
|
rowsCount int
|
|
|
|
|
|
|
|
// valuesLen is the total length of values in the current block
|
2024-05-12 16:33:29 +02:00
|
|
|
valuesLen int
|
|
|
|
}
|
|
|
|
|
|
|
|
func (wctx *pipeUniqWriteContext) writeRow(rowFields []Field) {
|
|
|
|
rcs := wctx.rcs
|
|
|
|
|
|
|
|
areEqualColumns := len(rcs) == len(rowFields)
|
|
|
|
if areEqualColumns {
|
|
|
|
for i, f := range rowFields {
|
|
|
|
if rcs[i].name != f.Name {
|
|
|
|
areEqualColumns = false
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if !areEqualColumns {
|
2024-05-25 21:36:16 +02:00
|
|
|
// send the current block to ppNext and construct a block with new set of columns
|
2024-05-12 16:33:29 +02:00
|
|
|
wctx.flush()
|
|
|
|
|
|
|
|
rcs = wctx.rcs[:0]
|
|
|
|
for _, f := range rowFields {
|
2024-05-20 04:08:30 +02:00
|
|
|
rcs = appendResultColumnWithName(rcs, f.Name)
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
|
|
|
wctx.rcs = rcs
|
|
|
|
}
|
|
|
|
|
|
|
|
for i, f := range rowFields {
|
|
|
|
v := f.Value
|
|
|
|
rcs[i].addValue(v)
|
|
|
|
wctx.valuesLen += len(v)
|
|
|
|
}
|
2024-05-22 21:01:20 +02:00
|
|
|
|
|
|
|
wctx.rowsCount++
|
2024-05-12 16:33:29 +02:00
|
|
|
if wctx.valuesLen >= 1_000_000 {
|
|
|
|
wctx.flush()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (wctx *pipeUniqWriteContext) flush() {
|
|
|
|
rcs := wctx.rcs
|
|
|
|
br := &wctx.br
|
|
|
|
|
|
|
|
wctx.valuesLen = 0
|
|
|
|
|
2024-05-25 21:36:16 +02:00
|
|
|
// Flush rcs to ppNext
|
2024-05-22 21:01:20 +02:00
|
|
|
br.setResultColumns(rcs, wctx.rowsCount)
|
|
|
|
wctx.rowsCount = 0
|
2024-10-17 23:44:38 +02:00
|
|
|
wctx.pup.ppNext.writeBlock(wctx.workerID, br)
|
2024-05-12 16:33:29 +02:00
|
|
|
br.reset()
|
|
|
|
for i := range rcs {
|
2024-05-20 04:08:30 +02:00
|
|
|
rcs[i].resetValues()
|
2024-05-12 16:33:29 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func parsePipeUniq(lex *lexer) (*pipeUniq, error) {
|
|
|
|
if !lex.isKeyword("uniq") {
|
|
|
|
return nil, fmt.Errorf("expecting 'uniq'; got %q", lex.token)
|
|
|
|
}
|
|
|
|
lex.nextToken()
|
|
|
|
|
|
|
|
var pu pipeUniq
|
2024-05-20 04:08:30 +02:00
|
|
|
if lex.isKeyword("by", "(") {
|
|
|
|
if lex.isKeyword("by") {
|
|
|
|
lex.nextToken()
|
|
|
|
}
|
2024-05-12 16:33:29 +02:00
|
|
|
bfs, err := parseFieldNamesInParens(lex)
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("cannot parse 'by' clause: %w", err)
|
|
|
|
}
|
|
|
|
if slices.Contains(bfs, "*") {
|
|
|
|
bfs = nil
|
|
|
|
}
|
|
|
|
pu.byFields = bfs
|
|
|
|
}
|
|
|
|
|
2024-05-25 21:36:16 +02:00
|
|
|
if lex.isKeyword("with") {
|
|
|
|
lex.nextToken()
|
|
|
|
if !lex.isKeyword("hits") {
|
|
|
|
return nil, fmt.Errorf("missing 'hits' after 'with'")
|
|
|
|
}
|
|
|
|
}
|
2024-05-24 03:06:55 +02:00
|
|
|
if lex.isKeyword("hits") {
|
|
|
|
lex.nextToken()
|
|
|
|
hitsFieldName := "hits"
|
|
|
|
for slices.Contains(pu.byFields, hitsFieldName) {
|
|
|
|
hitsFieldName += "s"
|
|
|
|
}
|
|
|
|
|
|
|
|
pu.hitsFieldName = hitsFieldName
|
|
|
|
}
|
|
|
|
|
2024-05-12 16:33:29 +02:00
|
|
|
if lex.isKeyword("limit") {
|
|
|
|
lex.nextToken()
|
|
|
|
n, ok := tryParseUint64(lex.token)
|
|
|
|
if !ok {
|
|
|
|
return nil, fmt.Errorf("cannot parse 'limit %s'", lex.token)
|
|
|
|
}
|
|
|
|
lex.nextToken()
|
|
|
|
pu.limit = n
|
|
|
|
}
|
|
|
|
|
|
|
|
return &pu, nil
|
|
|
|
}
|