diff --git a/pkg/metricstore/stats.go b/pkg/metricstore/stats.go index a253d777..64e8e555 100644 --- a/pkg/metricstore/stats.go +++ b/pkg/metricstore/stats.go @@ -55,9 +55,6 @@ func (b *buffer) stats(from, to int64) (Stats, int64, int64, error) { from = b.start } - // TODO: Check if b.closed and if so and the full buffer is queried, - // use b.statistics instead of iterating over the buffer. - samples := 0 sum, min, max := 0.0, math.MaxFloat32, -math.MaxFloat32 @@ -72,6 +69,31 @@ func (b *buffer) stats(from, to int64) (Stats, int64, int64, error) { idx = int((t - b.start) / b.frequency) } + // Fast path: standing at this buffer's first data point (idx 0) with the + // whole buffer inside [from, to). Fold in the cached aggregate and jump + // past the buffer's real data instead of scanning each point. Any slots + // between len(data) and cap are handled as gaps by the normal loop after + // t advances, so the returned `to` matches the scan semantics. + if idx <= 0 && t <= b.firstWrite() && b.end() <= to { + if !b.statsValid { + b.recomputeStats() + } + if b.statSamples > 0 { + sum += b.statSum + samples += b.statSamples + if b.statMin < min { + min = b.statMin + } + if b.statMax > max { + max = b.statMax + } + } + // Position t at the buffer's last real data point; the loop's + // t += frequency then advances into the trailing gap / next buffer. + t = b.end() - b.frequency + continue + } + if t < b.start || idx >= len(b.data) { continue } @@ -81,10 +103,14 @@ func (b *buffer) stats(from, to int64) (Stats, int64, int64, error) { continue } - samples += 1 + samples++ sum += xf - min = math.Min(min, xf) - max = math.Max(max, xf) + if xf < min { + min = xf + } + if xf > max { + max = xf + } } return Stats{ diff --git a/pkg/metricstore/stats_test.go b/pkg/metricstore/stats_test.go index 2dd33009..2fc31270 100644 --- a/pkg/metricstore/stats_test.go +++ b/pkg/metricstore/stats_test.go @@ -120,3 +120,93 @@ func TestRecomputeStatsEmptyBuffer(t *testing.T) { t.Errorf("empty buffer min/max = %v/%v, want sentinels", b.statMin, b.statMax) } } + +func TestStatsFullBufferMatchesScan(t *testing.T) { + b := newBuffer(100, 10) + b.write(100, schema.Float(3.0)) + b.write(110, schema.Float(1.0)) + b.write(120, schema.Float(5.0)) + // Range fully covers the buffer (from before firstWrite, to at/after end). + s, _, _, err := b.stats(100, 130) + if err != nil { + t.Fatalf("stats() error = %v", err) + } + if s.Samples != 3 { + t.Errorf("Samples = %d, want 3", s.Samples) + } + if s.Min != 1.0 || s.Max != 5.0 { + t.Errorf("Min/Max = %v/%v, want 1.0/5.0", s.Min, s.Max) + } + if s.Avg != 3.0 { + t.Errorf("Avg = %v, want 3.0", s.Avg) + } +} + +func TestStatsPartialRangeScans(t *testing.T) { + b := newBuffer(100, 10) + b.write(100, schema.Float(3.0)) // t=100 + b.write(110, schema.Float(1.0)) // t=110 + b.write(120, schema.Float(5.0)) // t=120 + // Partial range: only t=110 and t=120 (from excludes t=100). + s, _, _, err := b.stats(110, 130) + if err != nil { + t.Fatalf("stats() error = %v", err) + } + if s.Samples != 2 { + t.Errorf("Samples = %d, want 2", s.Samples) + } + if s.Min != 1.0 || s.Max != 5.0 { + t.Errorf("Min/Max = %v/%v, want 1.0/5.0", s.Min, s.Max) + } +} + +func TestStatsOverwriteThenFullQuery(t *testing.T) { + b := newBuffer(100, 10) + b.write(100, schema.Float(1.0)) + b.write(110, schema.Float(2.0)) + b.write(120, schema.Float(3.0)) + b.write(100, schema.Float(99.0)) // overwrite the running min -> statsValid false + + s, _, _, err := b.stats(100, 130) + if err != nil { + t.Fatalf("stats() error = %v", err) + } + if s.Min != 2.0 { + t.Errorf("Min = %v, want 2.0 (recomputed after overwrite)", s.Min) + } + if s.Max != 99.0 { + t.Errorf("Max = %v, want 99.0", s.Max) + } + if s.Samples != 3 { + t.Errorf("Samples = %d, want 3", s.Samples) + } + if !b.statsValid { + t.Error("statsValid should be true after the recompute triggered by the query") + } +} + +func TestStatsMultiBufferChain(t *testing.T) { + // Two full buffers of cap=3 (interior buffers are always full). + // b1: t=100,110,120 = 1,2,3 ; b2: t=130,140,150 = 4,5,6. + b1 := &buffer{data: make([]schema.Float, 0, 3), frequency: 10, start: 95} + b1.data = append(b1.data, 1.0, 2.0, 3.0) + b2 := &buffer{data: make([]schema.Float, 0, 3), frequency: 10, start: 125} + b2.data = append(b2.data, 4.0, 5.0, 6.0) + b2.prev = b1 + b1.next = b2 + // b1/b2 built bare: statsValid is false (zero value) -> stats() must recompute. + + s, _, _, err := b2.stats(100, 160) + if err != nil { + t.Fatalf("stats() error = %v", err) + } + if s.Samples != 6 { + t.Errorf("Samples = %d, want 6", s.Samples) + } + if s.Min != 1.0 || s.Max != 6.0 { + t.Errorf("Min/Max = %v/%v, want 1.0/6.0", s.Min, s.Max) + } + if s.Avg != 3.5 { + t.Errorf("Avg = %v, want 3.5", s.Avg) + } +}