perf: implement batch processing in iterateEvalTree

frac/active_index.go

@@ -167,11 +167,25 @@ func (p *activeIDsIndex) GetMID(lid seq.LID) seq.MID {
 	return seq.MID(p.mids[restoredLID])
 }
 
+func (p *activeIDsIndex) GetMIDs(lidsBatch []node.LID, out []seq.MID) []seq.MID {
+	for _, lid := range lidsBatch {
+		out = append(out, p.GetMID(lid.ToSeqLID()))
+	}
+	return out
+}
+
 func (p *activeIDsIndex) GetRID(lid seq.LID) seq.RID {
 	restoredLID := p.inverser.Revert(uint32(lid))
 	return seq.RID(p.rids[restoredLID])
 }
 
+func (p *activeIDsIndex) GetRIDs(lidsBatch []node.LID, out []seq.RID) []seq.RID {
+	for _, lid := range lidsBatch {
+		out = append(out, p.GetRID(lid.ToSeqLID()))
+	}
+	return out
+}
+
 func (p *activeIDsIndex) Len() int {
 	return p.inverser.Len()
 }

frac/fraction_test.go

@@ -1592,6 +1592,12 @@ func (s *FractionTestSuite) TestSearchLargeFrac() {
 		for _, ord := range orders {
 			histBuckets := make(map[string]uint64)
 			for _, doc := range testDocs {
+				if doc.timestamp.Before(fromTime) {
+					continue
+				}
+				if doc.timestamp.After(midTime) {
+					continue
+				}
 				if doc.service == "database" && doc.level == 3 {
 					bucketTime := doc.timestamp.Truncate(time.Second)
 					bucketKey := bucketTime.Format(time.RFC3339Nano)
@@ -1601,7 +1607,8 @@ func (s *FractionTestSuite) TestSearchLargeFrac() {
 
 			searchParams := s.query(
 				"service:database AND level:3",
-				withTo(toTime.Format(time.RFC3339Nano)),
+				withFrom(fromTime.Format(time.RFC3339Nano)),
+				withTo(midTime.Format(time.RFC3339Nano)),
 				withHist(1000))
 			searchParams.Order = ord
 

frac/processor/hist_map.go

@@ -0,0 +1,46 @@
+package processor
+
+import "github.com/ozontech/seq-db/seq"
+
+// HistMap is an optimized array-based map for histogram.
+type HistMap struct {
+	buckets  []uint64
+	start    seq.MID
+	interval seq.MID
+	base     uint64
+}
+
+func NewHistMap(from, to seq.MID, intervalMillis uint64) HistMap {
+	interval := seq.MillisToMID(intervalMillis)
+	base := uint64(from) / uint64(interval)
+	size := uint64(to)/uint64(interval) - base + 1
+	return HistMap{
+		buckets:  make([]uint64, size),
+		start:    from - from%interval,
+		interval: interval,
+		base:     base,
+	}
+}
+
+func (h *HistMap) Update(mids []seq.MID) {
+	// TODO(cheb0): unroll/vectorize/whatever when we optimize everything else
+	for _, mid := range mids {
+		bucketIndex := uint64(mid)/uint64(h.interval) - h.base
+		h.buckets[bucketIndex]++
+	}
+}
+
+func (h HistMap) ToMap() map[seq.MID]uint64 {
+	if len(h.buckets) == 0 {
+		return nil
+	}
+	res := make(map[seq.MID]uint64, len(h.buckets))
+	bucket := h.start
+	for _, cnt := range h.buckets {
+		if cnt > 0 {
+			res[bucket] = cnt
+		}
+		bucket += h.interval
+	}
+	return res
+}